第2章协作与通信¶

学习目标¶

理解 Agent 间通信协议（A2A）的设计要素与消息格式
掌握队列、共享内存、事件驱动三类通信机制及其适用场景
区分同步、异步、广播三种通信模式并能够正确选型
使用 Python 实现基于消息队列的 Agent 间通信
识别通信瓶颈并设计可扩展的分布式 Agent 通信架构

问题导入¶

某物流平台部署了订单、库存、配送、风控四个 Agent。最初采用直接函数调用（同步）方式：订单 Agent 调用库存 Agent，库存 Agent 调用配送 Agent……结果一旦配送 Agent 响应慢，整个链路阻塞，订单超时严重。更糟的是，新增"营销 Agent"需要修改订单 Agent 的代码才能接入。这说明：Agent 间的通信方式不是"能跑就行"的小事，而是决定系统可扩展性与健壮性的根基。本章系统讲解多 Agent 通信机制，帮你设计解耦、可扩展的协作架构。

核心概念¶

1. A2A（Agent-to-Agent）通信协议¶

A2A 通信协议定义 Agent 间交换消息的格式与规则。一条 A2A 消息通常包含：

字段	说明	示例
`sender`	发送方 Agent ID	`agent://order-001`
`receiver`	接收方 Agent ID 或 `broadcast`	`agent://inventory-002`
`msg_type`	消息类型	`request` / `response` / `event` / `handoff`
`content`	消息体（JSON/文本）	`{"sku": "A1", "qty": 10}`
`trace_id`	链路追踪 ID	`trace-abc123`
`timestamp`	时间戳	`2026-07-01T10:00:00Z`
`reply_to`	回复目标（可选）	`agent://order-001`

协议设计原则： - 自描述：消息含足够元数据，接收方无需额外上下文即可处理。 - 幂等：同一消息重复投递不应产生副作用（带 msg_id 去重）。 - 可追溯：每条消息带 trace_id，串联全链路日志。

2. 三类通信机制¶

直接调用（Direct Call）：Agent A 直接调用 Agent B 的函数/方法。最简单，但强耦合、易阻塞、难扩展。适合单机、简单、同步场景。

共享内存（Shared Memory / Blackboard）：所有 Agent 读写同一块共享状态空间（如 Redis、数据库）。Agent 间无直接耦合，通过"观察黑板"协同。适合需要共享大量上下文、Agent 数量适中的场景。

消息队列（Message Queue）：Agent 间通过队列（如 RabbitMQ、Kafka、Redis Stream）异步传递消息。发送方不等待接收方，天然解耦、可削峰、可扩展。适合分布式、高吞吐、需容错的场景。

3. 三种通信模式¶

模式	描述	优点	缺点	典型场景
同步	A 调用 B 并等待返回	简单、易调试	阻塞、耦合高	简单流水线、单机
异步	A 发消息给 B 后继续工作，B 完成后回调/通知	解耦、高吞吐	流程复杂、需回调	分布式、长任务
广播	A 把消息发给所有相关 Agent	联动性强	消息放大、需过滤	事件通知、状态同步

4. 事件驱动通信¶

事件驱动是异步通信的高级形式：Agent 不直接"调用"对方，而是"发布事件"，订阅了该事件的 Agent 自动响应。

事件总线（Event Bus）：中心化事件分发器，Agent 注册订阅、发布事件。
发布订阅（Pub/Sub）：解耦的经典模式，发布者不关心谁订阅。
适用场景：库存变化通知营销暂停、订单完成触发风控审计、异常告警多 Agent 联动。

5. 通信模式选型决策¶

任务线性、单机、低延迟要求 → 同步直接调用（CrewAI 顺序流程即此类）。
任务可并行、需解耦、分布式 → 异步消息队列。
一变多动、需多 Agent 联动 → 事件驱动广播。
需共享大上下文、Agent 数适中 → 共享黑板。

原理图¶

flowchart TB
    subgraph 同步调用
        S1[Agent A] -->|阻塞等待| S2[Agent B]
        S2 -->|返回| S1
    end

    subgraph 异步队列
        A1[Agent A] -->|发送消息| Q[(消息队列)]
        Q -->|投递| A2[Agent B]
        A2 -->|回调/事件| A1
    end

    subgraph 共享黑板
        B1[Agent X] -->|写| BB[(共享黑板)]
        B2[Agent Y] -->|读| BB
        B3[Agent Z] -->|读写| BB
    end

    subgraph 事件驱动广播
        E1[Agent A] -->|发布事件| EB((事件总线))
        EB -->|通知| E2[Agent B]
        EB -->|通知| E3[Agent C]
        EB -->|通知| E4[Agent D]
    end

图：四种多 Agent 通信机制对比

示例代码¶

使用 Redis 实现基于消息队列的异步 Agent 通信，演示订单 Agent 与库存 Agent 的解耦协作。

import json
import time
import uuid
import redis


class AgentMessage:
    """A2A 消息封装。"""

    def __init__(self, sender, receiver, msg_type, content, trace_id=None):
        self.msg_id = str(uuid.uuid4())
        self.sender = sender
        self.receiver = receiver
        self.msg_type = msg_type  # request / response / event
        self.content = content
        self.trace_id = trace_id or self.msg_id
        self.timestamp = time.time()

    def to_json(self):
        return json.dumps(self.__dict__, ensure_ascii=False)

    @classmethod
    def from_json(cls, s):
        d = json.loads(s)
        msg = cls(d["sender"], d["receiver"], d["msg_type"], d["content"], d.get("trace_id"))
        msg.msg_id = d["msg_id"]
        msg.timestamp = d.get("timestamp", time.time())
        return msg


class QueueAgent:
    """基于 Redis 队列的异步 Agent 基类。"""

    def __init__(self, agent_id, redis_client):
        self.agent_id = agent_id
        self.r = redis_client
        self.inbox = f"queue:{agent_id}"

    def send(self, receiver, msg_type, content, trace_id=None):
        """向其他 Agent 发送消息（异步，不阻塞）。"""
        msg = AgentMessage(self.agent_id, receiver, msg_type, content, trace_id)
        self.r.rpush(f"queue:{receiver}", msg.to_json())
        print(f"[{self.agent_id}] → [{receiver}] {msg_type}: {content}")
        return msg.msg_id

    def receive(self, timeout=5):
        """阻塞等待一条消息。"""
        result = self.r.blpop(self.inbox, timeout=timeout)
        if result is None:
            return None
        _, raw = result
        return AgentMessage.from_json(raw)

    def handle(self, msg):
        """子类重写：处理收到的消息。"""
        raise NotImplementedError


class OrderAgent(QueueAgent):
    def handle(self, msg):
        if msg.msg_type == "response":
            print(f"[{self.agent_id}] 收到库存回复: {msg.content}")
        elif msg.msg_type == "event":
            print(f"[{self.agent_id}] 收到事件: {msg.content}")


class InventoryAgent(QueueAgent):
    def handle(self, msg):
        if msg.msg_type == "request" and msg.content.get("action") == "check_stock":
            sku, qty = msg.content["sku"], msg.content["qty"]
            # 模拟库存检查
            available = qty <= 100
            self.send(msg.sender, "response",
                      {"sku": sku, "available": available}, msg.trace_id)


if __name__ == "__main__":
    r = redis.Redis(host="localhost", port=6379, decode_responses=True)
    # 清空测试队列
    r.flushdb()

    order = OrderAgent("order-001", r)
    inventory = InventoryAgent("inventory-002", r)

    # 订单 Agent 异步发送库存检查请求
    trace = order.send("inventory-002", "request",
                       {"action": "check_stock", "sku": "A1", "qty": 5})

    # 库存 Agent 接收并处理
    msg = inventory.receive(timeout=5)
    if msg:
        inventory.handle(msg)

    # 订单 Agent 接收回复
    reply = order.receive(timeout=5)
    if reply:
        order.handle(reply)

事件驱动广播示例（发布订阅）：

import redis

r = redis.Redis(host="localhost", port=6379, decode_responses=True)

# 订阅端：营销 Agent 监听库存事件
def marketing_listener():
    pubsub = r.pubsub()
    pubsub.subscribe("inventory.events")
    for msg in pubsub.listen():
        if msg["type"] == "message":
            print(f"[营销 Agent] 收到库存事件: {msg['data']}, 暂停推广")

# 发布端：库存 Agent 发布事件
r.publish("inventory.events", '{"sku":"A1","event":"low_stock","qty":2}')

企业案例¶

案例 1：分布式订单系统的异步改造¶

某电商平台原用同步调用串联订单、库存、配送 Agent，大促期间链路阻塞频发。改造为基于 Kafka 的异步消息队列：订单 Agent 发出"下单事件"后立即返回，库存、配送、风控 Agent 各自订阅处理。改造后订单接口 P99 延迟从 3.2s 降至 200ms，系统吞吐提升 8 倍。

案例 2：智能运营中心的共享黑板¶

某零售品牌部署售前、售后、库存、营销四个 Agent，通过 Redis 共享黑板交换状态。库存 Agent 把"即将断货"写入黑板，营销 Agent 轮询发现后自动暂停该商品推广，售前 Agent 读取后调整话术。黑板模式让 Agent 间无需直接调用即可联动，新增 Agent 只需读写黑板即可接入。

案例 3：风控审计的事件驱动联动¶

某金融平台部署风控 Agent 监听全平台"大额交易事件"。当事件发生，风控 Agent、合规 Agent、通知 Agent 同时响应：风控评估风险、合规记录留痕、通知推送人工复核。事件驱动让一次发布触发多 Agent 并行处理，避免串行调用造成的延迟。

常见问题¶

Q1：什么时候该用消息队列而不是直接调用？ A：当出现以下任一情况：①Agent 分布在不同进程/机器；②需要解耦以便独立扩展；③调用链长、需削峰填谷；④需保证消息持久化与重试。否则简单任务用直接调用更高效。

Q2：共享黑板会不会有并发冲突？ A：会。多 Agent 同时写黑板需用乐观锁（CAS）或分布式锁，避免覆盖。读多写少场景适合黑板，写冲突频繁的场景更适合队列。

Q3：广播会不会导致消息风暴？ A：可能。需通过主题过滤（topic）、消息 TTL、订阅者按需订阅来控制。生产环境建议用结构化主题（如 inventory.low_stock.*）精确路由。

Q4：如何保证消息不丢失？ A：使用支持持久化的队列（Kafka/RabbitMQ 持久化队列）、消费端确认机制（ack）、生产端重试与死信队列。关键业务消息需做幂等处理。

面试题¶

1. 对比同步、异步、广播三种 Agent 通信模式的优缺点与适用场景。 参考答案：同步简单易调试但阻塞耦合，适合单机简单流水线；异步解耦高吞吐但需回调与状态管理，适合分布式长任务；广播联动性强但消息放大，适合事件通知。生产环境多采用异步+事件驱动组合。

2. 设计一个支持水平扩展的多 Agent 通信架构。 答题要点：用消息队列（Kafka/Redis Stream）解耦；Agent 无状态化、可多副本；用 trace_id 串联链路；用服务发现动态注册 Agent；用负载均衡分发消息；关键消息持久化与重试。

3. A2A 消息协议应包含哪些关键字段？为什么？ 参考答案：sender/receiver 标识收发方；msg_type 区分请求/响应/事件；content 携带业务数据；trace_id 用于全链路追踪；msg_id 用于幂等去重；timestamp 用于时序与超时判断。

练习题¶

通信机制实操：用 Redis Stream 实现三个 Agent（A→B→C）的异步流水线，每步处理耗时不同，观察总耗时与同步方式的差异。
事件驱动设计：为"电商异常监控"场景设计事件主题与订阅关系，画出事件流图，说明哪些 Agent 订阅哪些事件。
思考题：如果消息队列宕机，多 Agent 系统应如何降级以保证核心业务可用？

小结¶

多 Agent 通信是协作的基石。同步直接调用简单但耦合高；异步消息队列解耦可扩展，是分布式生产环境的首选；共享黑板适合共享上下文；事件驱动广播适合多 Agent 联动。A2A 协议的自描述、幂等、可追溯三原则是设计健壮通信的关键。下一章将在此基础上讲解如何进行任务分工与协调。延伸阅读：Redis Streams 文档、Kafka 设计文档、论文《Multi-Agent Communication Protocols》。

第2章 协作与通信¶