22 KiB
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;metricocr-call-success已注册(research-wiki/metrics/ocr-call-success.md,含口径/埋点位置/基线,commite51f077),边 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,逐条为验收标准):
- 类结构仿
OpenAICompatTransport(openai_compat.py:207-224):client_factory可注入,默认 per-sourcehttpx.AsyncClient(base_url=source.base_url, timeout=source.timeout_s, trust_env=source.trust_env)——不带 Authorization 头(MonkeyOCR 无鉴权,api_key 是占位);_clients懒创建 dict;aclose()幂等释放全部 client。 recognize_text: POST/ocr/text,multipartfiles={"file": ("image.jpg", image, "image/jpeg")};2xx 后 JSON 解析,content缺失或非 str →ResultInvalidError;空串合法。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 解析。_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(缺失时用枚举序,防御)。- 错误翻译(独立 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)。 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 幂等。
- fixtures 脱敏规则(设计 §11): 以 2026-07-21 取证的真实响应为骨架二次构造——保留结构/数值/元素类型,一切 OCR 识别文本(lines/content/spans 内文字)替换为合成占位(如 "LINE-1");fixtures 以 Python 常量或 tests/fixtures/ocr/ 下 JSON 存放,构造一个含 text/table/image 三类块的
- 验证:
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 喂数)之外不得偏离。
实现要点:
- 构造签名(设计 §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 同款)。 recognize_text(image, *, session_id=None, parent_call_id=None) -> OcrTextResult与parse_layout(...) -> OcrLayoutResult: 共用私有_call(kind, image, ...)治理循环,kind 决定调 transport 哪个方法与结果组装。入参校验:isinstance(image, bytes)且非空,否则TypeError/ValueError(显式优于隐式)。- 治理循环逐分支(设计 §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 ANDprogress_age_s()超窗)→AllSourcesExhausted(stalled);否则 poll_interval 抖动等待。
- 成功:
- 遥测(设计 §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 内建)。 check_health() -> dict[str, bool]:asyncio.gather并发逐源transport.check_health(source=s),返回{name: bool};不吞 CancelledError(gather 天然穿透);不健康源 logger.warning。- 工厂:
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 对称。 - 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 验收出口一致):
- 真实
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 且不抛异常(合法无表语义)。 - tables/para_blocks 一致性断言(设计 §1.2 持续护栏): 同一样本直接裸调 /parse 下载
_middle.json,断言tables的 bbox 集合 == 库返回 elements 中 type=="table" 的 bbox 集合。 check_health(): 7866+7867 双源池 → 两 True;掺一个黑洞源(10.255.255.1)→ 该源 False 且 5s 级返回(不等 timeout_s)。- 真实 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 场景
.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。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 调用。tools/soak/run_soak.py: 装配OcrClient(P7 时替代 GatewayClient;复用_paced_dispatch有界分发);按 kind 分派两方法;失败打印含per_source_reasons链(既有格式)。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 无取消注入能力,不为此扩建)。- 跑 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 逐条)
- 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 状态 ✅。 - migrations/chsanalyzer.md: G1 行改"已闭(库字段齐备,项目侧留 10 行翻译 shim)";§4 表 OcrLayoutPort 消费 shim 更新(首个 type=="table" 元素 + int 四元组约 5 行);偏离表补 success!=true 有意修复与 429 细分有意放弃两行。
- migrations/video-tree-trm5.md: R9/R10 改已闭;§8 表 13 行裁决与设计 §10.1 对齐(check_health 形态、resp.ok→2xx 收紧)。
.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 收尾门(不可跳过)
- 全量 CI:
conda run -n PolyGateway make ci→ 0 失败,覆盖率 ≥80%;make lint→ 0 错误(radon 门在 hook)。 - 慢速真实等待用例(时间语义)不受本里程碑影响,但全量跑确认无回归。
- 独立 verifier(verification-before-completion): 全新上下文只读 subagent,输入=设计+本计划+分支 diff 区间,亲自跑测试,Critical/Important 清零后方可声称完成。
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 不做清单;任何"顺手重构"禁止。