模式接力中的审阅节点摘要在多阶段 AI 流水线Agent Pipeline中业界习惯把注意力放在“生成”环节模型写代码、模型写文档、模型做规划。但真正决定整条流水线质量上限的往往不是生成能力而是那个必须由人或一个独立审阅器动脑子的节点——Plan 确认。本文把这种“上游模式产出、下游模式消费、中间靠人工确认交接”的协作方式称为模式接力并深入剖析接力过程中唯一的审阅节点Plan 确认。文章会给出可运行的代码示例展示如何设计一个带审阅节点的流水线、如何定义 Plan 的结构化契约、如何实现审阅门控与回退重试以及如何避免把审阅节点做成“橡皮图章”。一、引言为什么 Plan 确认是整条流水线唯一需要动脑的地方先看一条典型的模式接力流水线需求澄清 → Plan 制定 → Plan 确认 → 代码生成 → 静态检查 → 测试生成 → 人工验收在这条链路上除了Plan 确认其余环节几乎都可以被自动化需求澄清可以由模型反复追问补全Plan 制定模型输出结构化计划代码生成模型按 Plan 逐条实现静态检查linter / type checker 自动跑测试生成模型按 Plan 的验收条件生成用例人工验收多数情况下只是点一下“通过”。唯独 Plan 确认不行。原因很朴素Plan 是整条流水线的“意图快照”。一旦 Plan 错了后面所有环节都会以极高的效率、极低的成本把一个错误的想法执行得非常彻底。代码生成越快错误扩散得越快测试生成越全错误被“验证通过”的概率越高。这就是所谓的“垃圾进、垃圾出而且出得飞快”。所以Plan 确认节点的价值不在于“再检查一遍”而在于它是唯一一个可以在错误被放大之前踩下刹车的地方。它需要动脑是因为它要回答三个模型自己回答不了的问题这个 Plan 解决的是不是真正的问题意图对齐这个 Plan 的边界和假设是否成立可行性判断这个 Plan 失败时代价是否可接受风险判断这三个问题没有标准答案只能靠人的判断。本文接下来会把“如何把这个审阅节点工程化”讲清楚。二、模式接力的本质模式之间的契约而非代码之间的调用2.1 什么是“模式接力”“模式”Mode在这里指流水线中一个具备独立职责的处理单元它有自己的输入契约、输出契约和失败语义。模式接力指的是上游模式的输出作为下游模式的输入中间通过一个显式的交接物传递。关键点在于模式之间不共享隐式状态只通过交接物通信。这和微服务的设计哲学一致——服务之间通过消息契约通信而不是共享内存。在 AI 流水线里这个交接物通常是一个结构化对象比如fromdataclassesimportdataclass,fieldfromtypingimportList,OptionalfromenumimportEnumclassPlanStatus(Enum):DRAFTdraft# 模型刚产出未审阅APPROVEDapproved# 审阅通过可进入下游REJECTEDrejected# 审阅驳回需重做REVISEDrevised# 审阅后修改过需二次确认dataclassclassPlanStep:Plan 中的单个步骤必须是可执行、可验证的step_id:strdescription:str# 做什么inputs:List[str]# 依赖哪些上游产物outputs:List[str]# 产出什么acceptance:str# 验收条件可被测试直接引用risk:strlow# low / medium / highdataclassclassPlan:模式接力的核心交接物plan_id:strgoal:str# 一句话说明要达成的目标assumptions:List[str]# 显式列出的假设out_of_scope:List[str]# 明确不做什么steps:List[PlanStep]status:PlanStatusPlanStatus.DRAFT reviewer:Optional[str]Nonereview_note:Optional[str]None2.2 为什么契约必须显式如果 Plan 只是一段自然语言审阅者只能“读一遍感觉一下”审阅质量完全取决于审阅者当天的状态。而一旦 Plan 被结构化审阅节点就可以做三件事逐字段审阅假设、范围、步骤、验收条件分别看机器辅助检查比如检查每个 step 是否都有 acceptance是否有循环依赖审阅留痕审阅意见绑定到具体字段下游可以追溯。下面这个校验函数就是“机器辅助检查”的例子它不替代人的判断但能把明显不合格的 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_idsanddepstep.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:int3):self.max_roundsmax_rounds self.rounds0defsubmit(self,plan:Plan,reviewer:str,decision:str,note:str)-Plan: decision: approve | reject | revise self.rounds1ifself.roundsself.max_rounds:# 防止无限打回强制升级到人工决策raiseRuntimeError(fPlan{plan.plan_id}已审阅{self.max_rounds}轮仍未通过需升级处理)plan.reviewerreviewer plan.review_notenoteifdecisionapprove:plan.statusPlanStatus.APPROVEDelifdecisionreject:plan.statusPlanStatus.REJECTEDelifdecisionrevise:plan.statusPlanStatus.REVISEDelse:raiseValueError(f未知审阅决定:{decision})returnplandefcan_proceed(self,plan:Plan)-bool:下游模式在启动前必须调用此方法returnplan.statusPlanStatus.APPROVED3.2 回退时必须携带审阅意见审阅驳回后Plan 会回到“Plan 制定”模式重做。这里有个关键细节回退必须携带审阅意见否则模型会重复犯同样的错误。defreplan(original:Plan,review_note:str,planner)-Plan:带着审阅意见重新制定 Planpromptf 你之前的 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_planplanner(prompt)new_plan.plan_idoriginal.plan_id# 保持同一 plan_id 便于追溯new_plan.statusPlanStatus.DRAFTreturnnew_plan3.3 审阅节点必须可追溯审阅意见是整条流水线最有价值的“负样本”。它应该被持久化用于后续同类任务的 Plan 制定参考。一个简单的追溯记录importjsonfromdatetimeimportdatetimedefrecord_review(plan:Plan,decision:str,note:str,path:strreview_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,encodingutf-8)asf:f.write(json.dumps(record,ensure_asciiFalse)\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_ids1,description实现用户登录,inputs[user_table],outputs[login_api],acceptance登录功能正常,# ❌ 不可执行)合格的例子PlanStep(step_ids1,description实现用户登录接口 POST /login,inputs[user_table],outputs[login_api,session_token],acceptance给定已注册用户当提交正确用户名密码则返回 200 与 session_token当密码错误则返回 401 且不返回 token,)4.4 风险与代价高风险的 step 是否有回退方案对于标记为riskhigh的步骤审阅者要确认如果这一步失败代价是什么有没有回退路径没有回退方案的高风险步骤应该被打回。把这四个维度固化成审阅清单可以显著降低审阅的随意性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):DRAFTdraftAPPROVEDapprovedREJECTEDrejectedREVISEDreviseddataclassclassPlanStep:step_id:strdescription:strinputs:List[str]outputs:List[str]acceptance:strrisk:strlowdataclassclassPlan:plan_id:strgoal:strassumptions:List[str]out_of_scope:List[str]steps:List[PlanStep]status:PlanStatusPlanStatus.DRAFT reviewer:strreview_note:str# ---------- 模式桩实现 ----------defplanner(prompt:str)-Plan:Plan 制定模式真实场景下调用 LLMreturnPlan(plan_idp-001,goal为订单服务增加幂等下单接口,assumptions[下游支付网关支持幂等键,订单表已有唯一索引],out_of_scope[不处理跨机房一致性问题,不改造历史订单数据],steps[PlanStep(step_ids1,description在 POST /orders 中接收 idempotency_key,inputs[orders_api],outputs[orders_api_v2],acceptance给定相同 idempotency_key 的两次请求则第二次返回与第一次相同的 order_id,),PlanStep(step_ids2,description为 idempotency_key 建立唯一索引,inputs[orders_table],outputs[migration_001],acceptance迁移可重复执行且不报错,riskmedium,),],)defcoder(plan:Plan)-str:代码生成模式真实场景下调用 LLMassertplan.statusPlanStatus.APPROVED,未审阅通过的 Plan 不得进入代码生成returnf// 按 Plan{plan.plan_id}生成代码共{len(plan.steps)}步# ---------- 审阅门控 ----------classReviewGate:def__init__(self,max_rounds:int3):self.max_roundsmax_rounds self.rounds0defsubmit(self,plan:Plan,reviewer:str,decision:str,note:str)-Plan:self.rounds1ifself.roundsself.max_rounds:raiseRuntimeError(审阅轮次超限需升级处理)plan.reviewerreviewer plan.review_notenote plan.status{approve:PlanStatus.APPROVED,reject:PlanStatus.REJECTED,revise:PlanStatus.REVISED,}[decision]returnplandefcan_proceed(self,plan:Plan)-bool:returnplan.statusPlanStatus.APPROVED# ---------- 流水线 ----------defrun_pipeline(plan:Plan,gate:ReviewGate,review_fn:Callable[[Plan],tuple]): review_fn:审阅函数返回(decision,note)真实场景下这是一个人工审阅界面测试中用