1. 为什么“说做完了”这件事值得单独做一条流水线多 Agent 协作编程现在是个热得发烫的方向但真正上手跑过几轮的人心里都清楚最让人血压升高的不是模型写不出代码而是它一脸笃定地告诉你“任务已完成”结果你去检查测试没跑、边界没覆盖、依赖没装、文件路径还是错的。这种“口头交付”在单 Agent 场景下顶多算个小麻烦放到多 Agent 流水线里就是灾难——上游 Agent 把半成品标记为完成下游 Agent 基于这个错误前提继续往下做错误会像滚雪球一样放大最后你拿到的是一坨看起来结构完整、实际根本跑不起来的东西。我给这套流水线起的内部代号叫GatePipeline核心思路只有一句话任何 Agent 声称的“完成”都必须经过一道独立的验收门禁门禁不通过就不算完成下游不许启动。这不是什么高深的技术本质上就是把软件工程里“Definition of Done”那套东西硬编码进多 Agent 的调度逻辑里。但恰恰是这种“笨办法”把多 Agent 编程从“看起来很美好”拉到了“真的能用”的水平。这篇文章适合三类人看一是已经在玩多 Agent 协作、被“假完成”坑过的开发者二是想给现有 AI 编程流程加一道质量闸门的工程负责人三是单纯对多 Agent 调度架构感兴趣、想看看别人怎么踩坑的技术爱好者。我会把整套流水线的设计思路、核心代码结构、门禁的具体实现、以及我在实际跑项目中总结的避坑经验全部摊开讲代码部分会给出可直接参考的 Python 实现。整套东西是开源的你可以在自己的项目里直接改吧改吧就用。先说清楚一个前提这套流水线不绑定任何特定的大模型厂商你用的是哪家 API 都行核心逻辑在于调度层和验收层的设计。我自己的测试环境里混用过好几个模型只要它们能返回结构化的输出这套门禁机制就能生效。2. 多 Agent 编程流水线的整体架构拆解2.1 为什么是“流水线 门禁”而不是“自由协作”多 Agent 协作目前主流有两种玩法。一种是自由协作式几个 Agent 互相聊天、互相提意见像一群人在会议室里头脑风暴最后收敛出一个结果。另一种是流水线式把任务拆成固定阶段每个阶段由专门的 Agent 负责阶段之间串行推进。我一开始也试过自由协作结论是在编程这种对正确性要求极高的场景下自由协作的不可控性太强了。Agent 之间会互相“客气”一个说“我觉得差不多了”另一个说“好的没问题”然后大家一起把一个半成品放行。更糟糕的是当出现分歧时它们会陷入无休止的讨论token 烧得飞快产出却寥寥。流水线式的优势在于职责边界清晰。每个 Agent 只对自己那一段负责做完就必须交出可验证的产物。而门禁的作用就是在每个阶段的出口处设一个“检查站”用客观标准判断这一阶段是否真的达标。这就像工厂的质检环节——不是工人说“我装好了”就算数而是要过检具、过测试台数据合格才放行到下一道工序。我最终确定的流水线阶段是这样的阶段负责 Agent核心产出门禁检查项需求解析Planner任务拆解清单、验收标准拆解是否覆盖原始需求、标准是否可量化代码实现Coder代码文件、依赖清单语法检查、静态分析、依赖可安装测试编写Tester测试用例、测试报告测试是否覆盖核心路径、是否全部通过集成验证Integrator可运行产物、运行日志端到端跑通、无报错、输出符合预期文档整理DocumenterREADME、使用说明文档是否包含运行步骤、是否与实际一致每个阶段的门禁不是简单的“通过/不通过”二值判断而是带反馈的验收。门禁失败时会把具体的失败原因结构化地返回给对应 Agent让它针对性修复而不是推倒重来。这一点很关键后面会详细讲。2.2 门禁的核心设计原则可验证、可复现、可追溯门禁要真正起作用必须满足三个条件。第一是可验证检查项必须是客观的、机器能判断的不能是“代码写得好不好”这种主观评价。比如“语法检查通过”是可验证的“代码风格优雅”就不是。第二是可复现同样的产物跑同样的门禁结果必须一致不能这次通过下次不通过。第三是可追溯每次门禁的检查结果、失败原因、修复记录都要落盘方便事后复盘。我见过一些团队的做法是让另一个 Agent 来“评审”前一个 Agent 的产出这本质上还是主观判断评审 Agent 的心情也就是随机性会影响结果。我的做法是能用确定性工具检查的绝不用模型判断。语法用编译器检查依赖用包管理器检查测试用测试框架跑只有那些确实需要语义理解的检查项比如“测试用例是否覆盖了需求里的边界条件”才交给模型而且要给模型非常明确的判断标准。这里有个实操心得门禁的检查脚本本身也要有测试。我一开始写的门禁脚本有个 bug导致某些情况下明明测试没跑却判定为通过结果放行了一堆有问题的代码。后来我给门禁脚本本身加了一套单元测试用故意构造的“坏产物”去验证门禁能不能正确拦截。这个习惯救了我很多次。2.3 状态机让每个任务都有明确的“当前状态”整套流水线用一个状态机来管理任务流转。每个任务从创建到完成会经历一系列状态状态之间的转换必须满足前置条件。这样做的好处是任何时候你都能清楚地知道一个任务卡在哪一步、为什么卡住。状态定义大致如下PENDING任务已创建等待调度PLANNING需求解析中PLAN_REVIEW需求解析门禁检查中CODING代码实现中CODE_REVIEW代码门禁检查中TESTING测试编写与执行中TEST_REVIEW测试门禁检查中INTEGRATING集成验证中INTEGRATION_REVIEW集成门禁检查中DOCUMENTING文档整理中DONE全部通过任务完成BLOCKED门禁连续失败超过阈值需要人工介入注意BLOCKED这个状态。门禁不能无限重试否则一个死活修不好的问题会让整个流水线卡死。我设置的阈值是同一个门禁连续失败 3 次就转BLOCKED把问题抛给人。实际跑下来大部分问题 Agent 能在 1 到 2 次修复内解决少数需要 3 次真正需要人工介入的比例大概在 5% 左右。3. 验收门禁的具体实现与核心代码3.1 门禁的抽象接口设计先看门禁的抽象基类。所有具体门禁都继承这个类实现check方法。返回值是一个结构化的结果对象包含是否通过、失败原因列表、以及给 Agent 的修复建议。from abc import ABC, abstractmethod from dataclasses import dataclass, field from typing import List, Optional dataclass class GateResult: passed: bool reasons: List[str] field(default_factorylist) suggestions: List[str] field(default_factorylist) raw_output: Optional[str] None class BaseGate(ABC): name: str base_gate abstractmethod def check(self, artifact: dict) - GateResult: artifact 是上游 Agent 的产出结构由各阶段约定 pass这个设计的关键在于suggestions字段。门禁不只是说“你错了”还要说“你可能哪里错了、往哪个方向修”。这些建议会作为下一轮 prompt 的一部分喂回给 Agent显著提高修复效率。我对比过带建议的修复轮次平均比不带建议少 0.8 次。3.2 代码门禁语法、静态分析、依赖三重检查代码阶段的门禁是最复杂的因为它要检查的东西最多。我把它拆成三个子检查任何一个不通过都算门禁失败。语法检查直接用 Python 内置的ast模块解析每个代码文件解析失败就说明有语法错误。这一步很快而且零依赖。import ast def check_syntax(files: dict) - GateResult: reasons [] for path, content in files.items(): if not path.endswith(.py): continue try: ast.parse(content) except SyntaxError as e: reasons.append(f{path} 第 {e.lineno} 行语法错误: {e.msg}) return GateResult(passedlen(reasons) 0, reasonsreasons)静态分析我用pyflakes做未使用变量、未定义名称的检查。这里有个坑pyflakes对动态生成的代码误报率较高所以我在配置里把一些常见的误报模式加了白名单。如果你用的是其他语言对应的工具换成eslint、golangci-lint之类即可思路是一样的。依赖检查是最容易被忽视但最重要的一环。Agent 经常写出import requests但忘了在requirements.txt里加requests的代码。我的做法是扫描所有代码文件的 import 语句提取出第三方包名然后和依赖清单做差集。差集不为空就说明有缺失依赖。import re def extract_imports(files: dict) - set: imports set() pattern re.compile(r^\s*(?:import|from)\s([a-zA-Z0-9_])) for content in files.values(): for line in content.splitlines(): m pattern.match(line) if m: imports.add(m.group(1)) return imports def check_dependencies(files: dict, requirements: str) - GateResult: stdlib set(sys.stdlib_module_names) declared set() for line in requirements.splitlines(): line line.strip() if line and not line.startswith(#): declared.add(re.split(r[!], line)[0].strip().lower()) used extract_imports(files) missing used - stdlib - declared # 过滤掉项目内部模块 missing {m for m in missing if not is_local_module(m, files)} return GateResult( passedlen(missing) 0, reasons[f缺少依赖声明: {m} for m in missing], suggestions[f在 requirements.txt 中添加 {m} for m in missing] )注意is_local_module这个函数要判断一个 import 名是不是项目内部的模块名否则会把本地模块误判为缺失依赖。我的实现是检查是否存在同名.py文件或同名目录下有__init__.py。3.3 测试门禁覆盖率与通过率的双重底线测试阶段的门禁有两个硬指标测试必须全部通过以及核心模块的测试覆盖率不低于设定阈值。第一个指标好理解第二个指标是为了防止 Agent 写一堆无关痛痒的测试来凑数。覆盖率我用coverage.py来测。具体做法是在门禁脚本里调用coverage run -m pytest然后coverage report --formatjson解析 JSON 拿到每个文件的覆盖率数据。阈值我设的是核心模块 70%非核心模块 50%。这个数字不是拍脑袋定的是我跑了十几个项目后总结的经验值——低于这个数通常意味着有重要的分支逻辑没被测试到。import json import subprocess def check_test_coverage(project_dir: str, core_modules: list, threshold: float 0.7) - GateResult: subprocess.run( [coverage, run, -m, pytest, --tbshort], cwdproject_dir, capture_outputTrue ) result subprocess.run( [coverage, report, --formatjson], cwdproject_dir, capture_outputTrue, textTrue ) data json.loads(result.stdout) reasons [] for module in core_modules: file_data data[files].get(module) if not file_data: reasons.append(f核心模块 {module} 没有覆盖率数据可能未被测试) continue cov file_data[summary][percent_covered] / 100 if cov threshold: reasons.append(f核心模块 {module} 覆盖率 {cov:.1%}低于阈值 {threshold:.0%}) return GateResult(passedlen(reasons) 0, reasonsreasons)这里有个实操细节pytest的退出码非零表示有测试失败但coverage run本身可能返回零。所以要先单独跑一次pytest拿退出码再跑coverage拿覆盖率数据。我一开始把两步混在一起结果测试失败了门禁却通过了排查了半天才发现是退出码的问题。3.4 集成门禁端到端跑通才算数集成阶段的门禁最直接把整个项目跑起来看能不能正常执行并产出预期结果。具体做法是执行项目定义的入口命令比如python main.py --input test_input.json检查退出码是否为零、输出文件是否存在、输出内容是否符合预期格式。这一步的难点在于“预期结果”怎么定义。我的做法是在需求解析阶段就让 Planner 产出一个acceptance_criteria.json里面用结构化格式描述验收标准。比如{ entry_command: python main.py --input {input} --output {output}, test_input: examples/sample_input.json, expected_output_schema: { type: object, required: [result, status], properties: { result: {type: array}, status: {type: string, enum: [success, partial]} } } }集成门禁读取这个文件按里面的命令执行然后用 JSON Schema 校验输出。这样验收标准就是可执行、可复现的而不是一句模糊的“功能正常”。4. 门禁失败后的反馈闭环怎么搭4.1 结构化反馈让 Agent 知道“错在哪、怎么修”门禁失败后最关键的是把失败信息有效地传回给 Agent。我试过直接把原始报错日志丢回去效果很差——Agent 面对一大坨 traceback 经常抓不住重点改了半天改错地方。后来我改成结构化反馈把失败信息整理成固定格式[门禁失败] 代码门禁 - 依赖检查 失败项 - 缺少依赖声明: requests - 缺少依赖声明: pandas 修复建议 - 在 requirements.txt 中添加 requests - 在 requirements.txt 中添加 pandas 相关文件 - src/data_loader.py (第 3 行 import requests) - src/analyzer.py (第 5 行 import pandas)这种格式 Agent 一看就懂修复准确率明显提升。我统计过结构化反馈下的首次修复成功率大约是 78%而原始日志反馈只有 40% 出头。4.2 重试策略有限重试 升级机制门禁失败后的重试不能是无脑循环。我的策略是指数退避 上下文累积。第一次失败把失败信息给 Agent 让它修第二次失败除了新的失败信息还把上一次的修复尝试也带上让 Agent 知道“你上次这么改没管用”第三次失败转BLOCKED生成一份详细的问题报告包含所有失败记录和修复尝试等人来看。这里有个细节每次重试都要重新跑完整的门禁而不是只跑失败的那一项。因为 Agent 的修复可能引入新的问题比如为了修依赖问题改了 import 结构结果把语法搞坏了。只跑失败项会漏掉这种回归。4.3 门禁日志事后复盘的金矿每次门禁的输入、输出、耗时、结果都落盘到gate_logs/目录下按任务 ID 和时间戳组织。这些日志在事后复盘时价值极高。我通过分析日志发现过几个规律比如代码门禁的依赖检查失败率最高占到所有门禁失败的 35%测试门禁的覆盖率不达标通常发生在涉及复杂条件分支的任务上集成门禁的失败大多和路径处理有关相对路径 vs 绝对路径。基于这些发现我在 Planner 的 prompt 里加了针对性的提醒比如“生成代码时注意声明所有第三方依赖”“涉及文件路径时统一使用 pathlib 处理”这些前置提醒让对应的门禁失败率下降了不少。5. 实际跑下来的效果与踩坑记录5.1 效果对比有门禁 vs 无门禁我在同一个项目上做了对比测试任务是用多 Agent 生成一个数据处理脚本。无门禁的情况下Agent 报告“完成”后我人工检查发现 10 个任务里有 6 个存在各种问题依赖缺失、测试没跑、输出格式不对。有门禁的情况下这 6 个问题在流水线内部就被拦截并修复了最终交付的 10 个任务全部通过人工验收。代价是耗时增加了约 40%主要是门禁检查和修复重试的时间。但考虑到省下的人工排查和返工时间这个投入完全值得。而且随着 Agent 对门禁反馈的“学习”通过 prompt 里的历史失败案例重试次数在逐渐减少。5.2 踩过的坑门禁太严和太松都是问题坑一门禁太严导致 Agent 陷入死循环。我一开始把代码风格检查也加进门禁要求行长度不超过 79 字符、函数不超过 20 行。结果 Agent 为了满足这些格式要求反复调整有几次改着改着把逻辑改错了。后来我把风格检查从门禁里移除只保留正确性相关的检查。风格问题交给格式化工具自动处理不占用 Agent 的修复轮次。坑二门禁太松导致漏网之鱼。早期版本的测试门禁只检查“测试是否通过”不检查覆盖率。结果 Agent 写了一个只测assert True的测试文件门禁轻松通过但实际代码根本没被验证。加上覆盖率检查后这个问题就消失了。坑三门禁脚本本身的 bug。前面提过我的依赖检查脚本一度把标准库里的json误判为缺失依赖因为sys.stdlib_module_names在某些 Python 版本里不包含它。这个 bug 导致 Agent 反复往requirements.txt里加标准库越改越乱。后来我加了一个硬编码的标准库白名单作为兜底。5.3 常见问题速查表问题现象可能原因排查方向解决方法门禁反复失败同一项Agent 修复方向错误检查反馈信息是否清晰优化 suggestions 的表述给出具体示例门禁通过但产物仍不可用检查项覆盖不全复盘漏检的问题类型补充对应的门禁检查项门禁耗时过长检查项太重分析各检查项耗时占比把耗时长的检查移到并行执行或降低频率Agent 修复引入新问题缺少回归检查对比修复前后的门禁结果每次重试跑完整门禁而非单项任务卡在 BLOCKED 无人处理缺少通知机制检查是否有告警配置加一个简单的邮件或 webhook 通知6. 把这套流水线用起来的几个建议如果你打算在自己的项目里落地这套东西我的建议是从最小可用版本开始别一上来就追求大而全。先实现代码门禁的语法检查和依赖检查这两个最基础、最容易出效果的项跑通整个“失败-反馈-修复-重试”的闭环然后再逐步往前后阶段扩展。我见过有人一上来就想把五个阶段的门禁全做了结果每个都做得半吊子反而拖慢了整体进度。另外门禁的检查项要跟着你的实际痛点走。不同项目容易出的问题不一样Web 项目可能更关注接口契约数据处理项目可能更关注输入输出格式算法项目可能更关注数值精度。我的这套检查项是基于我自己的项目经验总结的你直接拿去用可能需要调整。最好的做法是先用一段时间收集门禁日志看看你的项目里哪类问题最高频然后针对性地加强对应的检查。最后说一个我个人的体会门禁的价值不只是拦截错误更在于它把“完成”的定义从主观变成了客观。以前 Agent 说“做完了”你心里没底得自己去翻代码、跑测试。现在门禁说“通过了”你就知道至少语法没问题、依赖没缺失、测试跑通了、端到端能跑起来。这种确定性才是多 Agent 编程真正能进入生产环境的前提。整套代码我放在开源仓库里结构不算复杂核心逻辑加起来大概一千多行你可以直接 clone 下来改。