第二季第 4 篇：审查工作流编排——从 PDF 到报告，一条 DAG 跑完
📌 本文是《项目实战：从零搭建智能合同审查平台》（第二季）的第 4 篇🎯 读完本文你将：① 理解 DAG 工作流在合同审查中的编排逻辑 ② 实现条件分支 / 并行执行 / 失败重试 / 超时熔断 ③ 把前三篇的插件串成端到端流水线 ④ 对 5 份合同跑出完整审查报告（Markdown + 结构化 JSON）⏱️ 预计阅读时间：22 分钟 | 动手实践：30 分钟💻 前置要求：完成第二季第 1、2、3 篇🔗 原理对照：万悟平台的「工作流编排」能力讲的是"节点定义 → DAG 拓扑排序 → 调度执行 → 执行上下文"。本篇做的是同一件事——解析、规则审查、偏差对比、报告生成是四个节点，DAG 决定谁先谁后、谁能并行、谁失败了怎么重试。没读过万悟源码不影响本篇阅读。
一、本篇要解决什么问题？
1.1 前三篇的碎片
第 1 篇：parse_contract(text, contract_id) → ParsedContract第 2 篇：route_and_review(parsed, rules_by_type) → List[Finding]第 3 篇：evaluate_deviations(parsed, baselines) → DeviationReport verify_standards(parsed, baselines) → StandardReport
四个函数，四次手动调用。法务不会写 Python。
1.2 本篇目标
输入：一个 PDF 文件路径（或一批）输出： ├── review_report.md（人读） ├── review_report.json（机读，对接 OA / ERP） └── 控制台摘要（风险评分 + HIGH 条目）约束： ├── 端到端 < 3 秒（纯规则层）/ < 15 秒（含 LLM） ├── 单节点失败不阻塞全链路（降级继续） └── 支持批量（5 份合同并行审查）
1.3 为什么需要工作流引擎？
二、核心概念
2.1 DAG 拓扑
┌─────────────────┐ │ N1: 合同解析 │ │ (parser) │ └────────┬────────┘ │ ┌────────▼────────┐ │ N2: 规则审查 │──────────────┐ │ (含类型路由) │ │ ← N2 与 N3 并行（都只读 ParsedContract） └────────┬────────┘ │ │ ┌────────▼────────┐ ┌────────▼────────┐ │ N3: 偏差对比 │ │ N4: 标准校验 │ │ (deviation) │ │ (verify_standards) │ └────────┬────────┘ │ │ └────────┬────────┘ │ │ ┌────────▼────────┐ │ │ N5: LLM第二意见 │◀─条件：仅当 │ │ (可选,可关闭) │ 注入LLM客户端 │ └────────┬────────┘ │ │◀──────────────────────┘ ┌────────▼────────┐ │ N6: 构建RiskCard │ │ (build_risk_card)│ └─────────────────┘
关键设计：N2（规则审查）与 N3（偏差对比）无数据依赖，可以并行；N4（标准校验）对所有类型运行，但无基线 JSON 的类型（NDA/lease）自然返回空报告；N5（LLM 第二意见）是可选节点，没注入 LLM 客户端时直接 SKIPPED，不阻塞主链路。
2.2 节点状态机
PENDING → RUNNING → SUCCESS → FAILED → SKIPPED（标记失败，链路降级继续）
超时/重试是 DAGNode 的可配置字段（timeout_s 默认 30s、retry 默认 0）。教学版执行器以"节点隔离 + SKIPPED"为主，不实现自动重试循环——retry 字段预留给生产环境接指数退避。
2.3 并行策略
# N3 和 N4 并行执行with ThreadPoolExecutor(max_workers=2) as pool: f_review = pool.submit(review_contract, parsed) f_deviation = pool.submit(analyze_deviations, parsed, engine) review_result = f_review.result(timeout=5) deviation_report = f_deviation.result(timeout=10)
2.4 条件分支
# 标准校验节点对所有合同类型都会执行；# 但无基线 JSON 的类型（如 nda / lease）会自然返回空报告（0 条校验），不报错。CONDITIONS = {# 仅当注入了 LLM 客户端时才跑”LLM 第二意见”节点，否则 SKIPPED ”N5_llm_second_opinion”: lambda ctx: ctx.llm_client is not None,}# 不满足条件 → 节点状态 = SKIPPED，不执行
2.5 风险卡：一份输出、三处消费的枢纽
前三篇的产物散落在 ReviewResult（规则发现）、DeviationReport（偏差）、StandardVerification（标准验证）里。实战里，一份审查报告要同时被三种消费者用到：
- 机读：对接 OA / ERP / 审批流（本篇 JSON 报告）；
- 回流：人工反馈（采纳/驳回/改写）回流规则库（第 7 篇）。
如果每处消费端都各读各的结构，就会产生"同一结论三套表示"的混乱。本季把这类产物收敛成RiskCard（风险卡）作为统一枢纽——一份输出，三处消费：
@dataclassclass RiskCard: ”””风险卡：一份审查的枢纽产物（跨节点 / 跨消费端共享）””” contract_id: str contract_type: str tenant_id: str = ”” risk_score: int = 0# 由 score_findings(findings) 计算 findings: List[dict] = field(default_factory=list)# 规则 / 模型发现 deviations: List[dict] = field(default_factory=list)# 基线偏差 standard_verifications: List[dict] = field(default_factory=list) config_snapshot: str = ””# 配置 SHA256 快照（第 2 篇 2.6） degraded_sources: List[str] = field(default_factory=list)# 降级的外部源 llm_opinion: Optional[str] = None# LLM 第二意见（可选，可关闭） llm_degraded: bool = False feedback: List[dict] = field(default_factory=list)# 人工反馈预留位（第 7 篇） elapsed_ms: int = 0
RiskCard 既是第 7 篇可视化的输入，也是第 8 篇业务集成的输入——一篇生产、多篇消费，避免重复建模。第 5 节「报告生成」即把它渲染为 Markdown + JSON 两种形态。
2.6 外部服务并行调用与降级
第 3 篇 2.5 定义了「外部权威数据源接入层」。把它接到工作流里，就要解决并行 + 降级：多个外部源（主体核验 / 法规时效）应并行查，且任一不可用都不能拖垮整条链。
import asyncioasync def verify_external_sources(queries: List[str]) -> List[SourceResult]: ”””并行调用外部源，单源失败不影响整体””” async def _one(q): try:# 真实实现：await real_source.verify(q)，带独立超时 return mock_source.verify(q) except Exception: return SourceResult(source_id=”ext”, available=False, status=”暂不可用”, degraded=True)# return_exceptions：单源异常被捕获，不中断其余 return await asyncio.gather(*[_one(q) for q in queries], return_exceptions=True)# 落地为 DAG 节点（N5 之后的并行节点，或并入 N5）：# 服务不可用 → RiskCard.degraded_sources 追加该源，标「暂不可用」# 注意：降级时标「暂不可用」，**绝不**标「无风险」
实战要求（呼应第 2 篇 2.6 降级三原则）：
- 结构化降级：
RiskCard.degraded_sources 明确记录哪些源没拿到数据； - 宁少不误报：缺外部数据标「暂不可用」，不误判为「合规」。
三、核心代码拆解（完整代码见《附录-合同审查平台完整代码库精讲》（本季最后附录））
# workflow/review_pipeline.py”””合同审查 DAG 工作流引擎串联：解析 → 规则审查(含路由) ∥ 偏差对比 → 标准校验 → [可选 LLM 第二意见] → 构建 RiskCard”””import timefrom dataclasses import dataclass, fieldfrom typing import Callable, Dict, List, Optional // …（完整代码见《附录-合同审查平台完整代码库精讲》（本季最后附录））
四、运行结果（真实引擎输出，与第 2 篇 §四 一致）
💡 下表用真实引擎 engine.review_for_tenant(tenant_id, text, contract_id)（默认 tenant_yuanda_v1）对 5 份样本跑出的 规则审查风险评分，与第 2 篇 §四 同源、同口径。偏差比对 / 标准校验另见第 3 篇（NDA/lease 无基线 JSON，自然为空）。
审查总览（规则引擎 risk_score）┏━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━┳━━━━━━┳━━━━━┳━━━━━┓┃ 合同 ┃ 类型 ┃ 评分 ┃ HIGH┃ MED ┃┡━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━╇━━━━━━╇━━━━━╇━━━━━┩│ labor_contract_001 │ labor │ 30 │ 1 │ 0 ││ procurement_contract_001 │ procurement│ 30 │ 0 │ 2 ││ nda_001 │ nda │ 0 │ 0 │ 0 ││ lease_contract_001 │ lease │ 0 │ 0 │ 0 ││ construction_contract_001│ construction│ 60 │ 1 │ 2 │└──────────────────────────┴────────────┴──────┴─────┴─────┘
诚实边界（务必和第 2 篇一起看）：
- 采购合同 HIGH 规则（预付款 60%、赔偿上限 200%）未触发——与第 2 篇 §四 一致，源于解析精度（取条款首个百分比）的已知局限，不是规则写错。
- NDA / lease 评分 0：本季未内置这两类的专属规则集与基线，规则引擎无命中、偏差比对无基线可比，属预期边界。
- 采购合同的 2 条 MEDIUM 均为
YUANDA-PROC-003（违约金不对等：条款含"违约金"但缺"对等/一致/相同"措辞）。construction 的 1 条 HIGH 为 YUANDA-CONS-002（质保金 3% > 约定 1.5%）。
生成的 Markdown 报告（采购合同示例，真实命中）
# 合同审查报告| 项目 | 值 ||------|-----|| 合同编号 | procurement_contract_001 || 合同类型 | procurement || 风险评分 | **30/100** || HIGH 风险 | 0 条 || MEDIUM 风险 | 2 条 |---## 一、规则审查发现| 条款 | 规则 | 等级 | 风险描述 | 建议 ||------|------|------|----------|------|| 第七条 | YUANDA-PROC-003 | MEDIUM | 违约金条款未体现对等性（缺”对等/一致/相同”措辞） | 补充双方对等违约金约定 |（注：预付款 60%、赔偿上限 200% 的 HIGH 规则因解析精度未触发，见上方诚实边界）## 二、工作流执行日志（示意节点耗时）| 节点 | 状态 | 说明 ||------|------|------|| N1_parse | SUCCESS | 解析合同条款 || N2_review | SUCCESS | 规则审查（含类型路由） || N3_deviation | SUCCESS | 偏差比对（procurement 有基线） || N4_standard | SUCCESS | 标准校验（0 条：样本条款均已覆盖） || N5_llm | SKIPPED | 未注入 LLM 客户端（可关闭） || N6_build | SUCCESS | 构建 RiskCard |---*本报告由示例教学平台自动生成，仅供参考，不构成法律意见。*
五、踩坑记录
| | | | |
|---|
| | deviation_report | 两个线程同时写 ctx 属性，GIL 不保护复合赋值 | 每个节点写自己的字段（N3 写 review_result，N4 写 deviation_report），无交叉 |
| | | depends_on 只写了 ["N2_route"]，没等 N3/N4 | 改为 depends_on=["N3_review", "N4_deviation"] |
| | | nda/lease 没内置 baseline JSON，verify_standards 直接返回空 | 在 baseline/baselines/ 加对应.json（见第 3 篇 §2.1） |
| | | contract_id | 文件名加时间戳：{id}_{timestamp}.md |
| ThreadPoolExecutor | | Windows 默认 spawn 模式，子进程重新 import 触发 if __name__ | 加 if __name__ == "__main__": 保护 + freeze_support() |
六、与万悟平台能力对照
| 万悟平台·工作流编排（→ 第一季第 7 篇《工作流引擎》） 能力 | |
|---|
| | 解析 / 规则 / 偏差比对 / 标准校验 / 报告（LLM 第二意见可选） |
| | |
| | 是否注入 LLM 客户端 → 是否跑第二意见节点（标准校验对所有类型运行） |
| | |
| | |
七、小结 & 下一篇预告
⏱️ 30 秒速览
这篇你只需要记住 3 件事：
- 审查流程编排成 DAG：解析 → 规则审查 ∥ 偏差对比 → 标准校验 → [可选 LLM 第二意见] → 构建 RiskCard
- 节点可并行（规则审查 ∥ 偏差对比）、可降级（单节点失败标 SKIPPED 继续，无人工审批节点）
- 一份 RiskCard 被三处消费（Markdown 报告 / JSON 报告 / 可视化标注），避免重复建模
想深挖？
- 完整代码（DAG 编排 + RiskCard 枢纽）：见**《附录-合同审查平台完整代码库精讲》（本季最后附录）**；编排原理：§2 核心概念
本篇要点
- ✅ DAG 工作流：6 个节点，拓扑排序执行，规则审查 ∥ 偏差对比并行
- ✅ 标准校验对所有类型运行，无基线类型（NDA/lease）自然返回空报告
- ✅ 节点级隔离：单节点失败不阻塞全链路，标记 FAILED/SKIPPED 继续
- ✅ RiskCard 枢纽：Markdown（人读）+ JSON（机读）+ 可视化标注，三处消费
- ✅ 真实引擎 5 份输出：labor 30(H1) / procurement 30(M2) / construction 60(H1M2) / nda 0 / lease 0（详见 §4）
下一篇预告
第二季第 5 篇：批量性能与异步队列5 份合同 312ms 没问题。但法务部一次丢过来 200 份呢？本篇实现：异步任务队列（Celery/Redis）+ 进度推送（WebSocket）+ 背压控制 + 性能基准测试。目标：200 份合同 < 60 秒，P99 延迟 < 500ms/份。
🔗 原理对照：万悟平台的「工程实践·性能」能力
附录说明
本篇正文已讲清设计思路与关键片段；完整可运行源码库（解析 + 审查 / 批量 / 多租户 / 基线 / 可观测等全部模块，已测 63 passed）在《附录-合同审查平台完整代码库精讲》（本季最后附录）中逐一对应；源码可回复公众号关键词「第二季项目代码」，自动回复会给到付费文章《第二季·智能合同审查平台完整源码》（¥29）→ 付后解锁，文内附微云下载直链 + 二维码。
📦 获取方式：回复公众号关键词「第二季项目代码」→ 自动回复给付费文章《第二季·智能合同审查平台完整源码》（¥29）→ 付后解锁，文内附微云下载直链 + 二维码。
⚠️ 教学代码声明：本季配套源码 contract-review-platform 为示例教学代码，仅用于学习合同审查系统的设计思路与工程实现，不构成任何法律意见，也不保证适用于真实业务。作者不承担因使用、修改或部署本代码产生的任何问题解答义务或法律/商业后果；你可以在本示例代码基础上自行修改以满足你的需求。用于真实合同审查前，请务必由执业法律人士审核。
📱 关注公众号，追更不迷路
本系列文章首发于微信公众号「农夫三拳有点癫」，每周更新源码拆解与架构实战。
在微信扫描下方二维码即可关注：
