模式接力中的审阅节点

摘要

在多阶段 AI 流水线（Agent Pipeline）中，业界习惯把注意力放在“生成”环节：模型写代码、模型写文档、模型做规划。但真正决定整条流水线质量上限的，往往不是生成能力，而是那个必须由人（或一个独立审阅器）动脑子的节点——Plan 确认。本文把这种“上游模式产出、下游模式消费、中间靠人工确认交接”的协作方式称为模式接力，并深入剖析接力过程中唯一的审阅节点：Plan 确认。文章会给出可运行的代码示例，展示如何设计一个带审阅节点的流水线、如何定义 Plan 的结构化契约、如何实现审阅门控与回退重试，以及如何避免把审阅节点做成“橡皮图章”。

一、引言：为什么 Plan 确认是整条流水线唯一需要动脑的地方

先看一条典型的模式接力流水线：

需求澄清 → Plan 制定 → Plan 确认 → 代码生成 → 静态检查 → 测试生成 → 人工验收

在这条链路上，除了Plan 确认，其余环节几乎都可以被自动化：

• 需求澄清：可以由模型反复追问补全；
• Plan 制定：模型输出结构化计划；
• 代码生成：模型按 Plan 逐条实现；
• 静态检查：linter / type checker 自动跑；
• 测试生成：模型按 Plan 的验收条件生成用例；
• 人工验收：多数情况下只是点一下“通过”。

唯独 Plan 确认不行。原因很朴素：Plan 是整条流水线的“意图快照”。一旦 Plan 错了，后面所有环节都会以极高的效率、极低的成本，把一个错误的想法执行得非常彻底。代码生成越快，错误扩散得越快；测试生成越全，错误被“验证通过”的概率越高。这就是所谓的“垃圾进、垃圾出，而且出得飞快”。

所以，Plan 确认节点的价值不在于“再检查一遍”，而在于它是唯一一个可以在错误被放大之前踩下刹车的地方。它需要动脑，是因为它要回答三个模型自己回答不了的问题：

1. 这个 Plan 解决的是不是真正的问题？（意图对齐）
2. 这个 Plan 的边界和假设是否成立？（可行性判断）
3. 这个 Plan 失败时，代价是否可接受？（风险判断）

这三个问题没有标准答案，只能靠人的判断。本文接下来会把“如何把这个审阅节点工程化”讲清楚。

二、模式接力的本质：模式之间的契约，而非代码之间的调用

2.1 什么是“模式接力”

“模式”（Mode）在这里指流水线中一个具备独立职责的处理单元：它有自己的输入契约、输出契约和失败语义。模式接力指的是：上游模式的输出，作为下游模式的输入，中间通过一个显式的交接物传递。

关键点在于：模式之间不共享隐式状态，只通过交接物通信。这和微服务的设计哲学一致——服务之间通过消息契约通信，而不是共享内存。

在 AI 流水线里，这个交接物通常是一个结构化对象，比如：

fromdataclassesimportdataclass,fieldfromtypingimportList,OptionalfromenumimportEnumclassPlanStatus(Enum):DRAFT="draft"# 模型刚产出，未审阅APPROVED="approved"# 审阅通过，可进入下游REJECTED="rejected"# 审阅驳回，需重做REVISED="revised"# 审阅后修改过，需二次确认@dataclassclassPlanStep:"""Plan 中的单个步骤，必须是可执行、可验证的"""step_id:strdescription:str# 做什么inputs:List[str]# 依赖哪些上游产物outputs:List[str]# 产出什么acceptance:str# 验收条件（可被测试直接引用）risk:str="low"# low / medium / high@dataclassclassPlan:"""模式接力的核心交接物"""plan_id:strgoal:str# 一句话说明要达成的目标assumptions:List[str]# 显式列出的假设out_of_scope:List[str]# 明确不做什么steps:List[PlanStep]status:PlanStatus=PlanStatus.DRAFT reviewer:Optional[str]=Nonereview_note:Optional[str]=None

2.2 为什么契约必须显式

如果 Plan 只是一段自然语言，审阅者只能“读一遍感觉一下”，审阅质量完全取决于审阅者当天的状态。而一旦 Plan 被结构化，审阅节点就可以做三件事：

1. 逐字段审阅：假设、范围、步骤、验收条件分别看；
2. 机器辅助检查：比如检查每个 step 是否都有 acceptance，是否有循环依赖；
3. 审阅留痕：审阅意见绑定到具体字段，下游可以追溯。

下面这个校验函数就是“机器辅助检查”的例子，它不替代人的判断，但能把明显不合格的 Plan 挡在审阅者面前之前：

defprecheck_plan(plan:Plan)->List[str]:"""在人工审阅前做机械性检查，返回问题列表。 注意：这里只检查'形式合规'，不检查'内容正确'。 内容正确性必须由审阅节点的人来判断。 """problems=[]ifnotplan.goalorlen(plan.goal)<10:problems.append("goal 过于模糊，无法判断意图对齐")ifnotplan.assumptions:problems.append("未显式列出假设，审阅者无法判断前提是否成立")ifnotplan.out_of_scope:problems.append("未声明 out_of_scope，范围可能无限膨胀")step_ids={s.step_idforsinplan.steps}forstepinplan.steps:ifnotstep.acceptance:problems.append(f"步骤{step.step_id}缺少验收条件，无法验证")fordepinstep.inputs:# 依赖必须是上游步骤的产出，或者是外部输入ifdepinstep_idsanddep==step.step_id:problems.append(f"步骤{step.step_id}依赖自身")iflen(plan.steps)==0:problems.append("Plan 没有任何步骤")returnproblems

这段代码的定位很重要：它做的是“体检”，不是“诊断”。它能发现“没写验收条件”，但发现不了“验收条件写错了”。后者只能靠审阅节点。

三、审阅节点的设计：门控、回退与状态机

3.1 审阅节点是一个门控（Gate），不是一个步骤

很多流水线把“Plan 确认”实现成一个普通的流水线步骤，跑完继续往下。这是错误的。审阅节点应该是门控：下游能否启动，取决于门控的状态。

用一个简单的状态机来表达：

DRAFT ──审阅通过──> APPROVED ──> 下游启动 │ ├──审阅驳回──> REJECTED ──> 回到 Plan 制定（带审阅意见） │ └──审阅修改──> REVISED ──> 再次确认 ──> APPROVED / REJECTED

用代码实现这个门控：

classReviewGate:"""审阅门控：控制 Plan 能否放行到下游"""def__init__(self,max_rounds:int=3):self.max_rounds=max_rounds self.rounds=0defsubmit(self,plan:Plan,reviewer:str,decision:str,note:str="")->Plan:""" decision: 'approve' | 'reject' | 'revise' """self.rounds+=1ifself.rounds>self.max_rounds:# 防止无限打回，强制升级到人工决策raiseRuntimeError(f"Plan{plan.plan_id}已审阅{self.max_rounds}轮仍未通过，需升级处理")plan.reviewer=reviewer plan.review_note=noteifdecision=="approve":plan.status=PlanStatus.APPROVEDelifdecision=="reject":plan.status=PlanStatus.REJECTEDelifdecision=="revise":plan.status=PlanStatus.REVISEDelse:raiseValueError(f"未知审阅决定:{decision}")returnplandefcan_proceed(self,plan:Plan)->bool:"""下游模式在启动前必须调用此方法"""returnplan.status==PlanStatus.APPROVED

3.2 回退时必须携带审阅意见

审阅驳回后，Plan 会回到“Plan 制定”模式重做。这里有个关键细节：回退必须携带审阅意见，否则模型会重复犯同样的错误。

defreplan(original:Plan,review_note:str,planner)->Plan:"""带着审阅意见重新制定 Plan"""prompt=f""" 你之前的 Plan 被审阅驳回，驳回意见如下：{review_note}原 Plan： - 目标：{original.goal}- 假设：{original.assumptions}- 范围外：{original.out_of_scope}- 步骤：{[s.descriptionforsinoriginal.steps]}请针对驳回意见修订 Plan。注意： 1. 不要只改措辞，要改实质内容； 2. 如果驳回意见涉及假设不成立，请显式修改 assumptions； 3. 如果驳回意见涉及范围问题，请更新 out_of_scope。 """# planner 是一个封装了模型调用的函数，返回新的 Plan 对象new_plan=planner(prompt)new_plan.plan_id=original.plan_id# 保持同一 plan_id 便于追溯new_plan.status=PlanStatus.DRAFTreturnnew_plan

3.3 审阅节点必须可追溯

审阅意见是整条流水线最有价值的“负样本”。它应该被持久化，用于后续同类任务的 Plan 制定参考。一个简单的追溯记录：

importjsonfromdatetimeimportdatetimedefrecord_review(plan:Plan,decision:str,note:str,path:str="review_log.jsonl"):"""把审阅记录追加写入 JSONL，便于后续分析"""record={"ts":datetime.utcnow().isoformat(),"plan_id":plan.plan_id,"goal":plan.goal,"decision":decision,"note":note,"reviewer":plan.reviewer,"assumptions":plan.assumptions,"out_of_scope":plan.out_of_scope,}withopen(path,"a",encoding="utf-8")asf:f.write(json.dumps(record,ensure_ascii=False)+"\n")

这些记录积累起来之后，可以反哺 Plan 制定模式：把历史上被驳回最多的假设类型、最常见的范围遗漏，作为 few-shot 示例注入 planner 的 prompt。这是审阅节点从“成本中心”变成“资产中心”的关键路径。

四、审阅者该看什么：四个必查维度

审阅节点“需要动脑”，但动脑不能没有章法。实践中，审阅者应该按固定顺序检查四个维度，每个维度都有明确的通过标准。

4.1 意图对齐：Plan 解决的是不是真问题

检查goal和assumptions。问自己：

• 如果把goal单独拿给需求方看，他会说“对，就是这个”吗？
• assumptions里有没有一条，一旦不成立，整个 Plan 就失去意义？

第二条尤其重要。很多 Plan 的失败不是因为步骤写错，而是因为一个从未被检验的假设崩塌了。审阅者要把“致命假设”挑出来，要求 Plan 显式标注。

4.2 范围控制：out_of_scope是否足够具体

out_of_scope写成“不做无关的事”等于没写。合格的out_of_scope应该是可判定的，例如：

• ✅ “本次不处理并发写入场景”
• ✅ “不兼容 Python 3.8 以下版本”
• ❌ “不做过度设计”

审阅者要检查每条out_of_scope是否可判定。

4.3 步骤可验证性：每个 step 的 acceptance 是否可执行

这是最容易被忽略的一环。acceptance必须能被下游的测试生成模式直接翻译成断言。审阅者要检查：

• 是否能用“给定…当…则…”的形式复述？
• 是否有明确的输入输出？

不合格的例子：

PlanStep(step_id="s1",description="实现用户登录",inputs=["user_table"],outputs=["login_api"],acceptance="登录功能正常",# ❌ 不可执行)

合格的例子：

PlanStep(step_id="s1",description="实现用户登录接口 POST /login",inputs=["user_table"],outputs=["login_api","session_token"],acceptance="给定已注册用户，当提交正确用户名密码，则返回 200 与 session_token；""当密码错误，则返回 401 且不返回 token",)

4.4 风险与代价：高风险的 step 是否有回退方案

对于标记为risk="high"的步骤，审阅者要确认：如果这一步失败，代价是什么？有没有回退路径？没有回退方案的高风险步骤，应该被打回。

把这四个维度固化成审阅清单，可以显著降低审阅的随意性：

REVIEW_CHECKLIST={"intent":["goal 能被需求方一句话确认","致命假设已显式标注",],"scope":["每条 out_of_scope 可判定","没有隐含的范围扩张",],"verifiability":["每个 step 的 acceptance 可翻译为断言","steps 之间无循环依赖",],"risk":["高风险 step 有回退方案","整体失败代价可接受",],}defrender_checklist()->str:"""渲染审阅清单，供审阅界面展示"""lines=[]fordim,itemsinREVIEW_CHECKLIST.items():lines.append(f"##{dim}")foriteminitems:lines.append(f"- [ ]{item}")return"\n".join(lines)

五、完整示例：一个带审阅节点的最小流水线

把前面的片段串起来，给出一个可运行的最小流水线。为了不依赖真实模型，planner和coder用桩函数代替，但接口与真实实现一致。

fromdataclassesimportdataclass,fieldfromtypingimportList,CallablefromenumimportEnum# ---------- 数据契约 ----------classPlanStatus(Enum):DRAFT="draft"APPROVED="approved"REJECTED="rejected"REVISED="revised"@dataclassclassPlanStep:step_id:strdescription:strinputs:List[str]outputs:List[str]acceptance:strrisk:str="low"@dataclassclassPlan:plan_id:strgoal:strassumptions:List[str]out_of_scope:List[str]steps:List[PlanStep]status:PlanStatus=PlanStatus.DRAFT reviewer:str=""review_note:str=""# ---------- 模式（桩实现） ----------defplanner(prompt:str)->Plan:"""Plan 制定模式：真实场景下调用 LLM"""returnPlan(plan_id="p-001",goal="为订单服务增加幂等下单接口",assumptions=["下游支付网关支持幂等键","订单表已有唯一索引"],out_of_scope=["不处理跨机房一致性问题","不改造历史订单数据"],steps=[PlanStep(step_id="s1",description="在 POST /orders 中接收 idempotency_key",inputs=["orders_api"],outputs=["orders_api_v2"],acceptance="给定相同 idempotency_key 的两次请求，""则第二次返回与第一次相同的 order_id",),PlanStep(step_id="s2",description="为 idempotency_key 建立唯一索引",inputs=["orders_table"],outputs=["migration_001"],acceptance="迁移可重复执行且不报错",risk="medium",),],)defcoder(plan:Plan)->str:"""代码生成模式：真实场景下调用 LLM"""assertplan.status==PlanStatus.APPROVED,"未审阅通过的 Plan 不得进入代码生成"returnf"// 按 Plan{plan.plan_id}生成代码，共{len(plan.steps)}步"# ---------- 审阅门控 ----------classReviewGate:def__init__(self,max_rounds:int=3):self.max_rounds=max_rounds self.rounds=0defsubmit(self,plan:Plan,reviewer:str,decision:str,note:str="")->Plan:self.rounds+=1ifself.rounds>self.max_rounds:raiseRuntimeError("审阅轮次超限，需升级处理")plan.reviewer=reviewer plan.review_note=note plan.status={"approve":PlanStatus.APPROVED,"reject":PlanStatus.REJECTED,"revise":PlanStatus.REVISED,}[decision]returnplandefcan_proceed(self,plan:Plan)->bool:returnplan.status==PlanStatus.APPROVED# ---------- 流水线 ----------defrun_pipeline(plan:Plan,gate:ReviewGate,review_fn:Callable[[Plan],tuple]):""" review_fn:审阅函数，返回(decision,note)真实场景下这是一个人工审阅界面；测试中用