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.name=name self.system_prompt=system_prompt self.conversation_history=[]defsend_message(self,message:str,recipient)->str:# 将对话历史 + 新消息发给 LLMresponse=llm.chat(system=self.system_prompt,messages=self.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:response=llm.chat(system="你必须严格以指定的 JSON Schema 输出。不要添加任何其他文字。",messages=[{"role":"user","content":json.dumps(input_data)}],response_format={"type":"json_object"},)# 解析并验证parsed=AgentOutput.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.name=name self.blackboard=blackboarddefstep(self):# 1. 读取黑板上自己关心的数据context=self.blackboard.snapshot()# 2. LLM 推理，决定要写什么result=llm.chat(system=f"你是 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,})# 使用示例relay=ToolCallRelay()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:"""分析问题"""result=llm.chat(f"分析这个问题:{state['task']}")return{"analysis":result,"next_step":"fix"}deffixer_node(state:AgentState)->AgentState:"""修复问题"""result=llm.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"]# 构建图workflow=StateGraph(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)app=workflow.compile()result=app.invoke({"task":"修复 Bug #42","messages":[]})

优点：

• 明确的拓扑结构，可视化、可调试
• 状态是结构化的，可以随时检查"当前进展到哪一步"
• 支持条件分支、循环、并行
• 状态持久化使恢复和重放成为可能

缺点：

• 学习曲线最陡
• 对于简单的线性流程来说是过度设计
• State Schema 的设计需要前瞻性——后期改动成本高

适用场景：复杂工作流、需要条件分支和循环、需要状态持久化和可恢复性。

三、五种模式对比总览

四、选型决策指南

按场景选

反模式警告

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 系统的"神经系统"质量——好的通信让协作无缝，差的通信让系统瘫痪。