LLM和Agent——专题6:Multi Agent 入门(4)
多 Agent 协作的 5 种通信模式——从对话到结构化协议Agent 之间怎么说话从自然语言闲聊到结构化图状态传递这 5 种通信范式决定了你的 Multi-Agent 系统能有多可靠。一、引言搭建 Multi-Agent 系统时大多数人第一反应是关注每个 Agent 做什么角色定义但很快会撞上一面更隐蔽的墙Agent 之间怎么传递信息这个问题看似简单实则非常关键。通信方式直接影响可靠性信息传递过程中会不会失真可调试性出问题时你能追踪到哪个环节效率上下文是否被无效信息塞满可扩展性加了第三个 Agent 后通信会不会炸这篇文章系统性地梳理 Multi-Agent 系统中 5 种主流的通信模式帮你理解每种模式的工作原理、代码示例、适用场景和潜在陷阱。读完你会获得5 种通信模式的清晰定义和对比每种模式的代码示例按场景的选型建议常见的通信反模式二、五种通信模式详解2.1 模式一自然语言对话Natural Language Dialogue工作原理Agent 之间通过自然语言对话交流就像两个人聊天。上游 Agent 的输出直接作为下游 Agent 的输入对话历史的一部分。Agent A: 我分析了这个用户的问题他遇到了 Python 的 GIL 导致的性能瓶颈。我建议从以下几个角度回答1. GIL 是什么... Agent B: 好的我基于你的分析来撰写回复。关于 GIL我会重点解释...# 自然语言对话模式实现classDialogueBasedAgent:def__init__(self,name:str,system_prompt:str):self.namename self.system_promptsystem_prompt self.conversation_history[]defsend_message(self,message:str,recipient)-str:# 将对话历史 新消息发给 LLMresponsellm.chat(systemself.system_prompt,messagesself.conversation_history[{role:user,content:f[来自{self.name}]:{message}}])self.conversation_history.append({role:assistant,content:response})returnresponse优点最灵活不需要预定义任何结构Agent 可以表达不确定性、提出反问、进行澄清开发成本最低缺点信息容易失真A 说的话被 B 误解对话历史快速膨胀上下文窗口吃紧大量 Token 浪费在好的我理解了让我来处理这种无信息量的客套话上不可解析下游无法程序化处理适用场景探索性对话、需要灵活协商的多轮交互、人类在回路中的场景。2.2 模式二结构化 JSON 传递Structured JSON工作原理Agent 必须以预定义的 JSON Schema 输出下游 Agent 解析这个 JSON 结构来获取信息。{intent:code_debug,language:python,error_type:TypeError,root_cause:第 23 行传入的变量是 str 而非 int,suggested_fix:在调用前使用 int() 转换,confidence:0.85,references:[https://docs.python.org/3/library/functions.html#int]}# 结构化 JSON 传递实现frompydanticimportBaseModelfromtypingimportOptionalclassAgentOutput(BaseModel):analysis:strkey_findings:list[str]suggested_action:strconfidence:floatneeds_human_review:boolclassStructuredAgent:defprocess(self,input_data:dict)-AgentOutput:responsellm.chat(system你必须严格以指定的 JSON Schema 输出。不要添加任何其他文字。,messages[{role:user,content:json.dumps(input_data)}],response_format{type:json_object},)# 解析并验证parsedAgentOutput.model_validate_json(response)returnparsed优点结构化可被程序化解析和处理字段级的信息传递不会夹带废话便于验证——如果输出不符合 Schema可以自动重试调试清晰——出问题时能看到是哪个字段错了缺点灵活性受限无法表达复杂/模糊的信息Schema 设计本身就是一件很难的事——太松起不到约束作用太紧让 Agent 难以输出当 Schema 变化时所有相关 Agent 的 prompt 都要更新适用场景流水线任务、需要下游程序化处理结果的场景、对数据质量有严格要求的场景。2.3 模式三共享记忆/黑板模式Shared Memory / Blackboard工作原理多个 Agent 共享一个黑板共享数据空间每个 Agent 都可以读写黑板上的数据。Agent 之间不直接通信而是通过黑板间接交换信息。Agent A ─写─→ [共享记忆/黑板] ─读─→ Agent B Agent B ─写─→ [共享记忆/黑板] ─读─→ Agent C Agent C ─写─→ [共享记忆/黑板] ─读─→ Agent A# 共享记忆/黑板模式实现fromtypingimportAnyclassBlackboard:共享的黑板所有 Agent 可以读写def__init__(self):self.data:dict[str,Any]{}self.history:list[dict][]# 记录每次写入defwrite(self,agent_name:str,key:str,value:Any):self.data[key]value self.history.append({agent:agent_name,action:write,key:key,timestamp:time.time(),})defread(self,key:str)-Any:returnself.data.get(key)defsnapshot(self)-dict:return{data:self.data,last_updated_by:self.history[-1][agent]ifself.historyelseNone,}classBlackboardAgent:def__init__(self,name:str,blackboard:Blackboard):self.namename self.blackboardblackboarddefstep(self):# 1. 读取黑板上自己关心的数据contextself.blackboard.snapshot()# 2. LLM 推理决定要写什么resultllm.chat(systemf你是 Agent{self.name}。黑板上当前数据{context}。请执行你的任务并更新黑板。,)# 3. 写入黑板self.blackboard.write(self.name,f{self.name}_result,result)优点解耦Agent 不需要知道彼此的存在只需知道黑板容易增加新 Agent只需读写黑板天然支持并行——多个 Agent 可以同时读写需要并发控制缺点读写冲突两个 Agent 同时写同一个 key信息过载黑板上的数据越来越多Agent 需要自己筛选缺乏明确的完成信号——谁知道所有 Agent 都完成了适用场景多 Agent 并行处理、需要积累中间结果的长期任务、Agent 数量不固定的动态系统。2.4 模式四工具调用传递Tool Calling Pass-through工作原理下游 Agent 的 tool calling 结果直接作为上游 Agent 的输入材料而非由 Orchestrator 做中转。Agent A 调用 search_tool(Python GIL) → search_tool 返回结果 → 结果直接传给 Agent B 作为上下文 → Agent B 调用 code_executor(...) → 执行结果传给 Agent C# 工具调用传递模式实现classToolCallRelay:工具调用结果的中继器def__init__(self):self.observers:dict[str,list]{}# tool_name → [agents to notify]defregister_observer(self,tool_name:str,agent):iftool_namenotinself.observers:self.observers[tool_name][]self.observers[tool_name].append(agent)defon_tool_result(self,tool_name:str,result:Any,caller:str):# 通知所有关注该工具结果的 Agentforagentinself.observers.get(tool_name,[]):agent.receive_tool_result(tool_name,result,caller)classToolAwareAgent:defreceive_tool_result(self,tool_name:str,result:Any,caller:str):当其他 Agent 的某个工具调用返回结果时被触发self.context.append({type:tool_result,tool:tool_name,from_agent:caller,result:result,})# 使用示例relayToolCallRelay()relay.register_observer(web_search,agent_b)# Agent B 关注搜索工具的结果relay.register_observer(code_execution,agent_c)# Agent C 关注代码执行结果优点高效只传递工具调用的结果而非整个对话历史职责清晰每个 Agent 知道自己关注哪些工具结果减少上下文浪费缺点需要预先定义 Agent 对工具结果的关注关系只传递结果丢失了为什么调用这个工具的上下文框架支持有限往往需要自己实现适用场景Agent 之间有明确的数据依赖关系、需要减少 Token 消耗、工具调用是主要信息源。2.5 模式五图状态传递Graph State Passing工作原理这是 LangGraph 的核心设计模式。整个 Multi-Agent 系统被建模为一个有向图图的状态State在节点之间流转每个节点Agent读取状态、执行任务、更新状态。State { messages: [...], task: 修复 Bug #42, analysis: None, fix: None, test_result: None, } Analyzer 节点: State.analysis 类型错误第 23 行... ↓ Fixer 节点: State.fix diff: ... ↓ Tester 节点: State.test_result 通过# 图状态传递模式LangGraph 风格fromlanggraph.graphimportStateGraph,ENDfromtypingimportTypedDictclassAgentState(TypedDict):messages:listtask:stranalysis:str|Nonefix:str|Nonetest_result:str|Nonenext_step:strdefanalyzer_node(state:AgentState)-AgentState:分析问题resultllm.chat(f分析这个问题:{state[task]})return{analysis:result,next_step:fix}deffixer_node(state:AgentState)-AgentState:修复问题resultllm.chat(f根据分析修复:{state[analysis]})return{fix:result,next_step:test}deftester_node(state:AgentState)-AgentState:测试修复# 执行测试...return{test_result:通过,next_step:end}defrouter(state:AgentState)-str:returnstate[next_step]# 构建图workflowStateGraph(AgentState)workflow.add_node(analyzer,analyzer_node)workflow.add_node(fixer,fixer_node)workflow.add_node(tester,tester_node)workflow.set_entry_point(analyzer)workflow.add_conditional_edges(analyzer,router,{fix:fixer,end:END,})workflow.add_edge(fixer,tester)workflow.add_edge(tester,END)appworkflow.compile()resultapp.invoke({task:修复 Bug #42,messages:[]})优点明确的拓扑结构可视化、可调试状态是结构化的可以随时检查当前进展到哪一步支持条件分支、循环、并行状态持久化使恢复和重放成为可能缺点学习曲线最陡对于简单的线性流程来说是过度设计State Schema 的设计需要前瞻性——后期改动成本高适用场景复杂工作流、需要条件分支和循环、需要状态持久化和可恢复性。三、五种模式对比总览维度自然语言对话结构化 JSON共享记忆/黑板工具调用传递图状态传递灵活性★★★★★★★★☆☆★★★★☆★★☆☆☆★★★☆☆可靠性★★☆☆☆★★★★☆★★★☆☆★★★★☆★★★★★可调试性★☆☆☆☆★★★★☆★★★☆☆★★★☆☆★★★★★上下文效率★☆☆☆☆★★★★☆★★★☆☆★★★★★★★★★☆学习成本★☆☆☆☆★★☆☆☆★★★☆☆★★★★☆★★★★★框架支持CrewAI, AutoGen需自定义需自定义需自定义LangGraph四、选型决策指南按场景选场景推荐模式原因内容生成流水线结构化 JSON阶段产物需要结构化、可验证代码审查 修复循环图状态传递需要条件分支通过/不通过头脑风暴/探索自然语言对话需要最大的灵活性数据库查询管道结构化 JSON下游需要解析 SQL多路召回评测共享记忆/黑板多个评测 Agent 并行写入结果API 编排工具调用传递以工具结果作为核心数据流反模式警告1. 用自然语言传递结构化数据# 不要这样 Agent A: 用户 ID 是 42余额是 100.5 元等级是 VIP # Agent B 要费劲解析2. 所有 Agent 共用一个长长的对话历史上下文越滚越大到第 4 个 Agent 时已经包含了前面 3 个 Agent 的完整聊天记录。Agent 4 在 8000 Token 中找自己需要的 200 Token 关键信息。3. 通信模式和编排模式不匹配Hierarchical 编排却用图状态传递——调度者本身的决策逻辑在图里很难表达。五、实际选型组合示例以一个代码审查系统为例GitHub PR → 工作流入口 ↓ Agent 1 (代码分析): 输出结构化 JSON {issues: [...], severity: high} ↓ 通过图状态的 analysis 字段传递 Agent 2 (修复建议): 读取 analysis 字段 → 输出到 fix_suggestions 字段 ↓ 通过图状态传递 Agent 3 (验证): 读取 fix_suggestions → 沙箱执行 → test_result 字段 ↓ 如果 test_result fail → 回到 Agent 2条件循环 如果 test_result pass → 生成 Review Comment这里混合了图状态传递主流程结构化 JSON每个 Agent 的输出格式工具调用传递沙箱执行的结果。六、总结通信模式是 Multi-Agent 系统的隐形成本——模式选得不好上下文膨胀、信息失真、调试噩梦会接踵而来。没有最佳模式只有最匹配的模式。简单任务用自然语言对话就够复杂流水线用图状态传递更可靠。越结构化越可靠。如果可能Agent 间传递的信息尽量用结构化格式而非自然语言。实际系统往往是混合的。图状态做骨架结构化 JSON 做节点间传递自然语言只在需要灵活协商时使用。设计通信协议要像设计 API 一样认真。Agent 之间的接口同样需要版本管理、向下兼容、错误处理。通信模式决定了你 Multi-Agent 系统的神经系统质量——好的通信让协作无缝差的通信让系统瘫痪。