第四章：Provider 架构 -- 请求如何路由到正确的服务商
对应源文件：
packages/ai/src/api-registry.ts、packages/ai/src/models.ts、packages/ai/src/providers/register-builtins.ts
问题回顾
当你调用 stream(model, context) 时，系统需要回答一个问题：
这个 model 应该发给哪个 provider 适配器处理？
答案藏在 model.api 字段里。一个 model 的 api 字段告诉系统它使用哪种 API 协议。系统维护了一个注册表，把每种 API 协议映射到对应的处理函数。
API 注册表（API Registry）
什么是注册表模式
注册表模式是一种常见的设计模式。思路是：
- 维护一个"字典"（Map），key 是标识符，value 是处理逻辑
- 需要处理某个标识符时，从字典里查出对应的处理逻辑
- 新增处理逻辑只需要往字典里加一条记录
pi-ai 中的注册表：
注册表 (Map)
─────────────────────────────────────────
"anthropic-messages" → { stream: streamAnthropic, streamSimple: ... }
"openai-completions" → { stream: streamOpenAI, streamSimple: ... }
"openai-responses" → { stream: streamOpenAIResponses, streamSimple: ... }
"google-generative-ai" → { stream: streamGoogle, streamSimple: ... }
"mistral-conversations" → { stream: streamMistral, streamSimple: ... }
"bedrock-converse-stream" → { stream: streamBedrock, streamSimple: ... }
...
注册过程
每个注册项叫做一个 ApiProvider，它包含：
api：API 协议标识符stream：处理"原生选项"流式请求的函数streamSimple：处理"简化选项"流式请求的函数
注册就是往 Map 里放一条记录：
registerApiProvider({
api: "anthropic-messages",
stream: streamAnthropic,
streamSimple: streamSimpleAnthropic,
})
查找过程
当你调用 stream(model, context) 时：
1. 读取 model.api，比如 "anthropic-messages"
2. 在注册表中查找：registry.get("anthropic-messages")
3. 找到对应的 provider → { stream: streamAnthropic, ... }
4. 调用 provider.stream(model, context, options)
5. 返回 AssistantMessageEventStream
就这么简单。整个路由逻辑只有几行代码。
懒加载（Lazy Loading）
为什么需要懒加载
pi-ai 支持 20+ 个 provider，每个 provider 的适配器代码都不小（通常 1-3 万行）。如果启动时全部加载，会浪费时间和内存。实际上，用户通常只会用 1-2 个 provider。
懒加载的实现
pi-ai 的做法很巧妙：
- 启动时只注册一个"代理函数"（proxy），不加载实际的 provider 代码
- 当这个代理函数第一次被调用时，才去
import实际的 provider 模块 - 加载完成后，把实际模块的事件转发到调用者的事件流中
启动时注册的不是真正的 streamAnthropic，而是一个"懒代理"：
调用 stream() 时：
┌─────────────────────────────────────────────┐
│ 懒代理函数 │
│ 1. 创建一个空的 EventStream（outer） │
│ 2. 异步加载 anthropic.ts 模块 │
│ 3. 调用真正的 streamAnthropic() 得到 inner │
│ 4. 把 inner 的事件逐个转发到 outer │
│ 5. 立即返回 outer │
└─────────────────────────────────────────────┘
关键点：懒代理立即返回一个空的 EventStream，不需要等模块加载完成。调用者可以马上开始 for await，事件会在模块加载完成后陆续到达。
这是一种 "先承诺再兑现" 的设计：先给你一个"容器"，内容稍后到达。
加载失败怎么办
如果模块加载失败（比如某个 provider 的依赖没装），懒代理会生成一个 error 事件推到 outer stream 中。调用者会收到一个错误消息，而不是一个未捕获的异常。
这与前一章"错误不抛异常"的设计原则一致。
模型注册表（Model Registry）
API 注册表解决的是"如何路由到 provider"，模型注册表解决的是"有哪些模型可用"。
模型数据从哪来
models.generated.ts 是一个 384KB 的自动生成文件，包含所有已知 provider 的所有模型信息。它的结构大致是：
export const MODELS = {
"anthropic": {
"claude-sonnet-4-20250514": {
id: "claude-sonnet-4-20250514",
name: "Claude Sonnet 4",
api: "anthropic-messages",
provider: "anthropic",
baseUrl: "https://api.anthropic.com",
reasoning: true,
input: ["text", "image"],
cost: { input: 3, output: 15, cacheRead: 0.3, cacheWrite: 3.75 },
contextWindow: 200000,
maxTokens: 16384
},
// ... 更多 Anthropic 模型
},
"openai": {
"gpt-4o": { ... },
"gpt-4o-mini": { ... },
// ... 更多 OpenAI 模型
},
// ... 更多 provider
};
这个文件由 packages/ai/scripts/generate-models.ts 脚本生成。脚本会从各 provider 的 API 或文档中抓取最新的模型列表和定价信息。
模型查询
models.ts 提供了几个简单的查询函数：
// 获取所有 provider 列表
getProviders() → ["openai", "anthropic", "google", ...]
// 获取某个 provider 的所有模型
getModels("anthropic") → [model1, model2, ...]
// 获取一个具体模型
getModel("anthropic", "claude-sonnet-4-20250514") → Model 对象
这些函数都有完整的 TypeScript 类型提示。在 IDE 中输入 getModel(" 时会自动补全 provider 名称，输入第二个参数时会自动补全该 provider 下的模型 ID。
自定义模型
除了内置模型，用户还可以创建自定义模型。只要指定正确的 api 字段，注册表就知道用哪个适配器处理：
const myLocalModel = {
id: "llama-3.1-8b",
name: "Llama 3.1 8B (本地)",
api: "openai-completions", // 使用 OpenAI 兼容协议
provider: "ollama",
baseUrl: "http://localhost:11434/v1", // 本地 Ollama 服务
reasoning: false,
input: ["text"],
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 },
contextWindow: 128000,
maxTokens: 32000
};
// 直接传给 stream()，不需要注册
stream(myLocalModel, context);
因为 api: "openai-completions" 对应的适配器已经在注册表中了，所以这个自定义模型可以直接使用。
完整的调用链路
把前几章的内容串起来，一次完整的 LLM 调用的路径是：
你的代码
│
│ stream(model, context, options)
▼
stream.ts
│ 1. resolveApiProvider(model.api) → 在注册表中查找
│ 2. provider.stream(model, context, options)
▼
api-registry.ts
│ 3. 注册表返回懒代理函数
▼
register-builtins.ts (懒代理)
│ 4. 创建 outer EventStream
│ 5. 异步 import("./anthropic.js")
│ 6. 调用真正的 streamAnthropic(model, context, options)
▼
anthropic.ts (实际适配器)
│ 7. 把 Context 转换为 Anthropic API 的请求格式
│ 8. 发送 HTTP 请求到 api.anthropic.com
│ 9. 解析 Anthropic 的 SSE 响应
│ 10. 生成统一的 AssistantMessageEvent 序列
│ 11. push 到 EventStream
▼
返回给你的代码
│
│ for await (const event of stream) { ... }
消息转换（Message Transform）
在发送给 provider 之前，消息列表会经过一次 transformMessages() 处理：
它做了什么
-
图片降级：如果模型不支持图片输入（
model.input不包含"image"），自动把图片块替换为文字占位符："(image omitted: model does not support images)" -
思考块处理：
- 同模型回放：保留原始 thinking 块（因为有些 provider 需要 signature 用于多轮连续性）
- 跨模型切换：把 thinking 块转为普通文本（其他模型不认识 thinking 格式）
- 空思考块：直接丢弃
-
工具调用 ID 标准化：不同 provider 的 tool call ID 格式不同。比如 OpenAI Responses API 生成的 ID 有 450+ 字符且包含特殊字符，而 Anthropic 要求 ID 只能包含字母数字和短横线。需要跨 provider 回放对话时，就需要标准化这些 ID
-
孤立工具调用修复：如果对话历史中有个 assistant 调用了工具但没有对应的 toolResult（比如被中断了），自动插入一个合成的错误 toolResult，防止 API 报错
-
错误消息跳过：如果 assistant 消息的 stopReason 是 error 或 aborted，直接跳过不发给 LLM（因为不完整的消息可能导致 API 报错）
为什么需要消息转换
跨 provider 切换是 pi-ai 的核心卖点之一。用户在使用 pi 时可以中途切换模型（比如从 Claude 切到 GPT），之前的对话历史需要无缝衔接到新模型。消息转换层就是解决这个"翻译"问题的。
小结
┌──────────────────────────────────────────────────┐
│ 你的代码 │
│ stream(model, context) │
└──────────────────┬───────────────────────────────┘
│ model.api = "anthropic-messages"
▼
┌──────────────────────────────────────────────────┐
│ API 注册表 (Map) │
│ "anthropic-messages" → 懒代理 │
│ "openai-completions" → 懒代理 │
│ "google-generative-ai" → 懒代理 │
│ ... │
└──────────────────┬───────────────────────────────┘
│ 首次调用时加载模块
▼
┌──────────────────────────────────────────────────┐
│ Provider 适配器 (anthropic.ts) │
│ 1. transformMessages() 消息转换 │
│ 2. 构建 Anthropic API 请求体 │
│ 3. 发送 HTTP 请求 │
│ 4. 解析响应，生成统一事件序列 │
└──────────────────────────────────────────────────┘
下一章我们会深入看一个具体的 provider 适配器内部是怎么工作的。