8.2 消息历史与流式事件
本页分析版本earendil-works/pi@c13ffe12026-07-30本章解决什么问题：8.1 做出来的 FakeModel 是「憋完整句、再一次性返回」，用户要盯着空行等它写完。本章把模型抽象从「返回一个 Promise」换成「返回一串事件」，用
async function*让 FakeModel 一个 token 一个 token 地吐字，用for await..of在终端上做出打字机效果；并把一个容易被忽略的问题讲透——一边流一边显示的碎片，是在哪一刻、由谁、收拢成消息历史里的一条完整 assistant 消息的。前置知识：8.1 最小 CLI 与 Fake Model（
Message/Context类型与对话循环骨架）；2.6 异步迭代器与 for await（async function*、惰性求值）；2.3 union、字面量与可辨识联合；3.3 流式输出（事件协议的形状）；5.4 流式事件如何传播到界面（Pi 的三层事件）。学习目标：读完本章后你能
- 说清 step03 的
StreamEvent是哪 4 种事件、一次正常流与一次出错流各是什么序列；- 用
async function*写出一个逐块吐事件的模型，并解释它为什么是惰性的；- 说出
StreamFn的契约「不允许 throw，错误必须编码成error事件」以及这条契约买到了什么；- 指出 step03 里「消息历史收拢」发生在哪一行，并说明为什么
renderStream的返回类型是AssistantMessage | undefined；- 对照 Pi 的
message_update机制，说出 Pi 与 step03 在「半成品消息要不要先进历史」上的两种取舍。
建立直觉：一处改动，三个新问题
step02 与 step03 的主循环差别只有两行：
// step02/src/main.ts:46-48
const reply = await model(context);
context.messages.push(reply);
console.log(`助手> ${messageText(reply)}\n`);
// step03/src/main.ts:47-51
const message = await renderStream(model(context));
// 只有正常结束的回复才进上下文；出错的半截回复丢弃。
if (message) {
context.messages.push(message);
}2
3
4
5
6
7
8
9
10
11
await model(context) 变成了 renderStream(model(context))。表面上只是「等一个值」换成「消费一串事件」，但这一换会立刻逼出三个问题，本章的全部内容就是回答它们：
- 谁负责拼装？ 界面收到的是一块块碎片，历史里要存的是一条完整消息。碎片拼回整体，是模型做还是调用方做？
- 中途出错怎么办？ 前 19 个 token 已经打到屏幕上了，第 20 个的位置上发生了错误。这时候还能
throw吗？ - 什么时候入历史？ 一条正在生成的回复算不算「历史里的一条消息」？出错的半截回复要不要留下来？
最小示例：step03 的四个关键文件
实验目录：labs/mini-agent-harness/step03-streaming。安装并运行：
cd labs/mini-agent-harness/step03-streaming
npm install
npm start # 交互模式，每 token 间隔 30ms，看得见打字机节奏
npm run demo # 非交互模式，delayMs=0，输出与 expected-output.txt 可比对2
3
4
事件类型：一个 4 成员的可辨识联合
src/types.ts 新增的核心类型是 StreamEvent：
export type StreamEvent =
| { type: "start" }
/** 新吐出的一小段文本（一个或几个 token）。 */
| { type: "text_delta"; delta: string }
/** 流正常结束，附带拼装好的完整助手消息。 */
| { type: "done"; message: AssistantMessage }
/** 流因错误结束。注意：错误是**事件**，不是抛异常。 */
| { type: "error"; error: string };2
3
4
5
6
7
8
判别字段是 type，这是 2.3 的可辨识联合（Discriminated Union）：switch (event.type) 之后，每个分支里 TypeScript 都收窄到了具体形状——text_delta 分支能访问 event.delta，done 分支能访问 event.message，写错了编译期就报。
四种事件拼出的序列只有两种形状：
- 正常：
start → text_delta × N → done - 出错：
start → text_delta × N → error
done 事件里带着已经拼装好的完整消息，这是本章第一个问题的答案：拼装由模型侧负责。理由是每个调用方都自己拼一遍，既重复又容易各拼各的；让产生事件的一方顺手给出权威结果，消费方就只剩「显示」这一件事。
同一个文件里还新增了 StopReason 与 StreamFn：
export type StopReason = "stop" | "error" | "aborted";
export type StreamFn = (context: Context, options?: StreamOptions) => AsyncIterable<StreamEvent>;2
3
StreamFn 是整个 Mini Harness 最重要的一个抽象：给我一个 Context，还我一串事件。本步骤的 FakeModel 实现它，8.6 接入的真实 Provider（模型服务提供方）也实现它；从 8.3 起，Agent Loop（Agent 循环）只认这个类型，不关心背后是谁。StopReason 里的 "aborted" 到 8.5 才会真正用上，位置先留好。
生产端：async function*
src/fake-model.ts 与 step02 的结构性差别只有一处——函数体从 async (context) => {...} 变成 async function* (context) {...}：
return async function* (context: Context, callOptions): AsyncGenerator<StreamEvent> {
const delayMs = callOptions?.delayMs ?? defaultDelay;
yield { type: "start" };
const last = context.messages[context.messages.length - 1];
const input = last ? messageText(last) : "";
const rule = rules.find((r) => r.match === undefined || input.includes(r.match));
// 契约：不 throw，把错误编码成事件。
if (rule?.error !== undefined) {
yield { type: "error", error: rule.error };
return;
}
const reply = rule?.reply ?? "（FakeModel 没有命中任何规则，也没有兜底规则。）";
let text = "";
for (const token of tokenize(reply)) {
if (delayMs > 0) await sleep(delayMs);
text += token;
yield { type: "text_delta", delta: token };
}
const message: AssistantMessage = { role: "assistant", content: [{ type: "text", text }], stopReason: "stop" };
yield { type: "done", message };
};2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
三个细节值得指出：
tokenize()按 2 个字符一刀切「token」。真实模型的分词复杂得多，但对理解流式机制不重要——重要的只是「回复被切成了多块，一块一块地出来」。- 循环里同时维护了两样东西：
text累加出完整文本（给done用），yield吐出增量（给界面用）。同一份内容，两种粒度，各服务一个需求。 - 命中带
error的规则时，是yield一个error事件然后return，不是throw。
消费端：打字机在哪一行
src/render.ts 是事件的消费方，整个文件只有一个函数：
export async function renderStream(
stream: AsyncIterable<StreamEvent>,
): Promise<AssistantMessage | undefined> {
let message: AssistantMessage | undefined;
let wroteText = false;
for await (const event of stream) {
switch (event.type) {
case "start":
stdout.write("助手> ");
break;
case "text_delta":
// 关键：不是 console.log，而是 write——不换行，才有打字机效果。
stdout.write(event.delta);
wroteText = true;
break;
case "done":
message = event.message;
stdout.write("\n\n");
break;
case "error":
// 已经吐了半截文本的话先换行，避免错误信息接在半句话后面。
stdout.write(`${wroteText ? "\n" : ""}[错误] ${event.error}\n\n`);
break;
}
}
return message;
}2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
打字机效果就在 stdout.write(event.delta) 这一行：console.log 每次都会补一个换行，write 不会，于是一块接一块地贴在同一行上。
更值得注意的是返回类型 Promise<AssistantMessage | undefined>。渲染层不碰消息历史——它只把「这一轮有没有产出一条可以入历史的消息」作为返回值交出去，要不要 push 由调用方决定。模型只管产生事件、渲染只管显示、主循环只管维护历史，三者通过 StreamEvent 这个联合解耦：以后换模型不用改渲染，换界面不用改模型。
收拢：消息历史只在流结束时变长
src/main.ts 的主循环里，消息历史 context.messages 一共只被写两次：
context.messages.push(userText(line));
const message = await renderStream(model(context));
// 只有正常结束的回复才进上下文；出错的半截回复丢弃。
if (message) {
context.messages.push(message);
}2
3
4
5
6
7
流式期间，几十个 text_delta 全部只流向终端，一个都没有进历史。历史只在 done 事件到达、renderStream 返回之后长出一条完整的 assistant 消息。这就是本章开头第三个问题的答案：step03 的取舍是「要么是一条完整消息，要么什么都没有」，出错的半截回复直接丢弃。
这条取舍在 demo 的真实输出里可以数出来：
Mini Agent Harness · step03 streaming（demo）
你> 你好
助手> 你好！我是 FakeModel，正在一个 token 一个 token 地把这句话吐给你。
你> 流式输出是什么
助手> 流式输出的本质是：模型每生成一小段就发一个事件，调用方一边收一边打印。
你> 报错试试
助手> [错误] 模拟的模型错误：额度不足
你> /exit
再见。（本次对话共 5 条消息）2
3
4
5
6
7
8
9
10
11
12
13
最后一行是 5 条而不是 6 条：3 条用户消息 + 2 条助手消息。第三轮出错了，那条半截回复没有进历史。这个数字是验证你理解正确与否最省事的一个探针。
图解
图 8.2-1 step03 一轮对话的事件序列
阅读顺序：从上到下，时间向下流动。请重点关注三处。第一，箭头方向：`renderStream` 是**主动去拉**事件的（`R->>M`），生成器被拉一次才跑一次，这就是惰性；实线回箭头 `M-->>R` 是一次 `yield` 的到达。第二，`loop` 框里只有两条线——收到增量、写进终端，**没有任何一条通向消息历史 H**。第三，`alt` 的两条分支是全章的分水岭：走 `done` 分支时历史长出一条完整消息，走 `error` 分支时历史一动不动。图中的五个参与者分别对应实验里的 `src/main.ts`、`context.messages`（`src/types.ts` 的 `Context`）、`src/fake-model.ts` 的 `async function*`、`src/render.ts` 的 `renderStream`、`node:process` 的 `stdout`。
对照 Pi 源码：12 种事件与 message_update
step03 的 4 种事件是把 Pi 的协议缩小到「刚好够用」。Pi 的第一层事件协议长这样：
AssistantMessageEvent清点结果是 12 个成员（源码事实）：start、文本三件套、思考三件套、工具调用三件套、done、error。step03 的 4 种正好是它的骨架：start 与 done / error 一一对应，text_delta 对应文本三件套里的中间那个。剩下 8 种的去向是：工具调用三件套到 8.3 才登场（step04 用一个合并的 toolcall 事件代替这三个）；text_start / text_end 这类块边界，以及整组思考事件，Mini Harness 全程都没有对应物——因为我们的内容块只有一种，用不着「这一块属于第几个」这一维。
「不允许 throw」这条契约不是本书发明的，Pi 在两处都把它写成了明文：
StreamFunctionStreamFn这正是 step03 里 yield { type: "error", ... } 的出处（官方说明层级的事实：这两段是源码注释里的契约声明）。它买到的东西是：调用方只有一条代码路径。流是边收边用的，如果第 20 个 token 之后抛异常，消费者会同时处在「已经打印了半截」和「异常处理」两种状态里，要用 try/catch 包住 for await，还要和已打印的内容对齐；编码成事件后，switch 里多一个 case 就完了——render.ts 的 error 分支只有两行，这就是账。
Pi 把碎片放进历史的时机与 step03 相反
现在回到本章的主问题：半成品要不要先进消息历史。step03 的答案是「不进」。Pi 的答案是「进，而且原地替换」。
本书调研笔记 research/R02-agent-core.md 的结论（行号已复核）是：Agent 层的 AgentEvent 共 10 种，其中没有顶层的 text_delta——pi-ai 那 9 种块内事件被统一折叠进一个 message_update，原事件作为 assistantMessageEvent 字段随行。
AgentEvent折叠动作发生在 streamAssistantResponse() 的 for await 循环里，也正是在这里，Pi 对消息历史做了 step03 没做的事：
message_update// packages/agent/src/agent-loop.ts:317-359（有省略）
for await (const event of response) {
switch (event.type) {
case "start":
partialMessage = event.partial;
context.messages.push(partialMessage); // 半成品先占位
addedPartial = true;
await emit({ type: "message_start", message: { ...partialMessage } });
break;
case "text_start": case "text_delta": case "text_end":
// …（省略：thinking_* 与 toolcall_* 共 6 个 case 标签，与上面三个共用同一分支体）
if (partialMessage) {
partialMessage = event.partial;
context.messages[context.messages.length - 1] = partialMessage; // 原地替换
await emit({ type: "message_update", assistantMessageEvent: event, message: { ...partialMessage } });
}
break;
case "done":
case "error": {
const finalMessage = await response.result(); // 权威结果
if (addedPartial) context.messages[context.messages.length - 1] = finalMessage;
// …（省略：addedPartial 为 false 时补发 message_start）
await emit({ type: "message_end", message: finalMessage });
return finalMessage;
}
}
}2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
三处差别值得逐条对照 step03：
半成品占位：Pi 在
start时就把一条空的 assistant 消息推进context.messages，之后每个事件都把数组末尾那条替换掉。这样上层任何时刻去看messages，看到的都是「包含正在生成的这条」的完整上下文。step03 没有这个需求，因为它的渲染层直接写终端，用不着从历史里读半成品。权威结果不来自事件本身，而来自
await response.result()。AssistantMessageEventStream在done/error事件被 push 进来时就解析出最终消息，result()拿的是同一个值：packages/ai/src/utils/event-stream.ts ·AssistantMessageEventStream以 done 或 error 判定流结束，并从 done 的 message 或 error 的 error 字段取出最终 AssistantMessage 作为 result。这和 step03 的
done.message是同一个设计意图的两种写法：权威的完整消息由产生方给出。出错的半截回复留在历史里。Pi 的最终消息带
stopReason，出错时是"error"，被取消时是"aborted"，它照样留在context.messages里；随后循环靠这个字段决定收摊：packages/agent/src/agent-loop.ts ·stopReason拿到 assistant 消息后先看 stopReason：error 或 aborted 就发 turn_end 与 agent_end 直接返回，不再进入工具执行。step03 则直接丢弃。一种看法是：Pi 的做法保住了「用户看到过什么，历史里就有什么」，重试和 8.4 那样的持久化都需要这条记录；代价是历史里会出现「不完整的 assistant 消息」，后续每个消费者都得会看
stopReason。step03 的丢弃法省掉了这条心智负担，代价是「屏幕上出现过、历史里查不到」——等到 8.5 要处理「取消到一半」时，这笔账会重新摊开。
把两种收拢方式并排画出来：
图 8.2-2 两种把流收拢成消息的方式
阅读顺序：两个分组各自从上往下读，读完再横向对比。请关注两处。第一，左边 `A2` 与右边 `B2`、`B3` 的对照：step03 在流式期间**完全不动**消息历史，Pi 则让历史里一直有一条「正在长大的」assistant 消息。第二，两个菱形之后的分支：step03 的 `error` 分支让历史长度保持不变，Pi 的 `done` 与 `error` 走的是**同一条路**——都用 `result()` 的最终消息替换占位那条，区别只写在 `stopReason` 字段里。右边四个方框对应 `packages/agent/src/agent-loop.ts` 第 317–359 行的三个 `case` 分支；左边四个方框对应实验里 `src/render.ts` 的 `switch` 与 `src/main.ts` 的 `if (message)`。
最后一处对照留给 FakeModel 本身：真实项目同样需要一个可脚本化的假模型来做零成本、可复现的测试。Pi 的对应物是 faux provider，它的文本分块循环与 step03 的 tokenize + for 循环是同一件事的两种写法：
splitStringByTokenSize区别只在细节：Pi 按 token 数随机切块并按「每秒多少 token」计算等待时间，step03 按固定 2 个字符切、固定 delayMs 等待。机制完全一样。
实践任务
labs/mini-agent-harness/step03-streaming目标：跑通 step03，亲手验证事件序列、惰性求值与「消息历史何时收拢」三件事。实验目录：labs/mini-agent-harness/step03-streaming（全书实验索引见实践任务索引）。全程不联网、不需要任何 API Key。
步骤：
安装并跑非交互 demo，与
expected-output.txt比对：shcd labs/mini-agent-harness/step03-streaming npm install npm run demo1
2
3跑交互模式
npm start，依次输入「你好」「报错试试」，肉眼确认文字是逐字长出来的，以及错误提示前有换行。把
src/main.ts第 17 行的delayMs从 30 改成 300，重跑npm start。这一步是为了验证惰性：如果生成器不惰性，你会先干等再看到整段文字。在
src/render.ts的text_delta分支里，把stdout.write(event.delta)换成console.log(event.delta)，重跑并解释输出为什么基本上变成了每两个字符一行（tokenize按 2 个字符切，末尾那块可能只有 1 个字符）。改回去。在
src/render.ts的error分支末尾加一行message = { role: "assistant", content: [{ type: "text", text: "半截" }], stopReason: "error" };，重跑npm run demo，观察最后一行的消息条数从 5 变成 6，并说出这相当于把 step03 的取舍换成了 Pi 的哪一种。改回去。在
src/rules.ts里加一条自己的规则（match+reply），重跑npm start验证命中。
预期现象：第 1 步的输出与 expected-output.txt 完全一致，关键三行：
助手> 流式输出的本质是：模型每生成一小段就发一个事件，调用方一边收一边打印。
助手> [错误] 模拟的模型错误：额度不足
再见。（本次对话共 5 条消息）2
3
第 5 步改动后，最后一行变成「共 6 条消息」。
如何判断成功：你能不看代码画出正常流与出错流的两条事件序列；能指出打字机效果、消息收拢分别落在哪一行代码；能解释第 5 步为什么把 5 改成了 6。
常见错误：
- 看不到打字机效果：多半把输出重定向到了文件或管道（
npm start > out.txt、npm start | cat），请在终端里直接运行；npm run demo的delayMs是 0，本来就看不出节奏。 - 把
for await写成for：for..of走同步迭代协议，编译器会提示对象上没有Symbol.iterator。 - 在 FakeModel 里
throw而不是yield一个error事件：renderStream的for await会把异常原样抛给主循环，进程直接退出——这就是契约存在的理由。 switch分支忘了break：done分支穿透到error分支，屏幕上每轮都会多出一条错误提示。
对应源码位置：packages/ai/src/types.ts 第 493–513 行 AssistantMessageEvent（实验的 StreamEvent 是它的缩小版）与第 312–324 行 StreamFunction 契约；packages/agent/src/types.ts 第 18–32 行 StreamFn 契约、第 422–437 行 AgentEvent；packages/agent/src/agent-loop.ts 第 317–359 行（实验里 renderStream 的真实对应物）；packages/ai/src/providers/faux.ts 第 359–374 行（FakeModel 的真实对应物）。
本章小结
- 模型抽象从
(context) => Promise<AssistantMessage>换成StreamFn = (context, options?) => AsyncIterable<StreamEvent>，生产端是async function*，消费端是for await..of。主循环的骨架一行没改。 - step03 的
StreamEvent有 4 个成员：start、text_delta、done、error；序列只有两种形状——start → text_delta × N → done与start → text_delta × N → error。 - 拼装由模型侧负责：
done事件携带拼装好的完整消息，消费方不必自己累加。text_delta服务于界面实时性，done服务于状态正确性。 - 错误是事件不是异常。Pi 在
StreamFunction与StreamFn两处注释里把这条写成了明文契约，收益是消费方只有一条代码路径。 - 消息历史的收拢点在主循环那一句
if (message) context.messages.push(message)：流式期间历史一动不动，只有done到达才长出一条完整消息。demo 最后一行的「5 条消息」就是这条规则的探针。 - Pi 的取舍相反：
start时先把半成品推进context.messages，每个块内事件原地替换，done/error时用response.result()的最终消息再替换一次；出错与被取消的消息带着stopReason留在历史里。9 种块内事件在 Agent 层被折叠成一个message_update信封。
关键术语：流式输出（Streaming）、流式事件（Stream Event）、异步迭代器（Async Iterator）、异步生成器（async function*）、可辨识联合（Discriminated Union）、增量（delta）、终局事件（done / error）、StreamFn、StopReason、消息历史收拢、message_update
关键源码索引：packages/ai/src/types.ts 第 493–513 行 AssistantMessageEvent（12 种）、第 312–324 行 StreamFunction 契约；packages/agent/src/types.ts 第 18–32 行 StreamFn 契约、第 422–437 行 AgentEvent（10 种）；packages/agent/src/agent-loop.ts 第 317–359 行事件折叠与历史替换、第 193–200 行 stopReason 分流；packages/ai/src/utils/event-stream.ts 第 69–83 行 AssistantMessageEventStream；packages/ai/src/providers/faux.ts 第 359–374 行文本分块
自测问题：
renderStream的返回类型为什么是AssistantMessage | undefined而不是AssistantMessage？如果改成必返回一条消息，demo 最后一行会变成几条？- FakeModel 里如果把
yield { type: "error", ... }换成throw new Error(...)，程序在哪一行崩、屏幕上会留下什么？请照着render.ts与main.ts的代码走一遍。 done事件里已经有完整消息了，text_delta是不是多余的？如果只保留done，用户体验和 step02 有什么区别？- Pi 在
start事件时就把半成品消息推进了context.messages，step03 不这么做。各举一个「这样做更方便」的具体场景。
下一章预告：8.3 Agent Loop 与工具执行——到目前为止，模型只会说话。下一章给 ContentBlock 加入 toolCall、给 StreamEvent 加入 toolcall 事件、给 StopReason 加入 "toolUse"，还会出现第三种 role toolResult；「一问一答」于是变成「请求 → 执行工具 → 回填结果 → 再请求」的内层循环。本章那句「事件种类到 8.3 才补齐」的账，下一章开始还。