+
+
AuditCtx — 结构与调用速查
+
LeAudit 服务编排层的不可变上下文 · 2026-04-27
+
+
+ 是什么：
+ AuditCtx（src/leaudit/services/audit_ctx.py）是一次评查运行的不可变骨架。@dataclass(frozen=True)，构造一次，跨 service 流转；每个 stage 用 with_xxx() 派生新 ctx，旧 ctx 不变。
+
+
+
结构
+
+
+ | 字段 | 类 | 类型 | 说明 |
|---|
+
+ document_id | 身份 | str | 这次 run 的 ID |
+ rules_file | 身份 | RulesFile | None | 评查规则；None 时由分类器决定 |
+ file_path | 身份 | str | None | 本地文件路径 |
+ page_range | 身份 | tuple[int,...] | None | 子文档页面范围 |
+ services | 装配 | AuditServices | 所有 service / client 的容器 |
+ config | 装配 | AuditConfig | 运行期旋钮 |
+ normalized_doc | 产物 | NormalizedDocument | None | OCR + 分类 + 分段 |
+ extraction | 产物 | ExtractionBundle | None | 抽取字段 / 多实体 / 派生 |
+ phase | 产物 | str | None | draft 或 executed |
+ evaluation | 产物 | EvaluationResult | None | 每条规则的评查结论 |
+ fallback_tasks | 产物 | tuple[RescueTask,...] | 失败规则的修救任务 |
+ extraction_errors | 诊断 | tuple[str,...] | 抽取错误日志 |
+ timing | 诊断 | Mapping[str,float] | 每阶段耗时 |
+
+
+
+
不可变性：frozen=True 防字段重绑；__post_init__ 把 list/dict 强转 tuple / MappingProxyType，防 ctx.timing["x"] = 1 这种穿透改写。
+
+
调用
+
+
1. 构造
+
+
from leaudit.services.audit_ctx import AuditCtx
+from leaudit.services.audit_services import AuditServices
+from leaudit.config.audit_config import AuditConfig
+
+services = AuditServices(
+ llm_client=llm, vlm_client=vlm, ocr_client=ocr,
+ normalization=norm_svc, extraction=ext_svc, evaluation=eval_svc,
+)
+
+ctx = AuditCtx(
+ document_id="my_doc",
+ rules_file=rules,
+ services=services,
+ file_path="/path/to/doc.pdf",
+ config=AuditConfig(group_size=8),
+)
+
+
2. 演化（每个 stage 一次）
+
+
ctx = ctx.with_normalized_doc(ocr_result) # Stage 1
+ctx = ctx.with_extraction(extraction) # Stage 3
+ctx = ctx.with_phase("executed") # Stage 4
+ctx = ctx.with_evaluation(evaluation) # Stage 5
+ctx = ctx.with_fallback_task(task) # Stage 6（每条失败规则一次）
+ctx = ctx.with_timing(ocr=1.2, total=8.5) # 累加耗时
+
+
所有 with_xxx() 都是 dataclasses.replace() 的薄包装——返回新对象，旧 ctx 不变。要用 ctx = ctx.with_...() 接住返回值。
+
+
3. 读取
+
+
ocr_result = ctx.normalized_doc
+fields = ctx.extraction.fields
+phase = ctx.phase
+score = ctx.evaluation.total_score
+ocr_secs = ctx.timing.get("ocr", 0.0)
+text = ctx.source_text # 派生属性 = ctx.extraction.source_text
+
+
4. 两个桥接 API
+
+
+ | API | 用途 |
|---|
+
+
+ ctx.effective_config |
+ 合并 ctx.config + Settings。业务代码统一只读这个，禁止 get_settings() / os.getenv（架构 guard 在 tests/architecture/） |
+
+
+ ctx.as_rescue_inputs() |
+ 转成 engine 旧的 RescueInputs 形状，桥接 engine._try_ai_rescue，不用改 engine |
+
+
+
+
+
典型调用：AuditService.audit(ctx)
+
+
所有 stage 已经在 AuditService.audit() 里编排好了（src/leaudit/services/audit_service.py:152）。普通用法只要构造 ctx 然后调它：
+
+
ctx = AuditCtx(document_id="...", rules_file=rules, services=services,
+ file_path="...", config=cfg)
+ctx = await audit_service.audit(ctx)
+# 跑完 ctx 已经被填满 normalized_doc / extraction / phase / evaluation / fallback_tasks / timing
+
+
七个 stage 顺序：Normalize → Resolve rules → Extract → Phase 判定 → Evaluate → Rescue → Finalize。
+
+
规则三条
+
+
+ - 加产物字段 →
audit_ctx.py 加字段 + 一个 with_xxx helper
+ - 读运行配置 →
ctx.effective_config，不要直接 get_settings()
+ - 改 stage 行为 → 必须 return 新 ctx，不要原地改字段
+
+
+