iomgaa/PolyGateway
1
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects Releases 9 Wiki Activity
Files
main
PolyGateway/research-wiki/plans/2026-07-21-m3-ocr-plan.md
T

22 KiB
Raw Permalink Blame History

M3 OCR 实现计划

  • 目标: 按已批准设计 research-wiki/designs/2026-07-21-m3-ocr-design.md(下称"设计")落地 OCR 端口族: OcrClient 独立治理循环 + MonkeyOcrTransport 双端点 + 类型/端口/配置 + 遥测复用 + G1/R9/R10 销账 + 文档同步 + P7 OCR soak 验收。
  • 方案概述: 方案 A——仿 EmbeddingClient(src/polygateway/embedding.py)的独立精简治理循环,复用 QuotaGate/BreakerGate/backoff_delay/_failure_reason/SourceCooldownMemo/TelemetryEmitter/选源器;新增 OcrTransport 端口与 transports/monkey_ocr.py。不触碰 middleware/retry.py。
  • 涉及技术: Python 3.11 + httpx + zipfile/BytesIO(标准库,无新依赖);真实 MonkeyOCR 服务 10.77.0.20:7866/7867;分支 feature/m3-ocr;所有命令 conda run -n PolyGateway <cmd>。
  • 执行方式: 中等规模、任务间强顺序依赖 → 直接按计划实现,不用 subagent-driven-development。
  • structured-logging 已完成(SOP Phase1-2,先于本计划): 无新表,复用 llm_calls;metric ocr-call-success 已注册(research-wiki/metrics/ocr-call-success.md,含口径/埋点位置/基线,commit e51f077),边 metric→schema、schema→design 已连。埋点清单落 T5.4。
  • 测试结果门(全任务统一): 每个行为任务的新测试用例必须先于实现运行一次并出示失败证据(ImportError/AttributeError/断言失败),实现后转 PASS——T1-T8 的验证行均按此执行,不再逐条重复。

文件结构(锁定分解)

文件 动作 职责
src/polygateway/types.py 修改 新增 5 个 frozen dataclass(§T1)
src/polygateway/ports.py 修改 新增 3 个 Protocol(§T2)
src/polygateway/transports/monkey_ocr.py 创建 MonkeyOCR 协议细节: 双端点、两段协议、_middle.json 数值防御、错误翻译、check_health(§T3)
src/polygateway/config.py 修改 OcrSettings(§T4)
src/polygateway/ocr.py 创建 OcrClient 治理循环 + 工厂(§T5)
src/polygateway/__init__.py 修改 公共导出(§T6)
tests/unit/test_monkey_ocr.py / test_ocr_client.py 创建 §T3/§T5
tests/unit/test_types.py / test_ports.py / test_config.py / test_package.py 修改 随各任务扩展
tests/integration/test_monkey_ocr_live.py 创建 真实服务双端点 + 一致性断言(§T7)
tools/soak/scenarios.py / run_soak.py / scoreboard.py 修改 P7 OCR 场景(§T8)
research-wiki/ARCHITECTURE.md / ROADMAP.md / migrations/*.md / .env.example 修改 文档同步与销账(§T9)

T1 类型(types.py)

新增 5 个 frozen dataclass(设计 §3.1;字段顺序即公共承诺,docstring 中文):

@dataclass(frozen=True)
class OcrLayoutElement:
    type: str                                   # 开放字符串: table/image/text/...
    bbox: tuple[float, float, float, float]     # OCR 原生页面坐标 (x1,y1,x2,y2)
    page_index: int

@dataclass(frozen=True)
class OcrTextResult:
    text: str            # 空串 = 合法"无文字"
    source_name: str
    usage: Usage         # OCR 无计费: Usage(0, 0)
    latency_ms: int
    call_id: str
    raw: dict[str, Any]

OcrLayoutResult: elements: list[OcrLayoutElement]、page_sizes: list[tuple[float, float]](按 page_index 索引)+ 与 OcrTextResult 相同的溯源五件(source_name/usage/latency_ms/call_id/raw)。OcrTextTransportResult: text: str + raw: dict;OcrLayoutTransportResult: elements + page_sizes + raw。类型自身不做数值校验(校验归 transport 解析层,§T3)。

  • 测试(tests/unit/test_types.py 扩展): 5 类型可构造、frozen 不可变、字段默认值(无默认——全必填)。
  • 验证: conda run -n PolyGateway pytest tests/unit/test_types.py -v → PASS(新用例先于实现失败: ImportError)。
  • 提交 feat: add OCR result and transport types

T2 端口(ports.py)

新增 3 个 @runtime_checkable Protocol(设计 §3.2 逐字;import 处补 4 个 OCR 类型):

class OcrTextPort(Protocol):
    async def recognize_text(self, image: bytes) -> OcrTextResult: ...

class OcrLayoutPort(Protocol):
    async def parse_layout(self, image: bytes) -> OcrLayoutResult: ...

class OcrTransport(Protocol):
    async def recognize_text(self, *, image: bytes, source: SourceConfig, call_id: str) -> OcrTextTransportResult: ...
    async def parse_layout(self, *, image: bytes, source: SourceConfig, call_id: str) -> OcrLayoutTransportResult: ...
    async def check_health(self, *, source: SourceConfig) -> bool: ...
  • 测试(tests/unit/test_ports.py 扩展): 三 Protocol runtime_checkable;满足签名的桩类 isinstance 通过、缺方法的不通过。
  • 验证: conda run -n PolyGateway pytest tests/unit/test_ports.py -v → PASS。
  • 提交 feat: add OCR port family protocols

T3 MonkeyOcrTransport(transports/monkey_ocr.py)

保真校验任务: 实现时逐段比对 reference/CHSAnalyzer/app/providers/invokers.py:427-552(_finite_number/_parse_table_result/MonkeyOcrParseInvoker.locate_table 两段协议与错误翻译)与 reference/Video-Tree-TRM5/adapters/ocr.py(/ocr/text 请求形态、/health 判定)。设计 §4/§10 已声明的偏离(全元素提取、success!=true 附 status_code=200、429 细分放弃、resp.ok→2xx、显式 content 校验)之外不得改变任何语义。

实现要点(全部出自设计 §4,逐条为验收标准):

  1. 类结构仿 OpenAICompatTransport(openai_compat.py:207-224): client_factory 可注入,默认 per-source httpx.AsyncClient(base_url=source.base_url, timeout=source.timeout_s, trust_env=source.trust_env)——不带 Authorization 头(MonkeyOCR 无鉴权,api_key 是占位);_clients 懒创建 dict;aclose() 幂等释放全部 client。
  2. recognize_text: POST /ocr/text,multipart files={"file": ("image.jpg", image, "image/jpeg")};2xx 后 JSON 解析,content 缺失或非 str → ResultInvalidError;空串合法。
  3. parse_layout 两段: POST /parse 同 multipart → 校验 JSON dict、success is True(否则 RequestRejectedError 附 status_code=200)、download_url 非空 str(缺 → TransientError)→ GET download_url(以 / 开头或不含 :// 视为相对路径走同一 base_url client;http(s):// 绝对 URL 原样请求)→ ZIP 解析。
  4. _parse_middle_json(纯函数,bytes → tuple[elements, page_sizes]): ZIP 内首个 *_middle.json(无 → ResultInvalid);pdf_info 必须 list;每页必须 dict、page_size 为 2 元素正有限数(_finite_number 逐字移植: bool 拒绝、非有限拒绝)、para_blocks 缺省按空 list、每块必须 dict、type 必须非空 str、bbox 必须 4 有限数且 x2>x1、y2>y1 且 int() 整数化后不退化(注释写明: 此校验专为 CHS shim 的 int 裁剪路径兜底,勿当死代码删除);任何校验失败 → ResultInvalidError(捕获 ValueError/TypeError/KeyError/json.JSONDecodeError/zipfile.BadZipFile,CHS invokers.py:533-539 同款)。元素 page_index 取页面 page_idx(缺失时用枚举序,防御)。
  5. 错误翻译(独立 helper,httpx 异常 → 四分类): httpx.ConnectError/ConnectTimeout 等连接类 → TransientError(network_error);ReadTimeout/WriteTimeout/PoolTimeout → TransientError(timeout);HTTPStatusError: 5xx/429 → Transient、401/403 → SourceDeadError、其余 4xx → RequestRejectedError(全部附 source_name/status_code/operation)。
  6. check_health: GET {base_url}/health,固定 5s 超时(探测常量,不进配置),2xx → True;except Exception → False;asyncio.CancelledError 穿透(显式 except CancelledError: raise 在 Exception 之前)。
  • 测试(tests/unit/test_monkey_ocr.py 创建,transport 注入假 client_factory / httpx.MockTransport):
    • fixtures 脱敏规则(设计 §11): 以 2026-07-21 取证的真实响应为骨架二次构造——保留结构/数值/元素类型,一切 OCR 识别文本(lines/content/spans 内文字)替换为合成占位(如 "LINE-1");fixtures 以 Python 常量或 tests/fixtures/ocr/ 下 JSON 存放,构造一个含 text/table/image 三类块的 _middle.json 与一个真实结构 ZIP(zipfile 现场打包)。
    • 用例清单: /ocr/text 正常解析|content 缺失→ResultInvalid|content 非 str→ResultInvalid|空串合法;/parse 全链路正常(相对 download_url)|绝对 download_url|success=false→RequestRejected 且 status_code==200|download_url 缺→Transient|响应非 JSON→Transient;ZIP: 坏 ZIP→ResultInvalid|缺 _middle.json→ResultInvalid|pdf_info 非 list→ResultInvalid|page_size 非法(负数/NaN/bool)→ResultInvalid|bbox 顺序错→ResultInvalid|bbox 整数化退化(如 [1.2,1.2,1.8,5])→ResultInvalid|para_blocks 缺省→elements 空且合法;HTTP 错误翻译逐分类(mock 502/429/401/403/404/连接错/读超时);check_health 2xx→True|500→False|连接失败→False|取消穿透(mock 内 raise CancelledError,断言穿透而非 False);aclose 幂等。
  • 验证: conda run -n PolyGateway pytest tests/unit/test_monkey_ocr.py -v → PASS;先失败证据: 实现前先写协议解析用例跑一次(ModuleNotFoundError/断言失败)。
  • 提交 feat: add MonkeyOCR dual-endpoint transport

T4 配置(config.py OcrSettings)

仿 EmbeddingSettings(config.py:368-409): OcrSettings frozen dataclass,唯一字段 gateway: GatewaySettings;from_env(scope="OCR", env=None, *, env_file=".env") 委托 GatewaySettings.from_env(scope_u, env=env)——无 OCR 专用键。装配防御在 §T5 工厂做(config 不 import transport)。

  • 测试(tests/unit/test_config.py 扩展): OCR__MONKEY__1__* 最小键集(BASE_URL/API_KEY=none/MODEL/TIMEOUT_S)装配成功;per-scope 韧性键(OCR__RETRY__MAX_ATTEMPTS 等)生效(现 GatewaySettings 已支持,只加断言);缺 BASE_URL 报错。
  • 验证: conda run -n PolyGateway pytest tests/unit/test_config.py -v → PASS。
  • 提交 feat: add OcrSettings scope configuration

T5 OcrClient(src/polygateway/ocr.py)

保真校验任务: 循环形态与 embedding.py:169-329(_embed_batch/_pick_runnable/_on_no_runnable/_attempt/_gate_on_terminal/_record_quietly/_settle_and_release)逐段同构比对;治理语义与 reference/CHSAnalyzer/app/providers/governance.py:200-290 比对。设计 §5 声明的三处差异(settle 恒 0、无批处理、接 OutcomeAwareSelector 喂数)之外不得偏离。

实现要点:

  1. 构造签名(设计 §3.3): scope/sources/selector/limiter/breaker/transport(OcrTransport)/retry/backpressure/quota_full="wait"/telemetry=None/now/sleep/rng;无 pricing/batch/normalize/expected_dim。校验 quota_full 域、sources 非空不强制(空池调用时抛 no_sources,embedding 同款)。
  2. recognize_text(image, *, session_id=None, parent_call_id=None) -> OcrTextResult 与 parse_layout(...) -> OcrLayoutResult: 共用私有 _call(kind, image, ...) 治理循环,kind 决定调 transport 哪个方法与结果组装。入参校验: isinstance(image, bytes) 且非空,否则 TypeError/ValueError(显式优于隐式)。
  3. 治理循环逐分支(设计 §5;与 embedding 的 _embed_batch+_attempt 同构):
    • 成功: breaker.record_success(entry) + mark_progress() + selector.record_outcome(name, True)(isinstance OutcomeAwareSelector 判定一次,构造时缓存布尔)→ 组装结果(usage=Usage(0,0)、latency_ms、call_id、source_name、raw)。
    • ResultInvalidError / 带 status_code 的 RequestRejectedError: record_success(entry, count_attempt=False) 后直接上抛;不喂 selector(坏结果≠坏服务)。
    • 不带 status_code 的 RequestRejectedError: 探针则 release_probe,直接上抛。
    • CancelledError: 探针归还后 raise(遥测记 "cancelled")。
    • SourceDeadError/TransientError: record_failure(entry, reason, dead) + selector.record_outcome(name, False) + reasons[name]=reason;fails+=1,fails >= retry.max_attempts → AllSourcesExhausted(retry_exhausted, retry_after_s=backoff_base_s, per_source_reasons=reasons);否则非 dead 退避 backoff_delay(...) 后换源。
    • permit 在 finally settle(0) + release()(恒 0,OCR 无 token)。
    • 无可运行源: 全 gate 拒 → CircuitOpenError(retry_after_s=await breaker.retry_after_s(names));fail_fast → AllSourcesExhausted(quota_exhausted);双条件 stall(本地超 stall_window_s AND progress_age_s() 超窗)→ AllSourcesExhausted(stalled);否则 poll_interval 抖动等待。
  4. 遥测(设计 §7): 每次尝试(成败/取消)经 TelemetryEmitter.emit_attempt 单 helper;request 占位 messages 包成 ChatRequest(embedding.py _emit@345 先例)——[{"role":"user","content":f"<ocr:{kind} image_bytes={len(image)}>"}];成功 response 用 LLMResponse 包装: content = text 截断 200 字 / f"<elements n={len(elements)} pages={len(page_sizes)}>",tokens=0、ttft/inter=None、cache_hit=False;写失败降级不冒泡(Emitter 内建)。
  5. check_health() -> dict[str, bool]: asyncio.gather 并发逐源 transport.check_health(source=s),返回 {name: bool};不吞 CancelledError(gather 天然穿透);不健康源 logger.warning。
  6. 工厂: from_settings(OcrSettings, *, limiter=None, breaker=None, telemetry=None) 复用 client.py 的 _build_limiter/_build_breaker/_build_selector/_build_telemetry;装配防御: 任一 source.provider != "monkey" → ValueError(D9 GLM 预留档,严禁静默用 MonkeyOcrTransport 打别家端点);from_env(scope="OCR") 委托。aclose()/async context manager 与 EmbeddingClient 对称。
  7. G1 钉住: 上述三处 scope 级异常全部携带非空 per_source_reasons 与 retry_after_s(circuit_open/stalled 读熔断后端)。
  • 测试(tests/unit/test_ocr_client.py 创建,桩 OcrTransport + memory 后端;复用 test_retry.py 的 StaticSelector/RecordingSelector 桩形态):
    • 成功路径: 结果字段齐全(usage 0/latency/call_id/source_name/raw)、gate 记成功、mark_progress、selector 喂 True;
    • Transient 换源: 第一源失败第二源成功,fails 计数、退避走注入 sleep、selector 喂 False、reasons 记录;
    • SourceDead: 立即换源(immediate,不退避)、force_open;
    • ResultInvalid: 直接上抛、gate 记成功且 count_attempt=False(RecordingGate 桩断言)、不喂 selector;
    • RequestRejected 带 status_code: 同上直抛记成功;不带 status_code: 仅探针归还;
    • retry_exhausted: 抛 AllSourcesExhausted,per_source_reasons 非空、retry_after_s > 0(此三条断言即设计 §6 ③ 的 G1 契约钉,落在本单测文件,不另进 tests/contracts/——contracts 层是限流/熔断后端双实现契约,G1 是客户端循环行为);circuit_open 与 stalled 同断言;
    • stall 双条件: 本地超窗但 progress 新鲜 → 继续等(不抛);双超 → stalled;
    • 取消穿透: sleep 中取消、transport 调用中取消(探针归还断言)、permit finally 释放;
    • fail_fast: quota 满即抛 quota_exhausted;
    • check_health: 混合池返回逐源 dict、异常源 False、取消穿透;
    • 装配防御: provider="glm" 源 → ValueError;
    • 遥测: 假 Recorder 断言成功/失败/取消均有记录、占位 messages 不含图像字节。
  • 验证: conda run -n PolyGateway pytest tests/unit/test_ocr_client.py tests/unit/test_embedding.py -v → PASS(embedding 回归确认复用件未被改动)。
  • 提交 feat: add OcrClient governed loop with dual ports(可拆 2-3 个中间提交: 循环/工厂/check_health)

T6 公共导出(__init__.py)

__init__.py 新增导出: OcrClient、OcrSettings、OcrTextResult、OcrLayoutResult、OcrLayoutElement(Transport 结果类型与 Protocol 不出顶层——与 EmbeddingTransportResult 先例一致,高级用户从 polygateway.ports/polygateway.types 引用)。__all__ 按字母序插入。

  • 测试: tests/unit/test_package.py 扩展导出断言。
  • 验证: conda run -n PolyGateway pytest tests/unit/test_package.py -v → PASS。
  • 提交 feat: export OCR public API surface

T7 集成测试(真实服务 + 真实 Redis)

tests/integration/test_monkey_ocr_live.py 创建(标记 @pytest.mark.integration;服务不可达时 fail 而非 skip——验收必打真实,与 ROADMAP 验收出口一致):

  1. 真实 data/soak/chs_images/ 取 1 张有表样本(chs_0001)+ 1 张无表样本(chs_0002),OcrClient.from_env(scope="OCR") 配 7866 单源(测试内 env dict 注入,不落 .env): recognize_text 返回非空 text;parse_layout 有表样本 elements 含 type=="table" 且 bbox 4 元正序、page_sizes 非空;无表样本 elements 无 table 且不抛异常(合法无表语义)。
  2. tables/para_blocks 一致性断言(设计 §1.2 持续护栏): 同一样本直接裸调 /parse 下载 _middle.json,断言 tables 的 bbox 集合 == 库返回 elements 中 type=="table" 的 bbox 集合。
  3. check_health(): 7866+7867 双源池 → 两 True;掺一个黑洞源(10.255.255.1)→ 该源 False 且 5s 级返回(不等 timeout_s)。
  4. 真实 Redis(db3)后端组合: redis limiter+breaker 注入 OcrClient,打一次真实调用成功;黑洞源池验证熔断开路后 CircuitOpenError.retry_after_s > 0(G1 真实后端联调)。
  • 验证: conda run -n PolyGateway pytest tests/integration/test_monkey_ocr_live.py -v → PASS(注意: 与 soak 不并跑,Redis db3 冲突)。
  • 提交 test: add MonkeyOCR live integration suite

T8 P7 OCR soak 场景

  1. .env 增 OCR soak 池(先读现有 .env 再追加,严禁编造/覆盖;凭据只在 python 内经 dotenv 读): SOAK_OCR__MONKEY__{1..4}__*——源1=http://10.77.0.20:7866(真)、源2=http://10.77.0.20:7867(真)、源3=http://10.255.255.1:7866(黑洞)、源4=http://10.77.0.20:7899(坏端口,连接拒绝);api_key 全 "none";源级 RPM/并发参照 CHS 惯例(MAX_CONCURRENCY=4/RPM=120)+ TIMEOUT_S=120;scope 键 SOAK_OCR__RETRY__MAX_ATTEMPTS=3。
  2. tools/soak/scenarios.py: 新增 P7 生成器——语料 data/soak/chs_images/ 全量循环(实测 469 张,设计 §8 "约 300" 以实际为准),kind 新增 "ocr_text"/"ocr_layout"(权重 0.8/0.2),kwargs={"image": bytes};总量 1500 调用。
  3. tools/soak/run_soak.py: 装配 OcrClient(P7 时替代 GatewayClient;复用 _paced_dispatch 有界分发);按 kind 分派两方法;失败打印含 per_source_reasons 链(既有格式)。
  4. tools/soak/scoreboard.py: P7 口径裁剪——无 429/缓存/token 列;不变量: ① 成功率 ≥98%(双真源在池);② 坏源(源3/4)尝试占比 ≤15%(健康选源压制);③ 熔断对坏源开路且真源零误熔(gate 事件核对);④ RPM 逐源不超限(准入时刻+服务器钟偏移聚桶,复用既有函数);⑤ 全程 RSS 有界(< 500MB 绝对值,ps 当前值——新增检查函数,不改既有 inv_rss_stable 增长口径);⑥ tables/para_blocks 一致性抽查(layout 成功调用抽 20 条);⑦ 失败链全部可解释(reason ∈ 已知集合);⑧ 零 CancelledError 泄漏/零未分类异常(有意裁决: 设计 §8 的"取消穿透抽查"由 T5 单测主动注入覆盖,soak 只做被动泄漏核查——harness 无取消注入能力,不为此扩建)。
  5. 跑 P7(tmux + 禁日志缓存),结果落 research-wiki/findings/2026-07-21-p7-ocr-soak.md(阈值不达标 → systematic-debugging 迭代,测试环境不放水)。
  • 验证: 记分板 8 不变量全 PASS。
  • 提交 feat: add P7 OCR soak scenario and scoreboard(代码)+ docs: record P7 OCR soak acceptance(findings)

T9 文档同步与销账(设计 §12 逐条)

  1. ARCHITECTURE.md: D9(:208)与 §7.10(:459)"走同一中间件栈"措辞改为"复用同一套治理算法件与错误分类,循环形态同 EmbeddingClient 先例(M2 §7.1 有限重复裁决)";§7.10 补 check_health dict[str, bool] 签名与 download_url 相对路径协议事实;§5.1 OCR 类型清单按设计 §3.1 五件溯源更新。ROADMAP §4 ③ 措辞同步 + M3 状态 ✅。
  2. migrations/chsanalyzer.md: G1 行改"已闭(库字段齐备,项目侧留 10 行翻译 shim)";§4 表 OcrLayoutPort 消费 shim 更新(首个 type=="table" 元素 + int 四元组约 5 行);偏离表补 success!=true 有意修复与 429 细分有意放弃两行。
  3. migrations/video-tree-trm5.md: R9/R10 改已闭;§8 表 13 行裁决与设计 §10.1 对齐(check_health 形态、resp.ok→2xx 收紧)。
  4. .env.example: 补 OCR scope 样例段(CHS 单源 + VT 双实例两种写法,api_key="none" 惯例注释)。
  • 验证: grep 核对无残留"同一中间件栈"旧措辞指向 OCR;两迁移文档 G1/R9/R10 状态行更新。
  • 提交 docs: sync OCR architecture wording and close G1 R9 R10

T10 收尾门(不可跳过)

  1. 全量 CI: conda run -n PolyGateway make ci → 0 失败,覆盖率 ≥80%;make lint → 0 错误(radon 门在 hook)。
  2. 慢速真实等待用例(时间语义)不受本里程碑影响,但全量跑确认无回归。
  3. 独立 verifier(verification-before-completion): 全新上下文只读 subagent,输入=设计+本计划+分支 diff 区间,亲自跑测试,Critical/Important 清零后方可声称完成。
  4. finishing-a-development-branch: 合并方式呈人类决定。
  • 完成声明逐条对应本会话工具输出

保真校验总表

任务 蓝本 检查点
T3 CHS invokers.py:427-552 _finite_number 逐字;两段协议顺序与校验条件逐条;错误翻译分支与设计 §4 偏离声明外零改变
T3 VT ocr.py:41-70,108-119 multipart 形态;/health 判定(设计声明的 2xx 收紧);trust_env 语义
T5 embedding.py:169-329 循环分支逐段同构;_gate_on_terminal/_record_quietly/_settle_and_release 口径一致
T5 CHS governance.py:200-290 record_success/failure/release_probe 时机;reasons 聚合;stall 双条件

计划外事项(显式排除)

OCR 响应缓存、AIMD pacer、429 pushback、流式看门狗、GLM invoker、音频端口——设计 §5 不做清单;任何"顺手重构"禁止。

Reference in New Issue View Git Blame Copy Permalink