M04

第4章：模型调用 —— 一行代码驾驭多个模型

5921字 · 含 169 行代码 · 约 30 分钟

第 3 章我们追踪了 Agent Loop 的完整运转过程。其中最关键的一步是”调用模型”——Loop 把消息发给 LLM，拿到回复，再结合回复里的 toolCall、硬停止原因和宿主消息队列决定是否续转。

但当时我们用了一行代码就跳过了：

const stream = streamSimple(model, context, options);

这行代码看起来平平无奇。但如果你打开 packages/ai/src/api/ 目录，会发现里面有几十个文件、10 种 API 标识（9 种文本 API，另有 1 个 images 专用 API），各适配器的规模和复杂度并不相同。一行调用的简洁背后，是整个 AI 抽象层在承担协议差异。

这一章就来拆开这一行代码：Pi 怎么做到用同一个接口跨多种 Provider 调用不同模型？如果你想接入一个新模型，需要做什么？

校对口径：本章对应 Pi v0.80.2 的 packages/ai。该版本定义 9 种文本 KnownApi 和 35 个 KnownProvider 标识；Provider 标识包含区域或产品变体，不能直接解释成 35 家独立厂商。源码基线为 0201806。

一、问题：同一段对话，不同模型要”翻译”成不同格式

先说清楚要解决什么问题。

Agent Loop 要调用模型，而 v0.80.2 需要适配多种 API 协议与 Provider 配置——例如 Anthropic、OpenAI、Google、AWS Bedrock、Mistral。协议之间会用不同字段名和数据结构描述相近概念。

这个差异有多大？我们看一个最简单的例子。假设用户对 Agent 说了一句话：

“帮我读一下 main.ts”

这条消息在 Pi 内部是这样存的（统一格式）：

{ role: "user", content: "帮我读一下 main.ts", timestamp: 1748697600000 }

但要把这条消息发给不同 API，就必须”翻译”成各家要求的格式。纯文本这个最小例子里，有些协议表面上相同；一旦加入图片或工具结果，分歧会进一步扩大：

Anthropic Messages（Claude）——Pi 对纯文本使用字符串；包含图片时才转换成带 type 的内容块数组：

{ role: "user", content: "帮我读一下 main.ts" }
// 若同一条消息还带图片，content 才是 [{ type: "text", ... }, { type: "image", ... }]

OpenAI（GPT）——格式看起来差不多，但字段语义有细微差异（比如工具结果消息的处理方式完全不同）：

{ role: "user", content: "帮我读一下 main.ts" }
// 但如果消息里包含工具结果，OpenAI 要求单独的 { role: "tool" } 消息，
// 而 Anthropic 把工具结果合并到 user 消息里

Google（Gemini）——字段名从 content 变成了 parts，结构更扁平：

{ role: "user", parts: [{ text: "帮我读一下 main.ts" }] }

Bedrock（AWS）——在 AWS 自己的结构上又包了一层：

{ role: "user", content: [{ text: "帮我读一下 main.ts" }] }
// 注意：Bedrock 的 text 没有 type 字段，和 Anthropic 不一样

纯文本已经出现三种请求形状。 Anthropic 与 OpenAI 在这个最小例子里恰好都是字符串；Google 改用 parts[]，Bedrock 使用无 type 的 content[]。图片、工具调用和工具结果才会暴露更多协议差异。

配图说明：顶部是 Pi 内部统一格式，下面并列四种 API 的请求形状。Anthropic 纯文本会被适配器压成字符串，含图片时才使用带 type 的块数组；OpenAI 的差异主要在工具结果的独立 role: "tool"；Google 用 parts，Bedrock 的内容块没有 type 字段。

不只是消息格式——四个维度全都不一样

消息格式只是冰山一角。API 协议之间的差异是全方位的，主要在四个维度上：

维度	差在哪	举个例子
消息格式	同一条消息，字段名和结构不同	Anthropic 纯文本用 `content: string`、富内容用块数组；Google 用 `parts[]`
流式传输	增量返回机制不同	Anthropic 适配器读取原始 SSE，OpenAI 适配器消费 SDK 的结构化 chunk
思考模式	”让模型深度思考”的参数完全不同	Anthropic 用 `thinking.budget_tokens`，OpenAI 用 `reasoning_effort`
缓存控制	”标记不变内容”的方式不同	Anthropic 打 `cache_control` 标记；受支持的 Bedrock Claude 模型插入 `cachePoint` 节点

每个维度单独看都不复杂，但多种 API 协议、Provider 配置与模型能力组合起来，差异量就很大。

现在问题来了：Agent Loop 只有一行 streamSimple(model, context)，它不可能为每种 API 协议写一套不同的逻辑。那它怎么应对这么多差异？

二、解法：三层架构，各管一件事

一个自然的答案是：给所有 API 协议做一层封装，让它们看起来都一样——输入相同、输出相同。

没错，Pi 正是这么做的。 但具体怎么”让它们看起来都一样”？Pi 的方案是把这件事拆成三层，每层有明确的分工：

第一层 · 统一入口    →  "接收请求，查出该找谁处理"
第二层 · 事件协议    →  "约定输出格式——不管谁处理，交回来的都是这个样子"
第三层 · API 适配器   →  "真正干活的人——每个适配器精通一种 API 协议"

用一个类比来理解。想象一个国际翻译公司：

第一层（前台）：客户进门，前台问”你要翻译什么语言？“，然后查通讯录找到对应的翻译员，把工作派给他
第二层（标准报告模板）：不管翻译员翻译的是法语、日语还是阿拉伯语，最终交出来的报告都必须用公司统一的格式——封面、正文、审签栏，格式固定
第三层（翻译员们）：每个翻译员精通一种 API 协议，内部怎么翻译是他们的事，但输出必须符合第二层的标准格式

关键点：前台（第一层）和翻译员（第三层）之间靠标准报告（第二层）连接。前台不需要懂外语，翻译员不需要懂公司流程。

Agent Loop 就是那个”客户”——它把请求交给前台（第一层），拿到标准格式的报告（第二层），从不直接接触翻译员（第三层）。

配图说明：请求从 Agent Loop 进入顶层 streamSimple()：内置 Models 路径先按 Provider 找到运行时单元，compat 注册表路径则按 model.api 查找适配器；适配器的本地 streamSimple() 做参数映射后调用同模块的 stream()。响应方向则相反：各适配器按 12 种事件协议输出，统一事件流返回 Agent Loop。注册表不是第二次进入的中转站。

下面逐层展开。

第一层：统一入口

入口函数叫 stream()。下面保留源码里的两条路由路径：

// compat.ts
export function stream(model, context, options?) {
  if (shouldUseBuiltinModels(model)) {
    return compatModels.stream(model, context, options); // 内置模型兼容路径
  }

  const provider = resolveApiProvider(model.api);        // 注册表路径
  return provider.stream(
    model,
    context,
    withEnvApiKey(model, options),                       // 补入环境中的 API key
  );
}

对普通注册模型，主线仍是查表，派活；但入口并非只有这两句：内置模型可以走 compatModels 快捷路径，注册表路径还会尝试补入环境中的 API key。model.api 是一个 API 协议标识（如 "anthropic-messages"），系统启动时已经把内置适配器注册进”通讯录”。

这个”通讯录”长这样：

const BUILTIN_APIS = [
  ["anthropic-messages",       anthropicMessagesApi()],      // Claude 的翻译器
  ["openai-completions",       openAICompletionsApi()],      // GPT 的翻译器
  ["google-generative-ai",     googleGenerativeAIApi()],     // Gemini 的翻译器
  ["bedrock-converse-stream",   bedrockConverseStreamApi()], // Bedrock 的翻译器
  // ... 还有 5 个
];

左边是 key（“工号”），右边是翻译器（“翻译员”）。model.api 的值就是 key——拿 key 查表，找到翻译器，调用它。入口层不做消息或响应协议翻译；它负责选择调用路径、查找适配器，并在注册表路径补充环境凭据。

第二层：事件协议

翻译器把请求发给模型后，模型会以大小不一的增量片段流式返回内容。如果每个翻译员都按自己的方式输出，前台就无法统一处理。所以第二层约定了一套统一的输出格式。

Pi 规定：不管底层是哪家模型，翻译器都必须输出统一的事件流，事件类型一共 12 种：

AssistantMessageEvent（12 种）
│
├── start                              ← 流开始了
│
├── text_start → text_delta → ... → text_end       ← 模型在输出文字
├── thinking_start → thinking_delta → ... → thinking_end  ← 模型在思考
├── toolcall_start → toolcall_delta → ... → toolcall_end  ← 模型要调工具
│
├── done   (reason: stop / length / toolUse)   ← 正常结束
└── error  (reason: error / aborted)           ← 出错了

你不需要记住全部 12 种。只需要理解模式：模型回复的内容分三类（文字、思考、工具调用），每类都有“开始→增量→结束”三步，再加上流开始和流结束两个信号。

每类内容为什么是三步？因为这是流式输出——翻译器先发 text_start，再发送一个或多个 text_delta，最后发 text_end。delta 是 Provider 给出的增量块，不保证恰好对应一个 token 或一个字符；Agent Loop 仍可据此持续刷新 UI。

还有一个重要约定：start 以及 text / thinking / toolcall 的开始、增量和结束事件都携带 partial: AssistantMessage——当前消息的完整快照。终态 done / error 则分别携带最终 message / error。第 3 章讲的”原地替换”用的是前一类事件的 partial，终态再用最终消息收口。

第三层：翻译器

翻译器是真正干活的角色。每个翻译器对应一种 API 协议（如 Anthropic Messages、OpenAI Completions），负责把统一格式翻译成该 API 的请求格式，再把响应翻译回统一事件。Provider 则是更具体的运行时单元：它拥有模型目录、认证和端点配置，多个 Provider 可以复用同一个 API 适配器。

可以把内置翻译器的共同职责归纳成下面 5 步；具体文件未必机械地拆成五个同名函数：

配图说明：垂直 5 步管道——创建客户端→构建请求参数→发送请求→处理响应流→发送终止事件。第 2 步和第 4 步是工作量所在（红色边框高亮）。右侧失败分支汇聚成统一 error 事件。

翻译器(model, context, options)
│
├── 1. 创建客户端
│      初始化客户端与凭据。可能是 API Key、OAuth、AWS 凭据/区域等。
│
├── 2. 构建请求参数
│      把统一格式的消息、工具定义、系统提示，翻译成 API 的请求格式。
│      比如 Google 要 content → parts，这一步就做这个转换。
│
├── 3. 发送请求
│      通过 SDK 或直接 HTTP 发给模型。等模型开始响应。
│
├── 4. 处理响应流
│      模型流式返回内容。翻译器把 API 的私有事件格式，
│      翻译成第二层要求的 12 种统一事件。
│
└── 5. 发送终止事件
       成功 → push done；失败 → push error。流必须终止。

第 2 步（请求翻译）和第 4 步（响应翻译）通常是核心工作量所在；不同 API 的兼容分支数量不同，适配器规模也因此不同。

以 Anthropic 为例，第 4 步的翻译规则（把 Anthropic 的私有事件名翻译成 Pi 的统一事件名）：

Anthropic 私有事件                     →  Pi 统一事件
─────────────────                    ────────────
content_block_start (type: "text")    →  text_start
content_block_delta (text_delta)      →  text_delta
content_block_start (type: "tool_use") →  toolcall_start
content_block_delta (input_json)      →  toolcall_delta
message_delta (stop_reason)           →  更新 output.stopReason
原始流读取结束                            →  done（携带最终消息）

注意终止原因的映射：Anthropic 的 "end_turn" → Pi 的 "stop"，"tool_use" → "toolUse"。不同 API 的命名不一样，Pi 统一成了自己的术语。Agent Loop 只把 "error" / "aborted" 当作硬停止；其余响应仍从 content 提取 toolCall，再结合整批 terminate 和宿主队列决定是否续转，不能把 stopReason 当作唯一循环条件。

StreamFunction：翻译器的”入职要求”

前面说的三层架构要运转起来，有一个前提：所有翻译器必须遵守同一套规则。 Pi 用一个 TypeScript 类型来定义这套规则——它叫 StreamFunction，是整个抽象层的”宪法”：

export type StreamFunction<TApi extends Api, TOptions> = (
  model: Model<TApi>,       // 用哪个模型
  context: Context,         // 对话上下文（系统提示 + 消息 + 工具）
  options?: TOptions,       // 可选配置（思考级别、缓存等）
) => AssistantMessageEventStream;  // ← 必须返回统一事件流

这个签名定义了三条规则：

输入相同：所有翻译器接受同样的三个参数（model、context、options）
输出相同：必须返回 AssistantMessageEventStream——不管底层是原始 SSE 还是 SDK 流，对外都是这个类型
内置适配器把请求/流错误编码成事件：失败时尽量发 { type: "error" }，让消费端沿同一条流处理

第 3 条是内置适配器的运行时约定，不是 StreamFunction 类型能强制保证的性质。请求或读流失败时，适配器会把异常信息包装成 error 事件，并设置 stopReason: "error" 或 "aborted"；但调用方配置错误、事件监听器异常或自定义适配器主动抛错，仍可能越过这条边界。

有了 StreamFunction 这套规则，第一层可以按 model.api 把工作派给已注册的翻译器。它们共享请求与事件流的外形；错误如何进入终态事件则是内置适配器遵循、但类型本身无法强制的运行时约定。

三、怎么用：从调用到接入新模型

理解了三层架构，现在看实际使用。分两个场景：怎么调用已有模型，和怎么接入一个新模型。

场景一：调用模型

Agent Loop 实际调用的不是 stream()，而是 streamSimple()。为什么有两个版本？

stream() 是底层入口——它负责选择调用路径并分派到底层适配器，不做思考级别翻译。streamSimple() 会分派到对应适配器的便捷实现；该实现负责把统一 ThinkingLevel 翻译成具体 API 参数，部分预算型适配器还会连带调整 token 上限。简单说：

stream()：底层统一入口，仍会按 model.api 路由、补环境凭据，并由适配器完成请求与事件翻译；调用方要自己提供 API 特定选项，不享受统一思考级别映射
streamSimple()：路由到适配器的便捷实现，由它把统一思考级别翻译成 API 参数，是 Agent Loop 使用的入口

Agent Loop 使用 streamSimple() 的典型代码：

const stream = streamSimple(model, context, { reasoning: "high" });
//                                          ↑ 告诉它"用高级别思考"
//                                            streamSimple 会自动翻译成该 API 的具体参数

for await (const event of stream) {
  // 协议保证 start 在增量事件之前，并以 done 或 error 收口；
  // thinking / text / toolcall 块是否出现、以何种顺序出现由模型输出决定。
  switch (event.type) {
    case "text_delta":
      // 文字增量，显示到终端
      break;
    case "toolcall_end":
      // 模型要调工具，拿到完整的工具调用信息
      break;
    case "done":
      // 本轮模型调用结束；Loop 再看 toolCall 内容、terminate 与宿主队列
      break;
    case "error":
      // 出错了（网络超时、API错误等），event.reason 是 "error" 或 "aborted"
      break;
  }
}

Agent Loop 不需要为每种 API 写一套分支。 streamSimple() 返回统一事件流，所以主消费路径相同；模型能力、终止原因细节与 Provider 的认证、端点特性仍由适配层和模型元数据表达。

场景二：接入新模型——先判断是否真有新协议

“新模型”不等于”新 API”。动手前先问：它是否兼容 Pi 已支持的某种协议？

情况 A：复用已有 API 协议。 如果模型提供 OpenAI-compatible、Anthropic Messages 等已支持接口，就不必再写 StreamFunction。已有 Provider 增加一个模型时，通常只需补模型元数据；若同时新增 Provider，还要把认证、base URL 和模型列表接到现有 API 适配器上。下面是一个字段完整、可由 TypeScript 检查的 OpenAI-compatible 模型示例：

import type { Model } from "@earendil-works/pi-ai";

const yourModel = {
  id: "your-model-id",
  name: "Your Model",
  api: "openai-completions",
  provider: "your-provider",
  baseUrl: "https://api.your-provider.example/v1",
  reasoning: false,
  input: ["text"],
  cost: {
    input: 0,
    output: 0,
    cacheRead: 0,
    cacheWrite: 0,
  },
  contextWindow: 128_000,
  maxTokens: 16_384,
} satisfies Model<"openai-completions">;

这里的价格 0 只是合法占位，实际接入必须填写真实费率；如果兼容端点有行为差异，还应通过 compat 描述。关键是 api: "openai-completions" 直接复用现有协议实现，而 provider 仍标识具体的认证与端点归属。

情况 B：引入一种新的 API 协议。 只有请求结构或流式协议无法由现有适配器表达时，才需要做完整的三件事：

实现该 API 的 stream 与 streamSimple：把 Pi 消息翻译成请求，并把响应翻译成 12 种统一事件；请求、模型或读流失败应以 error 终态收口
把适配器挂到 Provider；如果仍使用 compat 全局入口，还要用 registerApiProvider() 把新的 api 标识注册到通讯录
添加完整的 Model<"your-model-api"> 元数据，以及 Provider 的认证、base URL 和模型列表

registerApiProvider({
  api: "your-model-api",
  stream: yourStreamFunction,
  streamSimple: yourSimpleFunction,
});

完成这些接线后，Agent Loop 的主流程通常不需要修改。分层带来的收益不是”每个新模型都写一遍翻译器”，而是同一协议的模型复用适配器，只有新协议才扩展适配层。

四、【进阶】翻译器内部：SSE 解析与思考模式方言

前三节覆盖了 AI 抽象层的核心设计。以下内容属于工程实现细节——如果你不需要自己写翻译器，可以跳过这一节。

流式传输：为什么 Anthropic 和 OpenAI 的解析方式完全不同？

第 3 章讲过，模型回复是流式的——内容以增量片段陆续到达。但各家的片段格式不同：

OpenAI 的适配器直接消费 SDK 返回的结构化 chunk 流。每个 chunk 里的 choices[0].delta 已经是解析后的数据。

Anthropic 适配器同样使用官方 SDK 创建请求，但调用 .asResponse() 取得原始 Response，随后自己逐行解析 SSE 的 event: / data: 字段，并对 JSON 做修复与完整性检查。

Anthropic 的解析链路（SDK 发请求，适配器解码原始响应）：
  SDK .asResponse()
    → 逐行读取
    → 分离 event 和 data
    → JSON 解析（含容错）
    → 内部事件

OpenAI 的解析链路（直接消费 SDK chunk）：
  client.chat.completions.create()
    → AsyncIterable<Chunk>
    → 结构化数据

为什么解析路径不统一？ SDK 能暴露的控制面不同，Pi 还要兼容自定义 base URL、OAuth 变体和流式边界检查。于是有的适配器直接迭代 SDK chunk，有的通过 SDK 拿到原始响应后自行解码。但不管内部怎么实现，最终都翻译成统一的 12 种事件。

思考模式：四种”思考”，四种参数

不同 API 对”让模型深度思考”这件事，概念和参数并不相同：

// Anthropic：给一个 token 预算，让模型在这个预算内思考
params.thinking = { type: "enabled", budget_tokens: 16384 };

// OpenAI：给一个努力程度（low/medium/high）
params.reasoning_effort = "high";

// Google Gemini 3 / Gemma 4：用大写 thinkingLevel
config.thinkingConfig = { includeThoughts: true, thinkingLevel: "HIGH" };

// 较早的 Gemini 模型走预算语义
config.thinkingConfig = { includeThoughts: true, thinkingBudget: 16384 };

甚至 Anthropic 自己都有两种模式——新版模型用”自适应思考”（模型自己决定思考多少），旧版模型用”预算思考”（给固定 token 上限）。

Pi 怎么统一：5 个思考级别，6 个可选状态

Pi 定义了一套统一的思考级别枚举：

配图说明：ThinkingLevel 本身是 minimal/low/medium/high/xhigh 五级；ModelThinkingLevel 再加入 off，所以调用侧一共看到六种状态。图中的 token 数只适用于映射到固定预算的模型，不能当成所有 Provider 的统一上限。

  off    minimal    low    medium    high    xhigh
  │        │         │       │        │        │
 不思考    └────────────── 统一语义级别 ──────────────┘

  固定预算适配器的默认示例：
          1024 tk  2048 tk  8192 tk  16384 tk
  （xhigh 以及各档实际参数由模型能力和适配器映射决定）

上层代码只需说 reasoning: "high"，适配层再根据模型能力和 thinkingLevelMap 翻译。没有映射表时还会使用 API 适配器的默认语义，因此不能把表理解为每个模型都必填。

如果请求的级别模型不支持（比如请求 xhigh 但只支持到 high），clampThinkingLevel() 函数做回退——先向上找（思考更多通常比更少安全），找不到再向下找。

streamSimple() 是这套翻译的统一便捷入口，但实现落在各 API 适配器里：通常会 clamp / 映射 reasoning；只有固定预算等需要为思考预留输出空间的适配器才调整 maxTokens，不能把它当作所有适配器共有的固定步骤。

五、【进阶】缓存控制与错误处理

缓存控制：让模型”算更少”

Agent 对话是”有状态的”——每轮都把之前的完整历史发给模型。如果你和 Agent 聊了 50 轮，每轮都重新计算前 49 轮的内容。缓存就是告诉模型服务器：“这些内容没变，别重新算了。”

但”怎么标记”各 API 不同：Anthropic 在消息块上追加 cache_control 标记；Bedrock 只对检测为支持缓存的 Claude 模型插入独立 cachePoint 节点（也可由兼容配置强制开启）。

Pi 把缓存控制抽象成三个语义级别：

type CacheRetention = "none" | "short" | "long";

上层只说”我要 long 缓存”，翻译器自己翻译成各 Provider 的具体标记方式。这和 ThinkingLevel 是同一种”统一枚举 + 各家翻译表”的套路，但比思考级别更有意思——四家对”怎么标记不变内容”这件事，思路完全不同。源码里的对照表大致是这样：

Provider	`none`	`short`	`long`	打点位置
Anthropic	不打标记	`cache_control: { type: "ephemeral" }`（默认 5 分钟 TTL）	加 `ttl: "1h"`（仅新模型支持）	system 末尾 + 最后一个 tool + 最后一条 user message（rolling）
Bedrock（受支持的 Claude 模型）	不插节点	在消息流里插一个独立对象 `{ cachePoint: { type: DEFAULT } }`	加 `ttl: ONE_HOUR`	system 块后面 + 最后一条消息后面
OpenAI Responses	不发 cache key	发送 `prompt_cache_key: sessionId`	加 `prompt_cache_retention: "24h"`	不打点——OpenAI 按 session key 自动匹配前缀
OpenAI 兼容（仅启用 `cacheControlFormat: "anthropic"`）	不打标记	按 Anthropic 风格打 `cache_control`	加 `ttl: "1h"`（如兼容端支持）	system + 最后一个 tool + 最后一条 user/assistant 文本

源码引用：Anthropic 的 system / message / tool 缓存标记在 anthropic-messages.ts:897-943 和 L1147-1200；Bedrock 的支持检测见 bedrock-converse-stream.ts:629-653，system 与 message 的 cachePoint 分别见 L668-683 和 L865-875；OpenAI Responses 的 prompt cache 参数在 openai-responses.ts:240-241；OpenAI-compatible 的条件开关与三处标记在 openai-completions.ts:729-834。

这里有个值得停下来想一下的对比：同样是”告诉服务器这段内容不变”，四家给出了四种完全不同的协议设计——

Anthropic 的做法像”贴便签”：在原有的内容块上附加一个 cache_control 字段。内容还是那些内容，只是多了一个”这段不变”的标签
Bedrock 的做法像”插路标”：在消息流的特定位置插入一个独立的、不含业务数据的 cachePoint 对象。它不依附在任何内容上，自己就是一个占位
OpenAI 原生的做法像”按会员卡号查记录”：你压根不打标记，只是每次请求带一个 prompt_cache_key（Pi 用 sessionId 当 key），OpenAI 后端自己识别前缀重合度
部分 OpenAI 兼容端点会显式声明 cacheControlFormat: "anthropic"；只有这类配置才会走 applyAnthropicCacheControl。不能把所有 DeepSeek、Qwen 或 OpenAI-compatible 端点一概归入这一行

第 3 章提到 Loop 每圈都会重新构造 llmContext。这不等于缓存一定失效：cacheRetention 让适配器在相应位置持续发送缓存标记，只要 Provider 认定的前缀仍匹配，就有机会复用缓存；滚动增长的消息和实际字节变化仍会影响命中。

这三个位置对应两类内容：相对稳定的 system/tools 前缀，以及随对话推进的最近 user 消息。它是一种适配 Provider 限制的实现选择，不应写成对所有工作负载都“最优”。

缓存可能显著降低重复前缀的延迟与费用，但折扣、TTL、写入价和命中规则会随 Provider 与模型变化。Pi 只负责记录模型价目和发送协议字段；实际账单应以对应 Provider 的当期文档为准。

错误处理：编码到流里，由 Loop 统一硬停

内置文本适配器会把调用异常归一化为终态 error 事件；可用下面的共同骨架理解：

try {
  // ... 正常流程：构建请求、发送、解析响应
  stream.push({ type: "done", reason: output.stopReason, message: output });
} catch (error) {
  // 错误不抛出，而是编码到流中
  output.stopReason = options?.signal?.aborted ? "aborted" : "error";
  output.errorMessage = error.message;
  stream.push({ type: "error", reason: output.stopReason, error: output });
}

这就是第 3 章所说 stopReason: "error" 和 "aborted" 的注入点。 模型 API 的原生错误先被适配器规范化为流事件；Agent Loop 收到后会发完当前 Turn 的结束事件并硬停止。重试或降级不是这段 Loop 自动完成的，需要更外层策略显式实现。

还有一个隐蔽的问题：上下文溢出。有时候请求没报错，但模型返回空输出——一种可能是输入 token 太多被服务端静默截断。Pi 的 isContextOverflow() 用三类线索做 best-effort 检测（错误消息模式匹配、token 数对比、输出为零且以 length 停止）。它能收敛多种已知表现，但无法保证识别未知错误文案、自定义 Provider 或没有可靠信号的静默截断。

六、回到那一行代码

回到开头的问题：streamSimple(model, context) 这一行代码背后发生了什么？

用一句话概括：Agent Loop 说了一句”给我用这个模型处理这段对话”，前台查出该找哪个翻译器，翻译器把请求翻译成 Provider 的格式发给模型，再把模型的流式响应翻译成统一事件返回——Agent Loop 自始至终只看到了统一事件流，不知道中间发生了多少翻译。

Agent Loop：streamSimple(model, context, { reasoning: "high" })
    │
    │  ① 顶层 streamSimple 按 Provider / model.api 分派
    │
    │  ② 适配器的 streamSimple clamp / 映射 reasoning；
    │     仅部分预算型适配器调整 maxTokens
    │
    │  ③ 调用同一适配器模块里的本地 stream()
    │     （不会重新进入顶层 stream() 或再查一次注册表）
    │
    │  ④ 适配器工作：
    │     · 统一格式 → API 请求格式（请求翻译）
    │     · 发给模型
    │     · API 私有响应 → 12 种统一事件（响应翻译）
    │
    │  ⑤ 返回 AssistantMessageEventStream
    │
    └── Agent Loop：for await (event of stream) { ... }  ← 消费统一事件

Agent Loop 最终看到的只有 ⑤——一个干净的事件流。 这就是三层架构的力量：复杂度被封装在翻译器里，外部接口保持简洁。

设计精华

三条核心思路值得带走：

1. “协议 > 基类”设计法。 Pi 没有要求翻译器继承 BaseProvider，而是定义事件协议（12 种事件）和 StreamFunction 签名。各适配器仍可共享辅助函数，但无需为了复用而塞进同一继承树；契约只约定输入输出，中间可以按协议特点组织。

2. “统一语义 + 能力映射”策略。 ThinkingLevel 有 5 级，ModelThinkingLevel 加上 off 共 6 种状态；ThinkingLevelMap 可覆盖具体模型映射。上层表达意图，适配层负责降级和协议翻译。

3. “语义统一，实现分散”。 cacheRetention（none/short/long）是语义接口——上层说”我要长期缓存”，不管底层是打标记还是插节点。语义接口描述”做什么”，机制接口描述”怎么做”。

七、下一站

AI 层把大部分协议差异收敛到统一接口——Agent Loop 主要消费 Model、Context 与标准事件，不直接处理各家 SSE 或请求结构。

但 Agent 怎么”做事”？模型返回了 ToolCall，谁来执行？执行过程中怎么转换并校验参数，宿主又能在哪里接入自己的权限或确认策略？第 3 章把这个过程当作黑盒跳过了。

下一章，我们打开这个黑盒——工具系统。

本章关键源码索引：

packages/ai/src/compat.ts:237-247 — stream() 入口（第一层·底层）

packages/ai/src/compat.ts:258-268 — streamSimple() 入口（第一层·便捷封装）

packages/ai/src/compat.ts:172-206 — Provider 注册（“通讯录”）

packages/ai/src/types.ts:304-308 — StreamFunction 签名（“宪法”）

packages/ai/src/types.ts:447-459 — 12 种 AssistantMessageEvent（第二层·事件协议）

packages/ai/src/api/anthropic-messages.ts — Anthropic 翻译器（第三层·5 步骨架）

packages/ai/src/api/openai-completions.ts — OpenAI 翻译器

packages/ai/src/types.ts:74-76 — ThinkingLevel / ThinkingLevelMap

packages/ai/src/models.ts:410-429 — clampThinkingLevel 回退策略

packages/ai/src/utils/overflow.ts:126-155 — isContextOverflow 三重检测