多Agent系统通信协议设计2026：从消息队列到共享状态的完整工程指南

单个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, Empty@dataclassclass AgentMessage: """标准Agent间消息格式""" message_id: str = field(default_factory=lambda: str(uuid.uuid4())) sender_id: str = "" recipient_id: str = "" message_type: str = "task" # task/result/status/error payload: Any = None metadata: Dict = field(default_factory=dict) timestamp: str = field(default_factory=lambda: 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_id=sender_id, recipient_id=self.sender_id, message_type="result", payload=payload, correlation_id=self.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(maxsize=100) async def send(self, message: AgentMessage): """发送消息到指定Agent""" if message.recipient_id not in self._queues: raise ValueError(f"Agent {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_id=message.sender_id, recipient_id=agent_id, message_type=message.message_type, payload=message.payload, correlation_id=message.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(), timeout=timeout ) except asyncio.TimeoutError: raise TimeoutError(f"Agent {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_id=self.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( model="gpt-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"}, source="orchestrator") 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"}, source="orchestrator") async def on_writing_done(self, event: dict): self.state["draft"] = event["data"]["draft"] # 触发审查 await self.bus.publish("task.review", {"draft": self.state["draft"]}, source="orchestrator") 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( f"task.{self.spec}.status", None, # 期望值：未被认领 "in_progress", # 新值 author=self.id ) if not claimed: print(f"{self.id}: 任务已被其他Agent认领，跳过") return try: # 执行研究 result = await self._research(topic, self.spec) # 写入结果 await shared_state.set(f"findings.{self.spec}", { "content": result, "agent": self.id, "timestamp": datetime.now().isoformat() }) await shared_state.set(f"task.{self.spec}.status", "completed") except Exception as e: await shared_state.set(f"task.{self.spec}.status", "failed") await shared_state.set(f"task.{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( model="gpt-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( model="gpt-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( model="gpt-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( model="gpt-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(default=5, ge=1, le=10) 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", mkstream=True) except Exception: pass # 组已存在 messages = await self.redis.xreadgroup( consumer_group, consumer_id, {stream: ">"}, count=count, block=5000 # 等待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. 可观测性先行，每条消息都要可追踪