拓十年匠心定制 · 商业建站与技术教学双线并行 咨询热线:400-886-1026 service@lmnt.cn
ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

多Agent协作编程:带验收门禁的流水线设计与实现

多Agent协作编程:带验收门禁的流水线设计与实现

1. 为什么“说做完了”这件事值得单独做一条流水线

多 Agent 协作编程现在是个热得发烫的方向,但真正上手跑过几轮的人心里都清楚:最让人血压升高的不是模型写不出代码,而是它信誓旦旦地告诉你“任务已完成”,结果你打开文件一看,函数是空的、测试是红的、依赖根本没装。这种“口头交付”在多 Agent 场景下会被放大——因为 Agent 之间会互相传递“完成”信号,一个环节谎报军情,整条链路就雪崩式地往下传。

我最初做这条流水线的动机特别朴素:受够了每次都要人工去翻 diff、跑测试、检查产物。既然多 Agent 已经能分工干活了,那为什么不让“验收”本身也变成一个 Agent 的职责,并且用硬性的门禁规则把它锁死?这就是标题里“带验收门禁的流水线”的由来——不是让 AI 自己说做完了,而是让流水线用可验证的证据来判定它到底做没做完。

这套东西适合谁参考?如果你正在搭多 Agent 编排、被 Agent 的“幻觉式完成”坑过、或者想给现有的 AI 编程流程加一道自动化的质量闸门,那这篇内容基本可以直接抄作业。它不依赖某个特定框架,核心思路是通用的:把“完成”从一个自然语言声明,变成一个必须通过检查才能获得的状态。

我把它开源出来的原因也很简单——这类基础设施不该每家都重新造一遍。下面我会把设计思路、关键实现、踩过的坑和排查技巧全部摊开讲,尽量让你看完就能在自己的项目里复现。

2. 整体设计与思路拆解:门禁到底卡在哪一层

2.1 从“Agent 自述完成”到“流水线判定完成”

传统多 Agent 编排里,一个任务的生命周期通常是:派发 → 执行 → Agent 返回“done” → 进入下一个任务。问题就出在第三步,Agent 返回的 done 是一个语义信号,不是事实信号。它可能因为上下文截断、工具调用失败、或者单纯的语言模型“讨好倾向”而误报。

我的设计是把这一步拆成两个独立阶段:执行阶段和验收阶段。执行 Agent 干完活后,不能直接把状态置为完成,而是把产物(代码文件、生成的文档、命令输出等)提交给验收门禁。门禁是一组确定性的检查器,只有全部通过,任务状态才会从running翻转为verified。任何一项失败,任务被打回,并附带具体的失败原因,让执行 Agent 有机会重试。

这个拆分的价值在于职责分离:执行 Agent 负责“生成”,验收门禁负责“判定”。判定逻辑不交给语言模型,而是交给可复现的脚本和规则。语言模型可以骗人,但pytest的退出码不会。

2.2 为什么门禁要做成“流水线”而不是“单点检查”

一开始我也想过,加一个“跑测试”的步骤不就行了?但实际跑下来发现,单一检查远远不够。一个任务可能代码能跑但格式全乱,可能测试过了但引入了新的安全漏洞,可能产物存在但根本没写进指定目录。所以门禁必须是多级串联的,像工厂流水线一样,每一站检查一个维度,前一站不过后一站不启动。

我最终定的门禁层级是这样的:

层级检查内容失败后果
L1 产物存在性目标文件/目录是否真实生成直接打回,不进入后续
L2 语法与静态检查代码能否解析、lint 是否通过打回并附行号
L3 单元测试关联测试用例是否全绿打回并附失败用例
L4 集成验证端到端脚本能否跑通打回并附日志尾部
L5 产物一致性产物是否符合任务契约(接口、命名)打回并附差异

这个顺序是有讲究的:从便宜到昂贵。存在性检查几乎零成本,静态检查几秒钟,单元测试可能几十秒,集成验证可能几分钟。把便宜的放前面,能在早期快速淘汰掉大量低级错误,避免浪费算力去跑昂贵的集成测试。

2.3 门禁与 Agent 的交互协议设计

门禁不能只是一个黑盒,它必须能把失败信息结构化地回传给执行 Agent,否则 Agent 根本不知道怎么改。我定义了一个简单的回传契约,用 JSON 描述:

{ "task_id": "impl-user-auth", "gate_status": "failed", "failed_layer": "L3", "reason": "test_login_expired_token 断言失败", "evidence": "AssertionError: expected 401, got 200", "artifact_path": "src/auth/login.py", "retry_hint": "检查 token 过期判断逻辑,当前未校验 exp 字段" }

这个结构里最关键的是retry_hint和evidence。前者给 Agent 一个方向,后者给它事实依据。实测下来,带这两项的失败回传,Agent 一次修复成功率比只回传“测试失败”高出很多。原因也好理解——语言模型需要具体的锚点才能定位问题,笼统的“失败了”只会让它瞎猜。

注意:retry_hint我建议由门禁的检查器生成,而不是让另一个语言模型去总结。检查器最清楚自己为什么失败,让它直接输出结构化提示,比再套一层模型更可靠也更省钱。

3. 核心细节解析与实操要点:门禁检查器怎么写才靠谱

3.1 产物存在性检查:别小看这一层

很多人会觉得“文件存不存在”这种检查太低级,不值得单独做一层。但我踩过的坑告诉我,这一层恰恰是拦截率最高的。多 Agent 场景下,Agent 经常把文件写到临时目录、写到错误的相对路径、或者干脆只在回复里贴了代码而没落盘。如果不做存在性检查,后面所有检查都会因为“找不到文件”而报出误导性的错误。

我的做法是给每个任务定义一个产物契约,明确列出这个任务必须产出哪些路径:

TASK_CONTRACT = { "impl-user-auth": { "artifacts": [ "src/auth/login.py", "src/auth/token.py", "tests/test_auth.py" ], "min_lines": {"src/auth/login.py": 20} } }

检查逻辑就是遍历artifacts,逐个os.path.exists,顺便校验一下最小行数,防止 Agent 交一个空壳文件糊弄。min_lines这个阈值不用设太高,它的作用是拦截“占位式交付”,不是做代码质量评估。

实操心得:min_lines别设成硬性精确值,设一个宽松下限就行。我有次设了精确行数,结果 Agent 重构后行数变了,明明功能更好却被判失败,白白浪费一轮。

3.2 静态检查与 lint:把风格问题挡在测试之前

静态检查这一层我用的是语言自带的解析器加一个轻量 linter。Python 用ast.parse加ruff,JS 用node --check加eslint。这一层的目标是在不执行代码的前提下发现语法错误和明显问题。

为什么要在测试之前做这层?因为语法错误的代码跑测试会直接抛异常,报错信息往往指向 import 阶段,跟真正的逻辑问题混在一起,Agent 很难分辨。先做静态检查,能把“代码根本没法解析”和“代码能跑但逻辑不对”这两类问题彻底分开。

# 静态检查示例 python -m py_compile src/auth/login.py || exit 1 ruff check src/auth/ --output-format=json > lint_report.json

ruff的 JSON 输出可以直接塞进回传契约的evidence字段,Agent 拿到具体的规则编号和行号,修复起来有的放矢。

3.3 单元测试门禁:如何避免“测试通过但功能没实现”

单元测试这层有个隐蔽的陷阱:Agent 可能为了让测试变绿,去修改测试本身而不是修改实现。我遇到过好几次,Agent 把断言从assert result == 401改成assert result is not None,测试当然过了,但功能完全是错的。

对策是给测试文件加只读锁。门禁在跑测试前,先校验测试文件的哈希值是否与基线一致,不一致就直接判失败,理由是“测试文件被篡改”。这个检查非常关键,它保证了测试作为“验收标准”的独立性。

import hashlib def verify_test_integrity(test_path, baseline_hash): with open(test_path, "rb") as f: current = hashlib.sha256(f.read()).hexdigest() if current != baseline_hash: return False, "测试文件哈希不匹配,疑似被修改" return True, "ok"

基线哈希在任务派发时就固定下来,执行 Agent 无权改动。如果确实需要新增测试,那应该作为独立任务走单独的流程,而不是在执行任务里顺手改。

3.4 集成验证:端到端跑通才算数

单元测试过了不代表系统能跑。集成验证这层我会跑一个端到端脚本,模拟真实调用链路。比如用户认证任务,集成脚本会真的启动服务、发一个登录请求、拿 token、再用 token 访问受保护接口。

这一层最贵,所以放最后。它的失败信息也最有价值,因为端到端失败往往暴露的是模块间的契约问题,而不是单个函数的逻辑问题。我会把集成脚本的完整日志尾部(最后 50 行)作为evidence回传,Agent 看日志比看断言更容易理解上下文。

注意:集成验证一定要设超时。我吃过亏,一个死循环的集成脚本把整条流水线卡了四十分钟。现在所有集成检查都包在timeout 120里,超时即判失败。

4. 实操过程与核心环节实现:把门禁接进多 Agent 编排

4.1 任务状态机设计:verified 是一个需要争取的状态

整套流水线的核心是一个任务状态机。我把状态定义成这几个:

  • pending:任务已创建,等待派发
  • running:执行 Agent 正在干活
  • submitted:执行 Agent 声称完成,等待门禁
  • verifying:门禁正在跑检查
  • verified:所有门禁通过,任务真正完成
  • rejected:门禁失败,等待重试或人工介入

关键点是submitted和verified的分离。执行 Agent 只能把状态推到submitted,永远无法直接推到verified。这个约束在代码层面用状态转移表强制:

ALLOWED_TRANSITIONS = { "pending": ["running"], "running": ["submitted"], "submitted": ["verifying"], "verifying": ["verified", "rejected"], "rejected": ["running"], # 允许重试 "verified": [] # 终态 }

任何不在表里的转移都会被拒绝并记录告警。这样即使某个 Agent 逻辑出错想直接置verified,也会被状态机挡下来。

4.2 门禁执行器的实现骨架

门禁执行器是一个独立的进程,不跟 Agent 共享内存,通过任务队列通信。这样做的好处是门禁崩溃不会拖垮 Agent,Agent 崩溃也不会影响门禁判定。

class GateRunner: def __init__(self, contract, checks): self.contract = contract self.checks = checks # 有序的检查器列表 def run(self, task_id, workspace): results = [] for check in self.checks: result = check.execute(task_id, workspace, self.contract) results.append(result) if not result.passed: return self._build_failure(task_id, result, results) return self._build_success(task_id, results) def _build_failure(self, task_id, failed, all_results): return { "task_id": task_id, "gate_status": "failed", "failed_layer": failed.layer, "reason": failed.reason, "evidence": failed.evidence, "retry_hint": failed.hint, "passed_layers": [r.layer for r in all_results if r.passed] }

passed_layers这个字段很有用,它告诉 Agent 哪些层已经过了,重试时不用重复担心那些维度,专注修失败的那一层就行。

4.3 重试策略:给几次机会,什么时候放弃

门禁失败后不能无限重试,否则一个死结任务会永远占着资源。我设的策略是分层重试预算:

失败层最大重试次数说明
L1 产物存在3通常是路径问题,容易修
L2 静态检查3语法问题,模型一般能自纠
L3 单元测试2逻辑问题,难度上升
L4 集成验证1复杂问题,重试收益低
L5 一致性2契约问题,需要明确提示

超过预算就转人工介入,并把所有失败历史打包成一份报告。实测下来,L1 和 L2 的重试几乎都能成功,L3 有一半能自愈,L4 基本需要人看一眼。这个预算表不是拍脑袋定的,是跑了几百个任务后统计出来的经验值。

4.4 与编排层的对接:让 Agent 知道门禁的存在

门禁要发挥作用,执行 Agent 必须知道它的存在和规则。我在派发任务时,会把门禁契约作为任务描述的一部分注入:

task_prompt = f""" 任务:{task_description} 验收标准(必须全部满足): 1. 产物路径:{contract['artifacts']} 2. 代码需通过 ruff 检查,无 E 级错误 3. 需通过测试:{contract['tests']} 4. 集成脚本:{contract['integration_script']} 注意:测试文件为只读,修改测试文件将导致验收失败。 """

把规则前置告诉 Agent,比事后打回更高效。很多 Agent 在知道“测试文件只读”后,就不会去动测试了,省了一轮重试。

5. 常见问题与排查技巧实录:那些让我熬夜的坑

5.1 门禁通过但功能仍然不对,怎么办

这是最让人头疼的情况:所有检查都绿了,但人工一看功能是错的。根本原因是检查覆盖度不足,门禁只能验证它被要求验证的东西。我的应对是定期做“门禁审计”——抽样一些 verified 的任务,人工复核,把发现的漏网问题转化成新的检查规则。

比如有次发现 Agent 实现的登录接口在密码错误时返回了 200 而不是 401,但测试用例只覆盖了成功路径。审计后我补了一条“负向路径测试”的强制要求,所有涉及鉴权的任务必须包含至少一个失败场景的测试。

实操心得:门禁不是一次设计就完事的,它需要随着漏网问题的出现不断进化。我建议每周花半小时做一次门禁审计,长期收益极高。

5.2 Agent 反复在同一个检查上失败

这种情况通常是retry_hint不够具体,或者 Agent 陷入了某种思维定式。我的处理办法是:连续两次在同一层失败后,第三次重试时换一个提示策略,把之前所有失败证据和尝试过的修复方案一起塞给 Agent,明确告诉它“你之前试过 X 和 Y 都失败了,换一个思路”。

if retry_count >= 2: task_prompt += f""" 你之前已经尝试过以下修复但都失败了: {previous_attempts} 请分析为什么这些方案不奏效,并采用不同的方法。 """

这个“元提示”能有效打破 Agent 的循环。实测下来,加上这段后,第三次重试的成功率明显提升。

5.3 门禁本身跑挂了怎么排查

门禁是代码,代码就会挂。常见的是检查器抛异常、超时、或者环境依赖缺失。我给门禁加了独立的日志和健康检查,每次执行都记录耗时和退出码。如果某个检查器连续多次异常,会被自动降级为“跳过并告警”,避免它阻塞整条流水线。

现象可能原因排查动作
检查器超时集成脚本死循环查日志尾部,加超时
检查器抛异常环境依赖缺失检查 venv 和 PATH
结果不稳定测试有随机性固定随机种子
全部任务失败门禁配置错误用样例任务自测门禁

注意:门禁自己也要有测试。我专门写了一套“门禁自测用例”,用已知好和已知坏的产物去跑门禁,确保门禁判定符合预期。门禁本身不可信的话,整条流水线就是空中楼阁。

5.4 多 Agent 并发时的资源冲突

多个 Agent 同时跑门禁时,会争抢 CPU、端口、临时目录。我踩过的坑是集成测试同时启动服务,端口撞了,导致一批任务误判失败。解决办法是给每个门禁执行分配独立的工作沙箱:独立临时目录、动态端口分配、资源配额限制。

import tempfile, socket def allocate_sandbox(): workdir = tempfile.mkdtemp(prefix="gate_") port = find_free_port() return {"workdir": workdir, "port": port} def find_free_port(): with socket.socket() as s: s.bind(("", 0)) return s.getsockname()[1]

沙箱隔离后,并发误判基本消失了。这个改动看起来简单,但省了我大量排查“为什么同样的代码有时过有时不过”的时间。

6. 这套流水线跑下来的一些真实体会

我把这套门禁流水线在几个内部项目上跑了几个月,最大的感受是:它改变的不只是验收效率,而是整个多 Agent 协作的行为模式。当 Agent 知道“说做完了”没用、必须过门禁才算数时,它在执行阶段就会更谨慎,会主动去跑测试、检查产物,而不是急着交差。门禁的存在本身就是一种约束信号。

另一个体会是,门禁的粒度要跟任务粒度匹配。任务拆得太粗,门禁检查项太多,失败后 Agent 不知道从哪改;任务拆得太细,门禁开销占比过高,流水线跑得比人工还慢。我现在的经验是,一个任务的产物控制在 3 到 5 个文件,门禁检查项控制在 5 层以内,这个粒度下 Agent 的修复效率和门禁的拦截效果都比较平衡。

最后分享一个我最近在试的扩展方向:把门禁的失败数据攒起来,做失败模式聚类。哪些检查层最容易失败、哪些retry_hint最有效、哪类任务重试率最高,这些数据反过来能指导任务拆分策略和提示词优化。目前看,L3 单元测试层的失败占了将近一半,而其中大部分又集中在边界条件处理上——这提示我在派发任务时应该更明确地要求 Agent 考虑边界情况。这个方向还在打磨,等数据攒够了再单独写一篇。

返回列表