消息与事件

在 AgentScope 2.x 中，Msg 与 Event 是两套独立但紧密耦合的数据单元：Msg 是 Agent 之间通信、上下文 存储与持久化的基本单元，承载"已经完成"的信息；Event 是 reply_stream 产出的流式与 Human-in-the-loop （HITL）单元，承载"正在发生"的过程。一次流式回复本质上是若干 Event 增量地拼装出一个 Msg 的过程。

版本基线：本文以 Python agentscope 2.0.4（PyPI，2026-07-07）为准，要求 Python >=3.11。 2.x 文档统一访问 https://docs.agentscope.io/versions/2.0.4/en/...，未带版本号的 /en/... 路径会 404。 本文所有类名、字段与导入路径均已与 src/agentscope/message/__init__.py、event/__init__.py 及 message/_base.py、message/_block.py 源码核对。

Msg 结构

Msg 是一个 Pydantic BaseModel，定义在 agentscope.message 模块。它不依赖任何运行时状态，可以序列化、 反序列化、跨进程传递，也能直接写入 AgentState 的上下文。

核心字段：

字段	类型	说明
name	str	发送者名称（用户名、Agent 名或 "system"）
content	list[ContentBlock]	消息内容，始终是 block 列表；传入纯字符串会自动包装
role	"user" | "assistant" | "system"	角色，决定允许哪些 block 类型
id	str	消息 ID，默认自动生成；流式重建时必须与 reply_id 对齐
metadata	dict	任意键值元数据，默认空 dict
created_at	str	创建时间（ISO 格式字符串）
finished_at	str | None	完成时间；assistant 消息在 ReplyEndEvent 时才盖戳
usage	Usage | None	Token 用量，含 input_tokens 与 output_tokens

角色与 block 约束

Msg 在 model_validator 中按 role 校验 content，约束来自源码而非文档承诺：

• UserMsg：只允许 TextBlock 与 DataBlock。不能塞 ToolCallBlock 或 ThinkingBlock。
• SystemMsg：只允许 TextBlock。系统消息不支持多模态。
• AssistantMsg：允许全部 ContentBlock 子类型。

试图给 UserMsg 传入 ToolCallBlock 会在构造时抛 ValueError。

构造消息

UserMsg / AssistantMsg / SystemMsg 是三个工厂函数，返回的都是 Msg 实例，只是 role 不同。content 既可以是纯字符串（自动包装成单个 TextBlock），也可以是显式的 block 列表。

import base64

from agentscope.message import (
    UserMsg,
    AssistantMsg,
    SystemMsg,
    TextBlock,
    DataBlock,
    Base64Source,
)

# 1. 纯文本用户消息：字符串自动包装为 TextBlock
user_msg = UserMsg(name="alice", content="帮我分析这张图")
print(user_msg.role)        # "user"
print(user_msg.content[0])  # TextBlock(type="text", text="帮我分析这张图")

# 2. 多模态用户消息：显式 block 列表，含 DataBlock
with open("chart.png", "rb") as f:
    img_b64 = base64.b64encode(f.read()).decode("ascii")

multimodal_msg = UserMsg(
    name="alice",
    content=[
        TextBlock(text="这张图说明了什么？"),
        DataBlock(
            source=Base64Source(data=img_b64, media_type="image/png"),
            name="chart.png",
        ),
    ],
)

# 3. 系统消息：只能是 TextBlock
system_msg = SystemMsg(name="system", content="你是一个数据分析助手")

# 4. 助手消息：通常由 Agent 产出，也可手动构造
assistant_msg = AssistantMsg(name="analyst", content="这张图展示的是 Q3 营收同比变化。")

读取消息内容

Msg 提供两个常用读取方法：

• get_text_content(separator="\n") -> str | None：拼接所有 TextBlock 的文本，没有则返回 None。
• get_content_blocks(block_type=None)：按类型过滤返回 block 列表，不传则返回全部。

from agentscope.message import Msg

# 假设 assistant_msg 已由 Agent 产出
text = assistant_msg.get_text_content()
tool_calls = assistant_msg.get_content_blocks("tool_call")
tool_results = assistant_msg.get_content_blocks("tool_result")

ContentBlock 体系

content 字段里的每个元素都是一个 ContentBlock 子类型。2.x 的 block 体系覆盖了文本、多模态、推理过程、 工具调用与工具结果，是 1.x 旧 Msg 形状的彻底重写。

Block	用途	关键字段
TextBlock	纯文本内容	text: str、id: str
DataBlock	二进制多模态数据（图片、音频、视频）	source: Base64Source | URLSource、name: str | None
ThinkingBlock	模型推理过程（如 Anthropic extended thinking）	thinking: str、id: str（extra="allow" 容纳 signature）
ToolCallBlock	Agent 发起的工具调用	name: str、input: str（原始 JSON）、state: ToolCallState
ToolResultBlock	工具执行结果，与 ToolCallBlock 配对	name: str、output: str | list[TextBlock | DataBlock]、state: ToolResultState
HintBlock	ReAct 循环中给模型的提示，传给 LLM 时转为 user 消息	hint: str | list[TextBlock | DataBlock]、source: str | None

几点源码级细节：

• DataBlock.source 既可以是 Base64Source(data, media_type)，也可以是 URLSource(url, media_type)。 流式 DataBlockDeltaEvent 到达时，框架会逐块 base64 解码、按字节拼接、再重新编码，避免裸字符串拼接 破坏字节流。
• ToolCallBlock.input 是原始 JSON 字符串，不是已解析的 dict。流式过程中 ToolCallDeltaEvent 的 delta 会追加到这个字符串上，全部到达后才得到完整 JSON。
• ToolCallBlock.state 是一个 StrEnum，状态机为 pending → asking / allowed / submitted → finished， 与 Permission 三态决策直接对应。ToolResultBlock.state 则是 success / error / interrupted / denied / running。
• ThinkingBlock 使用 extra="allow"，因此可以承载 provider 特有字段（如 Anthropic 的 signature）， 无需子类化。

Msg 与 Event 的关系

下图说明 Msg 如何由 ContentBlock 组成，以及 Event 流如何增量地把一个 Msg 拼装出来。ReplyStartEvent 负责创建一个空的 AssistantMsg，后续每一个 block 级 event 都通过 Msg.append_event 增量更新这个 Msg， 直到 ReplyEndEvent 盖上 finished_at。

关键关系：

• 一个 Msg 的 content 是一个有序 block 列表，block 的到达顺序就是列表顺序。
• Event 不直接持久化进上下文，而是通过 append_event 把变更落到 Msg 上，最终持久化的是 Msg。
• append_event 会校验 event.reply_id == self.id，不匹配的事件会被跳过并告警。因此流式重建时， AssistantMsg 的 id 必须取自 ReplyStartEvent.reply_id，不能自己生成。

Event 类型

agentscope.event 模块导出约 30 个事件类，全部继承自 EventBase，通过 EventType 枚举区分。下表列出 流式与 HITL 场景中最主要的事件类型，完整列表见 event/__init__.py。

事件	触发时机	携带的关键数据
ReplyStartEvent	一次 reply_stream 开始	name、reply_id
ModelCallStartEvent	调用模型前	reply_id、模型信息
ModelCallEndEvent	单次模型调用结束	input_tokens、output_tokens（累积进 Msg.usage）
TextBlockStartEvent	一段文本块开始	block_id
TextBlockDeltaEvent	文本块增量	block_id、delta
TextBlockEndEvent	文本块结束	block_id
ToolCallStartEvent	模型发起一个工具调用	tool_call_id、tool_call_name
ToolCallDeltaEvent	工具调用入参 JSON 增量	tool_call_id、delta
ToolResultEndEvent	工具执行完成	tool_call_id、state、metadata
ReplyEndEvent	一次回复结束	reply_id、reason: ReplyEndReason
RequireUserConfirmEvent	Permission 决策为 ask，等待用户确认	tool_calls（含 suggested_rules）
UserConfirmResultEvent	用户确认或拒绝	confirm_results
UserInterruptEvent	用户主动中断	中断上下文
ExceedMaxItersEvent	达到最大迭代次数	迭代计数
RequireExternalExecutionEvent	工具需外部执行（submitted 态）	tool_calls
ExternalExecutionResultEvent	外部执行结果回传	execution_results

HintBlockEvent 是一个"一次性"事件——与文本/数据/思考块的 Start-Delta-End 三段式不同，它的完整内容在 单个事件里到达，append_event 直接追加一个完整的 HintBlock。

流式重建 Msg

reply_stream 返回的是一个事件迭代器。要把事件流还原成一个完整的 Msg，标准做法是：在 ReplyStartEvent 时用事件带来的 name 与 reply_id 创建一个空的 AssistantMsg，然后对后续每一个 事件调用 msg.append_event(ev)。

import asyncio
import os

from agentscope.agent import Agent
from agentscope.model import DashScopeChatModel
from agentscope.credential import DashScopeCredential
from agentscope.message import UserMsg, AssistantMsg
from agentscope.event import (
    ReplyStartEvent,
    ReplyEndEvent,
    ExceedMaxItersEvent,
    RequireUserConfirmEvent,
)

# 模型与凭据通过环境变量注入，不要写死在代码里
agent = Agent(
    name="analyst",
    system_prompt="你是一个数据分析助手",
    model=DashScopeChatModel(
        credential=DashScopeCredential(api_key=os.environ["DASHSCOPE_API_KEY"]),
        model="qwen-plus",
    ),
)

user_input = "帮我总结一下今天的销售数据"


async def main() -> None:
    msg: AssistantMsg | None = None

    # reply_stream 是异步事件流，输入是 Msg 而非字符串
    async for event in agent.reply_stream(
        UserMsg(name="user", content=user_input),
    ):
        if isinstance(event, ReplyStartEvent):
            # 用事件的 name 与 reply_id 创建空 Msg，确保 append_event 的 id 校验通过
            msg = AssistantMsg(name=event.name, content=[], id=event.reply_id)

        elif isinstance(event, RequireUserConfirmEvent):
            # HITL 暂停：工具调用进入 asking 态，需要在外部决策后回灌 UserConfirmResultEvent
            print("等待用户确认的工具调用：", [tc.name for tc in event.tool_calls])

        elif isinstance(event, ExceedMaxItersEvent):
            print("达到最大迭代次数，提前结束")

        elif isinstance(event, ReplyEndEvent):
            print("回复结束，原因：", event.reason)

        # 对所有事件统一调用 append_event，让 Msg 增量更新
        if msg is not None:
            msg.append_event(event)

    # 流结束后，msg 就是一个完整的 AssistantMsg，可直接写入上下文或序列化
    if msg is not None:
        print(msg.get_text_content())
        print(msg.usage)  # 累积的 token 用量


asyncio.run(main())

几个容易踩的细节：

• AssistantMsg(name=..., content=[], id=ev.reply_id) 中的 id 必须取自事件，不能省略。省略后 AssistantMsg 会自动生成一个新 ID，与事件的 reply_id 不匹配，append_event 会跳过所有事件并告警。
• append_event 返回 Self（即 msg 本身），是原地变更，可以链式调用，也可以忽略返回值。
• DataBlockDeltaEvent 的处理做了 base64 字节级拼接，不要自己在外面拼字符串再塞回 block。
• ToolCallBlock.input 在全部 ToolCallDeltaEvent 到达前是不完整的 JSON 片段，不要中途解析它。

前端类比

如果你来自前端：

• 把 Msg 类比为 Redux 里的 action 对象 + state 快照：它是一个纯数据结构，记录"谁、说了什么、 用了多少 token"，可以被序列化、回放、存档。role 字段就像 action 的 type，决定了它能在上下文里 承载哪些字段。
• 把 ContentBlock 类比为 React Element 树的节点：一条消息不是一段扁平字符串，而是一棵由 TextBlock、DataBlock、ToolCallBlock 等节点组成的有序列表，每种节点有自己的 props。
• 把 Event 流类比为 SSE chunk：reply_stream 吐出的每个事件就是一个 chunk，前端可以逐帧渲染文本、 工具调用气泡、思考过程，而不是等整条消息拼完才显示。
• 把 append_event 类比为 reducer：它接收当前 Msg 状态与一个事件，返回更新后的 Msg，是一个 确定性的、可重放的折叠操作。

AgentScope 原生语义：Msg 并不是不可变的 action——它是一个 Pydantic BaseModel，append_event 会 原地变更 content、finished_at 与 usage。这与 Redux reducer "返回新 state"的纯函数语义不同：AgentScope 选择原地变更是为了在流式过程中避免反复拷贝 block 列表，但代价是你不能假设同一个 Msg 对象在 append_event 前后内容一致。如果需要快照，请用 msg.model_dump() 显式序列化。Event 也不是 纯粹的 UI 信号——RequireUserConfirmEvent、UserInterruptEvent 等会真正暂停 Agent 执行，是 HITL 控制 流的一部分，而不只是通知。

常见错误

• content 类型混用：给 UserMsg 传 ToolCallBlock，或给 SystemMsg 传 DataBlock，会在 model_validator 阶段抛 ValueError。构造前先确认 role 允许的 block 类型。
• 流式重建时忘记传 id：AssistantMsg(name=ev.name, content=[]) 不带 id 会自动生成新 ID， 导致 append_event 因 reply_id != self.id 跳过全部事件，最终 msg.content 始终为空。
• 中途解析 ToolCallBlock.input：在 ToolCallEndEvent 之前，input 是不完整的 JSON 片段， json.loads 会失败。需要等 ToolCallEndEvent 或 ReplyEndEvent 之后再解析。
• block 顺序假设：content 列表的顺序就是事件到达顺序，不要假设"所有 TextBlock 在前、ToolCallBlock 在后"。ReAct 循环中 TextBlock、ThinkingBlock、ToolCallBlock、ToolResultBlock 会交错出现。
• 把 Event 当持久化单元：持久化的是 Msg，不是 Event。事件流消费完即可丢弃，Msg 才是写入 AgentState 上下文与长期记忆的对象。

先修

• 快速上手

下一步

• 模型与格式化器

官方参考

• Message & Event（2.0.4）
• AgentScope GitHub 仓库
• message 模块源码
• event 模块源码