多 Agent 协作架构：Agent 之间如何通信、协调和分工

多 Agent 协作架构Agent 之间如何通信、协调和分工我在字节跳动 OpenViking 分析之后有一件事一直在脑子里转多 Agent 的协作到底难在哪里不是难在写代码而是难在把一堆各自为政的 Agent 变成一个真正能协作的系统。这篇文章是我自己踩坑之后的总结不是理论综述。为什么多 Agent 比单 Agent 难得多单 Agent 的问题很简单上下文窗口塞满就炸不知道什么时候停偶尔幻觉。多 Agent 的问题复杂一个数量级。先看一个数字一个包含 5 个 Agent 的系统如果每个 Agent 每步有 95% 的准确率5 步之后系统整体准确率是0.95^25 ≈ 0.28。这不是理论推导这是我在实际部署里观察到的现象——单个 Agent 表现不错但串起来之后系统质量断崖式下跌。根本原因是错误传播Agent A 的一个小错误会变成 Agent B 的输入前提然后被 C 放大到 D 的时候已经跑偏了方向。这不是 bug这是分布式系统的基本性质。三个核心挑战状态同步Agent A 做了什么Agent B 怎么知道共享内存消息传递文件每种方式都有代价。任务分发谁决定哪个 Agent 做什么Orchestrator 集中决策还是 Agent 自己抢任务阻塞问题Agent B 在等 Agent AAgent A 在等用户输入整个系统死锁。主要协作模式Orchestrator/Worker 模式这是最常见的模式也是我用得最多的。一个 Orchestrator 负责任务分解和调度Worker 负责执行。用户请求 ↓ Orchestrator ├── 任务分解 ├── Worker A ← 子任务1 ├── Worker B ← 子任务2 └── Worker C ← 子任务3 ↓ 结果汇总 Orchestrator → 最终输出优势集中控制状态清晰容错简单Worker 失败Orchestrator 重试或换策略并行执行效率高劣势Orchestrator 是单点它的决策质量决定整体上限Orchestrator 上下文压力大需要跟踪所有 Worker 的状态Worker 之间没有直接通信信息必须过 Orchestrator 中转# 伪代码Orchestrator 任务分发asyncdeforchestrate(task:str)-str:subtasksawaitdecompose(task)resultsawaitasyncio.gather(*[worker.execute(subtask)forsubtaskinsubtasks])returnawaitsynthesize(results)实际中 Orchestrator 需要处理的远比这复杂Worker 返回不完整结果怎么办Worker 超时怎么办Worker 的结果互相矛盾怎么办Peer-to-Peer 模式Agent 之间直接通信没有中心调度。适合需要 Agent 互相 review、互相纠错的场景。Agent A ←→ Agent B ↑ ↑ └── Agent C ┘真实案例让两个 Agent 互相 review 代码。Agent A 写代码Agent B 找 bugAgent A 根据 B 的反馈修复再让 B 验证。这种 adversarial 协作在某些任务上效果显著好于单 Agent。# Agent 对话循环asyncdefadversarial_review(code:str)-str:coderAgent(roledeveloper)reviewerAgent(rolecritic)forroundinrange(max_rounds):issuesawaitreviewer.review(code)ifnotissues:breakcodeawaitcoder.fix(code,issues)returncode问题容易出现互相讨好的情况——两个 Agent 都倾向于认可对方而不是真的批判性审查。需要在 prompt 里明确强制批评性角色。Pipeline 模式线性流水线每个 Agent 处理上一个的输出。Raw Input → Extractor → Analyzer → Formatter → Output最简单最容易实现也最脆弱。中间任何一环挂掉整条 pipeline 断。但对于格式转换、逐步精炼的任务非常合适。分层专家模式一个路由层把任务分发给领域专家 Agent。用户输入 → Router Agent ├── Code Expert ├── Data Expert ├── Writing Expert └── Search Expert路由本身是个分类问题可以用简单 Agent 做省 token也可以嵌入语义路由准确但贵。我的经验是路由错误是这个模式最大的痛点需要设计兜底机制。Agent 间通信的核心挑战状态共享的四种方式1. 共享文件系统最简单最通用。Agent 把结果写到约定的文件路径其他 Agent 读取。# Agent A 写结果echo{status: done, result: ...}/tmp/agent_a_result.json# Agent B 等待并读取while[!-f/tmp/agent_a_result.json];dosleep1;donecat/tmp/agent_a_result.json适合长期状态、大量数据、跨进程通信不适合高频通信、需要事务性更新的场景踩坑文件锁。两个 Agent 同时写同一个文件会导致数据损坏。必须用文件锁或者 atomic rename。importos,json,tempfiledefatomic_write(path:str,data:dict):diros.path.dirname(path)withtempfile.NamedTemporaryFile(modew,dirdir,deleteFalse,suffix.tmp)asf:json.dump(data,f)tmp_pathf.name os.rename(tmp_path,path)# atomic on POSIX2. 消息队列Agent 通过消息队列异步通信。发送方不需要知道接收方的存在。# 发布任务queue.publish(task_queue,{task_id:abc123,type:analyze,payload:data})# 消费任务whileTrue:taskqueue.consume(task_queue)resultprocess(task)queue.publish(fresult_{task[task_id]},result)适合高并发、解耦、容错不适合需要强一致性、调试复杂度高3. 共享内存/上下文在同一个进程内Agent 直接共享内存对象。最高效但限制最多。shared_stateSharedState()asyncdefagent_a(state:SharedState):asyncwithstate.lock(analysis):state.analysis_resultawaitanalyze(state.raw_data)asyncdefagent_b(state:SharedState):# 等待 A 完成awaitstate.wait_for(analysis_result)resultawaitformat(state.analysis_result)4. 结构化 APIAgent 通过 HTTP API 或 RPC 通信。最正式最可维护但也最重。# Agent A 暴露 APIapp.post(/analyze)asyncdefanalyze(request:AnalyzeRequest)-AnalyzeResponse:resultawaitdo_analysis(request.data)returnAnalyzeResponse(resultresult,confidence0.95)# Agent B 调用responseawaithttp_client.post(http://agent-a:8000/analyze,json{data:my_data})任务分发的难点静态分发 vs 动态分发静态分发任务开始前决定好谁做什么。简单预期行为稳定但灵活性差。动态分发运行中根据 Agent 负载、能力、当前状态决定。复杂但能适应变化。classDynamicScheduler:asyncdefassign(self,task:Task)-Agent:available[agentforagentinself.agentsifagent.can_handle(task)andnotagent.is_busy()]ifnotavailable:# 等待 or 排队 or 失败raiseNoAvailableAgentError(task)# 选择最合适的returnmin(available,keylambdaa:a.estimated_latency(task))任务依赖图当任务之间有依赖关系时需要 DAG有向无环图调度。A → B → D ↓ ↑ C ──┘B 依赖 AC 依赖 BD 依赖 B 和 C。必须等 A 完成才能跑 B等 B 完成才能并行跑 C 和某些 D 前置步骤。asyncdefexecute_dag(dag:Dict[str,List[str]],tasks:Dict[str,Task]):completedset()asyncdefrun_task(task_id:str):depsdag.get(task_id,[])awaitasyncio.gather(*[wait_for_completion(dep,completed)fordepindeps])resultawaitexecute(tasks[task_id])completed.add(task_id)returnresultreturnawaitasyncio.gather(*[run_task(tid)fortidintasks])阻塞问题阻塞是多 Agent 系统最难调试的问题。死锁场景Agent A 等 Agent B 的结果Agent B 等 Agent A 的输入系统永远卡住解决方案超时 断路器asyncdefwith_timeout(coro,timeout_seconds:float,fallbackNone):try:returnawaitasyncio.wait_for(coro,timeouttimeout_seconds)exceptasyncio.TimeoutError:iffallbackisnotNone:returnfallbackraise# 在关键依赖上加超时resultawaitwith_timeout(agent_b.process(data),timeout_seconds30,fallback{status:timeout,partial:True})活锁场景更难发现Agent A 不断重试Agent B 不断返回相同错误系统在动但没有进展通信协议ACP、MCP 和文件共享的对比MCPModel Context ProtocolAnthropic 推的工具调用协议。本质是让 LLM 能调用外部工具但也可以用来做 Agent 间通信。优势标准化生态好Claude 原生支持劣势设计目标是LLM 调用工具不是Agent 与 Agent 通信。状态管理弱长对话里上下文会膨胀。{type:tool_use,name:call_agent_b,input:{task:analyze the following data,data:...}}ACPAgent Communication ProtocolIBM/BeeAI 推的 Agent 间通信协议。REST-based专门为 Agent 设计。优势设计目标准确支持流式、多模态有标准的 Agent 发现机制劣势生态还小tooling 不成熟部署复杂度高# ACP 风格的 Agent 调用responseawaitacp_client.run(agent_iddata-analyzer,messages[{role:user,content:task_description}],streamTrue)asyncforchunkinresponse:process(chunk)文件共享最简单也最常被低估对于很多实际场景文件系统就够了。Agent A 把结果写到/workspace/results/step1.jsonAgent B 读取。真实优势天然的持久化系统崩溃重启状态还在可审计所有中间状态都有记录调试容易语言无关Python Agent 写的文件Node.js Agent 也能读真实劣势实时性差需要轮询或 inotify大量小文件性能差没有标准化的错误/状态格式我自己的经验对于异步、低频的 Agent 协作文件共享是性价比最高的选择。对于需要低延迟实时通信的场景才值得引入消息队列或 RPC。现有框架的实际评估LangGraph图结构的 Agent 编排把 Agent 流程建模为有向图。适合需要明确控制流、有循环的复杂流程不适合动态任务结构、需要运行时修改流程fromlanggraph.graphimportStateGraph graphStateGraph(State)graph.add_node(research,research_agent)graph.add_node(write,write_agent)graph.add_node(review,review_agent)graph.add_edge(research,write)graph.add_conditional_edges(review,lambdas:writeifs.needs_revisionelseEND)AutoGenMicrosoft多 Agent 对话框架Agent 之间通过自然语言对话协作。适合需要 Agent 互相讨论、协商的任务不适合高精度任务对话容易漂移需要控制成本的场景实际测试AutoGen 的对话质量上限不低但 token 消耗是单 Agent 的 3-8 倍而且很难预测对话什么时候结束。CrewAI角色扮演风格的多 Agent每个 Agent 有明确的职责和目标。适合任务天然能对应到不同职能角色不适合需要精确控制执行流程的场景我认为未来方向是什么标准化状态协议现在最大的问题是没有标准。每个框架都有自己的状态格式、通信方式、错误处理。Agent A 的完成和 Agent B 的完成可能完全不同语义。我认为会出现类似 HTTP 状态码那样的标准化 Agent 状态协议不只是成功/失败还要包含置信度、部分完成度、下一步建议。{status:partial,confidence:0.7,completed_steps:[research,outline],pending_steps:[write,review],blocking_on:user_approval,estimated_completion:2min}异步优先的设计现在大多数框架默认同步调用 Agent等结果继续。但真正高效的多 Agent 系统应该是异步优先的——发出任务去做别的收到完成通知再处理。这需要更好的 event-driven 架构支持以及 Agent 能主动唤醒等待它的上游。验证层的标准化每个 Agent 的输出都应该有标准的验证步骤不是 Agent 自己验证自己而是独立的验证 Agent。类似 CI/CD 里的 test pipeline代码写完不算完测试过才算完。Agent 输出不算完验证过才算完。classVerifiedAgent:def__init__(self,agent:Agent,validator:Agent):self.agentagent self.validatorvalidatorasyncdefrun(self,task:str,max_retries:int3)-Result:forattemptinrange(max_retries):resultawaitself.agent.execute(task)validationawaitself.validator.validate(result)ifvalidation.passed:returnresult taskf{task}\n\n上次结果的问题{validation.issues}raiseMaxRetriesExceededError()从Agent 工具到Agent 网络现在的 Agent 系统大多是星形结构一个主 Agent调用一堆工具其中一些工具是其他 Agent。未来会是真正的网络结构Agent 有自己的专长有自己的 reputation根据任务需求自动发现和组合。类似微服务但每个服务是一个有自主性的 Agent。这需要解决的核心问题Agent 发现怎么知道有哪些 Agent、能力描述怎么知道这个 Agent 擅长什么、信任机制怎么知道这个 Agent 可靠。实践建议如果你现在要搭一个多 Agent 系统我的具体建议先解决最小可行协作不要一开始就设计完美的多 Agent 架构。先让两个 Agent 能可靠协作再扩展到三个、四个。把验证做成标准步骤每个 Agent 完成任务后必须有验证步骤。不是自我报告完成了而是可度量的验收标准计数、格式检查、一致性验证。保持状态可见选择让中间状态可观测的方案。文件比内存好可持久化结构化日志比自然语言好可查询。调试多 Agent 系统最大的挑战是发生了什么而不是怎么修复。设计失败路径每个 Agent 调用都应该有明确的失败处理超时后干什么返回错误后干什么部分成功怎么处理。不要假设 Agent 会成功。多 Agent 协作还远没有成熟但方向是对的。单 Agent 的上限太明显了——上下文窗口、单点失败、无法并行。多 Agent 系统的复杂度代价是真实的但随着工具链成熟这个代价会越来越小。现在的关键不是等待完美的框架而是在实际任务里踩坑、总结、沉淀。