多 Agent 协作踩坑记：拆解、共享与回滚
单 Agent 处理长任务容易跑偏，于是把任务拆给多个 Agent 协作——然后掉进了一堆新坑。拆解粒度、上下文共享和失败回滚，三个问题三个教训。
去年我把团队的研报生成流程从"一个超级 Prompt 包打天下"改成了"主管 Agent 派活给三个工人 Agent"。上线第一周，吞吐量确实上去了，但第二周就出了三起事故：两篇报告内容重复了将近三分之一，一篇报告里股票价格和正文结论对不上，还有一次一个工人崩了，另外两个工人还在傻乎乎地继续干活。那次复盘给我的教训很直接：多 Agent 系统的坑和普通微服务完全不是一回事，它的问题不在网络、不在超时，而在"意图的传递"本身。
这篇文章不是来安利多 Agent 架构的。恰恰相反，我想先把踩过的坑摊开讲，再聊什么场景它才真的划算。如果你正在考虑把单体 Agent 拆成多个，或者已经拆了但总觉得哪里不对，下面这些应该能帮你少走几步弯路。
我后来把多 Agent 协作的故障归纳成三类，它们几乎覆盖了我在生产环境里见过的八成问题。先放一张总览，再逐个拆开讲现象、根因和对策。需要提前说明：这些坑几乎没有一个能用"加几行代码"解决，它们都长在编排设计里。
三大坑概览
多 Agent 协作的本质，是"把一个有自主性的目标，交给多个有自主性的执行体去共同完成"。这个定义里有两个"自主性"，正是所有麻烦的源头。主管 Agent 不是函数调用，它给工人的指令是自然语言、是可被重新诠释的；工人 Agent 也不是纯函数，它会基于自己的上下文做取舍。
- 任务拆解幻觉：主管把任务切得太碎或边界模糊，工人要么重复劳动，要么留下真空地带没人管。
- 共享状态冲突：多个 Agent 并发改写同一份中间产物，最后谁的结果算数说不清。
- 联动回滚缺失：一个 Agent 失败时，其他 Agent 既不知道也不撤，导致半成品污染下游。
这三类坑有个共同点：它们都源于"多个有自主性的执行体共享一个目标"，而不是源于某段具体代码写错了。所以纯靠加单测救不了，得从编排设计上动手。我把它们当成架构问题而非代码问题来治，事后看是对的。
坑一：任务拆解幻觉
现象
最典型的现象是"重复劳动"和"真空地带"并存。比如我让主管把"写一份新能源行业研报"拆成子任务，它可能会同时交给两个工人去"搜集竞品信息"和"整理市场格局"：这两个子任务在工人眼里高度重叠，于是两人在不同对话里各自搜了一遍，产出了两份百分之七八十相似的内容。更隐蔽的是反过来的情况：主管觉得"数据清洗"是小事没单列，结果写稿的工人以为数据已经干净，分析工人以为数据会有人洗，最后谁都没做，报告里直接用了原始脏数据。
我印象最深的一次，是主管把一个"综述"任务拆成了"中文综述"和"英文综述"两个工人，又补了一个"整合"工人。结果整合工人拿到的两份综述各有各的引用编号体系，它自己又重新编了一套，读者根本没法对照原文。这种"拆了但没对齐"的代价，往往比不拆更贵。
根因
根因在于，我们用"人类团队协作"的直觉去设计 Agent 拆解，但 Agent 没有共享心智模型。两个人开会说一句"你负责市场、我负责技术"，彼此能脑补出大量隐含边界；Agent 之间只有那条被序列化的指令，任何没写进去的边界都等于不存在。再加上 LLM 在拆分任务时天然倾向于"多列几个显得周到"，过度拆解会指数级放大模糊边界的概率。
还有一个被忽视的根因：主管在拆任务时，并不真的知道每个工人"会怎么理解"。它只是在预测一组合理的子目标，而工人拿到子目标后还会再做一次自己的解读。两层解读叠加，边界就会漂移。这本质上是"语义压缩-解压"过程中的信息损耗，不是提示词写细一点就能根治的。
对策
我的第一条经验是：拆解必须产出"契约"，而不是"描述"。每个子任务要显式声明三件事，它负责什么、它依赖哪些前置产物、它的产出格式是什么。下面这段是我现在在用的计划函数，它强制把边界写死，并且用依赖图去检测重叠和真空。
from dataclasses import dataclass, field
from typing import List
@dataclass
class SubTask:
id: str
owner: str
inputs: List[str] = field(default_factory=list)
outputs: List[str] = field(default_factory=list)
depends_on: List[str] = field(default_factory=list)
def plan(topic: str) -> List[SubTask]:
# 显式声明每个子任务的输入、输出与依赖，避免边界漂移
return [
SubTask(id="outline", owner="planner", outputs=["outline.md"]),
SubTask(id="market", owner="researcher",
inputs=["outline.md"], outputs=["market.md"],
depends_on=["outline"]),
SubTask(id="finance", owner="researcher",
inputs=["outline.md"], outputs=["finance.md"],
depends_on=["outline"]),
SubTask(id="draft", owner="writer",
inputs=["outline.md", "market.md", "finance.md"],
outputs=["draft.md"],
depends_on=["market", "finance"]),
SubTask(id="edit", owner="editor",
inputs=["draft.md"], outputs=["final.md"],
depends_on=["draft"]),
]
def detect_overlap(tasks: List[SubTask]) -> List[str]:
# 检查是否有多个任务声明了同名输出，提前拦截重复劳动
seen = {}
warnings = []
for t in tasks:
for o in t.outputs:
if o in seen:
warnings.append(f"输出 {o} 被 {seen[o]} 与 {t.id} 同时声明")
seen[o] = t.id
return warnings第二条经验是：让主管只做"切分"，把"去重与校验"交给一个确定性脚本。上面的 detect_overlap 就是纯函数，它不依赖 LLM，能在派活前拦掉大部分重叠。确定性校验层放在 LLM 编排层之外，是降低幻觉最划算的一笔投入。我还额外加了一个契约校验：每个工人的输出必须能通过一份 JSON Schema，通不过就打回重做，而不是让下游默默接住脏数据。
坑二：共享状态冲突
现象
当多个工人需要往同一份草稿里写东西时，冲突几乎是必然的。我遇到过最离谱的一次：两个工人分别负责"补充案例"和"润色语言"，它们共享一个 draft.md 文件。润色工人读取了版本 A，案例工人也读取了版本 A，案例工人先写回版本 B，润色工人后写回版本 C，而它的写回基于自己手里的版本 A，于是案例工人的补充被整个覆盖掉了。报告交上去，编辑发现案例章节凭空消失。
更微妙的是"读偏"型冲突：两个工人读到的不是同一时刻的状态，但都认为自己看到的是最新版，于是各自在旧基础上推进，最后合并出来的东西逻辑自相矛盾。这种问题在跑批时偶发，复现极难，是最折磨人的一类。
根因
根因是把 Agent 当成了"无状态函数"，但它们操作的是有状态的共享资源。文件系统、数据库行、甚至一段共享的上下文窗口，本质上都是临界区。LLM 没有"并发"的概念，它不会自己加锁，也不会在写之前先 rebase。当两个 Agent 的"读-改-写"周期交错，最后写者胜出，中间的所有修改都丢了。
这里还有个认知陷阱：很多人以为"用数据库事务就行"。但 Agent 的修改往往不是一条 SQL，而是一段文本编辑或一次工具调用，事务边界罩不住"读文档、思考、再写文档"这段跨越多次推理的周期。真正需要串行化的是这段"思考周期"，而不是单次 IO。
对策
最朴素也最有效的办法，是给共享资源加一个串行化 gate。我更倾向于用"单一写者 + 消息队列"而不是让所有人直接读写文件：每个工人只往自己的私有工作区写，由一个 merge 环节按确定顺序合并。下面这段示意把并发写转成了串行提交。
import asyncio
from collections import deque
class SharedDraft:
def __init__(self):
self._content = ""
self._queue = asyncio.Queue()
self._merger = asyncio.create_task(self._run())
async def _run(self):
# 所有修改请求串行进入合并器，杜绝交错写
while True:
patch = await self._queue.get()
self._content = self._apply(self._content, patch)
self._queue.task_done()
async def submit(self, agent_id: str, patch: str):
await self._queue.put((agent_id, patch))
def _apply(self, base: str, patch: str) -> str:
# 真实场景可用结构化 diff，这里简化为按来源追加
return base + f"\n[from {patch}]"如果共享的是数据库，那就更简单了：把写操作收敛到一个带事务的服务接口，Agent 只调用接口、不直接碰表。核心是"共享状态的所有权必须唯一"，谁拥有、谁串行化，不要让多个自主实体同时拥有写权。我后来把这条写进了团队的 Agent 开发规范第一条：凡是要落盘的共享资源，必须有一个唯一 owner 服务。
坑三：联动回滚缺失
现象
这是最阴险的一类。一个工人失败了，主管和其他工人毫无察觉，继续往前推进。结果就是：下游拿到的前置产物是残缺的，但没人报错，整条流水线安静地吐出了一份"看起来完整、实则错误"的成品。我在一次批量生成里，因为一个检索工人触发了速率限制而返回空结果，写稿工人没接到任何异常，直接基于空数据编了一段看起来合理的分析，读者来信说"这一段和公开财报对不上"。
更糟的是级联污染：失败的任务在中间产物里留了个洞，后面的工人基于这个洞补出了一个自洽但错误的结论，等人工审到时已经很难定位最初是哪一步塌的。
根因
根因是"失败传播"在 Agent 编排里是隐式的。微服务里有超时、有熔断、有显式的错误码返回；而 Agent 之间的衔接往往是文本，文本里没有"错误类型"这个字段。一个工人说"抱歉我没查到"，在主管看来这和"我查完了，结论是暂无"几乎无法区分，于是失败被静默吞掉了。
还有一个结构性的根因：我们把"完成"定义成了"没有抛异常"，而不是"产出了符合契约的产物"。很多 Agent 即便拿不到数据，也会努力生成一个"能看"的结果，因为它被训练得倾向于 helpful。这种"善意的补全"在协作系统里是毒药。
对策
我的做法是在每个任务上挂一个结构化的状态机，失败是显式的一等公民，不是靠主管去猜。下面这段主管逻辑要求：任何子任务失败，都要触发关联任务的回滚，并且整个作业进入 FAILED 而非 DONE。
from enum import Enum
class Status(Enum):
PENDING = "pending"
RUNNING = "running"
DONE = "done"
FAILED = "failed"
async def supervisor(tasks):
state = {t.id: Status.PENDING for t in tasks}
completed = {}
try:
for t in order_by_dependency(tasks):
state[t.id] = Status.RUNNING
completed[t.id] = await run(t)
state[t.id] = Status.DONE
except Exception as e:
# 失败要显式传播，并回滚已完成的关联任务
for tid, result in completed.items():
await rollback(tid, result)
state[t.id] = Status.FAILED
raise RuntimeError(f"任务 {t.id} 失败，已回滚 {len(completed)} 个前置") from e
return completed
async def rollback(tid, result):
# 真实实现按任务类型决定如何撤销，例如删除草稿、标记作废
print(f"rollback {tid}")关键是把"失败"从一段自然语言，变成一条带类型、带可追溯 ID 的事件。这样主管才能做出"撤不撤、怎么撤"的确定性决策，而不是赌 LLM 能读懂工人的潜台词。我还会给每个工人加一条硬规则：拿不到关键前置数据时，必须显式返回 MISSING 状态，禁止自行补全。这条规则上线后，级联污染类事故几乎清零。
单 Agent + 多工具 何时更合适
讲完三个坑，我想泼一盆冷水：很多你以为需要多 Agent 的场景，其实用"单 Agent + 多工具"更稳。判断标准很简单，如果子任务之间需要频繁共享中间状态、或者依赖关系高度耦合，拆开只会增加冲突面。
- 适合单 Agent 多工具：任务本质是一条线性流水线，工具之间只是"调用"，不需要各自维护独立上下文。比如"读 URL、提取正文、生成摘要、写数据库"，一个 Agent 持有一个上下文就够了，拆成四个 Agent 纯属制造边界问题。
- 适合多 Agent：子任务真的彼此正交，且各自的上下文会互相干扰。比如"同时用不同人设审同一篇稿子"，让三个 Agent 各自带不同角色设定，反而比塞进一个上下文更干净。
一句话：拆分的成本由"边界管理"承担，只有当边界管理的成本低于共享上下文的噪声成本时，拆分才划算。我给自己定的经验线是：如果拆解后子任务间需要交换的状态超过三处，先别拆，先用单 Agent 加更多工具试试。
主管-工人 与 流水线 两种编排
这两种是最常用的编排模式，但适用面完全不同，我见过太多人把它们混用。
主管-工人（Supervisor-Worker）是中心化的：主管持有全局计划，动态派活给工人，工人之间不直接通信。它适合"任务不确定、需要中途调整"的场景，比如探索性研究。优点是灵活，缺点是主管是单点瓶颈，也是单点故障，主管一旦抽风，全队停摆。
流水线（Pipeline）是去中心化的：每个环节只认上游产物，按固定顺序流动。它适合"步骤固定、追求吞吐"的场景，比如批量生成。优点是稳定、好测试、好扩并发，缺点是不灵活，上游一变下游全断，且难以处理"中途发现需要回头改上游"的情况。
# 主管-工人：主管动态决策下一个派给谁
async def supervisor_loop(plan, workers):
while not plan.all_done():
next_task = plan.pick_next() # 动态选择
worker = workers[next_task.owner]
await worker.run(next_task)
# 流水线：固定顺序，无中央决策
async def pipeline(stages, payload):
data = payload
for stage in stages: # 顺序固定
data = await stage(data)
return data我现在的实践倾向是：先用流水线把稳定的部分固化下来，把不确定的探索部分包成一个"主管-工人"子模块挂在某一个流水线段上。这样既保住了吞吐，又留了灵活性。举个例子，固定流程是"抓取→清洗→生成→发布"，但"生成"这步内部用主管-工人去协调选题、写作、配图三个工人，外层仍然是一条可预测的流水线。
怎么判断你的场景该不该拆
我总结了一个粗略的评分卡，帮团队在动手前先想清楚。满足越多条，越适合多 Agent；反之，老实做单 Agent。
- 子任务能否用一句话说清边界，且彼此输出不重叠？说不清就别拆。
- 子任务是否真的需要不同角色设定或不同工具集？不需要就别拆。
- 失败是否能被清晰地局部化，不影响其他分支？不能就先别拆。
- 你是否愿意为它额外写确定性校验层（去重、冲突、回滚）？不愿意就别拆。
这套卡帮我省下了至少两个不该拆的项目。其中一个，我们本来想拆成"选题 Agent + 写作 Agent + 审校 Agent"，后来发现三者共享同一份不断变化的提纲，冲突面太大，改成单 Agent 带三个工具后，质量反而更稳，因为上下文不再被切碎。
多 Agent 系统的测试策略
多 Agent 最难测的不是单点，而是"编排"。我现在的做法是把确定性和非确定性分层测：确定性部分（计划生成、依赖校验、回滚逻辑、串行化 gate）写普通单测，这部分必须百分之百覆盖；非确定性部分（每个工人的实际输出质量）用离线回放测，录下真实跑过的任务，定期用新模型重跑对比。
下面这段是一个最小化的契约测试，它不验证"写得好不好"，只验证"有没有按契约交差"，这恰恰是协作系统里最该先守住的底线。
import json
from pathlib import Path
def assert_contract(task_id: str, output_path: str, schema: dict):
# 只校验产出是否符合契约，不评价内容优劣
if not Path(output_path).exists():
raise AssertionError(f"{task_id} 未产出 {output_path}")
data = json.loads(Path(output_path).read_text())
for field_name in schema.get("required", []):
if field_name not in data:
raise AssertionError(f"{task_id} 缺少字段 {field_name}")我特别想强调：不要试图用单测去断言 Agent 的输出"语义正确"，那是离线评估的事。单测只该守住"结构正确、边界正确、失败可见"，把语义判断交给回放和人工抽检。混淆了这两层，测试会既脆弱又无用。
一个最小可运行的编排骨架
把前面三类的对策合起来，下面是一个我实际在用的骨架：它用契约做计划、用队列做串行写、用状态机做失败传播。它不长，但三个坑的护栏都在里面。
import asyncio
from dataclasses import dataclass, field
from typing import List
@dataclass
class Job:
id: str
owner: str
outputs: List[str] = field(default_factory=list)
depends_on: List[str] = field(default_factory=list)
class Orchestrator:
def __init__(self):
self._bus = asyncio.Queue()
async def run_all(self, jobs: List[Job]):
done = {}
try:
for j in sorted(jobs, key=lambda x: len(x.depends_on)):
result = await self._execute(j)
done[j.id] = result
except Exception:
for jid, r in done.items():
await self._compensate(jid, r)
raise
return done
async def _execute(self, job: Job):
await self._bus.put(job.id)
return f"result-of-{job.id}"
async def _compensate(self, jid: str, result):
print(f"compensate {jid}")
# 用法
async def main():
jobs = [Job("a", "w1"), Job("b", "w2", depends_on=["a"])]
await Orchestrator().run_all(jobs)这个骨架的价值不在于它多完整，而在于它把"边界、所有权、失败"三件事变成了代码里绕不开的结构。新人接手时，想制造前面那三种坑都得多写几行，而不是随手就能写出竞态。
结语
多 Agent 不是银弹，它把"单体里的隐性耦合"显式化成了"编排层的显性问题"。坑不在模型聪明不聪明，而在我们是否认真设计过边界、所有权和失败传播。把确定性校验、串行化写、显式失败这三件事做扎实，多 Agent 才真正值得上。否则，一个带着好工具的单体 Agent，往往比你手忙脚乱的 Agent 小队更可靠。我现在的默认选项是"能不拆就不拆"，只有当单 Agent 的上下文噪声确实撑不住时才拆，而且拆之前，先把这三类坑的护栏写好。