单个AI Agent能力有限多Agent协作才能解决复杂问题。但Agent间如何通信决定了整个系统的性能、可靠性和可扩展性。本文深度解析多Agent通信架构的设计模式与工程实践。一、多Agent通信的核心挑战多Agent系统面临的通信挑战不同于传统微服务非确定性问题Agent的输出不可预测A agent产出的结果B agent可能无法理解上下文传递Agent需要共享足够的上下文但又不能超出token限制协调问题多个Agent并发执行时如何避免冲突、重复工作失败处理一个Agent失败如何影响整个系统如何优雅降级可观测性分布式Agent的状态难以追踪## 二、三种基础通信模式### 模式一直接消息传递点对点pythonfrom dataclasses import dataclass, fieldfrom typing import Optional, Any, Dictfrom datetime import datetimeimport uuidimport asynciofrom queue import Queue, Emptydataclassclass AgentMessage: 标准Agent间消息格式 message_id: str field(default_factorylambda: str(uuid.uuid4())) sender_id: str recipient_id: str message_type: str task # task/result/status/error payload: Any None metadata: Dict field(default_factorydict) timestamp: str field(default_factorylambda: datetime.now().isoformat()) correlation_id: Optional[str] None # 关联原始请求 reply_to: Optional[str] None # 回复目标地址 def reply(self, payload: Any, sender_id: str) - AgentMessage: 创建回复消息 return AgentMessage( sender_idsender_id, recipient_idself.sender_id, message_typeresult, payloadpayload, correlation_idself.message_id )class MessageBus: 轻量级Agent消息总线 def __init__(self): self._queues: Dict[str, asyncio.Queue] {} self._subscribers: Dict[str, list] {} def register_agent(self, agent_id: str): 注册Agent创建专属消息队列 self._queues[agent_id] asyncio.Queue(maxsize100) async def send(self, message: AgentMessage): 发送消息到指定Agent if message.recipient_id not in self._queues: raise ValueError(fAgent {message.recipient_id} not registered) await self._queues[message.recipient_id].put(message) async def broadcast(self, message: AgentMessage, exclude: list None): 广播消息给所有Agent exclude exclude or [] tasks [] for agent_id, queue in self._queues.items(): if agent_id ! message.sender_id and agent_id not in exclude: broadcast_msg AgentMessage( sender_idmessage.sender_id, recipient_idagent_id, message_typemessage.message_type, payloadmessage.payload, correlation_idmessage.correlation_id ) tasks.append(queue.put(broadcast_msg)) await asyncio.gather(*tasks) async def receive(self, agent_id: str, timeout: float 30.0) - AgentMessage: 接收消息带超时 try: return await asyncio.wait_for( self._queues[agent_id].get(), timeouttimeout ) except asyncio.TimeoutError: raise TimeoutError(fAgent {agent_id} 等待消息超时) def pending_count(self, agent_id: str) - int: 查看待处理消息数量 return self._queues[agent_id].qsize() if agent_id in self._queues else 0# 使用示例bus MessageBus()class ResearcherAgent: def __init__(self, agent_id: str, bus: MessageBus): self.id agent_id self.bus bus bus.register_agent(agent_id) async def run(self): while True: msg await self.bus.receive(self.id) if msg.message_type task: # 执行研究任务 research_result await self._do_research(msg.payload[query]) # 发送结果给请求者 reply msg.reply( payload{research: research_result}, sender_idself.id ) await self.bus.send(reply) async def _do_research(self, query: str) - str: # 实际的研究逻辑 from openai import AsyncOpenAI client AsyncOpenAI() response await client.chat.completions.create( modelgpt-4o, messages[{role: user, content: f研究: {query}}] ) return response.choices[0].message.content### 模式二发布-订阅事件驱动pythonfrom typing import Callablefrom collections import defaultdictimport asyncioclass EventBus: Agent事件总线Pub/Sub模式 def __init__(self): self._handlers: Dict[str, list[Callable]] defaultdict(list) self._event_history: list [] self._max_history 1000 def subscribe(self, event_type: str, handler: Callable): 订阅特定类型的事件 self._handlers[event_type].append(handler) def subscribe_all(self, handler: Callable): 订阅所有事件用于监控/日志 self._handlers[*].append(handler) async def publish(self, event_type: str, data: Any, source: str ): 发布事件 event { type: event_type, data: data, source: source, timestamp: datetime.now().isoformat(), event_id: str(uuid.uuid4()) } # 记录历史 self._event_history.append(event) if len(self._event_history) self._max_history: self._event_history.pop(0) # 触发处理器 handlers self._handlers.get(event_type, []) self._handlers.get(*, []) if handlers: await asyncio.gather(*[ self._safe_handle(h, event) for h in handlers ]) async def _safe_handle(self, handler: Callable, event: dict): try: if asyncio.iscoroutinefunction(handler): await handler(event) else: handler(event) except Exception as e: print(f事件处理器错误: {e}) def get_history(self, event_type: str None, limit: int 50) - list: events self._event_history if event_type: events [e for e in events if e[type] event_type] return events[-limit:]# 多Agent协作的事件流event_bus EventBus()class OrchestratorAgent: 编排Agent通过事件协调其他Agent def __init__(self, bus: EventBus): self.bus bus # 订阅子Agent的完成事件 bus.subscribe(research.completed, self.on_research_done) bus.subscribe(writing.completed, self.on_writing_done) bus.subscribe(review.completed, self.on_review_done) self.state { research_done: False, draft: None, review_done: False } async def start_workflow(self, topic: str): 启动工作流 print(f开始工作流{topic}) # 触发研究 await self.bus.publish(task.research, {topic: topic, depth: deep}, sourceorchestrator) async def on_research_done(self, event: dict): self.state[research_done] True research_data event[data][result] # 触发写作 await self.bus.publish(task.write, {research: research_data, format: blog}, sourceorchestrator) async def on_writing_done(self, event: dict): self.state[draft] event[data][draft] # 触发审查 await self.bus.publish(task.review, {draft: self.state[draft]}, sourceorchestrator) async def on_review_done(self, event: dict): self.state[review_done] True print(✅ 工作流完成) print(f最终内容: {event[data][approved_draft][:200]}...)### 模式三共享状态黑板模式pythonfrom typing import Any, Optionalimport asynciofrom datetime import datetimeclass SharedStateStore: 多Agent共享状态存储黑板模式 def __init__(self): self._state: Dict[str, Any] {} self._locks: Dict[str, asyncio.Lock] {} self._watchers: Dict[str, list[Callable]] defaultdict(list) self._history: Dict[str, list] defaultdict(list) self._global_lock asyncio.Lock() async def set(self, key: str, value: Any, author: str ): 设置状态值 async with self._get_lock(key): old_value self._state.get(key) self._state[key] value # 记录历史 self._history[key].append({ value: value, author: author, timestamp: datetime.now().isoformat() }) # 通知观察者 if old_value ! value and key in self._watchers: await asyncio.gather(*[ w(key, old_value, value) for w in self._watchers[key] ]) async def get(self, key: str, default: Any None) - Any: return self._state.get(key, default) async def compare_and_swap( self, key: str, expected: Any, new_value: Any, author: str ) - bool: 原子性CAS操作防止并发冲突 async with self._get_lock(key): current self._state.get(key) if current expected: self._state[key] new_value return True return False async def update(self, key: str, updater: Callable, author: str ) - Any: 原子性更新读-修改-写 async with self._get_lock(key): current self._state.get(key) new_value updater(current) self._state[key] new_value self._history[key].append({ value: new_value, author: author, timestamp: datetime.now().isoformat() }) return new_value def watch(self, key: str, callback: Callable): 监听状态变化 self._watchers[key].append(callback) def _get_lock(self, key: str) - asyncio.Lock: if key not in self._locks: self._locks[key] asyncio.Lock() return self._locks[key] def snapshot(self) - dict: 获取当前状态快照 return dict(self._state)# 多Agent使用共享状态shared_state SharedStateStore()class ParallelResearchAgent: 并行研究Agent共享发现结果 def __init__(self, agent_id: str, specialization: str): self.id agent_id self.spec specialization async def research_and_share(self, topic: str): 执行研究并将结果写入共享状态 # 声明开始工作避免重复 claimed await shared_state.compare_and_swap( ftask.{self.spec}.status, None, # 期望值未被认领 in_progress, # 新值 authorself.id ) if not claimed: print(f{self.id}: 任务已被其他Agent认领跳过) return try: # 执行研究 result await self._research(topic, self.spec) # 写入结果 await shared_state.set(ffindings.{self.spec}, { content: result, agent: self.id, timestamp: datetime.now().isoformat() }) await shared_state.set(ftask.{self.spec}.status, completed) except Exception as e: await shared_state.set(ftask.{self.spec}.status, failed) await shared_state.set(ftask.{self.spec}.error, str(e)) raise async def _research(self, topic: str, aspect: str) - str: from openai import AsyncOpenAI client AsyncOpenAI() response await client.chat.completions.create( modelgpt-4o, messages[{ role: user, content: f关于{topic}重点分析{aspect}方面500字以内 }] ) return response.choices[0].message.content## 三、LangGraph状态机实现pythonfrom langgraph.graph import StateGraph, ENDfrom typing import TypedDict, Annotated, Listimport operatorclass ResearchState(TypedDict): 工作流状态定义 topic: str research_results: Annotated[List[str], operator.add] # 累积列表 draft: str review_feedback: str final_content: str iteration_count: int status: strdef create_content_pipeline() - StateGraph: 创建内容生产流水线 workflow StateGraph(ResearchState) # 添加节点 workflow.add_node(researcher, research_node) workflow.add_node(writer, writer_node) workflow.add_node(reviewer, reviewer_node) workflow.add_node(reviser, reviser_node) # 设置入口 workflow.set_entry_point(researcher) # 定义边 workflow.add_edge(researcher, writer) workflow.add_edge(writer, reviewer) # 条件边审查通过则结束否则继续修改 workflow.add_conditional_edges( reviewer, should_revise, { revise: reviser, approve: END } ) workflow.add_edge(reviser, reviewer) return workflow.compile()async def research_node(state: ResearchState) - dict: 研究节点 from openai import AsyncOpenAI client AsyncOpenAI() response await client.chat.completions.create( modelgpt-4o, messages[{ role: user, content: f请对{state[topic]}进行深入研究提供5个核心观点。 }] ) return { research_results: [response.choices[0].message.content], status: researched }async def writer_node(state: ResearchState) - dict: 写作节点 from openai import AsyncOpenAI client AsyncOpenAI() research_summary \n.join(state[research_results]) response await client.chat.completions.create( modelgpt-4o, messages[{ role: user, content: f基于以下研究写一篇2000字的技术文章\n\n{research_summary} }] ) return { draft: response.choices[0].message.content, status: drafted }async def reviewer_node(state: ResearchState) - dict: 审查节点 from openai import AsyncOpenAI client AsyncOpenAI() response await client.chat.completions.create( modelgpt-4o, messages[{ role: user, content: f审查以下文章第{state.get(iteration_count, 0)1}次审查{state[draft]}输出格式通过/修改: [通过 或 需要修改]反馈: [具体的修改建议如果通过则为空] }] ) review response.choices[0].message.content approved 通过 in review and 需要修改 not in review return { review_feedback: review, final_content: state[draft] if approved else , iteration_count: state.get(iteration_count, 0) 1, status: approved if approved else needs_revision }def should_revise(state: ResearchState) - str: 决定是否需要修改 # 最多修改3次防止无限循环 if state.get(iteration_count, 0) 3: return approve # 强制通过 return revise if state[status] needs_revision else approve## 四、通信协议最佳实践### 4.1 消息格式标准pythonfrom pydantic import BaseModel, Fieldfrom typing import Literal, Unionclass TaskMessage(BaseModel): type: Literal[task] task_id: str task_type: str input: dict priority: int Field(default5, ge1, le10) deadline: Optional[str] Noneclass ResultMessage(BaseModel): type: Literal[result] task_id: str # 关联原任务 success: bool output: Any error: Optional[str] None metadata: dict {}class StatusMessage(BaseModel): type: Literal[status] agent_id: str status: Literal[idle, busy, error, shutting_down] current_task: Optional[str] None queue_depth: int 0# 联合类型用于消息路由AgentMessagePayload Union[TaskMessage, ResultMessage, StatusMessage]### 4.2 通信模式选型建议任务场景 → 推荐模式1. 流水线处理A→B→C依次 → 直接消息传递 明确的输入/输出格式2. 扇出/扇入一个任务分多个子任务并行 → 消息总线 结果聚合等待3. 事件驱动协作状态变化触发操作 → 发布/订阅模式4. 竞争执行多个Agent抢同一批任务 → 共享状态 CAS原子操作5. 复杂状态机多轮交互、条件分支 → LangGraph状态机6. 大规模生产100个Agent → 外部消息队列Redis Stream/Kafka/RabbitMQ### 4.3 生产级消息队列集成pythonimport redis.asyncio as redisimport jsonclass RedisAgentQueue: 基于Redis Stream的生产级Agent消息队列 def __init__(self, redis_url: str redis://localhost:6379): self.redis redis.from_url(redis_url) async def publish(self, stream: str, message: dict): 发布消息到Redis Stream msg_data {k: json.dumps(v) if not isinstance(v, str) else v for k, v in message.items()} await self.redis.xadd(stream, msg_data) async def consume(self, stream: str, consumer_group: str, consumer_id: str, count: int 1) - list: 消费消息支持消费者组自动负载均衡 try: # 创建消费者组如果不存在 await self.redis.xgroup_create(stream, consumer_group, 0, mkstreamTrue) except Exception: pass # 组已存在 messages await self.redis.xreadgroup( consumer_group, consumer_id, {stream: }, countcount, block5000 # 等待5秒 ) result [] for stream_name, msgs in (messages or []): for msg_id, msg_data in msgs: parsed {k: self._try_parse(v) for k, v in msg_data.items()} result.append((msg_id, parsed)) return result async def ack(self, stream: str, consumer_group: str, msg_id: str): 确认消息处理完成 await self.redis.xack(stream, consumer_group, msg_id) def _try_parse(self, value): try: return json.loads(value) except Exception: return value## 五、总结多Agent通信架构的选择直接决定系统的可维护性和可扩展性| 规模 | 推荐方案 ||------|---------|| 2-5个Agent原型 | 直接函数调用 共享Python对象 || 5-20个Agent单机 | 异步消息总线 LangGraph || 20个Agent分布式 | Redis Stream / Kafka 事件溯源 |核心原则1. 消息格式要标准化用Pydantic验证2. 总是处理超时和失败不要假设通信成功3. 实现幂等性允许消息重复投递4. 从简单开始确认需求再引入消息队列5. 可观测性先行每条消息都要可追踪