Python 阅读说明：本版与 TypeScript 版共享同一份事实与正文结构。下列 Python 代码只用于解释 TypeScript 源码的控制流，并非可安装的 Pi Python SDK；字段名和类型以链接的 v0.80.2 TypeScript 源码为准。
第 3 章我们追踪了 Agent Loop 的完整运转过程。其中最关键的一步是”调用模型”——Loop 把消息发给 LLM，拿到回复，再结合回复里的 toolCall、硬停止原因和宿主消息队列决定是否续转。
但当时我们用了一行代码就跳过了：
# ============================================================
# 【Python 改写】第3章里被跳过的那一行调用
# 原文 TS: const stream = streamSimple(model, context, options);
# ============================================================
stream = stream_simple(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 内部是这样存的（统一格式）：
# ============================================================
# 【Python 改写】Pi 内部统一格式的消息
# 原文 TS: { role: "user", content: "帮我读一下 main.ts", timestamp: 1748697600000 }
# ============================================================
{"role": "user", "content": "帮我读一下 main.ts", "timestamp": 1748697600000}但要把这条消息发给不同 API，就必须”翻译”成各家要求的格式。纯文本这个最小例子里，有些协议表面上相同；一旦加入图片或工具结果，分歧会进一步扩大：
Anthropic Messages（Claude）——Pi 对纯文本使用字符串；包含图片时才转换成带 type 的内容块数组：
# ============================================================
# 【Python 改写】Anthropic 消息格式
# 原文 TS:
# { role: "user", content: "帮我读一下 main.ts" }
# // 若同一条消息还带图片，content 才是带 type 的 block 数组
# ============================================================
{"role": "user", "content": "帮我读一下 main.ts"}
# 若同一条消息还带图片，content 才是带 type 的 block 列表OpenAI（GPT）——格式看起来差不多，但字段语义有细微差异（比如工具结果消息的处理方式完全不同）：
# ============================================================
# 【Python 改写】OpenAI 消息格式
# 原文 TS:
# { role: "user", content: "帮我读一下 main.ts" }
# // 但如果消息里包含工具结果，OpenAI 要求单独的 { role: "tool" } 消息，
# // 而 Anthropic 把工具结果合并到 user 消息里
# ============================================================
{"role": "user", "content": "帮我读一下 main.ts"}
# 但如果消息里包含工具结果，OpenAI 要求单独的 {"role": "tool"} 消息，
# 而 Anthropic 把工具结果合并到 user 消息里Google（Gemini）——字段名从 content 变成了 parts，结构更扁平：
# ============================================================
# 【Python 改写】Google Gemini 消息格式
# 原文 TS: { role: "user", parts: [{ text: "帮我读一下 main.ts" }] }
# ============================================================
{"role": "user", "parts": [{"text": "帮我读一下 main.ts"}]}Bedrock（AWS）——在 AWS 自己的结构上又包了一层：
# ============================================================
# 【Python 改写】Bedrock 消息格式
# 原文 TS:
# { role: "user", content: [{ text: "帮我读一下 main.ts" }] }
# // 注意：Bedrock 的 text 没有 type 字段，和 Anthropic 不一样
# ============================================================
{"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()。下面保留源码里的两条路由路径：
# ============================================================
# 【Python 改写】compat.ts 第一层入口
# 原文 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));
# }
# ============================================================
# 概念对照：TS 的可选参数 options? → Python 的 options=None；
# 入口先判断内置兼容路径，普通注册模型再"查表 + 补环境凭据 + 派活"
def stream(model, context, options=None):
if should_use_builtin_models(model):
return compat_models.stream(model, context, options)
provider = resolve_api_provider(model.api)
return provider.stream(model, context, with_env_api_key(model, options))对普通注册模型，主线仍是查表，派活；但入口并非只有这两句：内置模型可以走 compatModels 快捷路径，注册表路径还会尝试补入环境中的 API key。model.api 是一个 API 协议标识（如 "anthropic-messages"），系统启动时已经把内置适配器注册进”通讯录”。
这个”通讯录”长这样：
# ============================================================
# 【Python 改写】BUILTIN_APIS 注册表
# 原文 TS:
# const BUILTIN_APIS = [
# ["anthropic-messages", anthropicMessagesApi()],
# ["openai-completions", openAICompletionsApi()],
# ["google-generative-ai", googleGenerativeAIApi()],
# ["bedrock-converse-stream", bedrockConverseStreamApi()],
# // ... 还有 5 个
# ];
# ============================================================
# 概念对照：TS 的 [key, value] 元组数组 → Python 的 dict（或 list of tuples）
# 这就是个字符串 → API 实例 的映射
BUILTIN_APIS = {
"anthropic-messages": anthropic_messages_api(), # Claude 的翻译器
"openai-completions": openai_completions_api(), # GPT 的翻译器
"google-generative-ai": google_generative_ai_api(), # Gemini 的翻译器
"bedrock-converse-stream": bedrock_converse_stream_api(), # 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，是整个抽象层的”宪法”：
# ============================================================
# 【Python 改写】StreamFunction 签名（"宪法"）
# 原文 TS:
# export type StreamFunction<TApi extends Api, TOptions> = (
# model: Model<TApi>,
# context: Context,
# options?: TOptions,
# ) => AssistantMessageEventStream;
# ============================================================
# 概念对照：TS 的泛型类型别名 → Python 用 Callable[[...], ...] 或 Protocol；
# 这里用 Callable 表达"任何符合此签名的可调用对象"
from typing import Callable, Optional, TypeVar
from typing_extensions import Protocol
TApi = TypeVar("TApi", bound="Api")
TOptions = TypeVar("TOptions")
# 签名约定：输入三个参数（model, context, options），输出统一事件流
StreamFunction = Callable[
["Model[TApi]", "Context", Optional[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() 的典型代码：
# ============================================================
# 【Python 改写】Agent Loop 用 streamSimple 消费事件流
# 原文 TS:
# 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;
# }
# }
# ============================================================
# 概念对照：TS 的 for await...of → Python 的 async for...in；
# TS 的 switch → Python 的 if/elif（或 match/case，Python 3.10+）
stream = stream_simple(model, context, {"reasoning": "high"})
# ↑ 告诉它"用高级别思考"
# stream_simple 会自动翻译成该 API 的具体参数
async for event in stream:
# 协议保证 start 在增量事件之前，并以 done 或 error 收口；
# thinking / text / toolcall 是否出现、以何种顺序出现由模型输出决定。
if event.type == "text_delta":
# 文字增量，显示到终端
...
elif event.type == "toolcall_end":
# 模型要调工具，拿到完整的工具调用信息
...
elif event.type == "done":
# 本轮模型调用结束；Loop 再看 toolCall、terminate 与宿主队列
...
elif event.type == "error":
# 出错了（网络超时、API 错误等），event.reason 是 "error" 或 "aborted"
...Agent Loop 不需要为每种 API 写一套分支。 streamSimple() 返回统一事件流，所以主消费路径相同；模型能力、终止原因细节与 Provider 的认证、端点特性仍由适配层和模型元数据表达。
场景二：接入新模型——先判断是否真有新协议
“新模型”不等于”新 API”。动手前先问：它是否兼容 Pi 已支持的某种协议？
情况 A：复用已有 API 协议。 如果模型提供 OpenAI-compatible、Anthropic Messages 等已支持接口，就不必再写 StreamFunction。已有 Provider 增加一个模型时，通常只需补模型元数据；若同时新增 Provider，还要把认证、base URL 和模型列表接到现有 API 适配器上。下面是一个字段完整、可由 TypeScript 检查的 OpenAI-compatible 模型示例：
# ============================================================
# 【Python 改写】复用现有 API 协议的完整 Model 元数据
# 原文 TS:
# 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">;
# ============================================================
# 概念对照：Python 版没有真实 SDK；此字典只展示相同的必填元数据。
your_model = {
"id": "your-model-id",
"name": "Your Model",
"api": "openai-completions",
"provider": "your-provider",
"base_url": "https://api.your-provider.example/v1",
"reasoning": False,
"input": ["text"],
"cost": {
"input": 0,
"output": 0,
"cache_read": 0,
"cache_write": 0,
},
"context_window": 128_000,
"max_tokens": 16_384,
}这里的价格 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 和模型列表
# ============================================================
# 【Python 改写】注册一种新的 API 协议
# 原文 TS:
# registerApiProvider({
# api: "your-model-api",
# stream: yourStreamFunction,
# streamSimple: yourSimpleFunction,
# });
# ============================================================
register_api_provider({
"api": "your-model-api",
"stream": your_stream_function,
"stream_simple": your_simple_function,
})完成这些接线后，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 对”让模型深度思考”这件事，概念和参数并不相同：
# ============================================================
# 【Python 改写】四家 Provider 的思考模式参数对照
# 原文 TS:
# params.thinking = { type: "enabled", budget_tokens: 16384 };
# params.reasoning_effort = "high";
# config.thinkingConfig = { includeThoughts: true, thinkingLevel: "HIGH" };
# config.thinkingConfig = { includeThoughts: true, thinkingBudget: 16384 };
# ============================================================
# 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 把缓存控制抽象成三个语义级别：
# ============================================================
# 【Python 改写】CacheRetention 语义枚举
# 原文 TS: type CacheRetention = "none" | "short" | "long";
# ============================================================
# 概念对照：TS 的联合字符串字面量类型 → Python 用 Literal 或 Enum
from typing import Literal
CacheRetention = Literal["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 事件；可用下面的共同骨架理解：
# ============================================================
# 【Python 改写】翻译器错误处理通用模式
# 原文 TS:
# 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 });
# }
# ============================================================
# 概念对照：TS 的 try/catch → Python 的 try/except；
# TS 的三元运算符 → Python 的条件表达式
try:
# ... 正常流程：构建请求、发送、解析响应
stream.push({"type": "done", "reason": output.stop_reason, "message": output})
except Exception as error:
# 错误不抛出，而是编码到流中
output.stop_reason = "aborted" if (options and options.signal and options.signal.aborted) else "error"
output.error_message = str(error)
stream.push({"type": "error", "reason": output.stop_reason, "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/ThinkingLevelMappackages/ai/src/models.ts:410-429—clampThinkingLevel回退策略packages/ai/src/utils/overflow.ts:126-155—isContextOverflow三重检测