第五章：Provider 适配器内部 -- 以 Anthropic 为例

对应源文件：packages/ai/src/providers/anthropic.ts

一个 provider 适配器要做什么

每个 provider 适配器本质上是一个"翻译器"，需要完成两个方向的翻译：

发送方向：pi-ai 统一类型  →  provider 原生 API 格式
接收方向：provider 原生响应  →  pi-ai 统一事件序列

具体来说，一个适配器需要实现以下步骤：

第一步：消息转换（Outbound）

把 pi-ai 的 Context（systemPrompt + messages + tools）转换为 provider 特定的请求格式。

例如，Anthropic API 要求：

• system prompt 单独放在顶层 system 字段
• messages 数组只包含 user 和 assistant 消息
• tool results 嵌套在 user 消息内部
• thinking 块需要特殊的 content block 格式
• 工具定义放在顶层 tools 数组

而 OpenAI API 要求：

• system prompt 是 messages 数组中 role 为 system（或 developer）的消息
• tool results 是独立的 role: "tool" 消息
• 工具定义放在顶层 tools 数组，schema 包在 function.parameters 里

每个适配器都有大量的"格式转换"代码来处理这些差异。

第二步：发送 HTTP 请求

大多数 LLM API 使用 SSE（Server-Sent Events） 协议进行流式传输。SSE 是 HTTP 之上的一种简单协议：

HTTP/1.1 200 OK
Content-Type: text/event-stream

event: message_start
data: {"type":"message_start","message":{...}}

event: content_block_delta
data: {"type":"content_block_delta","delta":{"text":"你好"}}

event: content_block_delta
data: {"type":"content_block_delta","delta":{"text":"世界"}}

event: message_stop
data: {"type":"message_stop"}

每行以 data: 开头，后面是一个 JSON 对象。事件之间用空行分隔。

第三步：解析响应（Inbound）

把 provider 返回的原生事件翻译成 pi-ai 的统一 AssistantMessageEvent 序列。

例如，Anthropic 返回 content_block_delta 类型包含 {"text": "你好"} 时，适配器需要把它翻译成 pi-ai 的 text_delta 事件。

第四步：计算使用量

从响应中提取 token 使用信息（input tokens、output tokens、cache tokens），结合 Model 的定价信息，计算出费用。

Provider 适配器的共同模式

虽然每个 provider 的 API 格式不同，但适配器的代码结构都遵循类似的模式：

导出两个函数：
  streamXxx()       -- 接受 provider 特定选项
  streamSimpleXxx() -- 接受统一简化选项，内部映射为特定选项

内部流程：
  1. 转换消息（统一格式 → 原生格式）
  2. 构建请求体
  3. 发送流式 HTTP 请求
  4. 逐块解析响应
  5. 生成统一事件序列
  6. push 到 EventStream
  7. 最终生成 AssistantMessage 并以 done/error 事件结束

streamSimple 如何映射到 stream

streamSimple 是给调用者用的"简化版"。它内部做的事情是把统一的 reasoning 参数映射为各 provider 的特有参数：

每个 provider 对"思考"的控制方式不同：

• Anthropic：通过 token 预算（thinkingBudgetTokens）控制思考长度
• OpenAI：通过 reasoningEffort 字符串级别控制
• Google：通过 thinking.budgetTokens 控制

streamSimple 把这些差异隐藏起来了。

兼容性处理（Compat）

Model 类型有一个可选的 compat 字段，用于处理各种 API 兼容性差异。这在 OpenAI 兼容 API 中尤其复杂。

为什么需要 compat

很多 provider（xAI、Groq、DeepSeek、OpenRouter 等）声称"兼容 OpenAI API"，但实际上总有一些微小差异：

• 有的不支持 store 字段
• 有的不支持 developer 角色（只接受 system）
• 有的不支持在流式响应中返回 usage 信息
• 有的 max tokens 字段叫 max_tokens 而不是 max_completion_tokens
• 有的 thinking 参数格式完全不同

compat 字段让每个模型可以声明自己的 API 有哪些"偏差"，适配器会据此调整请求格式。

例如：

compat: {
  supportsStore: false,              // 不支持 store
  supportsDeveloperRole: false,      // 不支持 developer 角色
  thinkingFormat: "deepseek",        // 使用 DeepSeek 特有的思考格式
  requiresToolResultName: true,      // 工具结果必须带 name 字段
}

这避免了为每个"稍有不同"的 provider 写一个全新的适配器。相反，一个 openai-completions 适配器通过 compat 配置就能适配十几个不同的兼容 provider。

Faux Provider -- 测试利器

packages/ai/src/providers/faux.ts 是一个特殊的"假 provider"，用于测试：

• 不发送任何网络请求
• 你预设好 AI 应该返回什么，它就返回什么
• 完整模拟流式事件序列（包括逐 token 的 delta 事件）
• 可以控制输出速度（tokensPerSecond）

const reg = registerFauxProvider();
reg.setResponses([
  fauxAssistantMessage([
    fauxThinking("我在思考..."),
    fauxText("这是回答"),
    fauxToolCall("read", { path: "foo.ts" })
  ], { stopReason: "toolUse" })
]);

const model = reg.getModel();
for await (const event of stream(model, context)) {
  // 会收到完整的事件序列，就像真的 provider 一样
}

这对于测试 Agent 逻辑很有价值：你可以精确控制 LLM 的"行为"，验证你的 Agent 是否正确处理了各种场景，而不需要花真钱调用真实的 LLM API。

小结

Provider 适配器的职责：

       统一世界                              原生世界
┌─────────────────┐                  ┌─────────────────┐
│  Context        │  ─── 转换 ───►  │  Anthropic 请求   │
│  (systemPrompt, │                  │  (system, messages│
│   messages,     │                  │   tools, ...)     │
│   tools)        │                  │                   │
└─────────────────┘                  └──────┬────────────┘
                                           │ HTTP SSE
                                           ▼
┌─────────────────┐                  ┌─────────────────┐
│ AssistantMessage │  ◄── 翻译 ───  │  Anthropic 响应   │
│ Event 序列       │                  │  (SSE events)    │
│ (text_delta,    │                  │                   │
│  toolcall_end,  │                  │                   │
│  done)          │                  │                   │
└─────────────────┘                  └─────────────────┘

核心技巧：

1. 懒加载：首次使用时才加载 provider 模块
2. Compat 配置：用配置而非代码处理 API 差异
3. streamSimple 映射：统一参数到各 provider 特有参数
4. 消息转换：处理跨 provider 的格式差异（思考块、工具 ID、图片）