pi-ai:统一多-provider LLM 层

30 秒导读: pi-ai 是 Pi 项目的最底层基座——一个 TypeScript 库,把 40 多家大模型 供应商(Anthropic、OpenAI、Google、Bedrock、Mistral、xAI、DeepSeek、各种网关……)、 8 种互不兼容的 HTTP/WebSocket 线缆协议,统一成一个函数调用:给它一个 model 和一段 对话 Context,它返回一个标准化的流式事件流。上层的 agent 循环(见 02-agent-loop) 因此完全不用关心底下是谁、说的是哪种"方言"。

---

1. 这是什么(零基础也能懂)

一句话定义: pi-ai 是一个"大模型万能翻译插座"——你按同一种方式插进去,它负责把请求 翻译成每家供应商各自的电压和插脚。

解决什么问题 / 给谁用。 假设你在写一个编码 agent,想让用户既能用 Claude、也能用 GPT、 还能用本地跑的 Llama。三家的 HTTP 接口完全不一样:

• Anthropic 用 /v1/messages,thinking(思考)是一种独立的 content block。
• OpenAI 有两套接口:老的 /chat/completions 和新的 /responses,reasoning 的表达方式又不同。
• Google 用 generateContent,消息角色叫 model 不叫 assistant。
• Amazon Bedrock 走的是 AWS SigV4 签名的 Converse Stream,连认证方式都不一样。

如果每加一家供应商,你的 agent 主循环就得改一次,那是灾难。pi-ai 就是把这堆差异吸收进一层, 让上层只面对一个稳定接口。

它能做什么(功能):

• 规范化的会话数据模型:一套 Context / Message / AssistantMessage,所有供应商共用。
• 每种线缆协议一个适配器:把规范模型 ↔ 该协议的 JSON 来回翻译。
• 统一的流式接口:stream() / streamSimple() 返回同一种 AssistantMessageEventStream。
• 内置的模型目录(40+ provider、数百个模型),带价格、上下文窗口、能力标记。
• 认证解析:OAuth token、API key、AWS profile、环境变量……自动挑对的那个。
• 成本 / token 统计、重试分类、上下文溢出检测等横切工具。

用起来什么样。 最小调用(示意,真实 API 见下文):

// 示意,非源码 —— 展示"上层眼里 pi-ai 有多简单"
import { builtinModels } from "@earendil-works/pi-ai/providers/all";

const models = builtinModels();                       // 装好所有内置 provider
const model = models.getModel("anthropic", "claude-opus-4-7");

// 同一个调用,换成 openai / google / bedrock 只需改上面两行
const stream = models.streamSimple(model, {
  systemPrompt: "You are helpful.",
  messages: [{ role: "user", content: "你好", timestamp: Date.now() }],
});

for await (const event of stream) {
  if (event.type === "text_delta") process.stdout.write(event.delta);
}
const final = await stream.result();                  // 完整 AssistantMessage

一句话直觉 / 类比。 把 pi-ai 想成旅行万能转换插头:你的电器(agent 循环)只有一种插头, 墙上的插座(供应商)有几十种。转换插头内部对每种墙插各有一套铜片布局(适配器),但对你露出的 永远是同一个孔。

---

2. 顶层全景(它大概怎么转)

pi-ai 的整个价值,浓缩成一句话:"规范模型"进,"规范事件流"出,中间用一层可替换的适配器 兑换成某家供应商的方言。 全景如下(从左到右是一次请求的数据流):

上层(agent 循环)
      │  给出: Model + Context(规范模型)
      ▼
┌───────────────────────────────────────────────────────┐
│  Models 集合 (models.ts)                                │
│   · 按 model.provider 找到 Provider                     │
│   · 解析认证 (auth/): OAuth? api-key? AWS profile?      │
│   · 把 apiKey/headers/baseUrl 注入 options              │
└───────────────────────────────────────────────────────┘
      │  委托给拥有该 model 的 Provider
      ▼
┌───────────────────────────────────────────────────────┐
│  Provider (providers/anthropic.ts …)                    │
│   · 绑定到某个 "API 格式" (= wire 协议适配器)           │
│   · 按 model.api 分发到对应适配器                       │
└───────────────────────────────────────────────────────┘
      │  model.api = "anthropic-messages"
      ▼
┌───────────────────────────────────────────────────────┐
│  API 适配器 (api/anthropic-messages.ts …)               │
│   规范 Context ──► 该协议 JSON  (buildParams)          │
│   供应商 SSE 分片 ──► 规范事件  (convertMessages 反向)  │
└───────────────────────────────────────────────────────┘
      │  逐块推送
      ▼
   AssistantMessageEventStream(统一事件流)──► 回到上层

部件一句话职责:

部件	干什么	在哪个文件
规范数据模型	定义所有供应商共用的会话类型	packages/ai/src/types.ts
API 适配器	每种 wire 协议一个,做规范↔方言的双向翻译	packages/ai/src/api/*.ts
Provider	把"某家供应商"绑定到"某个 API 格式"	packages/ai/src/providers/*.ts
Models 集合	注册 provider、解析认证、按 model 分发请求	packages/ai/src/models.ts
api-registry	旧版全局 stream() 的 api→实现 注册表	packages/ai/src/compat.ts
事件流	统一的异步可迭代 + 最终结果 promise	packages/ai/src/utils/event-stream.ts
模型目录	生成的静态目录(价格/窗口/能力)	packages/ai/src/models.generated.ts
认证	凭据存储 + OAuth/api-key 解析	packages/ai/src/auth/*.ts

主线走一遍(高层,不进代码):

1. 上层拿一个 Model(描述"我要用哪家的哪个模型")和一个 Context(系统提示 + 消息 + 工具)。
2. Models.streamSimple(model, context) 按 model.provider 找到对应 Provider,先解析认证。
3. Provider 看 model.api(如 "anthropic-messages"),分发给对应的 API 适配器。
4. 适配器把规范 Context 翻译成该协议的请求 JSON,发出去,再把回来的 SSE 分片翻译回规范事件。
5. 上层拿到的永远是同一种 AssistantMessageEventStream——不管底下是谁。

两条平行的入口。 pi-ai 有两套并存的分发路径:新路径是面向对象的 Models / Provider(models.ts),旧路径是全局函数 stream() + api-registry(compat.ts, 文件头自称"临时兼容入口")。本章两者都讲,因为它们共享同一批适配器和同一套数据模型。

---

3. 核心原理(逐个机制,由浅入深)

3.1 规范化的会话数据模型(一切的地基)

它要解决的小问题。 每家供应商的消息 JSON 都长得不一样。如果上层直接面对这些 JSON, 换供应商就得重写。所以第一步是:定义一套中立的、谁都能映射过去的会话类型。

思路 / 直觉。 找出"对话"这件事的最小公共骨架:一段系统提示、一串消息、一组工具;消息分 用户、助手、工具结果三种;助手消息的内容是"文本 / 思考 / 工具调用"三类块的序列。任何供应商的 格式都能双向映射到这套骨架。

核心类型一览(packages/ai/src/types.ts):

类型	是什么	关键字段
Context	一次请求的全部输入	systemPrompt / messages / tools
Message	三种消息的联合	UserMessage | AssistantMessage | ToolResultMessage
AssistantMessage	模型的一轮输出	content[] / usage / stopReason / api / provider
ToolResultMessage	工具执行结果回填	toolCallId / content[] / isError
Usage	token 与成本核算	input / output / cacheRead / cacheWrite / cost
Model<TApi>	描述"用哪家哪个模型"	provider / api / baseUrl / cost / contextWindow

助手消息的内容是"内容块序列",这是关键设计:

// 示意,非源码 —— 重点看 content 是三种块混排的数组
interface AssistantMessage {
  content: (TextContent | ThinkingContent | ToolCall)[];  // 顺序有意义
  api: Api;          // 这条消息是"哪种协议"产出的
  provider: string;  // "哪家"产出的
  model: string;     // "哪个模型"产出的
  usage: Usage;
  stopReason: "stop" | "length" | "toolUse" | "error" | "aborted";
}

真实定义见 types.ts:383 AssistantMessage、types.ts:322 TextContent、 types.ts:328 ThinkingContent、types.ts:344 ToolCall。

关键细节 / 巧妙处:

• 每条 AssistantMessage 都记着 api/provider/model 三个来源标签(types.ts:386-388)。 这不是冗余——它让"同一段对话历史里混着不同模型产出的消息"成为可能(会话中途换模型), 下游据此判断某条历史消息是不是"本模型自己产的",决定要不要保留只有本模型认得的签名。
• ThinkingContent 带 thinkingSignature 和 redacted(types.ts:328-336):思考内容常常是 供应商返回的不透明加密串,回放给同一模型时必须原样带回,跨模型时必须丢弃——数据模型 预留了这些字段,翻译层才有地方安放。
• Usage.cost 是结构化的(types.ts:366-372),按 input/output/cacheRead/cacheWrite 分项, 由 calculateCost() 依据模型单价填充(见 §3.5)。

3.2 "API 格式"适配层:一种 wire 协议一个适配器

它要解决的小问题。 有了规范模型,还需要把它翻译成每家真实的 HTTP 请求,并把真实的 流式响应翻译回来。这是全项目工程含量最高的地方——难点不是"发个请求",而是"把几十种细节 差异一一对齐,还不让上层看见"。

思路 / 直觉。 关键洞察:"供应商"和"线缆协议"不是一一对应的。 很多供应商其实说的是 同一种协议方言,只是 URL 和认证不同。所以 pi-ai 不按"供应商"切,而按协议切——一种 wire 协议写一个适配器,几十家供应商复用这 8 个适配器。

8 种内置 API 格式(types.ts:15 的 KnownApi,每个对应 api/ 下一个文件):

API 格式 (api id)	适配器文件	谁在用(举例)
anthropic-messages	api/anthropic-messages.ts	Anthropic、Bedrock 上的 Claude、GitHub Copilot、Fireworks
openai-completions	api/openai-completions.ts	一大堆 OpenAI 兼容供应商(DeepSeek、Groq、xAI、Together…)
openai-responses	api/openai-responses.ts	OpenAI 新接口
openai-codex-responses	api/openai-codex-responses.ts	OpenAI Codex(可走 WebSocket)
azure-openai-responses	api/azure-openai-responses.ts	Azure 托管的 OpenAI
google-generative-ai	api/google-generative-ai.ts	Google Gemini(API key)
google-vertex	api/google-vertex.ts	Vertex AI(ADC/项目)
mistral-conversations	api/mistral-conversations.ts	Mistral
bedrock-converse-stream	api/bedrock-converse-stream.ts	Amazon Bedrock(SigV4)

每个适配器模块只需导出两个函数,这就是统一契约(types.ts:222-225 ProviderStreams):

// 示意,非源码 —— 每个 api/*.ts 都满足这个形状
interface ProviderStreams {
  stream(model, context, options?): AssistantMessageEventStream;        // 全功能,带 provider 专属 options
  streamSimple(model, context, options?): AssistantMessageEventStream;  // 简化,只认统一的 reasoning 等级
}

真实签名:api/anthropic-messages.ts:468 的 export const stream 和 :767 的 export const streamSimple。其余 7 个适配器结构完全一致(见前面 grep 出的一排 export const stream)。

适配器内部干两件事,以 anthropic 适配器为例:

1. 规范 → 方言(出站)。 buildParams()(anthropic-messages.ts:901)把 Context 装成 Anthropic SDK 的 MessageCreateParamsStreaming:系统提示拆成带 cache_control 的 text 块、 消息经 convertMessages()(:1011)映射、工具转成 Anthropic tool schema、thinking 按 adaptive / budget 两种模式配置。
2. 方言 → 规范(入站)。 消费 SDK 的 RawMessageStreamEvent 分片,增量拼出 AssistantMessage 的 content 块,并沿途 push 出统一的 AssistantMessageEvent。

transformMessages:一段被所有适配器共享的"历史消息清洗" (api/transform-messages.ts:64)。这是适配层最精华的一块通用逻辑,它在正式翻译前先把消息历史 "打扫干净",专治跨模型 / 跨协议回放的坑:

• 不支持视觉的模型,图片降级成占位文本(downgradeUnsupportedImages,transform-messages.ts:35), 避免把 base64 图片喂给纯文本模型报错。
• 跨模型时,只有本模型认得的东西要处理:redacted 加密思考直接丢、带签名的思考块保留、 普通思考转成纯文本、Google 的 thoughtSignature 删掉(transform-messages.ts:97-142)。判断依据 正是 §3.1 说的那三个来源标签(isSameModel)。
• 工具调用 ID 归一化:OpenAI Responses 的 ID 可长达 450+ 字符带 |,而 Anthropic 只接受 ^[a-zA-Z0-9_-]+$ 且 ≤64 字符(normalizeToolCallId,anthropic-messages.ts 内);映射表保证 对应的 toolResult 也改成同样的新 ID。
• 给"孤儿工具调用"补合成结果(transform-messages.ts:160):如果历史里有工具调用却没有对应 结果(比如上一轮被中断),就插一条 isError: true 的 "No result provided",满足多数 API "工具调用必须配工具结果"的硬性要求。
• 跳过 error/aborted 的助手消息(transform-messages.ts:192):失败的半截回合不回放,免得触发 "reasoning without following item" 这类 API 错误。

buildBaseOptions / simple-options.ts:统一 options 的降解。 streamSimple 拿到的是中立的 SimpleStreamOptions(只有 reasoning: "low"|"medium"|... 这种等级),得翻译成各家的具体参数。 api/simple-options.ts:21 的 buildBaseOptions 收拢公共字段;:51 的 adjustMaxTokensForThinking 把统一的 reasoning 等级换算成 token 预算(minimal=1024 … high=16384),再塞进 max_tokens。 Anthropic 的 streamSimple(anthropic-messages.ts:767)就是调这些函数,把等级映射成 effort (adaptive 模型)或 budget(老模型)。

3.3 provider 绑定到 API 格式:注册与分发

它要解决的小问题。 适配器是按协议切的,但用户想的是"我要用 anthropic 这家"。需要一层 把"供应商 id"绑到"某个 API 格式",并处理这家的 baseUrl、认证、模型列表。

思路 / 直觉。 一个 Provider = 供应商元数据(id/name/baseUrl)+ 认证方式 + 模型列表 + 一个(或一组)API 适配器。绑定极其轻量——看 anthropic provider 的全部代码:

// providers/anthropic.ts:7 —— 一个 provider 就这么点
export function anthropicProvider(): Provider<"anthropic-messages"> {
  return createProvider({
    id: "anthropic",
    baseUrl: "https://api.anthropic.com",
    auth: { apiKey: envApiKeyAuth(...), oauth: lazyOAuth(...) },  // 见 §3.6
    models: Object.values(ANTHROPIC_MODELS),
    api: anthropicMessagesApi(),        // ← 绑定到 anthropic-messages 适配器
  });
}

分发机制(新路径,models.ts):

• createProvider()(models.ts:323)接受 api 参数,可以是单个适配器(所有模型走它), 也可以是一张 model.api → 适配器 的映射表(混合协议的供应商)。内部的 dispatch() (models.ts:333)按 model.api 挑适配器;挑不到就返回一个"该协议无实现"的错误流。
• Models 集合(models.ts 的 ModelsImpl)持有一堆 provider。请求进来时:requireProvider() 按 model.provider 找到 provider →applyAuth()(models.ts:230)解析并注入认证 →委托 provider.stream()。builtinModels()(providers/all.ts:111)一次性把 35 个内置 provider 全 setProvider 进去。

分发机制(旧路径,compat.ts 的 api-registry):

这是一个更"扁"的全局注册表,给旧版全局 stream() 函数用:

                     apiProviderRegistry (Map<api, 实现>)
  "anthropic-messages" ─► anthropicMessagesApi()
  "openai-completions" ─► openAICompletionsApi()
  "google-generative-ai" ─► googleGenerativeAIApi()
        …8 个内置,启动时 registerBuiltInApiProviders() 灌入

• ApiProvider 结构:{ api, stream, streamSimple }(compat.ts:77)。
• registerApiProvider()(compat.ts:120)登记时用 wrapStream 加一道 model.api === api 的断言。
• 全局 stream()(compat.ts:237)按 model.api 从注册表取实现;若模型是内置模型,则改走 新路径的 compatModels(compat.ts:208 的 builtinModels()),否则用 api-registry + 环境变量 API key 注入(withEnvApiKey,compat.ts:214)。

一句话记住两条路径的分工: 新路径按 provider 分发(先找家、再找协议),旧路径按 api 分发(直接找协议)。两者最终都落到同一批 api/*.ts 适配器上。

3.4 统一的流式接口:一次调用,一个事件流

它要解决的小问题。 各家的流式响应格式天差地别(Anthropic 的 content_block_delta、OpenAI 的 chunk、Google 的 candidates…)。上层想要一种流,边收边处理,最后还能拿到完整结果。

思路 / 直觉。 定义一个既能 for await 增量迭代、又能 await .result() 拿终态的对象: AssistantMessageEventStream。所有适配器都往里 push 同一套事件。

统一事件协议(types.ts:453-465 的 AssistantMessageEvent):

start ──► (text_start ─ text_delta* ─ text_end)
          (thinking_start ─ thinking_delta* ─ thinking_end)
          (toolcall_start ─ toolcall_delta* ─ toolcall_end)  ──► done | error

• *_delta 携带增量文本;每个事件都带一份 partial(到目前为止拼好的 AssistantMessage)。
• 流必然以 done(成功,stopReason ∈ stop/length/toolUse)或 error(失败,携带 stopReason error/aborted 的消息)终止——契约写在 types.ts:447-452 的注释里。

底层实现极简(utils/event-stream.ts):EventStream<T,R>(:4)是一个手写的 异步生产者-消费者队列——push 时若有等待的消费者就直接投递,否则入队;result() 返回一个在 "完成事件"到来时兑现的 promise。AssistantMessageEventStream(:69)只是指定"完成 = done/error 事件、结果 = 里面那条 message"。

关键细节 / 巧妙处:

• lazyStream:先给流,再异步 setup(api/lazy.ts:39)。认证解析、动态 import 适配器都是 异步的,但 stream() 的签名是同步返回一个流。lazyStream 的办法是:立刻造一个空流交出去, 背后 setup().then(转发).catch(把错误编码成 error 事件)。于是"setup 失败"也变成流里的一个 error 事件,而不是抛异常——上层永远只需处理流,不需要 try/catch(types.ts:298-303 的契约)。
• lazyApi:适配器按需加载(api/lazy.ts:63)。anthropicMessagesApi() (api/anthropic-messages.lazy.ts:4)其实只是 lazyApi(() => import("./anthropic-messages.ts"))。 真正的适配器(以及它拖着的 @anthropic-ai/sdk)首次调用才 import,让浏览器 / 精简构建 不必打包用不到的供应商 SDK。

3.5 重试分类 与 上下文溢出检测(两个纯函数横切工具)

它要解决的小问题。 请求会失败。上层需要判断:这个失败该不该重试?是不是上下文塞爆了 (该压缩历史而不是重试)?但各家的错误文案五花八门。

思路 / 直觉。 pi-ai 不实现重试策略(那是上层的事),只提供两个分类纯函数,把几十种 供应商错误文案归成"可重试 / 溢出 / 都不是"。

• isRetryableAssistantError()(utils/retry.ts:91):对着两张正则表——先排除"额度耗尽 / 账单"这类不可重试的限额错误(retry.ts:7),再匹配 overloaded / 429 / 5xx / 网络断开 / "stream ended before message_stop" 等可重试的瞬时错误(retry.ts:26)。注释里每条正则都标了 它专治哪家哪个 issue。
• isContextOverflow()(utils/overflow.ts:126):用 OVERFLOW_PATTERNS(overflow.ts:35) 匹配二十来种"prompt 太长"文案,还处理静默溢出——有的供应商(z.ai)不报错而是照单全收, 就靠 usage.input > contextWindow 判断;还有(Xiaomi MiMo)截断输入后 stopReason: "length" 且 output=0,也单独识别。

关键细节。 这两个函数是"错误方言"的翻译器,和数据方言、协议方言是同一种哲学:把供应商的 不一致吸收进一层可测试的纯函数,上层只看归一化后的布尔结论。

3.6 认证:一个凭据抽象吃下 OAuth / api-key / 环境变量 / AWS

它要解决的小问题。 认证方式比协议还乱:Anthropic 可 OAuth 也可 API key;Bedrock 用 AWS profile / IAM / bearer;Vertex 用 ADC 文件;还有一堆用环境变量。要用一套抽象表达全部。

思路 / 直觉。 把认证拆成三层:凭据存储(存什么)、认证方式(怎么解析)、解析编排 (先查谁后查谁)。

• ProviderAuth(auth/types.ts:179):每个 provider 声明自己支持 apiKey 和/或 oauth。
• ApiKeyAuth / OAuthAuth(auth/types.ts:129 / :154):前者 resolve() 出 apiKey;后者 拆成 refresh(换 token)+ toAuth(从凭据推出请求认证),让上层能把"加锁刷新"逻辑收在一处。
• CredentialStore(auth/types.ts:47):按 provider id 存一条凭据,唯一写路径是 modify() ——一个序列化的读-改-写,保证并发请求不会重复刷新同一个 OAuth token。

解析顺序(auth/resolve.ts:40 的 resolveProviderAuth),一句话:"存了的凭据说了算,没存 才看环境。"

请求带了显式 apiKey?  ──► 用它
        │否
存储里有凭据?
   ├─ oauth 型 ──► resolveStoredOAuth:过期就"双检加锁"刷新一次(auth/resolve.ts:86)
   └─ api_key 型 ──► 用存的 key
        │都没有
Ambient 兜底 ──► 环境变量 / AWS profile / ADC 文件

具体到环境变量(env-api-keys.ts):getApiKeyEnvVars()(:64)是一张 "provider → 环境变量名"的映射(OPENAI_API_KEY、GEMINI_API_KEY……),anthropic 特意让 ANTHROPIC_OAUTH_TOKEN 优先于 ANTHROPIC_API_KEY。Bedrock / Vertex 这类"环境里没显式 key 但有 凭据文件"的,getEnvApiKey()(:136)返回哨兵字符串 "<authenticated>" 表示"已配好"。

便捷工厂: envApiKeyAuth()(auth/helpers.ts:9)一行造出"存储 key 优先、否则按序试环境变量" 的标准 ApiKeyAuth;lazyOAuth()(auth/helpers.ts:34)让 provider 定义能"宣告支持 OAuth"却 不立刻 import 实现(Node-only 的 OAuth 流程首次用到才加载)。utils/oauth/ 是具体实现, src/oauth.ts 只是它的 re-export。

3.7 模型目录与生成物:代码即目录

它要解决的小问题。 数百个模型的价格、上下文窗口、能力标记,手写会过时会出错。

思路 / 直觉。 不手写目录,而是从上游元数据生成。 scripts/generate-models.ts 从 models.dev 拉元数据(并对若干 provider 现拉 /models 端点,如 nvidia、openrouter),生成:

• providers/<name>.models.ts:每个 provider 一份,导出 XXX_MODELS 常量(如 anthropic.models.ts:6 的 ANTHROPIC_MODELS),每个模型是一个 satisfies Model<...> 的字面量。
• models.generated.ts:把上面全部汇总成一张 MODELS 大表(models.generated.ts:40), getBuiltinModel()(providers/all.ts:48)据此做类型安全的静态查表。
• 图像模型走平行的一套:scripts/generate-image-models.ts → image-models.generated.ts → image-models.ts 的注册与查询。

为什么这么设计? 生成的目录是普通 TS 源码——可被类型检查、可 tree-shake、可 grep、 上游更新只需重跑脚本。文件头都印着"auto-generated, do not edit"(models.generated.ts:1)。

注意:目录只是元数据。 Model 只描述"这个模型是什么、多少钱、走哪个 api";真正发请求的是 §3.2 的适配器。目录只收窄到支持工具调用的模型(README 明说),因为 Pi 是 agent 框架。

---

4. 深入实现(一次 streamSimple 的端到端走读)

以 models.streamSimple(anthropicModel, ctx) 为例,把新路径串起来:

1. ModelsImpl.streamSimple(models.ts:278)立刻 return lazyStream(model, async () => {...}) ——同步交出空流,后台跑 setup。
2. setup 里先 requireProvider(model)(models.ts:222)按 model.provider="anthropic" 取 provider; 取不到抛 ModelsError("provider", ...),被 lazyStream 的 catch 编码成 error 事件。
3. applyAuth(models.ts:230)调 resolveProviderAuth,拿到 { apiKey, headers, baseUrl }, 按字段合并进 options(显式请求值优先,headers/env 逐键 merge)。若 auth 带了 baseUrl (如 Copilot 每凭据不同的端点),就 {...model, baseUrl} 覆盖。
4. provider.streamSimple(requestModel, ctx, requestOptions) → createProvider 里的 dispatch (models.ts:333)按 model.api="anthropic-messages" 挑到 anthropic 适配器,调其 streamSimple。
5. 适配器 streamSimple(anthropic-messages.ts:767):assertRequestAuth 校验 → buildBaseOptions 收拢公共 options → 按 reasoning 映射成 effort/budget → 调本模块的 stream()。
6. 适配器 stream(anthropic-messages.ts:468):建 AssistantMessageEventStream,后台 async 里 createClient(选 OAuth Bearer 还是 api-key,注入 Claude Code 身份头 / beta 头)→ buildParams 翻译请求 → 消费 SDK 流式事件 → 逐块 push 统一事件 → 终止时 push done/error。
7. 中途每一层的异步失败都被 lazyStream 的 catch 兜住,统一变成流里的 error 事件。上层只 for await + await stream.result()。

内层还有一层重试。 适配器把 maxRetries / timeoutMs 透传给供应商 SDK(如 Anthropic SDK 默认重试 2 次,见 types.ts:160-172 的 options 注释)。所以重试有两级:SDK 内建的瞬时重试, 和上层基于 §3.5 isRetryableAssistantError 的回合级重启——pi-ai 只负责前者的透传和后者的分类。

---

5. 巧妙之处(可借鉴的技术)

• 按"协议"切而非按"供应商"切。 8 个适配器复用给 40+ 供应商,新增一家 OpenAI 兼容供应商 往往只需加一份 *.models.ts + 一个 provider 工厂,零新代码。补充手段是 openai-completions 适配器的 detectCompat()(api/openai-completions.ts:1179)——从 baseUrl 自动识别该套哪种 兼容开关(z.ai/together/moonshot/openrouter…),把"同协议下的细微方言"再收一层。
• "先给流、后台 setup"消灭了同步/异步的阻抗失配(api/lazy.ts:39)。所有失败都成为流里的 一个 error 事件,上层代码路径只有一条。
• 三个来源标签让"会话中途换模型"可行(types.ts:386-388 + transform-messages.ts 的 isSameModel):历史里混着不同模型的输出也能安全回放,只把"只有原模型认得的签名"择出来。
• OAuth 双检加锁刷新(auth/resolve.ts:86):乐观判过期→加锁内再判一次→全局只刷一次→存回轮换 后的凭据。并发请求不会打爆刷新端点。有效 token 零加锁开销。
• 错误分类是纯函数 + 带 issue 注释的正则表(utils/retry.ts、utils/overflow.ts):把"供应商 错误方言"当成又一种要翻译的东西,可单测、可增量维护。
• 目录即生成源码:类型安全、可 tree-shake、可 grep,靠重跑脚本跟上游。

---

6. 边界与局限

• 只收支持工具调用的模型(README 明说)——纯文本 completion 模型不在目录里。
• pi-ai 不做重试策略、不做上下文压缩、不做 agent 循环:它只提供分类工具和一次请求的翻译。 重试预算、历史压缩、工具执行都在上层(见 02、03)。
• compat.ts 是"临时兼容入口"(文件头自述),计划随 coding-agent 的 ModelManager 迁移删除; 新代码应走 createModels() + provider 工厂。
• 溢出/重试检测依赖错误文案匹配,对未收录的自定义供应商可能漏判(overflow.ts 注释给了 自行添加正则的步骤)。
• 静默溢出天然不可靠:靠 usage.input > contextWindow 猜,供应商若不报 usage 就无从判断。

---

7. 横向对比

pi-ai 与常见"统一 LLM 层"(如 LiteLLM、Vercel AI SDK)的取舍差异:

维度	pi-ai	典型同类
切分维度	按 wire 协议切适配器,供应商复用	多按供应商切
数据模型	自有规范模型,消息带 api/provider/model 来源标签	多贴近 OpenAI 消息格式
流式	自有 AssistantMessageEventStream + "先给流后台 setup"	多用 Web ReadableStream / async generator
认证	内建 OAuth 加锁刷新 + AWS/ADC ambient 兜底	多只覆盖 api-key
目录	从 models.dev 生成的静态 TS 目录	多为运行时拉取或手维护

Pi 生态内部,pi-ai 是最底座:02-agent-loop 消费它的 stream() 跑 agent 循环,03-harness-context 在其上做会话树与上下文压缩。本章只讲到"一次 请求怎么翻译并流回",循环怎么用留给 02。

---

8. 代码地图(导航索引)

主题	文件	关键符号
规范数据模型	packages/ai/src/types.ts	Context · AssistantMessage · Usage · Model · AssistantMessageEvent · ProviderStreams
消息清洗(跨模型回放)	packages/ai/src/api/transform-messages.ts	transformMessages · downgradeUnsupportedImages
统一 options 降解	packages/ai/src/api/simple-options.ts	buildBaseOptions · adjustMaxTokensForThinking
Anthropic 适配器	packages/ai/src/api/anthropic-messages.ts	stream · streamSimple · buildParams · convertMessages
适配器懒加载契约	packages/ai/src/api/lazy.ts	lazyStream · lazyApi
适配器 api id 集合	packages/ai/src/compat.ts	KnownApi(types.ts)· ApiProvider · registerApiProvider · stream
Provider / Models 分发	packages/ai/src/models.ts	createProvider · Provider · ModelsImpl · applyAuth · dispatch
provider 工厂(示例)	packages/ai/src/providers/anthropic.ts	anthropicProvider
全部内置 provider	packages/ai/src/providers/all.ts	builtinProviders · builtinModels · getBuiltinModel
事件流实现	packages/ai/src/utils/event-stream.ts	EventStream · AssistantMessageEventStream
重试分类	packages/ai/src/utils/retry.ts	isRetryableAssistantError
上下文溢出检测	packages/ai/src/utils/overflow.ts	isContextOverflow · OVERFLOW_PATTERNS
认证解析	packages/ai/src/auth/resolve.ts	resolveProviderAuth · resolveStoredOAuth
认证类型	packages/ai/src/auth/types.ts	ProviderAuth · ApiKeyAuth · OAuthAuth · CredentialStore
认证工厂	packages/ai/src/auth/helpers.ts	envApiKeyAuth · lazyOAuth
环境变量 key	packages/ai/src/env-api-keys.ts	getEnvApiKey · getApiKeyEnvVars
模型目录(生成)	packages/ai/src/models.generated.ts · providers/*.models.ts	MODELS · ANTHROPIC_MODELS
目录生成脚本	packages/ai/scripts/generate-models.ts	(从 models.dev 生成)