2.4 通信机制与消息传递

2.4 通信机制与消息传递

在体系中的位置：上一节讲了 Agent 长什么样、怎么被框架管理，这一节回答"多个 Agent 之间到底怎么说话"。它决定了你搭出来的多智能体系统是"几个人各说各话"，还是"一支能接话茬、能传话、能分工的队伍"。

先说一个常见的翻车现场。有人用三个 Agent 搭了一个"调研小分队"：A 负责搜资料，B 负责写综述，C 负责查漏。跑起来之后发现 B 拿到的是 A 第一次的输出——A 后来修正过的结论根本没到 B 手里；C 等 B 等得超时，直接空手返回，把"没找到问题"当成"没问题"写进了报告。排查到最后，问题不在任何单个 Agent 的逻辑，而在消息这一层：没人约定"消息长什么样、发给谁、丢了怎么办"。这一节就是把这三件事讲清楚。

一条消息的最小结构：先让"说话"有格式

AgentScope 里所有交互都走消息对象 Msg，它不让你发裸字符串。原因很实际：多智能体系统里一条消息要被路由、被记录、被回放、被不同 Agent 理解，没有固定结构，这些全都做不了。

from agentscope.message import Msg

# 一条最小可用的消息：谁说的、说了什么、以什么角色说的
msg = Msg(
    name="researcher",          # 发送者标识
    content="已确认 LLaMA-3.1 的上下文窗口是 128K",
    role="assistant",           # assistant / user / system，用于区分视角
)

# 调试时最常用的一招：直接打印看整条消息
print(msg)
​

结构上只有三样是必须的：name（谁发的）、content（说了什么）、role（以什么身份说的）。剩下的都塞进 metadata——时间戳、任务编号、置信度，随你用。这里有个容易被忽略的设计点：消息里没有"发给谁"这个字段。这不是遗漏，是刻意的。在 AgentScope 里，发送者只说"我有什么要告诉大家"，至于谁接、怎么接，是下一层的事。

消息怎么走：总线、路由与三种典型模式

"发给谁"由框架的消息总线负责。你可以把总线想成办公室的公共白板：有人在上面贴便签，不指名道姓；谁关心这张便签，谁自己过来看。这种"贴上去"而不是"塞到你手里"的设计，换来的是 Agent 之间彻底解耦——新增一个 Agent 只需要让它订阅它关心的消息类型，不用改任何既有代码。

路由靠消息类型区分。常见三种，按场景选：

# 广播示例：调度器通知所有 Agent 换用新数据源
from agentscope.message import Msg

broadcast = Msg(
    name="dispatcher",
    content="数据源已切换为 v2，请清空各自缓存",
    role="system",
    metadata={"type": "data_source_change", "version": "v2"},
)
for agent in agents:
    agent.observe(broadcast)   # observe 是 Agent 接收消息的统一入口
​

流水线接力是发布-订阅用得最顺的场景：A 发布"草稿完成"，B 订阅草稿主题写评论，C 订阅评论主题定终稿。每个 Agent 只关心自己订阅的那一跳，中间的增删改完全不动其他人。这也是上一节说的"以消息为中心"在协作场景下的直接体现。

同步还是异步：先把"等不等"想清楚

通信模式里最影响系统形态的决策，是等还是不等。同步请求-响应写起来最省心——发一条、等一条、拿到就用，适合"这一步的产出是下一步的输入"的强依赖场景。但代价是等待期间这个 Agent 干不了别的，多个强依赖串起来，整条链路的耗时就是各段之和。

异步则让 Agent 发完消息就继续干自己的活，回复到了再处理。吞吐上去了，但你要自己处理两件麻烦事：一是消息可能丢（进程崩了、队列满了），二是回复可能乱序（先发的后到）。工程上通常的折中是：异步发送 + 在消息里带序号或任务 ID，接收方按 ID 归并，超时未到就当失败重发。下面这段示意了"带任务 ID 的异步归并"这个套路：

import uuid
from agentscope.message import Msg

task_id = str(uuid.uuid4())

# 发信方：带上任务 ID，不阻塞等待
req = Msg(name="planner", content="请评估方案 A 的落地成本",
          metadata={"task_id": task_id})
agent_b.observe(req)

# 收信方逻辑（示意）：回复时原样带回 task_id
reply = Msg(name="estimator", content="约 12 人天",
            metadata={"task_id": task_id, "reply_to": "planner"})
​

判断口诀：如果两个 Agent 必须按顺序接力，用同步；如果它们可以各自推进、偶尔汇合，用异步加任务 ID。 见过太多系统先把所有通信改成异步图"性能"，结果调试时连"这条消息到底是谁发的"都查不出来——异步省下的时间全赔在排查上了。

消息会出什么事故：三个高频坑

第一个坑是把日志当消息。有人为了让 Agent"记得"某件事，往系统日志里写一行，指望另一个 Agent 去翻日志——这不是通信，这是碰运气。通信必须走 Msg，日志只留给人类排查用。

第二个坑是广播上瘾。所有更新都广播，Agent 少的时候没事，Agent 一多，每个 Agent 都被无关消息轰炸，observe 里全是过滤逻辑。正确姿势是广播只用于"群体状态变化"这类真的人人都要知道的事，定向信息一律点对点或订阅。

第三个坑是丢了回复的关联。异步模式下，如果不带任务 ID 就发多条请求，回复回来根本不知道是对应哪一条。方案就是上面那段代码里的套路：metadata 里带 task_id 和 reply_to，一劳永逸。这个习惯养成之后，排查消息问题会轻松非常多——你永远能从一条消息回答三个问题：谁发的、回给谁、对应哪次请求。

一句话收束

通信层的设计目标就一个：让 Agent 之间只依赖"消息协议"，不依赖彼此的实现。把消息结构定好、把"等还是不等"想清楚、把任务 ID 带上，你的多智能体系统就从"几个聊天的循环"变成了"一支能分工的队伍"。下一节的环境（Environment）会在这个基础上回答另一个问题：除了 Agent 之间，系统怎么和外部世界（工具、数据、用户）交互——通信的边界会从"队伍内部"扩展到"队伍与外界"。