6.1 pi-ai：统一的模型接口
本页分析版本earendil-works/pi@c13ffe12026-07-30本章解决什么问题：Pi 支持几十家模型服务商，但 Agent Loop 里只有一行
models.stream(...)。这中间的「统一层」到底统一了什么、怎么实现的。 前置知识：2.6 异步迭代器与 for await、3.3 流式输出、3.4 工具调用、4.3 monorepo 与 package 地图。 学习目标：① 说出 pi-ai 的四层结构与各层职责；② 看懂统一消息类型与 12 种流式事件的协议约定；③ 讲清complete()与stream()的关系以及EventStream的缓冲机制；④ 跟着一条完整调用链读完 Anthropic 协议实现的 SSE 转换；⑤ 不用 API Key 就能用 faux provider 跑出一次完整的假对话。
建立直觉：几十家 provider，一个形状
Anthropic 的流式接口返回 content_block_delta，OpenAI 的 Responses 接口返回 response.output_text.delta，Google 返回的又是另一套字段名。它们描述的是同一件事——「模型刚刚多吐了几个字」——但字段名、嵌套层次、停止原因的取值全都不同。
没有统一层会怎样？Agent Loop 里就必须写 if (provider === "anthropic") { ... } else if (provider === "openai") { ... }，而且每加一家服务商，Agent Loop、工具执行、界面渲染、Session 存储都要跟着改一遍。上下文里存下来的历史消息也会带上厂商烙印，换个模型继续对话就变成了数据迁移工程。
pi-ai 的做法是在中间插一层「形状统一」：进去的是统一的 Context，出来的是统一的 AssistantMessageEvent 事件流，中间由谁去调哪家 HTTP 接口，调用方完全不需要知道。这一层可以拆成四块来理解：
types.ts）：消息、内容块、上下文长什么样。协议层（
types.ts 的事件定义 + utils/event-stream.ts）：一次生成过程如何被拆成事件、如何被消费。路由层（
models.ts）：`Models` 集合持有若干 `Provider`，按 `model.provider` 把请求派发给对应的 provider，顺带解析凭据。实现层（
api/*.ts）：每个文件是一种线上协议（wire protocol）的适配器，负责把厂商响应翻译成统一事件。 包本身是独立发布的：npm 包名 @earendil-works/pi-ai，锁定版本 0.83.0，自述为 "Unified LLM API with automatic model discovery and provider configuration"（官方说明，见 packages/ai/package.json:2-4）。它不依赖仓库里任何其他包——这是 4.3 那张依赖图的最底层。
门面：Models 集合
pi-ai 的对外入口不是一堆全局函数，而是一个可以创建多个实例的集合对象。createModels() 造一个空集合，builtinModels() 造一个已注册全部内置 provider 的集合（源码事实）：
// packages/ai/src/providers/all.ts:131-137
export function builtinModels(options?: CreateModelsOptions): MutableModels {
const models = createModels(options);
for (const provider of builtinProviders()) {
models.setProvider(provider);
}
return models;
}builtinModels集合上的方法分三类：查目录（getProviders / getModels / getModel / refresh）、管凭据（getAuth / login / logout / checkAuth）、发请求（stream / complete / streamSimple / completeSimple）。
Models四个发请求的方法其实只有两个维度的差别。第一个维度是「流还是结果」——这一点被实现得非常直白：
// packages/ai/src/models.ts:504-510
async complete<TApi extends Api>(
model: Model<TApi>,
context: Context,
options?: ModelsApiStreamOptions<TApi>,
): Promise<AssistantMessage> {
return this.stream(model, context, options).result();
}complete这句话值得停一下：pi-ai 里没有「非流式」的代码路径。所有 provider 实现都只写流式一种，非流式只是「订阅这条流的最终结果」。一种看法是：这样每个 provider 少写一半实现，两条路径也不会各自长出不同的 bug；代价是即使只想要一句完整回答，也要付出建流、入队、事件分发的开销。
第二个维度是 stream 与 streamSimple 的区别：前者接受该 provider 协议专属的选项（例如 Anthropic 的 thinkingEnabled / thinkingBudgetTokens），后者接受一个统一的推理档位 ThinkingLevel = "minimal" | "low" | "medium" | "high" | "xhigh" | "max"（types.ts:79），由各实现自己翻译成厂商参数。Anthropic 的翻译逻辑在 api/anthropic-messages.ts:801-841：没请求推理就直接 thinkingEnabled: false（809-811 行），支持自适应推理的模型映射到 effort（815-822 行），老模型则换算成 token 预算（826-840 行）。翻译完照样回落到同文件的 stream。
数据层：一条消息长什么样
统一层的第一块地基是消息类型。Pi 只有三种消息，联合成 Message（源码事实）：
// packages/ai/src/types.ts:393-433（节选，UserMessage 压成一行）
export interface UserMessage { role: "user"; content: string | (TextContent | ImageContent)[]; timestamp: number }
export interface AssistantMessage {
role: "assistant";
content: (TextContent | ThinkingContent | ToolCall)[];
api: Api;
provider: ProviderId;
model: string;
// …（省略：responseModel / responseId / diagnostics / errorMessage 等）
usage: Usage;
stopReason: StopReason;
timestamp: number;
}
export interface ToolResultMessage<TDetails = any> {
role: "toolResult";
toolCallId: string;
toolName: string;
content: (TextContent | ImageContent)[];
// …（省略：details / usage / addedToolNames）
isError: boolean;
timestamp: number;
}
export type Message = UserMessage | AssistantMessage | ToolResultMessage;Message这是一个典型的可辨识联合（2.3 讲过）：role 是判别字段。几个值得注意的设计（分析解释）：
- 助手消息的内容块只允许三种：文本、思考（thinking）、工具调用（
types.ts:401）。图片只能出现在用户消息和工具结果里——在这套对话类型里模型不输出图片（图像生成是另一套AssistantImages类型，types.ts:444-454），这条约束直接写进了类型而不是靠约定。 - 工具调用是内容块，不是独立消息。
ToolCall.arguments是已经解析好的对象（types.ts:360-366），调用方拿到的不是 JSON 字符串。 - 工具结果是独立消息（
role: "toolResult"），用toolCallId回指。这解释了 Agent Loop 为什么能把一轮工具调用拆成「助手消息 + 若干工具结果消息」追加进历史。 - 每条消息都带
timestamp，助手消息还带api/provider/model/usage/stopReason——历史记录里保存的是「这句话是谁在什么配置下生成的」，而不只是文本。这是 Session 能跨模型接力的前提。
请求的输入侧同样简单：Context = { systemPrompt?, messages, tools? }（types.ts:487-491），工具用 TypeBox schema 描述参数（types.ts:480-485）。整个 pi-ai 对外暴露的输入就这三个字段。
协议层：12 种事件
生成过程被拆成 12 种事件，全部定义在一个联合类型里：
AssistantMessageEvent| 事件 | 何时发出 | 关键字段 |
|---|---|---|
start | 流开始，此时 partial.content 还是空数组 | partial |
text_start | 一个文本块开始 | contentIndex |
text_delta | 文本块多了一段 | delta、contentIndex |
text_end | 文本块结束 | content（完整文本） |
thinking_start | 一个思考块开始 | contentIndex |
thinking_delta | 思考块多了一段 | delta |
thinking_end | 思考块结束 | content |
toolcall_start | 一次工具调用开始 | contentIndex |
toolcall_delta | 工具参数的 JSON 又来了一截 | delta（JSON 片段） |
toolcall_end | 工具调用完整 | toolCall（含 id/name/arguments） |
done | 正常终止 | reason：stop / length / toolUse；message |
error | 异常终止 | reason：aborted / error；error（同样是一条 AssistantMessage） |
三条约定必须记住：
- 每个事件都带
partial，即「截至此刻已经拼好的那条AssistantMessage」。界面层不需要自己攒字符串，直接渲染partial即可（5.4 流式事件如何传播到界面会看到 Pi 的 TUI 就是这么做的）。 - 块之间可能交错。官方文档明确说明：不同内容块的事件不保证连续，可能出现
text_start→text_delta→toolcall_start→text_delta这种顺序，消费者必须用contentIndex把 delta 关联回它所属的块（packages/ai/README.md:641-661）。 - 终止事件二选一，且
error也携带一条完整消息。也就是说「失败」不是抛异常，而是一条stopReason为error或aborted的助手消息。types.ts:313-319把这一点写成了StreamFunction的契约：一旦被调用，失败必须编码进流里，不许 throw。这是整个 pi-ai 错误模型的基石——调用方永远只需要处理流，不需要同时处理 try/catch 和流。
EventStream：推与拉之间的那个缓冲
事件是 provider 实现「推」出来的，调用方是用 for await 一条条「拉」的。两边速度不一样：网络一次可能到来 5 个事件，而消费者还在渲染第一个。中间必须有个缓冲。
最小示例：30 行手写一个
先脱离 Pi 看这个模式的本质。把下面的内容存成 mini-stream.mjs，用 node mini-stream.mjs 运行：
class MiniStream {
queue = [];
waiting = [];
done = false;
constructor() { this.result = new Promise((r) => (this.resolveResult = r)); }
push(event) {
if (this.done) return;
if (event.type === "done") { this.done = true; this.resolveResult(event.text); }
const waiter = this.waiting.shift();
if (waiter) waiter({ value: event, done: false }); // 有人在等 → 直接交付
else this.queue.push(event); // 没人等 → 先排队
}
async *[Symbol.asyncIterator]() {
while (true) {
if (this.queue.length > 0) yield this.queue.shift();
else if (this.done) return;
else {
const r = await new Promise((resolve) => this.waiting.push(resolve));
if (r.done) return;
yield r.value;
}
}
}
}
const s = new MiniStream();
s.push({ type: "delta", text: "你" }); // 生产者一口气推完，不等消费者
s.push({ type: "delta", text: "好" });
s.push({ type: "done", text: "你好" });
for await (const e of s) console.log(e.type, e.text); // 消费者迟到，但一条不丢
console.log("result =", await s.result);实际运行输出（真实运行，Node 26）：
delta 你
delta 好
done 你好
result = 你好三个要点：队列接住消费者来不及取的事件；waiting 数组存放消费者的「等待凭证」，生产者一到就直接兑现；resolveResult 让「只要最终结果」的调用方不必遍历整条流。
回到 Pi 源码
Pi 的实现就是这个结构，只是把「什么算终止」和「终止时取什么」做成了构造参数：
EventStreamAssistantMessageEventStream（同文件 69-83 行）只是给它填上具体规则：done 或 error 算终止，终止时取出 event.message 或 event.error 作为最终的 AssistantMessage。前面那句 complete() = stream().result() 到这里就闭环了——result() 返回的正是这个 finalResultPromise。
error 事件本身就是终止事件，result() 会正常 resolve 出一条 stopReason: "error" 的消息，而不是 reject。所以 await models.complete(...) 不会因为模型报错而抛异常——你必须检查返回消息的 stopReason。这一点在 6.3 pi-agent-core 的 Agent Loop 里非常关键。 图解：请求怎么走完这四层
图 6.1-1 pi-ai 的四层与一次请求的去程回程
从上往下的实线是去程，带「AssistantMessageEvent 事件流」标签的那条是回程：事件直接回到调用方，不被中间层再包一层。虚线表示路由层与实现层都以 types.ts 为共同契约。
读图要关注两处。第一，auth 在路由层解析，不在实现层：applyAuth（models.ts:463-487）解析出 apiKey、合并 headers、必要时替换 baseUrl，再把结果作为普通选项传给 provider。协议实现只管「有 key 就发请求」，不关心 key 从环境变量、凭据文件还是 OAuth 刷新来。第二，实现层是懒加载的：anthropicMessagesApi() 全文只有一行 lazyApi(() => import("./anthropic-messages.ts"))（api/anthropic-messages.lazy.ts:4），厂商 SDK 直到第一次真正发起请求才被加载。lazyStream（api/lazy.ts:46-61）让 stream() 仍然同步返回一条流，异步的加载与鉴权在背后进行，失败时转成流内的 error 事件——又一次践行「错误进流不抛出」的契约。
完整调用链（源码事实，可用 grep 逐环复核）：
models.stream()（models.ts:489-502）→ lazyStream()（api/lazy.ts:46-61）→ applyAuth()（models.ts:463-487）→ provider.stream（models.ts:619）→ dispatch()（models.ts:576-587）→ lazyApi().stream（api/lazy.ts:70-72）→ stream（api/anthropic-messages.ts:487）→ iterateAnthropicEvents()（同文件 446-485）→ iterateSseMessages()（同文件 387-444）→ 事件循环 push 统一事件（同文件 573-745）→ done（758 行）或 error（768 行）→ EventStream.result()（utils/event-stream.ts:64-66）。
解剖一个实现：anthropic-messages.ts
api/ 下的每个模块恰好导出 stream 与 streamSimple 两个函数，因此模块本身就满足 ProviderStreams 接口（types.ts:228-239 的注释明确写了这一点）。挑 Anthropic 这个看，因为它的 SSE 处理是手写的，读起来最完整。
第一步：把字节流切成 SSE 事件
服务端发回的是 SSE（Server-Sent Events，服务器推送事件）文本流，形如 event: content_block_delta 换行 data: {...} 换行空行。pi-ai 没有用 SDK 自带的流封装，而是自己解码：
// packages/ai/src/api/anthropic-messages.ts:387-444（节选，签名已折行压缩）
async function* iterateSseMessages(body: ReadableStream<Uint8Array>, signal?: AbortSignal) {
const reader = body.getReader();
const decoder = new TextDecoder();
const state: SseDecoderState = { event: null, data: [], raw: [] };
let buffer = "";
try {
while (true) {
if (signal?.aborted) throw new Error("Request was aborted");
const { value, done } = await reader.read();
if (done) break;
buffer += decoder.decode(value, { stream: true });
let consumed = consumeLine(buffer);
while (consumed) {
buffer = consumed.rest;
const event = decodeSseLine(consumed.line, state);
if (event) yield event; // 只有空行触发 flush 时才有事件产出
consumed = consumeLine(buffer);
}
}
// …（省略：419-440 行，流读完后 flush 残留 buffer 与最后一个未闭合事件）
} finally {
reader.releaseLock();
}
}iterateSseMessages外面还套了一层过滤：iterateAnthropicEvents()（446-485 行）只放行 6 种 Anthropic 消息事件（ANTHROPIC_MESSAGE_EVENTS，307-314 行），用 parseJsonWithRepair 解析（467 行），并在流结束时校验「见过 message_start 就必须见到 message_stop」（482-484 行），否则报「流提前结束」。
第二步：六类上游事件 → 统一事件
主循环在 573-745 行，是一棵大 if/else 树。对照表如下（源码事实）：
| Anthropic 事件 | 做了什么 | push 出的统一事件 |
|---|---|---|
message_start | 记录 responseId、初始 usage 并算成本 | 无（start 已在 568 行发出） |
content_block_start | 按 text / thinking / redacted_thinking / tool_use 建块 | text_start / thinking_start / toolcall_start |
content_block_delta | 按 delta 子类型追加内容 | text_delta / thinking_delta / toolcall_delta；signature_delta 只累积签名，不发事件 |
content_block_stop | 收尾、清理临时字段 | text_end / thinking_end / toolcall_end |
message_delta | mapStopReason 映射停止原因、累计 usage | 无 |
message_stop | 由 iterateAnthropicEvents 用于完整性校验 | 无 |
停止原因的映射在 mapStopReason（1325-1351 行）：end_turn → stop，max_tokens → length，tool_use → toolUse，refusal 与 sensitive → error 并带上说明文案，未知取值直接抛异常（宁可显式失败也不装作正常）。
图 6.1-2 一次 Anthropic 流式请求的事件转换
从上到下是时间顺序。注意 start 事件在 HTTP 响应头拿到之后、任何数据到达之前就发出，让界面能立刻显示「正在生成」。
这张图对应四段真实源码：A->>H 是 559-566 行经 retryProviderRequest 包裹的请求；A-->>U: start 是 568 行；中间三段 delta 来自 573-745 行的主循环；最后的 done 在 758 行、error 在 768 行。要留意 start 的位置——它在拿到响应之后立刻发出，晚于建立连接但早于第一个 token。
第三步：工具参数的增量 JSON
最需要单独说的是工具调用参数。Anthropic 把参数 JSON 拆成 input_json_delta 一片片发，中途的字符串必然是残缺的（例如 {"path":"READ）。pi-ai 每收到一片就尝试解析一次：
// packages/ai/src/api/anthropic-messages.ts:654-666
} else if (event.delta.type === "input_json_delta") {
const index = blocks.findIndex((b) => b.index === event.index);
const block = blocks[index];
if (block && block.type === "toolCall") {
block.partialJson += event.delta.partial_json;
block.arguments = parseStreamingJson(block.partialJson);
stream.push({
type: "toolcall_delta",
contentIndex: index,
delta: event.delta.partial_json,
partial: output,
});
}input_json_deltaparseStreamingJson 有四层兜底（源码事实）：先按标准 JSON 解析；失败则修复非法转义与裸控制字符后再解析；再失败交给 partial-json 这个第三方库做残缺解析（依赖声明见 packages/ai/package.json:72）；仍失败就返回空对象，绝不抛异常。
parseStreamingJson为什么要费这个劲？因为界面希望在参数流完之前就显示「正在读取文件 README.md」。有了增量解析，partial.content[i].arguments 在中途就已经是 { path: "READ" } 这样的半成品对象，界面可以直接读字段而不必自己拼 JSON。到 content_block_stop 时再整体重解析一次并删掉 partialJson 缓冲（695-698 行），保证存进 Session 的只有干净的参数对象。
要注意这属于 Anthropic 实现的行为：下面实践任务用的 faux provider 只按块发出 JSON 字符串片段，不做增量解析，参数对象是在 toolcall_end 之前整体赋值的（faux.ts:377 与 faux.ts:389）。同一个事件协议，不同实现对「中途的 partial 有多准」可以有不同投入。
相关测试：packages/ai/test/anthropic-sse-parsing.test.ts:82 的用例 "repairs malformed SSE JSON and malformed streamed tool JSON" 直接喂进畸形 SSE 与畸形工具 JSON，验证这条修复链路。
faux provider：把「假模型」做成正式设施
3.1 讲过用假模型来学习和测试的想法。pi-ai 把它工程化成了一个正式的 provider：fauxProvider()。
// packages/ai/src/providers/faux.ts:523-531（节选）
export function fauxProvider(options: RegisterFauxProviderOptions = {}): FauxProviderHandle {
const core = createFauxCore(options);
const provider = createProvider({
id: core.provider,
auth: { apiKey: { name: "Faux", resolve: async () => ({ auth: {} }) } },
models: core.models,
api: { stream: core.stream, streamSimple: core.streamSimple },
});
// …（省略：531-540 行返回句柄，含 setResponses / appendResponses / state 等）fauxProvider关键在于它不是 mock，而是一个真 provider：注册进 Models 之后，models.stream() 的每一环——鉴权、dispatch、事件流——都真实执行，只有最后一步「发 HTTP 请求」被替换成「从队列里取一条预设消息」。因此它能验证的东西远多于一个假函数：
- 脚本化响应：
setResponses([...])排队，stream每次调用shift一条（faux.ts:447）；队列空了会推出一条errorMessage为"No more faux responses queued"的error事件（faux.ts:453-464）。队列元素还可以是函数，按上下文动态决定回什么（faux.ts:96-103）。 - 真的流式：
streamWithDeltas()（faux.ts:308-404）把消息按 3–5 个 token（约 12–20 字符）随机切块（faux.ts:253-263），逐块发出全套*_start/*_delta/*_end事件，还响应AbortSignal——中途取消会得到stopReason: "aborted"。 - 估算用量：
withUsageEstimate()（faux.ts:213-251）按约 4 字符 = 1 token 估算，并且在给了sessionId时用「前缀公共长度」模拟 prompt cache 的命中与写入。
注意 faux 不在 builtinProviders() 的 38 个内置 provider 列表里（providers/all.ts:87-128），必须自己 models.setProvider(faux.provider)。它从根入口导出（packages/ai/src/index.ts:36），所以 import 时不需要子路径。
Pi 自己就靠它测试：packages/ai/test/faux-provider.test.ts 有 23 个用例，其中 382 行的 "streams an exact event order for fixed-size chunks" 用固定块大小断言了完整事件序列；coding-agent 的测试脚手架也把它当默认模型用（packages/coding-agent/test/test-harness.ts:46-50）。
实践任务
目标：不需要任何 API Key，亲眼看到 models.stream() 吐出的统一事件序列，并验证「complete 就是 stream 的结果」这条设计。
准备：确认 Pi 仓库已装好依赖（仓库根目录存在 node_modules/.bin/tsx）。
步骤 1：在 Pi 仓库之外新建文件 faux-demo.mts（后缀必须是 .mts，原因见下方常见错误），把第一行的路径换成你本机 Pi 仓库的绝对路径：
import { createModels, fauxAssistantMessage, fauxProvider, fauxText, fauxToolCall } from "/绝对路径/pi/packages/ai/src/index.ts";
const faux = fauxProvider();
const models = createModels();
models.setProvider(faux.provider);
faux.setResponses([fauxAssistantMessage([fauxText("先看看 README"), fauxToolCall("read_file", { path: "README.md" }, { id: "call_1" })], { stopReason: "toolUse" })]);
const stream = models.stream(faux.getModel() as any, { messages: [{ role: "user", content: "介绍一下这个仓库", timestamp: Date.now() }] });
for await (const event of stream) console.log(event.type, "delta" in event ? JSON.stringify(event.delta) : "");
const final = await stream.result();
console.log("stopReason =", final.stopReason, "| totalTokens =", final.usage.totalTokens);步骤 2：在 Pi 仓库根目录执行（把路径换成你的文件位置）：
./node_modules/.bin/tsx ~/faux-demo.mts预期现象（本书作者真实运行输出，Node 26 + 锁定版本源码）：
start
text_start
text_delta "先看看 README"
text_end
toolcall_start
toolcall_delta "{\"path\":\"README."
toolcall_delta "md\"}"
toolcall_end
done
stopReason = toolUse | totalTokens = 15如何判断成功：① 事件序列符合 start → 各块的 *_start/*_delta/*_end → done；② 最后一行 stopReason = toolUse，与你在 fauxAssistantMessage 里写的一致；③ 你能解释 totalTokens 为什么不是 0（提示：faux.ts:213-251 的 4 字符/token 估算）。注意 toolcall_delta 的条数与切分位置每次运行都可能不同——分块大小是随机的 3–5 个 token、即 12–20 个字符（faux.ts:253-263），作者多次运行分别得到过 "{\"path\":\"REA" + "DME.md\"}"、"{\"path\":\"README." + "md\"}"，以及一次就发完的 "{\"path\":\"README.md\"}"；事件类型的先后顺序则是固定的。
常见错误：
- 文件后缀写成
.ts且放在没有"type": "module"的目录下，会报Top-level await is currently not supported with the "cjs" output format。改成.mts即可（这是作者真实踩到的报错）。 - 忘了调用
setResponses：只会看到一个error事件，errorMessage为No more faux responses queued（faux.ts:453-464）——这本身也是一次有用的观察。 - 改用
models.complete(...)却期待看到事件：complete只返回最终消息，中间事件不会打印。想两者都要，就照上面这样先遍历流、再await stream.result()。
对应源码位置：packages/ai/src/providers/faux.ts:523-541（装配）、445-479（从队列取一条并交给流）、308-404（切块并推送事件）；packages/ai/src/models.ts:489-502（Models.stream）、504-510（complete）。
本章小结
- pi-ai 分四层：数据（
types.ts的消息与内容块）、协议（12 种AssistantMessageEvent+EventStream）、路由（models.ts的Models与Provider）、实现（api/*.ts每个文件一种线上协议）。 - 对外只有一个集合对象：
createModels()/builtinModels()造出来，方法分为查目录、管凭据、发请求三类。complete()就是stream().result()，包里没有第二条非流式代码路径。 - 事件协议的三条约定：每个事件带
partial；块可能交错、必须用contentIndex关联；终止只有done/error两种，且失败编码进流而不是抛异常。 EventStream是一个「推-拉桥接」：队列接住早到的事件，waiting 数组兑现晚到的消费者，finalResultPromise服务只要结果的调用方。- Anthropic 实现的完整链路：手写 SSE 解码 → 事件过滤与 JSON 修复 → 六类上游事件映射成统一事件 →
mapStopReason归一化停止原因；工具参数用parseStreamingJson做四级兜底的增量解析。 - faux provider 是一个真 provider，只把「发 HTTP」换成「读队列」，因此能在没有 API Key 的情况下跑通除网络之外的全部路径。
- 关键术语：统一消息类型（Unified Message）、流事件协议（
AssistantMessageEvent）、事件流缓冲（EventStream）、线上协议（wire protocol）、懒加载 API（lazyApi）、增量 JSON 解析（partial JSON）、faux provider。 - 关键源码索引：
packages/ai/src/types.ts:338-433（内容块与消息）、487-491（Context）、501-513（事件协议）、313-319（StreamFunction 契约）packages/ai/src/models.ts:127-187（Models）、463-510（applyAuth / stream / complete）、556-623（createProvider 与 dispatch）packages/ai/src/utils/event-stream.ts:4-88、packages/ai/src/utils/json-parse.ts:104-124packages/ai/src/api/anthropic-messages.ts:387-444（SSE）、573-745（事件映射）、1325-1351（停止原因）、801-841（streamSimple）packages/ai/src/providers/faux.ts:308-541、packages/ai/src/providers/all.ts:87-137- 测试：
packages/ai/test/anthropic-sse-parsing.test.ts、packages/ai/test/faux-provider.test.ts
- 自测问题：① 为什么说 pi-ai 里没有非流式代码路径？② 模型拒答（refusal）时，
await models.complete(...)会抛异常吗？返回的消息里哪个字段能看出来？③ 工具参数流到一半时，partial里的arguments是什么？由哪个函数保证它一定是个对象？④ faux provider 和「把 stream 函数换成一个假函数」相比，多验证了哪些环节？ - 下一章：6.2 Provider 与模型注册——38 个内置 provider 是怎么组装的、
models.generated.ts这份模型目录从哪来、自定义 provider 需要提供什么。本章尚未展开的内容还有：transformMessages的跨模型接力（历史消息如何在换模型时被改写）、models-store.ts的动态模型目录持久化、以及 OAuth 登录流程——前两者在 6.2 讲，OAuth 属于支线，本书只在 7.4 自定义 Provider 顺带提及。