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

第 3 章里，Agent Loop 调用模型只用了一行代码：

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

但如果你打开 packages/ai/src/api/ 目录，会发现里面有几十个文件、9 种对话 API 翻译器 + 1 种 images 专用翻译器，每个都上千行。一行调用的简洁背后，是整个 AI 抽象层的精心设计。

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

假设用户对 Agent 说"帮我读一下 main.ts"，Pi 内部统一格式是：

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

但要发给不同模型，必须翻译成各家要求的格式：

Anthropic：content 必须是数组，每项带 type——{ role: "user", content: [{ type: "text", text: "..." }] }
OpenAI：纯文本可以直接是字符串；但工具结果必须是独立的 { role: "tool" } 消息（Anthropic 则合并进 user 消息）
Google：字段名从 content 变成 parts——{ role: "user", parts: [{ text: "..." }] }
Bedrock：在 AWS 自己的结构上包一层，text 没有 type 字段

连最简单的纯文本消息都有四种写法。 而且消息格式只是冰山一角，Provider 的差异是全方位的：

维度	差在哪	例子
消息格式	字段名和结构不同	Anthropic 用 `content[]`，Google 用 `parts[]`
流式传输	"逐字返回"机制不同	有的用 SDK 结构化 chunk，有的是原始 SSE 文本要自己解析
思考模式	"深度思考"参数完全不同	Anthropic 用 `thinking.budget_tokens`，OpenAI 用 `reasoning_effort`
缓存控制	"标记不变内容"的方式不同	Anthropic 打 `cache_control` 标记，Bedrock 插入 `cachePoint` 节点

四个维度乘以 30 多个 Provider，组合差异量是巨大的。Agent Loop 不可能为每个 Provider 写一套逻辑——怎么办？

四个 Provider 的消息格式差异

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

用"国际翻译公司"类比：

第一层（前台/统一入口）：客户进门，查通讯录找到对应翻译员，把活派给他；
第二层（标准报告模板/事件协议）：不管翻译的是什么语言，交出来的报告必须用公司统一格式；
第三层（翻译员/翻译器）：每个翻译员精通一种方言，内部怎么翻是自己的事，输出必须符合第二层标准。

前台不需要懂外语，翻译员不需要懂公司流程——两者靠标准报告连接。 Agent Loop 就是那个客户，从不直接接触翻译器。

翻译公司三层架构

第一层：统一入口

// compat.ts
export function stream(model, context, options?) {
  const provider = resolveApiProvider(model.api);  // 查表：这个 model 该找谁？
  return provider.stream(model, context, options);  // 把活派给翻译器
}

两步：查表，派活。 "通讯录"就是一个注册表：

const BUILTIN_APIS = [
  ["anthropic-messages",       anthropicMessagesApi()],
  ["openai-completions",       openAICompletionsApi()],
  ["google-generative-ai",     googleGenerativeAIApi()],
  ["bedrock-converse-stream",  bedrockConverseStreamApi()],
  // ... 还有 5 个对话 API + 1 个 images 专用
];

model.api 的值就是 key。入口层不处理任何业务逻辑，纯粹是路由器。

第二层：事件协议

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

├── 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)             ← 出错

模式很简单：内容分三类（文字、思考、工具调用），每类都有"开始→逐字增量→结束"三步，再加流开始和流结束两个信号。 每个事件还携带 partial: AssistantMessage（当前消息的完整快照）——第 3 章讲的"原地替换"就是靠它：收到事件就用 partial 覆盖 context 最后一条消息，不用自己拼接增量。

第三层：翻译器

所有翻译器遵循同一个 5 步工作流程：

1. 创建客户端      用 API Key 初始化连接
2. 构建请求参数    统一格式 → Provider 私有格式（核心工作量之一）
3. 发送请求        通过 SDK 或直接 HTTP 发出
4. 处理响应流      Provider 私有事件 → 12 种统一事件（核心工作量之二）
5. 发送终止事件    成功 push done；失败 push error。流必须终止

翻译器内部 5 步骨架

第 2 步和第 4 步是每个翻译器上千行代码的主要原因。以 Anthropic 为例，第 4 步的翻译规则：

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)            →  done（终止原因映射）

终止原因也要映射：Anthropic 的 "end_turn" → Pi 的 "stop"，"tool_use" → "toolUse"。第 3 章 Agent Loop 检查的 stopReason，就是经过这层翻译后的统一术语。

StreamFunction：翻译器的"宪法"

export type StreamFunction<TApi extends Api, TOptions> = (
  model: Model<TApi>,
  context: Context,
  options?: TOptions,
) => AssistantMessageEventStream;   // ← 必须返回统一事件流

三条规则：输入相同（同样三个参数）、输出相同（不管底层是原始 SSE 还是 SDK 流，对外都是 AssistantMessageEventStream）、错误不抛异常（失败时不 throw，而是 push 一个 { type: "error" } 事件——第 3 章的 stopReason: "error" / "aborted" 就是翻译器的 catch 块注入的）。

怎么用

调用：stream vs streamSimple

stream() 是底层入口，只做查表派活；streamSimple() 在上面包了一层便捷功能——自动处理思考级别翻译（查表 + clamp + 调整 maxTokens）。实际开发用后者：

const stream = streamSimple(model, context, { reasoning: "high" });
//                                            ↑ streamSimple 自动翻译成各 Provider 的具体参数

for await (const event of stream) {
  switch (event.type) {
    case "text_delta":   /* 逐字渲染 */ break;
    case "toolcall_end": /* 拿到完整工具调用 */ break;
    case "done":         /* 看 stopReason 决定是否继续循环 */ break;
    case "error":        /* errorReason 是 "error" 或 "aborted" */ break;
  }
}

Agent Loop 不需要关心底层是哪个模型——Claude、GPT、Gemini 返回的都是同一种事件流，消费方式完全一样。

接入一个新模型：三步

写一个翻译器——一个符合 StreamFunction 签名的函数：请求方向把统一格式翻译成目标 API 格式，响应方向把流式响应翻译成 12 种统一事件，异常时 push error 事件而不 throw。这是工作量最大的一步，但是一次性工作。
注册翻译器——registerApiProvider({ api: "your-model-api", stream: yourStreamFunction, streamSimple: yourSimpleFunction })，往通讯录里加一条记录。
配置模型信息——Model 对象的 api 字段指向第 2 步注册的名字，第一层靠它查表路由。

完成。Agent Loop 一行都不用改，事件系统、会话管理、压缩算法全部自动适配——新模型的接入成本被限制在"写一个翻译器"这一个点上。

进阶：翻译器内部的两个硬骨头

流式解析：SDK 与原始 SSE

OpenAI 的 SDK 封装得很好：client.chat.completions.create() 直接返回结构化的 chunk 流，choices[0].delta 拿来就能用。Anthropic 虽然也有官方 SDK，但 Pi 的翻译器选择直接解析原始 SSE 文本流，以获得更细粒度的控制——自己逐行读 event: 和 data: 字段、自己做 JSON 解析（含容错处理，有些 chunk 的 JSON 可能不完整）。

不管内部怎么实现，最终都翻译成统一的 12 种事件——差异被完全关在翻译器内部。

思考模式：统一成 ThinkingLevel 六级刻度

各家"让模型深度思考"的参数完全不同：Anthropic 给 token 预算（thinking.budget_tokens，且新版模型还有自适应模式），OpenAI 给努力程度（reasoning_effort），Google 用 thinkingConfig。Pi 定义统一的六级刻度：

  off    minimal    low    medium    high    xhigh
  │        │         │       │        │        │
 不思考  ~1024 tk  ~2048 tk ~8192 tk ~16384 tk 模型最大值
       （各级对应 token 值为示意值，实际以各模型的 thinkingLevelMap 为准）

上层只说 reasoning: "high"，每个 Model 对象自带的 thinkingLevelMap（翻译表）声明"我这台模型每级对应什么参数"，翻译器查表翻译。请求的级别不支持时（如请求 xhigh 但只支持到 high），clampThinkingLevel() 做回退——先向上找（思考更多通常比更少安全），找不到再向下。

ThinkingLevel 六级思考刻度

缓存控制：四种完全不同的协议设计

上层只说 cacheRetention: "none" | "short" | "long"，翻译器翻译成各家的具体标记——四家的思路截然不同：

Provider	`short` 的做法	打点位置
Anthropic（贴便签）	内容块上附 `cache_control: { type: "ephemeral" }`	system 末尾 + 最后一个 tool + 最后一条 user message（rolling）
Bedrock（插路标）	消息流里插入独立的 `{ cachePoint }` 对象	system 块后 + 最后一条消息后
OpenAI Responses（按会员卡号查记录）	请求带 `prompt_cache_key: sessionId`，后端自动匹配前缀	不打点
OpenAI 兼容厂商（DeepSeek/Qwen 等）	直接复用 Anthropic 风格标记	复用 Anthropic 三处策略

源码位置：Anthropic 三处打点在 anthropic-messages.ts（L922/1208/1157），Bedrock 的 cachePoint 在 bedrock-converse-stream.ts:696，OpenAI 的 prompt_cache_key 在 openai-responses.ts:229。

这回答了第 3 章的悬念：每圈重建 llmContext 为什么不破坏 cache？ 因为 Anthropic 按内容字节做 prefix 匹配，system + tools 内容不变、标记位置一致，命中就是命中，跟"你是第几圈调用"无关。为什么选那三个位置打点？system + tools 是整条 Trace 最稳定的前缀，最后一条 user message 是 rolling 截止线；Anthropic 限每请求最多 4 个 breakpoint，这三个位置是收益最大化又不超限额的最优解。

缓存的经济价值很高：命中的 token 按缓存读取价计费，通常是正常输入价的 1/10——200K token 的对话历史，缓存后能省约 90% 的输入成本。

错误处理与上下文溢出

所有翻译器的错误处理是同一个模式：try 里正常流程末尾 push done；catch 里设 output.stopReason = signal?.aborted ? "aborted" : "error" 并 push error 事件。异常被编码进流，Agent Loop 收到后可以重试、降级或报告用户，循环永远不被异常打断。

还有个隐蔽问题：上下文溢出时有些 Provider 不报错，只是静默截断返回空输出。Pi 的 isContextOverflow() 做三重检测（错误消息模式匹配、token 数对比、输出为零 + length 停止），统一捕获各家的溢出表现。

设计精华

"协议 > 实现"：不用 BaseProvider 抽象类，而是定义事件协议 + StreamFunction 签名。为什么不用继承？翻译器之间几乎没有共同点——Anthropic 和 Google 连字段名都不一样，继承找不到共同代码；协议只约定输入输出，不管中间怎么处理。
"统一枚举 + 映射表"：ThinkingLevel 是统一枚举，每个 Model 自带翻译表。比"取各家交集"灵活（交集可能只剩 off），比"暴露原生参数"简洁（上层不用知道 20 种参数名）。
"语义统一，实现分散"：cacheRetention 是语义接口——上层说"我要长期缓存"，不管底层是打标记、插节点还是发 session key。

下一章打开另一个黑盒：模型返回 ToolCall 之后，工具系统怎么保证参数正确、执行安全？

本章关键源码索引：compat.ts:237-247（stream 入口）、compat.ts:172-206（Provider 注册表）、types.ts:304-308（StreamFunction）、types.ts:453-465（12 种事件）、models.ts:410-429（clampThinkingLevel）、utils/overflow.ts:126-155（isContextOverflow）。本文改编自 CC-BY-SA-4.0 许可的开源教程。

模型调用：一行代码驾驭多个模型

带着问题读