Agent 间通信模式

章节：专家篇第13章第2节
难度：⭐⭐⭐⭐ 专家级
阅读时间：25-30 分钟
核心收获：掌握多Agent间的通信方式和协议设计

16.6 理论要点：Agent通信的本质

16.6.1 为什么需要通信协议

问题：多个Agent独立运行，如何协调工作？

通信协议的价值： - 标准化消息格式 - 明确职责边界 - 支持异步解耦 - 便于监控和调试

16.6.2 通信模式分类

16.7 实战案例：消息队列实现

16.7.1 场景：热点选题任务的消息流

Planner → Executor1: {task: "fetch_wechat"}
Planner → Executor2: {task: "fetch_github"}
Planner → Executor3: {task: "fetch_search"}
Planner → Executor4: {task: "fetch_aihot"}

Executor1 → Aggregator: {source: "wechat", data: [...]}
Executor2 → Aggregator: {source: "github", data: [...]}
Executor3 → Aggregator: {source: "search", data: [...]}
Executor4 → Aggregator: {source: "aihot", data: [...]}

Aggregator → Filter: {items: [...], sources: {...}}
Filter → Formatter: {filtered_items: [...]}
Formatter → Deliver: {formatted_output: "..."}

16.7.2 消息格式设计

{
  "message_id": "uuid",
  "correlation_id": "batch-uuid",
  "sender": "planner",
  "receiver": "executor_wechat",
  "timestamp": "2026-07-10T09:00:00+08:00",
  "type": "task",
  "priority": "high",
  "payload": {
    "task": "fetch_hotspots",
    "params": {
      "date": "2026-07-10",
      "max_items": 50
    }
  },
  "metadata": {
    "timeout": 30,
    "retry_count": 0,
    "max_retries": 2
  }
}

16.7.3 消息队列实现

import asyncio
from collections import deque

class MessageQueue:
    def __init__(self):
        self.queues = {}
        self.handlers = {}

    def register(self, agent_id, handler):
        """注册Agent和处理器"""
        self.queues[agent_id] = deque()
        self.handlers[agent_id] = handler

    async def send(self, to, message):
        """发送消息"""
        if to in self.queues:
            self.queues[to].append(message)
            return True
        return False

    async def process(self, agent_id):
        """处理消息"""
        while self.queues[agent_id]:
            message = self.queues[agent_id].popleft()
            await self.handlersagent_id

16.8 图文说明

📌 待补充：建议绘制以下示意图 1. 五种通信模式对比图 2. 消息序列图 3. 消息队列架构图 4. 错误处理流程图

16.9 常见错误

📝 版本迭代记录