第 29 章：消息为什么是唯一接口

发表于 分类于 阅读次数： 本文字数： 4.8k 阅读时长 ≈ 4 分钟
深入剖析 AgentScope 为何选择单一 Msg 类作为全框架唯一通信接口，而非多种消息类型或纯字符串方案。通过对比 LangChain 的多类型设计和 AutoGen 的字典方案，分析统一接口在序列化、扩展性和互操作方面的优势，以及运行时类型检查和角色语义模糊的代价。
个人公众号

难度：中等

AgentScope 中，Agent、Model、Tool、Memory 全部通过 Msg 对象通信。为什么不让 Agent 直接返回字符串？为什么不定义多种消息类型？

上一章：第 28 章 终章——集成实战

决策回顾

打开 src/agentscope/message/_message_base.py:21：

1
2
3
4
5
6
7
8
class Msg:
    def __init__(self, name, content, role, metadata=None, ...):
        self.name = name           # 发送者名称
        self.content = content     # str 或 ContentBlock 列表
        self.role = role           # user / assistant / system
        self.metadata = metadata   # 结构化输出等附加数据
        self.id = shortuuid.uuid()
        self.timestamp = ...

Msg 是系统中唯一的消息类型。所有模块通过它传递数据：

1
2
3
4
5
6
7
8
9
10
flowchart LR
    A["用户"] -->|"Msg"| B["Agent"]
    B -->|"Msg"| C["Memory"]
    B -->|"Msg 列表"| D["Formatter"]
    D -->|"dict 列表"| E["Model"]
    E -->|"ChatResponse"| D
    D -->|"Msg"| B
    B -->|"ToolUseBlock"| F["Toolkit"]
    F -->|"ToolResponse"| B
    B -->|"Msg"| A

注意：Msg 在 Agent ↔ Memory、Agent ↔ Formatter 之间传递。Model 不直接接收 Msg，而是接收 Formatter 转换后的 dict。

被否方案一：多种消息类型

方案：为不同场景定义不同的消息类型：

1
2
3
4
class UserMessage: ...
class AssistantMessage: ...
class ToolMessage: ...
class SystemMessage: ...

LangChain 就是这样做的——HumanMessage、AIMessage、ToolMessage、SystemMessage 各一个类。

问题：

  1. 类型爆炸：每种消息需要独立的序列化/反序列化逻辑
  2. 接口割裂：Memory 的 add() 方法需要处理 4 种类型：
    1
    2
    async def add(self, msg: UserMessage | AssistantMessage | ToolMessage | SystemMessage):
        ...
  3. 转换成本：Agent 需要在 Msg → ModelMessage → Msg 之间来回转换

AgentScope 的选择：一个 Msg 类 + role 字段区分角色。content 字段统一承载所有内容类型（文本、工具调用、图片……）。

被否方案二：纯字符串

方案：Agent 直接返回字符串：

1
2
async def reply(self, msg: str) -> str:
    ...

问题：

  1. 无法携带工具调用：模型返回的 ToolUseBlock 放不进字符串
  2. 无法携带元数据：token 用量、结构化输出等无处存放
  3. 无法追踪来源：不知道消息来自哪个 Agent

AgentScope 的选择：content 可以是字符串（简单场景）或 ContentBlock 列表（复杂场景）：

1
2
3
4
5
6
7
8
# 简单场景
Msg("user", "你好", "user")

# 复杂场景：文本 + 工具调用
Msg("assistant", [
    TextBlock(type="text", text="我来查一下天气"),
    ToolUseBlock(type="tool_use", name="get_weather", input={"city": "北京"}),
], "assistant")

后果分析

好处

  1. 统一接口：所有模块只需理解一种类型
  2. 序列化简单：to_dict() / from_dict() 只需实现一次
  3. 扩展容易：添加新的 ContentBlock 类型不需要改 Msg 本身
  4. 泛化能力：Msg 可以承载文本、图片、音频、视频、工具调用

麻烦

  1. 类型不够严格：content 是 str | list[ContentBlock]，需要运行时检查
  2. 角色语义模糊：role 只有三个值，但实际消息的语义更丰富（工具结果是 “user” 角色，但这只是 OpenAI API 的约定）
  3. Metadata 滥用风险：metadata 是自由字典，可能变成”什么都能往里塞”的垃圾桶

横向对比

框架消息类型优点缺点
AgentScope1 个 Msg 类接口简单类型不够严格
LangChain4+ 消息类类型安全类型爆炸，转换成本
AutoGen字典灵活无类型检查
CrewAI字符串 + 元组极简无法承载复杂内容

AgentScope 官方文档的 Basic Concepts > Message 页面详细展示了 Msg 的创建方法和 7 种 ContentBlock 类型（TextBlock、ThinkingBlock、ImageBlock、AudioBlock、VideoBlock、ToolUseBlock、ToolResultBlock），并说明了 Msg 在 Agent、用户和工具之间传递信息的核心作用。

AgentScope 1.0 论文对这一设计的说明是：

“we abstract foundational components essential for agentic applications and provide unified interfaces and extensible modules”

— AgentScope 1.0: A Comprehensive Framework for Building Agentic Applications, arXiv:2508.16279, Section 2

统一消息格式（unified message format）是框架的核心设计目标——所有组件通过同一种 Msg 类型通信，确保无缝互操作。

验证性实验：测试 Msg 作为唯一接口

这个实验不需要 API key。

目标：验证所有组件确实通过 Msg 通信。

步骤：

  1. 在 src/agentscope/agent/_agent_base.py 的 __call__ 中加一行 print(f"[接口] Agent 收到: {type(msg).__name__}")。

  2. 用同样的方法在 Model、Memory、Toolkit 的入口打印接收参数的类型。

  3. 运行天气 Agent，观察所有组件是否都接收 Msg 类型。

思考：如果将来要添加一个新组件（如 Planner），它是否也应该用 Msg 做接口？用 Msg 的好处和限制分别是什么？

你的判断

开放性问题：

  1. 如果要支持”消息路由”（根据消息类型分发到不同处理器），Msg 的单一类型是否够用？
  2. metadata 字段是否应该用 Pydantic BaseModel 替代 dict，以获得类型检查？

Msg 的支撑 API

“唯一接口”的可行性依赖于 Msg 自身提供足够的工具方法。打开 _message_base.py：

类型安全的内容访问

get_content_blocks() 用 @overload 实现类型安全的返回值：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# _message_base.py:149-229
@overload
def get_content_blocks(self, block_type: Literal["text"]) -> Sequence[TextBlock]: ...

@overload
def get_content_blocks(self, block_type: Literal["tool_use"]) -> Sequence[ToolUseBlock]: ...

def get_content_blocks(self, block_type=None) -> Sequence[ContentBlock]:
    blocks = []
    if isinstance(self.content, str):
        blocks.append(TextBlock(type="text", text=self.content))
    else:
        blocks = self.content or []
    if isinstance(block_type, str):
        blocks = [_ for _ in blocks if _["type"] == block_type]
    return blocks

调用 msg.get_content_blocks("tool_use") 时，类型检查器知道返回的是 Sequence[ToolUseBlock]——字段 name、input 都可以直接访问。

便捷文本提取

get_text_content()（第 123 行）把 content 的双态统一为字符串：

1
2
3
# content = "hello" → 返回 "hello"
# content = [TextBlock(text="a"), TextBlock(text="b")] → 返回 "a\nb"
# content = [ToolUseBlock(...)] → 返回 None（无文本块）

序列化

to_dict()（第 75 行）和 from_dict()（第 86 行）直接序列化/反序列化所有字段。因为 ContentBlock 是 TypedDict（就是 dict），content 字段天然支持 JSON——不需要额外的转换逻辑。

1
2
3
4
5
6
msg = Msg("user", [TextBlock(type="text", text="你好")], "user")
d = msg.to_dict()
# {"id": "...", "name": "user", "content": [{"type": "text", "text": "你好"}], ...}

msg2 = Msg.from_dict(d)
# 完整恢复，包括 id

这套 API 让 Msg 作为”唯一接口”不是一句空话——它提供了足够的方法让消费者不需要知道 content 的具体结构。

设计一瞥：get_content_blocks 的 @overload 是一种”穷人版多态”——不用子类也能实现类型安全的返回值。这和第 33 章的 TypedDict Union 选择一脉相承：用类型系统（Union + overload）代替 OOP 继承来提供多态。

下一章预告

Msg 是统一接口。但工具函数的注册方式——为什么是 toolkit.register_tool_function(func) 而不是在函数上加 @tool 装饰器？下一章我们看注册方式的选择。

下一章：第 30 章 为什么不用装饰器注册工具