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 | NoneOCR + 分类 + 分段
extraction产物ExtractionBundle | None抽取字段 / 多实体 / 派生
phase产物str | Nonedraft 或 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。

规则三条