流式与事件

reply_stream 是 AgentScope 2.x 把 ReAct 循环"拆开看"的入口。它不是简单的 token 管道，而是一条 结构化事件流：文本块、思考块、工具调用、工具结果、HITL 暂停、迭代超限都以独立事件类型产出，消费方可以逐帧渲染 UI、捕获用户确认、重建完整 Msg。本页在 Agent 核心的 reply_stream 一节基础上展开，覆盖事件流的完整使用模式。

版本基线：本文以 Python agentscope 2.0.4（PyPI，2026-07-07）为准，要求 Python >=3.11。 2.x 文档站存在版本化陷阱：非版本化的 /en/ URL 会 404，本页所有官方链接均使用 versions/2.0.4/en/ 形式。本文涉及的事件类名、导入路径与字段（reply_id、delta）均已与 src/agentscope/event/__init__.py 源码核对，未在该源码中出现的字段名不会出现在本页。

为什么需要流式

如果只用 await agent.reply(msg)，调用方会一直阻塞到 Agent 完成全部推理与工具调用后才拿到最终 Msg。在真实产品里这会带来三个问题：

逐 token 渲染无法实现--用户要等几秒甚至几十秒才看到第一个字，体验断崖式下降。
HITL 中断没有入口--工具调用需要用户确认时，reply 会暂停并返回当前累积的 Msg，但"暂停" 这一信号本身必须通过事件流才能传递给前端。
进度不可见--模型在思考、在调用工具、在等工具结果，这些阶段对调用方完全黑盒，无法做进度提示与可观测性埋点。

reply_stream 把上述过程拆成事件流，让消费方在每个阶段都能介入。

前端类比

如果你来自前端：

把 reply_stream 类比为 SSE（Server-Sent Events）chunk 流：每个事件是一个 chunk，前端逐帧渲染文本、工具调用气泡、思考过程，而不是等整条消息拼完才显示。
把事件消费循环类比为 ReadableStream 的 getReader() + while(true) 模式：async for event in agent.reply_stream(msg) 就是 reader.read() 循环的 Python 版，每个 event 对应一个解码后的帧。
把 Msg.append_event 类比为 reducer：它接收当前 Msg 状态与一个事件，原地更新 content，是一个确定性的折叠操作，把事件流收敛成最终的 Msg。

AgentScope 原生语义：reply_stream 不是把模型输出做一层 SSE 转发，而是把整个 ReAct 循环-- 包括模型调用、Permission 决策、工具执行、HITL 暂停--都拆成事件。因此事件流里不仅有文本增量，还有工具调用生命周期、工具结果、HITL 请求与中断信号。append_event 也是原地变更而非 Redux 风格的 "返回新 state"，这与前端 reducer 的纯函数语义不同：AgentScope 选择原地变更是为了避免反复拷贝 block 列表，但代价是你不能假设同一个 Msg 对象在 append_event 前后内容一致。如果需要快照，请用 msg.model_dump() 显式序列化。

reply_stream 基本用法

reply_stream 是 async generator，输入是一个 Msg（或恢复事件，见 HITL 事件流），不是字符串。消费方式只能是 async for，不能用同步 for 或 agent(msg)。

最小示例：打印文本增量，同时用 append_event 重建完整 Msg。

python

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,
    TextBlockDeltaEvent,
    ReplyEndEvent,
)

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 是 async generator，输入是 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)
            print("--- 回复开始 ---")

        elif isinstance(event, TextBlockDeltaEvent):
            # 逐 token 渲染：delta 是本次增量文本
            print(event.delta, end="", flush=True)

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

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

    # 流结束后，msg 就是一个完整的 AssistantMsg
    if msg is not None:
        print("\n最终文本：", msg.get_text_content())
        print("token 用量：", msg.usage)


asyncio.run(main())

几个关键点：

输入必须是 Msg：UserMsg(name=..., content=...) 是标准入口。直接传字符串会抛类型错误。
ReplyStartEvent 创建空 Msg：AssistantMsg 的 id 必须取自 event.reply_id，不能省略。省略后会自动生成新 ID，与事件流的 reply_id 不匹配，append_event 会跳过所有事件并告警。
TextBlockDeltaEvent.delta 是增量：它是本次到达的文本片段，不是累积值。累积值由 append_event 内部维护到 TextBlock.text 上。
append_event 是原地变更：返回 Self（即 msg 本身），可以链式调用也可以忽略返回值。事件流消费完后，msg 即为完整的 AssistantMsg，可直接写入 AgentState 上下文或序列化。

事件重建的完整语义（包括 DataBlock 的 base64 字节拼接、ToolCallBlock.input 的 JSON 片段累积）见消息与事件页。

事件时序

下面这张时序图说明一次带工具调用的 reply_stream 如何按事件顺序展开。图前先说明：消费方在每个事件上做两件事--一是渲染 UI（打印文本、显示工具气泡），二是调用 append_event 把变更落到 Msg 上。

图后解释：

ReplyStartEvent 是重建起点：它带来 name 与 reply_id，消费方据此创建空 AssistantMsg。没有 ReplyStartEvent 就无法开始重建。
ModelCallStart/End 包裹一次模型调用：ModelCallEndEvent 携带 input_tokens 与 output_tokens，append_event 会累积进 Msg.usage。ReAct 循环里模型可能被调用多次，因此这对事件可能出现多轮。
文本块三段式：TextBlockStartEvent -> 若干 TextBlockDeltaEvent -> TextBlockEndEvent。 delta 只在 Delta 事件上出现，Start/End 主要用于划定边界。
工具调用与工具结果配对：ToolCallStartEvent（带 tool_call_id 与工具名）开始， ToolCallDeltaEvent 追加入参 JSON 片段，ToolCallEndEvent 结束调用声明；随后 ToolResultStartEvent -> ToolResultTextDeltaEvent / ToolResultDataDeltaEvent -> ToolResultEndEvent（带 state）。结果回灌后 Agent 进入下一轮推理，可能再次产出 ModelCallStartEvent。
ReplyEndEvent 收尾：携带 reply_id 与 reason: ReplyEndReason。reason 区分正常完成、 HITL 暂停、迭代超限等终止原因，append_event 据此盖上 finished_at。

事件类型分组

agentscope.event 模块导出约 30 个事件类，全部继承自 EventBase，通过 EventType 枚举区分。下表按生命周期阶段分组，列出流式与 HITL 场景中最主要的事件类型。完整列表见 event 模块源码。

分组	事件	触发时机
回复生命周期	`ReplyStartEvent`	一次 `reply_stream` 开始，携带 `name` 与 `reply_id`
	`ReplyEndEvent`	一次回复结束，携带 `reason: ReplyEndReason`
模型调用	`ModelCallStartEvent`	调用模型前
	`ModelCallEndEvent`	单次模型调用结束，携带 token 用量
文本块	`TextBlockStartEvent`	一段文本块开始
	`TextBlockDeltaEvent`	文本块增量，携带 `delta`
	`TextBlockEndEvent`	文本块结束
思考块	`ThinkingBlockStartEvent`	推理过程块开始（如 extended thinking）
	`ThinkingBlockDeltaEvent`	推理过程增量
	`ThinkingBlockEndEvent`	推理过程块结束
数据块	`DataBlockStartEvent`	多模态数据块开始
	`DataBlockDeltaEvent`	多模态数据增量（base64 字节级拼接）
	`DataBlockEndEvent`	多模态数据块结束
工具调用	`ToolCallStartEvent`	模型发起一个工具调用，携带 `tool_call_id` 与工具名
	`ToolCallDeltaEvent`	工具调用入参 JSON 增量
	`ToolCallEndEvent`	工具调用声明结束
工具结果	`ToolResultStartEvent`	工具执行结果开始
	`ToolResultTextDeltaEvent`	文本型结果增量
	`ToolResultDataDeltaEvent`	数据型结果增量
	`ToolResultEndEvent`	工具执行完成，携带 `state`
HITL	`RequireUserConfirmEvent`	Permission 决策为 `ask`，等待用户确认
	`UserConfirmResultEvent`	用户确认或拒绝结果回灌
	`RequireExternalExecutionEvent`	工具需外部执行（`is_external_tool=True`）
	`ExternalExecutionResultEvent`	外部执行结果回传
	`UserInterruptEvent`	用户主动中断
其他	`ExceedMaxItersEvent`	达到 `ReActConfig` 最大迭代次数
	`HintBlockEvent`	ReAct 循环中给模型的提示，一次性到达
	`CustomEvent`	自定义事件，用于中间件或扩展

几点说明：

文本/数据/思考块都是三段式：Start -> 若干 Delta -> End。HintBlockEvent 是例外，它的完整内容在单个事件里到达，append_event 直接追加一个完整的 HintBlock。
工具调用与工具结果是两条独立的三段式：先有 ToolCallStart/Delta/End 声明"要调用什么"，再有 ToolResultStart/Delta/End 报告"执行结果是什么"。两者通过 tool_call_id 关联。
HITL 事件会真正暂停 Agent：RequireUserConfirmEvent / RequireExternalExecutionEvent 不是通知，而是控制流--Agent 会在产出这些事件后暂停，等待消费方把结果事件作为 inputs 重新调用 reply / reply_stream 恢复（见 HITL 事件流）。

事件重建 Msg

事件流消费完后，需要得到一个完整的 Msg 写入上下文或序列化。标准模式在消息与事件页有完整示例，本页只列要点：

在 ReplyStartEvent 时用 event.name 与 event.reply_id 创建空 AssistantMsg。
对后续每个事件调用 msg.append_event(event)。
ReplyEndEvent 后 msg 即为完整的 AssistantMsg。

python

from agentscope.event import ReplyStartEvent, ReplyEndEvent
from agentscope.message import AssistantMsg

msg: AssistantMsg | None = None

async for event in agent.reply_stream(user_msg):
    if isinstance(event, ReplyStartEvent):
        # id 必须取自 event.reply_id，不能省略
        msg = AssistantMsg(name=event.name, content=[], id=event.reply_id)
    # 对所有事件统一调用 append_event
    if msg is not None:
        msg.append_event(event)

关键约束：

id 必须取自 event.reply_id：append_event 会校验 event.reply_id == self.id，不匹配的事件会被跳过并告警。省略 id 会自动生成新 ID，导致全部事件被跳过，msg.content 始终为空。
不要中途解析 ToolCallBlock.input：在 ToolCallEndEvent 之前，input 是不完整的 JSON 片段， json.loads 会失败。等 ReplyEndEvent 之后再解析。
DataBlockDeltaEvent 不要自己拼字符串：append_event 内部做了 base64 字节级拼接，自己在外面拼字符串再塞回 block 会破坏字节流。

详细语义（包括 role 与 block 约束、usage 累积、finished_at 盖戳）见消息与事件页。

HITL 事件流

HITL（Human-in-the-Loop）事件是 reply_stream 区别于普通 token 流的核心能力。Agent 在工具执行前经 Permission 三态决策（allow / deny / ask），当决策为 ask 或工具被标记为 is_external_tool=True 时，Agent 会暂停执行并产出 HITL 事件，等待消费方决策后回灌结果事件恢复。

RequireUserConfirmEvent 模式

Permission 返回 ask 时触发 RequireUserConfirmEvent，事件携带 tool_calls（含 suggested_rules）。消费方需要：

检测 RequireUserConfirmEvent，展示工具调用信息给用户。
用户决策（allow / deny），构造 UserConfirmResultEvent。
把 UserConfirmResultEvent 作为 inputs 重新调用 reply_stream 恢复。

python

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,
    RequireUserConfirmEvent,
    UserConfirmResultEvent,
)
from agentscope.event import ConfirmResult

agent = Agent(
    name="ops_agent",
    system_prompt="你是一个运维助手，可以执行只读命令。",
    model=DashScopeChatModel(
        credential=DashScopeCredential(api_key=os.environ["DASHSCOPE_API_KEY"]),
        model="qwen-plus",
    ),
    # toolkit 与 permission 配置省略，详见"工具与 MCP"页
)


async def run_with_hitl() -> None:
    msg: AssistantMsg | None = None
    pending_confirm: RequireUserConfirmEvent | None = None

    async for event in agent.reply_stream(
        UserMsg(name="user", content="查看 /etc/hosts 内容"),
    ):
        if isinstance(event, ReplyStartEvent):
            msg = AssistantMsg(name=event.name, content=[], id=event.reply_id)

        elif isinstance(event, RequireUserConfirmEvent):
            # Agent 暂停：工具调用进入 asking 态，需要外部决策
            pending_confirm = event
            print("等待用户确认的工具调用：", [tc.name for tc in event.tool_calls])
            # 这里同步拿到用户决策；真实场景通常跨请求异步完成
            # 假设用户允许
            break  # 跳出当前事件循环，准备恢复

        if msg is not None:
            msg.append_event(event)

    if pending_confirm is not None:
        # 构造确认结果：每个 tool_call 对应一个 ConfirmResult
        results = [
            ConfirmResult(
                tool_call_id=tc.tool_call_id,
                allow=True,  # 用户允许
            )
            for tc in pending_confirm.tool_calls
        ]
        resume_event = UserConfirmResultEvent(
            reply_id=pending_confirm.reply_id,
            confirm_results=results,
        )

        # 用结果事件作为 inputs 恢复 reply_stream
        async for event in agent.reply_stream(resume_event):
            if isinstance(event, ReplyEndEvent):
                print("恢复后回复结束，原因：", event.reason)
            if msg is not None:
                msg.append_event(event)


asyncio.run(run_with_hitl())

要点：

reply_id 必须匹配：UserConfirmResultEvent 的 reply_id 必须与暂停时的 agent.state.reply_id（即 RequireUserConfirmEvent.reply_id）一致，否则无法定位挂起状态。
ConfirmResult 可带 suggested_rules：用户允许时可以附带规则，让后续同类调用自动放行，避免每次都打断用户。
恢复后事件流继续：恢复调用 reply_stream 会从工具执行阶段继续，产出 ToolResultStartEvent 等事件，直到 ReplyEndEvent。

UserInterruptEvent 中断

如果用户不想继续，可以传入 UserInterruptEvent 主动中断暂停态的 Agent：

python

from agentscope.event import UserInterruptEvent

# Agent 此前已因 RequireUserConfirmEvent 暂停
async for event in agent.reply_stream(
    UserInterruptEvent(reply_id=agent.state.reply_id),
):
    # Agent 会为每个未完成工具调用合成"被用户中断"的 ToolResultBlock
    # 依次产出 ToolResultStartEvent / ToolResultTextDeltaEvent / ToolResultEndEvent
    # 最后产出 ReplyEndEvent，回到可接收新输入的状态
    print(event)

UserInterruptEvent 会丢弃挂起状态、为每个未完成工具调用合成一个标记为"被用户中断"的 ToolResultBlock，依次产出 ToolResultStartEvent / ToolResultTextDeltaEvent / ToolResultEndEvent 与最终 ReplyEndEvent，让 Agent 回到可接收新输入的状态。若 Agent 未处于暂停态，该事件是 no-op，立即返回。

RequireExternalExecutionEvent 模式

当工具被标记为 is_external_tool=True 时，工具逻辑在 Agent 外部执行（人工操作或外部系统）。 Agent 产出 RequireExternalExecutionEvent，消费方在外部执行后把结果包装成 ToolResultBlock，通过 ExternalExecutionResultEvent 传回。模式与 RequireUserConfirmEvent 一致：检测事件 -> 外部执行 -> 构造结果事件 -> 作为 inputs 恢复 reply_stream。

完整 HITL 语义（Permission 三态、suggested_rules 累积、外部工具标记）见 Agent 核心 - HITL 暂停与恢复。

SSE 对外

AgentScope 的 Agent Service 把 reply_stream 封装为 SSE 端点，前端可以直接订阅：

GET /sessions/{session_id}/stream

这个端点把 Agent 内部的事件流透传为标准 SSE chunk，前端用 EventSource 或 fetch + ReadableStream 消费即可。SSE 层做的事情是：

把 Python 端的事件对象序列化为 JSON chunk。
按 SSE 协议添加 data: 前缀与事件边界。
保持连接直到 ReplyEndEvent 或客户端断开。

完整的 Agent Service 部署、SSE 端点配置、前端对接代码见部署与 Agent Service 页。本页只需记住：前端消费 SSE 时看到的 chunk 结构与本文描述的事件一一对应，前端按事件类型分发渲染即可。

常见错误

同步遍历 reply_stream：for event in agent.reply_stream(msg) 会抛类型错误。reply_stream 是 async generator，必须用 async for。这是从 1.x 迁移时最常见的错误，1.x 的 __call__ 同步接口在 2.x 已移除。
忘记 append_event：只打印了 TextBlockDeltaEvent.delta 却没有调用 append_event，最终 msg.content 会是空的。append_event 是把事件落到 Msg 上的唯一入口，消费方如果不调用它， Msg 就不会更新。
HITL 未回灌 UserConfirmResultEvent 导致卡住：检测到 RequireUserConfirmEvent 后如果不构造结果事件恢复，Agent 会一直停留在暂停态，reply_stream 不会产出 ReplyEndEvent。前端表现是 "请求一直 pending"。正确做法是把用户决策包装成 UserConfirmResultEvent，作为 inputs 重新调用 reply_stream。
reply_id 不匹配：恢复事件（UserConfirmResultEvent / ExternalExecutionResultEvent / UserInterruptEvent）的 reply_id 必须与暂停时的 agent.state.reply_id 一致，否则无法定位挂起状态，Agent 会报错或忽略恢复请求。
中途解析 ToolCallBlock.input：在 ToolCallEndEvent 之前，input 是不完整的 JSON 片段， json.loads 会失败。需要等 ToolCallEndEvent 或 ReplyEndEvent 之后再解析。
把 Event 当持久化单元：持久化的是 Msg，不是 Event。事件流消费完即可丢弃，Msg 才是写入 AgentState 上下文与长期记忆的对象。

先修

状态与会话 - 理解 AgentState、reply_id 与中断恢复的上下文载体

下一步

可观测性 - 基于事件流做 tracing、metrics 与日志埋点

流式与事件 ​

为什么需要流式 ​

前端类比 ​

reply_stream 基本用法 ​

事件时序 ​

事件类型分组 ​

事件重建 Msg ​

HITL 事件流 ​

RequireUserConfirmEvent 模式 ​

UserInterruptEvent 中断 ​

RequireExternalExecutionEvent 模式 ​

SSE 对外 ​

常见错误 ​

先修 ​

下一步 ​

官方参考 ​

流式与事件

为什么需要流式

前端类比

reply_stream 基本用法

事件时序

事件类型分组

事件重建 Msg

HITL 事件流

RequireUserConfirmEvent 模式

UserInterruptEvent 中断

RequireExternalExecutionEvent 模式

SSE 对外

常见错误

先修

下一步

官方参考