7.1 流式事件模型
前六章里，minagent 一直是个「输入 → 等 → 输出」的黑箱。你调一次 complete()，它闷头跑完所有步骤，最后把一句完整的话丢回来。中间那几次工具调用、那几秒的等待，调用方一概看不见。3.3 加的 on_step / on_tool_call 回调算是开了扇小窗，但它是「一个钩子管一类事件」的形态：要订阅三种事件就得挂三个钩子，每个各写一套处理逻辑，事件之间的先后顺序还得自己拼。真实产品需要的不是几扇小窗，而是一条统一、有序、可订阅的事件流。
本节把 minagent 推进到「可订阅的事件流」：新增
minagent/events.py定义Event，让FakeModel与Model通过stream()逐 token 产出事件，并把 3.3 的on_step/on_tool_call两个回调收束成单一生成器agent.stream()。
7.1.1 三种可观察性方案
从「完全看不见」到「事件流」，我们其实走过了三档：
| 方案 | 数据形态 | 订阅方式 | 问题 |
|---|---|---|---|
| 一次性返回 | RunResult | 无（跑完才有） | 过程全黑箱，长任务体验差 |
| 多回调 | 多个 on_* 函数 | 每类事件挂一个钩子 | 钩子数量随事件类型膨胀 |
| 单一事件流 | Iterator[Event] | 一个 for 循环 | 无（本节的落点） |
回调那一档的毛病，用过就懂：on_step 给的是拼好的字符串摘要，on_tool_call 给的是原始对象，两者字段不同、时机不同、还不在同一个序列里。前端想「按发生顺序渲染一条时间线」，就得在调用方自己维护一个队列，把两路回调的内容手工合并。事件流把这层合并下沉到运行时——谁先发生谁先出现在流里，调用方只管顺序消费。
7.1.2 Event：一种类型，六种事件
事件模型的核心是一个极小的 dataclass：
# minagent/events.py
from dataclasses import dataclass, field
from typing import Any, Literal
EventType = Literal["token", "tool_call", "tool_result", "step", "final", "error"]
@dataclass
class Event:
type: EventType
data: dict[str, Any] = field(default_factory=dict)
只做两件事：用 type 标明「这是什么事件」，用 data 装「这次事件带了什么」。type 用 Literal 而不是裸 str，是为了让类型检查器和 IDE 都能在写错事件名时立刻报警——事件名一旦拼错，下游 if ev.type == ... 会静默失配，这是最难查的一类 bug。
六种类型各司其职：
type | 触发时机 | data 关键字段 |
|---|---|---|
token | 模型吐出一个（近似）token | text |
tool_call | 模型发起一次工具调用 | id / name / arguments |
tool_result | 工具执行完毕 | id / name / result |
step | 一个循环步骤结束 | index / kind |
final | 整轮收敛 | content |
error | 未收敛或异常 | detail |
设计上有意把「模型侧」和「运行时侧」的事件分开：token 与 tool_call 由模型产出，tool_result、step、final、error 由 Agent 循环产出。这样上层永远能从 type 判断这条事件是谁发的，而不必去猜 data 的形状。
7.1.3 Model.stream()：把 complete 拆成 token
Model 协议原本只有 complete()。我们给它加一个姊妹方法 stream()，产出 Event 而不是 ModelResponse：
# minagent/fake.py（新增）
import re
_TOKEN_RE = re.compile(r"[A-Za-z0-9]+|\s+|[^\sA-Za-z0-9]")
def _tokenize(text: str) -> list[str]:
"""把文本切成近似 token 的片段：ASCII 连续段成块，CJK 逐字。"""
return _TOKEN_RE.findall(text) or [""]
def stream(self, messages, tools):
self.calls.append((list(messages), list(tools)))
if self._cursor >= len(self.script):
raise RuntimeError("FakeModel 脚本耗尽：Agent 没有收敛")
item = self.script[self._cursor]
self._cursor += 1
if isinstance(item, str):
for chunk in _tokenize(item):
yield Event(type="token", data={"text": chunk})
return
if isinstance(item, ModelResponse):
if item.message.tool_calls:
for tc in item.message.tool_calls:
yield Event(type="tool_call",
data={"id": tc.id, "name": tc.name, "arguments": tc.arguments})
else:
for chunk in _tokenize(item.message.content):
yield Event(type="token", data={"text": chunk})
return
for tc in item:
yield Event(type="tool_call",
data={"id": tc.id, "name": tc.name, "arguments": tc.arguments})
三个要点：
stream()是个生成器。它不返回列表，而是用yield一个个吐事件——调用方for ev in model.stream(...)就能边收边处理，不必等整轮跑完。complete()与stream()共用同一个游标。脚本消费是线性的，两个方法只是同一份脚本的两种投影：complete把一项脚本压成一次ModelResponse，stream把它摊成一串Event。_tokenize是假模型的近似。它按「ASCII 连续段成块、CJK 逐字」切分，让输出看起来像真 token 流。真实 tokenizer 的切法不同，但这不影响本节的结论——事件边界只是粒度问题。
注意 stream() 只吐 token 和 tool_call 两类事件。tool_result / step / final 是 Agent 循环才知道的信息，模型无权生产——职责边界要划清。
7.1.4 Agent.stream()：从回调到生成器
有了 model.stream()，把 Agent 的回调升级成事件流就顺理成章了：
# minagent/agent.py（新增 stream 方法）
def stream(self, user_input: str):
messages = list(self.history)
if self.system and not messages:
messages.append(Message(role="system", content=self.system))
messages.append(Message(role="user", content=user_input))
steps: list[Step] = []
for i in range(self.max_steps):
msg = Message(role="assistant")
for ev in self.model.stream(messages, self.registry.schemas()):
if ev.type == "token":
msg.content += ev.data["text"]
yield ev
elif ev.type == "tool_call":
msg.tool_calls.append(ToolCall(
id=ev.data["id"], name=ev.data["name"], arguments=ev.data["arguments"]))
yield ev
messages.append(msg)
if not msg.tool_calls:
steps.append(Step(index=i, kind="final", detail=msg.content))
self.history = messages
yield Event(type="step", data={"index": i, "kind": "final"})
yield Event(type="final", data={"content": msg.content})
return
for tc in msg.tool_calls:
result = self.registry.call(tc)
messages.append(Message(role="tool", content=result, tool_call_id=tc.id))
yield Event(type="tool_result",
data={"id": tc.id, "name": tc.name, "result": result})
steps.append(Step(index=i, kind="tool_call", detail=f"{tc.name} -> {result}"))
yield Event(type="step", data={"index": i, "kind": "tool_call"})
self.history = messages
yield Event(type="error", data={"detail": "达到 max_steps 仍未收敛"})
和 3.3 的 run() 对照，有两处关键差别：
yield代替了self._emit(...)。3.3 是「Agent 主动调回调」，现在是「Agent 把事件交给调用方拉取」。控制权反转了：拉取方决定什么时候取下一个事件，也可以随时break提前中止。- 边收 token 边拼消息。
msg.content += ev.data["text"]在循环里逐片累加，流结束后这条assistant消息正好是完整文本。流式产出与完整消息是同一份数据的两种视角——这一点保证了流式不会牺牲后续逻辑所需的完整性。
history 的写回时机也值得留意：只在收敛或撞上限时才 self.history = messages，中途 break 不会污染历史。这与 3.3 的设计一脉相承。
7.1.5 跑一遍：订阅完整事件流
把假模型脚本设成「add → multiply → 文本收敛」，订阅整条流：
for ev in agent.stream("先算 2+3，再把结果乘 4"):
if ev.type == "token":
print(ev.data["text"], end="")
elif ev.type == "tool_call":
print(f"[tool_call] {ev.data['name']}{ev.data['arguments']}")
elif ev.type == "tool_result":
print(f"[tool_result] {ev.data['name']} -> {ev.data['result']}")
elif ev.type == "step":
print(f"[step] index={ev.data['index']} kind={ev.data['kind']}")
elif ev.type == "final":
print(f"[final] {ev.data['content']}")
本机实测（Python 3.14.6）的事件序列：
[tool_call] add{'a': 2, 'b': 3}
[tool_result] add -> 5
[step] index=0 kind=tool_call
[tool_call] multiply{'a': 5, 'b': 4}
[tool_result] multiply -> 20
[step] index=1 kind=tool_call
[step] index=2 kind=final
[final] 2 + 3 = 5，再乘 4 得 20。
事件计数: {'tool_call': 2, 'tool_result': 2, 'step': 3, 'token': 19, 'final': 1}
拼接文本: 2 + 3 = 5，再乘 4 得 20。
history: 7
事件计数可以精确对账：两次 tool_call（add、multiply）各配一次 tool_result；三个 step（两个工具步 + 一个收敛步）；19 个 token 拼回原文；一个 final。history 仍是 7 条，说明事件流没有改变消息累积的语义。
一个容易忽略的细节：tool_call 事件先于 tool_result 出现，中间隔着一次真实的工具执行。调用方拿到 tool_call 就能先渲染「正在调用 add……」，等 tool_result 回来再补上结果。这正是「边跑边显示」的机制所在。
7.1.6 流式对可观察性的意义
把「一次性返回」和「事件流」摆在一起看，差距不在快慢，而在可见性：
| 维度 | 一次性返回 | 事件流 |
|---|---|---|
| 首字节延迟 | 整轮结束才有 | 第一个 token 即可显示 |
| 过程可见 | 只有最终答案 | token / 工具调用 / 步骤全程可见 |
| 可中止 | 只能等 | 随时 break |
| 前端渲染 | 结果一次性替换 | 可做打字机、进度条、工具时间线 |
| 调试 | 事后看 RunResult | 实时观察每一步 |
「首字节延迟」是体验的分水岭。一次真实调用可能要十几秒，用户盯着空白屏幕十几秒和盯着文字一个个冒出来，感受完全不同。而「可中止」是成本的分水岭——用户看到模型理解偏了，可以在第一个 token 冒头时就 break，省下后面所有的 token 费用。
更重要的是，事件流让 Agent 从「一个函数」变成「一条可观测的管道」。第 15 章的可观测性、第 17 章的 SSE 服务化，都要架在这条流上：SSE 无非是把 Event 序列化成 data: 行推给浏览器，span 树无非是给每条事件盖个时间戳与父子关系。流是底座，观测与服务都只是它的消费者。
7.1.7 生成器与回调：控制权反转
3.3 的回调是「推」：Agent 跑到某处，主动调用 on_step。本节的生成器是「拉」：Agent 把事件挂在那，调用方 next() 一次才取一个。这一推一拉，能力上多了两样东西：
| 能力 | 回调（推） | 生成器（拉） |
|---|---|---|
| 背压 | 无，回调被硬调 | 有，调用方决定取多快 |
| 提前中止 | 难，得靠回调内抛异常 | 易，break 即可 |
| 顺序保证 | 靠调用方自己排 | 天然有序 |
| 可组合 | 差 | 好，能接 itertools、async for |
「背压」是最容易被低估的一条：回调是同步强推的，Agent 吐多快调用方就得处理多快，慢消费者会拖慢整个循环；生成器则让消费方按自己的节奏 next()，快慢解耦。break 提前中止更是只有生成器才有的顺手——订阅方一看风向不对就跳出循环，生成器停在 yield 处，后续步骤（包括还没执行的工具）不再进行。
代价也有：生成器是惰性的，不迭代就什么都不发生。如果你 agent.stream(...) 之后忘了写 for，整段逻辑一步都不会跑。这是新手最常踩的坑——生成器函数被调用时只是造了个生成器对象，真正的执行推迟到第一次 next()。
7.1.8 真实模型下会怎样
假模型的 token 是我们用正则切出来的，节奏完全确定。真实模型的流有三点不同：
- token 边界由供应商决定。
stream()收到的可能是半个词、一个表情符号、甚至一段 JSON 的碎片。所以真实适配器必须能把碎片拼回完整结构，再交给上层——这正是工具调用流式化的难点：arguments是逐片到达的 JSON，要等它闭合才能解析。 - 事件类型更细。Anthropic 的流式协议除了文本增量，还有
message_start、content_block_start、input_json_delta等一长串事件。我们的六种类型是归一化后的结果，适配器的职责就是把供应商的原始事件翻译成这六种，让上层永远只看到统一形状。 - 可能中途断流。网络抖动会让流在半个句子处戛然而止。上层必须把「流意外结束」当成
error事件处理，而不是当成正常收敛——否则会把半截回答当成完整答案。第 14 章的超时与取消会补上这层防护。
一句话：真实场景里 stream() 的形状不变，变的是它背后要做多少归一化与容错。事件模型的价值恰恰在于——无论供应商怎么变，上层订阅的那条流始终长这样。
至此，一次运行已经能被完整订阅。但事件流解决的是「过程可见」，还没解决「状态可携带」：history 仍焊在 Agent 实例里，一个 Agent 只能服务一段对话。下一节 7.2 会话状态对象
会把状态从 Agent 里剥出来，交给独立的 Session。想回看回调的地基，翻 6.3 结构化输出约束与解析兜底
。
阅读导航：上一节：6.3 结构化输出约束与解析兜底 · 下一节：7.2 会话状态对象 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。