ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

多Agent编程流水线GatePipeline:用验收门禁杜绝假完成

多Agent编程流水线GatePipeline:用验收门禁杜绝假完成

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_factory=list) suggestions: List[str] = field(default_factory=list) 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(passed=len(reasons) == 0, reasons=reasons)

静态分析我用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( passed=len(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 --format=json,解析 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", "--tb=short"], cwd=project_dir, capture_output=True ) result = subprocess.run( ["coverage", "report", "--format=json"], cwd=project_dir, capture_output=True, text=True ) 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(passed=len(reasons) == 0, reasons=reasons)

这里有个实操细节: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 下来改。

RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进