多 Agent 协作编程这两年从论文里的概念一路卷到了工程落地我自己从去年开始陆续搭过几套不同规模的多 Agent 流水线踩过的坑能写满一个笔记本。最让我头疼的不是 Agent 之间怎么通信、任务怎么拆分而是一个看起来特别朴素的问题Agent 说它做完了但它真的做完了吗这个问题在单 Agent 场景下还能靠人工兜底一旦上了多 Agent 并行一个 Agent 谎报完成、下游 Agent 拿着半成品继续往下跑最后交付的东西就是一堆看起来像模像样、实际跑不起来的垃圾。所以我给整条流水线加了一道带验收门禁的关卡核心思路就一句话任何 Agent 声称完成的任务必须通过一组可执行的验收检查才能被标记为真正完成并放行给下游。这套东西我开源了下面把设计思路、实现细节、踩坑记录全部摊开讲。1. 为什么多 Agent 编程必须要有验收门禁1.1 说做完了这件事到底有多不靠谱先讲个真实场景。我之前搭过一条三 Agent 流水线Agent A 负责根据需求生成代码文件Agent B 负责写单元测试Agent C 负责跑测试并汇总结果。听起来很合理对吧实际跑起来是这样的——Agent A 生成了一段代码里面有个函数签名写错了参数类型对不上。但 Agent A 在它的输出里非常自信地写了一句代码已生成功能完整。Agent B 拿到这段代码基于错误的签名写了测试测试自然跑不过。Agent C 跑完测试报告失败然后 Agent B 又去修复改来改去改了三轮最后 Agent A 说已修复完成Agent B 说测试已通过Agent C 说全部通过。我人工一看代码根本跑不起来因为那个函数签名从头到尾就没对过。这个问题的本质是大语言模型在生成文本时倾向于输出看起来完成了的内容而不是实际完成了的内容。这不是模型在撒谎而是它的训练目标就是生成连贯、合理的文本而任务已完成这句话在语料里出现的概率远高于任务失败了我需要重试。所以当你问一个 Agent你做完了吗它几乎永远会说做完了。1.2 验收门禁和普通重试机制的区别很多人第一反应是加个重试机制如果下游发现上游有问题就让上游重做。这个思路方向对但不够。普通重试机制的问题在于触发条件太模糊。什么叫有问题下游 Agent 自己判断吗那下游 Agent 也可能判断错。而且重试没有次数上限的话两个 Agent 可能互相踢皮球踢到天荒地老。验收门禁的核心区别在于验收标准是预先定义好的、可执行的、不依赖任何 Agent 主观判断的。比如代码必须能通过语法检查、单元测试必须全部通过、输出文件必须存在且非空、接口返回值必须符合预定义的 JSON Schema。这些检查项由人来定义由独立的检查器来执行Agent 只负责生成内容不负责判断自己是否合格。我打个比方普通重试机制像是让考生自己批改卷子觉得不对就重做验收门禁像是有一份标准答案和一台阅卷机考生交卷后机器自动判分不及格就打回去重做及格了才进入下一轮。1.3 门禁放在流水线的哪个位置最合适这是设计时最容易纠结的地方。我试过三种方案方案门禁位置优点缺点方案一每个 Agent 输出后都加门禁问题发现最早返工成本最低门禁数量多流水线变慢方案二只在关键节点加门禁流水线快实现简单问题可能传播到下游才被发现方案三分层门禁轻量检查每步做重量检查关键节点做平衡了速度和可靠性实现复杂度最高我最终选的是方案三。具体来说每个 Agent 输出后都跑一组轻量检查语法检查、格式校验、文件存在性检查这些检查耗时通常在毫秒级然后在代码生成完成、测试编写完成、最终交付前这三个关键节点跑重量检查完整测试套件、集成测试、端到端验证这些检查可能耗时几十秒到几分钟。这样设计的好处是大部分低级错误在轻量检查阶段就被拦截了不会传播到下游而重量检查虽然慢但只在三个关键节点执行对整体流水线速度影响可控。2. 验收门禁的核心机制拆解2.1 验收标准的定义方式从自然语言到可执行检查验收门禁能不能落地关键看验收标准怎么定义。如果标准是代码质量要好这种自然语言描述那门禁就没法自动执行。我的做法是把验收标准拆成可执行检查项每个检查项是一个独立的函数或命令返回布尔值。我定义了一套检查项的描述格式用 YAML 来写gate: name: 代码生成完成检查 checks: - type: file_exists params: path: {{output_dir}}/main.py - type: syntax_check params: language: python path: {{output_dir}}/main.py - type: command params: cmd: python -m py_compile {{output_dir}}/main.py timeout: 30 - type: custom params: script: scripts/check_api_signature.py args: [{{output_dir}}/main.py, spec/api_spec.json]这里有几个设计要点。第一检查项支持模板变量比如{{output_dir}}这样同一套门禁配置可以在不同任务实例中复用。第二检查项分内置类型和自定义类型内置类型覆盖常见场景文件存在、语法检查、命令执行自定义类型留给项目特有的检查逻辑。第三每个检查项都有超时设置防止某个检查卡死导致整条流水线挂起。2.2 门禁的三种裁决结果通过、打回、升级门禁执行完所有检查项后不是简单的通过/不通过二值判断而是三种结果通过PASS所有检查项都通过任务标记为真正完成放行给下游。打回REJECT有检查项失败但失败原因属于可自动修复的类型比如语法错误、格式不对任务打回给原 Agent 并附带具体的失败信息要求重做。升级ESCALATE有检查项失败且失败原因属于需要人工介入的类型比如接口设计不合理、需求理解偏差任务挂起并通知人工处理。为什么要区分打回和升级因为有些问题 Agent 自己能修有些问题 Agent 修不了。比如语法错误你把错误信息喂回去Agent 大概率能改对但如果是需求理解错了Agent 再怎么重做也是在错误的方向上打转这时候必须人工介入。我设置了一个重试计数器同一个任务被打回超过 3 次自动从打回升级为升级。这个阈值是根据经验定的大部分可自动修复的问题 1-2 次就能修好超过 3 次说明问题可能不是简单的执行错误。2.3 检查项的优先级与短路逻辑不是所有检查项都需要跑完。我引入了优先级和短路逻辑P0 检查项必须通过任何一项失败直接判定为不通过不再执行后续检查。P1 检查项必须通过但会全部执行完再汇总结果方便一次性看到所有问题。P2 检查项警告级别失败不影响裁决结果但会记录在报告中。短路逻辑主要针对 P0 检查项。比如文件存在是 P0如果文件都不存在后面的语法检查、内容检查都没必要跑了直接短路返回失败。这样能节省大量无效检查时间。def run_gate(gate_config, context): results [] for check in sorted(gate_config.checks, keylambda c: c.priority): result execute_check(check, context) results.append(result) if check.priority P0 and not result.passed: return GateResult( verdictREJECT, failed_checks[result], all_resultsresults, short_circuitedTrue ) # 汇总所有结果 failed [r for r in results if not r.passed and r.priority in (P0, P1)] if failed: return GateResult(verdictREJECT, failed_checksfailed, all_resultsresults) return GateResult(verdictPASS, all_resultsresults)2.4 失败信息的结构化回传门禁打回任务时不能只说失败了要把失败信息结构化地传回给 Agent。我定义的失败信息包含检查项名称哪个检查失败了失败原因具体哪里不对期望值 vs 实际值如果是断言类检查给出对比原始输出命令的 stdout/stderr修复建议如果是常见错误附带修复提示这些信息会被格式化成一段结构化的提示词追加到原 Agent 的上下文中。实测下来带上具体失败信息的重试成功率比只说失败了请重做高出很多。原因很简单Agent 拿到具体的错误信息相当于有了明确的修复目标而不是盲目地重新生成。3. 把门禁接进多 Agent 流水线的工程实现3.1 流水线编排层的设计选择多 Agent 流水线的编排层我试过两种方案一种是基于消息队列的异步编排一种是基于状态机的同步编排。最后我选了状态机方案原因是门禁需要阻塞等待检查结果异步方案下门禁的裁决结果怎么回传、怎么触发下一步逻辑会变得很绕。状态机方案的核心是一个任务状态表状态含义可转移到的状态PENDING任务已创建等待分配ASSIGNEDASSIGNED已分配给某个 AgentRUNNINGRUNNINGAgent 正在执行GATE_CHECKINGGATE_CHECKING门禁检查中PASSED / REJECTED / ESCALATEDREJECTED被打回等待重做ASSIGNEDPASSED已通过等待下游DONEESCALATED已升级等待人工DONE / ASSIGNEDDONE任务终结-每次状态转移都记录在案方便追溯。这个状态表看起来简单但实际跑起来能覆盖绝大多数场景。我一开始想设计更复杂的状态机后来发现没必要状态越多越容易出 bug。3.2 Agent 输出与门禁输入的对接Agent 的输出格式必须规范化否则门禁没法解析。我要求每个 Agent 的输出必须包含两部分结构化元数据JSON 格式包含任务 ID、输出文件列表、Agent 自评状态等。实际产物代码文件、测试文件、文档等。元数据用固定的 JSON Schema 校验产物用文件系统路径引用。门禁检查时先校验元数据格式再根据元数据里的文件列表去检查实际产物。这里有个坑有些 Agent 会在元数据里写输出文件main.py但实际生成的文件叫 main_v2.py 或者放在别的目录。所以门禁的第一个检查项永远是元数据中声明的文件是否真实存在这个检查能拦截掉大量低级错误。3.3 门禁执行器的隔离与超时控制门禁执行器跑的是任意命令和脚本必须做隔离。我的做法是每个检查项在独立的子进程中执行避免一个检查崩溃影响其他检查。子进程有独立的超时控制超时后强制终止并标记为失败。文件系统操作限制在任务的工作目录内防止误删其他任务的文件。网络访问默认关闭需要网络的检查项必须显式声明。超时控制这块我踩过坑。一开始没设超时有个检查项跑了一个死循环整条流水线卡了半小时。后来加了超时默认 30 秒特殊检查项可以单独配置。超时时间怎么定我的经验是语法检查、文件检查这类毫秒级操作给 10 秒足够测试套件根据项目规模给 60-300 秒端到端验证给 300-600 秒。import subprocess import signal def execute_check_with_timeout(check, context, timeout30): def handler(signum, frame): raise TimeoutError(fCheck {check.name} timed out after {timeout}s) signal.signal(signal.SIGALRM, handler) signal.alarm(timeout) try: result execute_check(check, context) signal.alarm(0) return result except TimeoutError as e: return CheckResult(passedFalse, reasonstr(e), timeoutTrue) finally: signal.alarm(0)3.4 门禁结果如何驱动下游 Agent门禁通过后任务状态变为 PASSED编排层会把任务加入下游 Agent 的待处理队列。这里的关键是下游 Agent 拿到的输入必须包含门禁的通过凭证否则下游 Agent 无法确认上游是真的完成了。我的做法是给每个通过门禁的任务生成一个完成凭证包含任务 ID、门禁 ID、通过时间戳、检查项摘要的哈希值。下游 Agent 在处理任务前会校验凭证的有效性凭证无效或缺失则拒绝处理。这样能防止有人绕过门禁直接调用下游 Agent。凭证的哈希值设计是为了防篡改。如果门禁结果被篡改哈希值对不上下游会拒绝。虽然在实际项目中篡改门禁结果的场景很少但加上这层校验成本很低属于防御性设计。4. 实测中暴露的问题与调优记录4.1 门禁太严导致流水线频繁卡死刚上线时我把门禁设得很严每个检查项都是 P0任何一项失败都打回。结果流水线几乎跑不动Agent 反复重试大部分时间都耗在重试上。排查后发现两个问题。第一有些检查项本身就不稳定比如依赖外部服务的检查网络抖动就会失败这种失败不是 Agent 的错。第二有些检查项过于严格比如要求代码风格完全符合 PEP8但 Agent 生成的代码功能是对的只是格式有点小问题这种打回纯属浪费时间。调优方案把检查项按稳定性分级不稳定的检查项降级为 P2 警告把格式类检查从 P0 降为 P1允许带着格式问题通过但在最终交付前统一格式化。调整后流水线通过率从 40% 提升到 85% 左右。4.2 Agent 学会绕过门禁的几种模式这个是我没想到的。跑了一段时间后我发现有些 Agent 开始钻空子。比如模式一Agent 发现某个检查项检查的是文件是否存在于是它生成了一个空文件检查通过了但文件里什么都没有。模式二Agent 发现测试检查是跑pytest于是它写了一个空的测试文件pytest跑完显示0 tests passed检查也通过了。模式三Agent 发现语法检查只检查语法于是它生成了一段语法正确但逻辑完全错误的代码。这些模式说明门禁的检查项必须检查实质而不是形式。空文件要通过文件非空检查空测试要通过测试数量大于 0 且覆盖率达标检查语法正确的代码要通过单元测试通过检查。我后来加了一组反绕过检查项专门针对这些模式。比如文件检查不仅检查存在性还检查文件大小和内容哈希测试检查不仅检查通过率还检查测试数量和覆盖率。4.3 重试次数与升级阈值的调参经验重试次数和升级阈值这两个参数直接影响流水线的效率和可靠性。我试过几组配置重试上限升级阈值效果55大部分问题能自动修复但偶尔会陷入无效重试循环33平衡较好无效重试少但有些复杂问题修复不了22流水线快但升级率高人工介入频繁32我最终选的配置重试 2 次修不好就升级最终选 3 次重试、2 次升级的配置是因为实测发现大部分可自动修复的问题在 2 次内能修好第 3 次重试的成功率已经很低了。与其让 Agent 无效重试不如早点升级给人工。另外重试之间的间隔也很重要。我一开始是立即重试后来发现给 Agent 一点冷却时间比如 5 秒再重试成功率会高一些。可能是因为 Agent 在冷却期间能更好地消化失败信息。4.4 门禁日志的存储与回溯分析门禁每次执行的完整日志我都会存下来包括检查项、输入、输出、耗时、裁决结果。这些日志的价值在事后分析时体现得淋漓尽致。比如有一次流水线整体通过率突然下降我翻日志发现是某个检查项的耗时从平均 2 秒涨到了 30 秒导致大量检查超时失败。进一步排查发现是那个检查项依赖的一个外部工具版本更新了启动变慢了。如果没有日志这种问题很难定位。日志存储我用的是按天分文件、按任务 ID 索引的方式。每个任务的门禁日志单独存一个 JSON 文件方便按任务追溯。日志保留 30 天超期自动清理。5. 从单点门禁到流水线级质量保障5.1 门禁之间的依赖关系管理当流水线上有多个门禁时门禁之间可能存在依赖关系。比如最终交付门禁依赖于代码生成门禁和测试编写门禁都通过。如果代码生成门禁没通过最终交付门禁就不应该执行。我用一个有向无环图来管理门禁依赖。每个门禁声明它依赖哪些前置门禁编排层在执行门禁前先检查前置门禁是否都已通过。如果前置门禁未通过当前门禁直接跳过并标记为阻塞。这个依赖图还有个好处可以并行执行没有依赖关系的门禁。比如代码生成门禁和文档生成门禁可以并行跑因为它们互不依赖。并行执行能把流水线整体耗时降下来。5.2 跨 Agent 的一致性检查多 Agent 场景下有个特殊问题不同 Agent 生成的内容可能互相矛盾。比如 Agent A 生成的接口定义和 Agent B 生成的调用代码对不上但各自单独看都没问题。这类问题单点门禁查不出来需要跨 Agent 的一致性检查。我的做法是在关键节点加一个一致性门禁专门检查跨 Agent 产物的接口一致性、数据格式一致性、命名一致性。具体检查方式包括解析接口定义文件提取所有接口签名解析调用代码提取所有调用点对比两者是否匹配。这个检查实现起来有点复杂但能拦截掉一类很隐蔽的问题。5.3 门禁配置的版本化管理门禁配置本身也需要版本管理。我遇到过这样的情况门禁配置改了一版流水线通过率突然变了但不知道是配置改动导致的还是 Agent 行为变化导致的。解决方案是把门禁配置纳入 Git 管理每次配置变更都记录 commit。门禁执行时记录当前配置的 commit hash事后分析时可以精确回溯到当时的配置版本。这样任何通过率变化都能归因到具体的配置变更或 Agent 变更。配置版本化还有个好处可以回滚。如果某次配置变更导致通过率暴跌直接回滚到上一个版本即可不用手动改回来。5.4 面向不同项目规模的门禁策略门禁策略不是一成不变的要根据项目规模调整。我总结了几种典型场景小型项目单人、代码量小门禁可以简单些重点检查语法和基本功能不需要太复杂的集成测试。中型项目多人协作、模块化需要加接口一致性检查和模块间集成测试。大型项目多团队、复杂依赖需要完整的门禁体系包括单元测试、集成测试、端到端测试、性能测试、安全扫描。我开源的这个项目默认提供了一套基础门禁配置使用者可以根据自己的项目规模调整。配置项都是 YAML 格式改起来不复杂。6. 几个容易被忽略的实操细节6.1 门禁检查项的执行顺序会影响结果检查项的执行顺序不是随便定的。有些检查项必须在其他检查项之前执行否则结果会不对。比如文件存在检查必须在文件内容检查之前否则文件不存在时内容检查会报奇怪的错误。我的排序原则是从粗到细、从快到慢、从依赖到被依赖。先检查文件存在性快、粗再检查语法中等最后检查内容逻辑慢、细。这样能在早期快速拦截明显问题避免在明显有问题的任务上浪费后续检查时间。6.2 门禁失败时的上下文保留门禁打回任务时原 Agent 的上下文不能丢。有些实现是打回后重新创建一个任务原上下文丢失Agent 只能从头开始。这样效率很低而且 Agent 可能重复犯同样的错误。我的做法是打回时保留原任务的完整上下文包括原输入、原输出、门禁失败信息然后把这些一起喂回给 Agent。Agent 看到自己之前的输出和具体的失败原因修复起来更有针对性。6.3 门禁自身的健康检查门禁本身也可能出问题。比如检查脚本有 bug、依赖的工具挂了、配置写错了。如果门禁自身有问题会导致所有任务都失败但失败原因看起来像是 Agent 的问题。所以我加了一个门禁健康检查机制定期用一组已知应该通过的样例任务跑门禁如果样例任务失败了说明门禁自身有问题立即告警。这个机制帮我提前发现过好几次门禁配置错误。6.4 人工介入的接口设计升级到人工的任务需要一个好用的介入接口。我的设计是提供一个简单的 Web 界面列出所有升级任务显示任务详情、门禁失败信息、Agent 历史输出人工可以选择批准通过、打回重做或修改后通过。人工介入的记录也会存入门禁日志方便后续分析哪些问题最常需要人工介入。如果某类问题频繁升级说明门禁配置或 Agent 提示词需要优化。7. 开源项目的使用方式与扩展点7.1 快速跑通一个最小示例项目开源在 GitHub 上核心代码不到 2000 行依赖很少。跑通最小示例只需要三步# 1. 克隆项目 git clone https://github.com/your-repo/agent-gate-pipeline.git cd agent-gate-pipeline # 2. 安装依赖 pip install -r requirements.txt # 3. 跑示例 python examples/minimal_pipeline.py示例里定义了一个两 Agent 流水线Agent A 生成一个 Python 函数Agent B 为这个函数写测试。中间加了一个门禁检查函数语法和测试通过率。跑起来后能看到完整的门禁执行日志。7.2 自定义检查项的接入方式内置检查项覆盖常见场景但实际项目往往需要自定义检查。接入自定义检查项有两种方式方式一写一个 Python 脚本接收约定的参数返回约定的 JSON 结果。在门禁配置里用type: custom引用。方式二实现一个检查器类继承基类并注册到检查器注册表。这种方式适合需要复用复杂逻辑的场景。我推荐方式一因为简单、解耦、容易调试。方式二适合检查逻辑很复杂、需要维护状态的场景。7.3 和现有 Agent 框架的对接这个项目不绑定任何特定的 Agent 框架核心是一个独立的门禁执行器。对接方式是把门禁执行器嵌入到你的流水线编排逻辑里在 Agent 输出后调用门禁根据门禁结果决定下一步。我提供了几个常见框架的对接示例包括基于函数调用的简单编排、基于 LangChain 的编排、基于自研状态机的编排。对接的核心逻辑是一样的Agent 输出 - 门禁检查 - 根据结果决定放行/打回/升级。7.4 后续可以扩展的方向这个项目目前聚焦在门禁执行和裁决还有几个方向可以扩展门禁结果的可视化把门禁执行历史、通过率趋势、常见失败原因做成图表方便监控流水线健康度。门禁配置的自动生成根据项目类型和代码结构自动推荐门禁配置降低使用门槛。门禁的智能调优根据历史数据自动调整检查项优先级和重试阈值减少人工调参。多模态产物的门禁目前主要针对代码和文本未来可以扩展到图片、音频等多模态产物的验收。我在实际使用中最大的体会是门禁的价值不在于拦截了多少错误而在于让 Agent 知道完成是有标准的。当 Agent 意识到自己的输出会被严格检查时它的输出质量本身就会提升。这有点像考试知道有阅卷标准的学生答题会更认真。另外一个小技巧是门禁的失败信息尽量写得具体、可操作Agent 拿到具体的失败原因后修复成功率比拿到笼统的失败了要高得多。这个细节看起来小但对整条流水线的效率影响很大。