L1ngg's Home

它要抹平哪些差异#

这一节看厂商之间具体差在哪——后面所有设计都在回应这四条：

• 鉴权方式：最简单的是填一个 API key；订阅制产品走 OAuth（浏览器跳转登录，用订阅额度，比如 Claude Pro/Max、ChatGPT Plus/Pro、GitHub Copilot）；云上服务用环境凭证链（AWS profile、gcloud ADC）。一个库要同时装下这三种。
• thinking 参数：Anthropic 收 thinking.budget_tokens，OpenAI 收 reasoning_effort，DeepSeek 收 thinking.type，Qwen 收 enable_thinking——同一件事，四种参数名。
• 工具调用 ID 格式：OpenAI Responses 生成的 ID 有 450+ 字符还带 |，Anthropic 要求匹配 ^[a-zA-Z0-9_-]+$ 且不超过 64 字符。A 家产生的历史消息直接喂给 B 家，会被接口拒绝。
• 流式事件的粒度和命名：每家的事件结构都不同，消费端原本要各写一套解析。

thinking 是什么

模型正式回答前的内部推理过程。有的厂商把它单独返回、可以回看，有的不暴露。各家开启它的参数格式完全不同，是统一接口最难抹平的部分之一。

先跑起来：不用 API key 的 agent 循环#

这一节不调任何真实模型，把 pi-ai 的完整交互流程跑一遍，建立直觉。

pi-ai 内置了一个假 provider：fauxProvider。它按你预先排好的「剧本」（一串写死的回复）回应请求，所以不需要任何密钥。先明确什么叫 agent 循环：模型回复 → 模型要求调用工具 → 你执行工具、把结果喂回去 → 模型继续，如此来回。

下面这段代码我在本机实际跑过（Node 24 + pi-ai 0.84.2），走完了「思考 → 调工具 → 收结果 → 继续回答」的完整循环：

1
import {
2
  createModels,
3
  fauxProvider,
4
  fauxAssistantMessage,
5
  fauxThinking,
6
  fauxText,
7
  fauxToolCall,
8
} from '@earendil-works/pi-ai';
9

10
const faux = fauxProvider({ tokensPerSecond: 200 });
11
const models = createModels();
12
models.setProvider(faux.provider);
13

14
const model = faux.getModel();
15
const context = {
16
  systemPrompt: 'You are helpful.',
17
  messages: [{ role: 'user', content: 'echo hi then summarize', timestamp: Date.now() }],
18
};
19

20
// 第一轮排一个「思考 + 工具调用」的剧本
21
faux.setResponses([
22
  fauxAssistantMessage(
23
    [fauxThinking('Need to call echo first.'), fauxToolCall('echo', { text: 'hi' })],
24
    { stopReason: 'toolUse' },
25
  ),
26
]);
27

28
const s = models.stream(model, context);
29
for await (const event of s) {
30
  if (event.type === 'thinking_delta') console.log('[think]', event.delta);
31
  if (event.type === 'toolcall_end') console.log('[toolcall]', event.toolCall.name, event.toolCall.arguments);
32
  if (event.type === 'done') console.log('[done]', event.reason);
33
}
34

35
const msg = await s.result();
36
console.log('usage:', msg.usage.input, 'in /', msg.usage.output, 'out');
37

38
// 执行工具，把结果塞回 context，再排第二轮
39
context.messages.push(msg);
40
context.messages.push({
41
  role: 'toolResult',
42
  toolCallId: msg.content.find((b) => b.type === 'toolCall').id,
43
  toolName: 'echo',
44
  content: [{ type: 'text', text: 'hi' }],
45
  isError: false,
46
  timestamp: Date.now(),
47
});
48

49
faux.setResponses([fauxAssistantMessage([fauxText('The echo said hi.')])]);
50
const next = await models.complete(model, context);
51
console.log('final:', next.content.find((b) => b.type === 'text').text);
52
console.log('context bytes:', JSON.stringify(context).length);

真实输出：

1
[think] Need to call echo fi
2
[think] rst.
3
[toolcall] echo { text: 'hi' }
4
[done] toolUse
5
usage: 13 in / 11 out
6
final: The echo said hi.
7
context bytes: 753

注意 thinking 的增量是半个词半个词到的（fi / rst.）——流式消费代码必须按增量拼接，不能假设语义完整的 chunk。两个实现细节：faux 的 usage 按 4 字符 ≈ 1 token 估算；剧本队列掏空后，再请求会收到 errorMessage: "No more faux responses queued" 的错误消息，同样不 throw。

pi-ai 的全部交互模型就是这样：把消息推进 context，从流里收事件。写 agent 的测试时 faux provider 价值很大——工具循环、中断恢复、跨模型切换的断言都可以做成确定性的。下一节把假模型换成真的。

接入真实模型的最小骨架#

换回真实模型，只要改两样东西：注册方式、密钥。先装包：

1
npm install @earendil-works/pi-ai

最省事的全量版：

1
import { Type, type Context, type Tool } from '@earendil-works/pi-ai';
2
import { builtinModels } from '@earendil-works/pi-ai/providers/all';
3

4
const models = builtinModels();  // 40 家全注册；在意包体积就改用单家工厂
5
const model = models.getModel('anthropic', 'claude-sonnet-4-5')!;
6

7
const context: Context = {
8
  systemPrompt: 'You are a helpful assistant.',
9
  messages: [{ role: 'user', content: 'What time is it?', timestamp: Date.now() }],
10
  tools: [/* TypeBox 工具 */],
11
};
12

13
// ANTHROPIC_API_KEY 从环境自动解析；显式 options.apiKey 永远优先
14
const s = models.stream(model, context);
15
for await (const event of s) {
16
  if (event.type === 'text_delta') process.stdout.write(event.delta);
17
  if (event.type === 'toolcall_end') {
18
    // 执行工具后往 context.messages 推 toolResult，再 complete 一轮
19
  }
20
}
21
context.messages.push(await s.result());

tools 数组留空也能跑；工具怎么定义，见后面「工具、思考等级与 Simple API」一节。只关心包体积的单家版：createModels() + anthropicProvider()（来自 @earendil-works/pi-ai/providers/anthropic）+ models.setProvider(...)。

接本地推理服务不用写协议代码，声明一个自定义 provider 就行：

1
import { createModels, createProvider, type Model } from '@earendil-works/pi-ai';
2
import { openAICompletionsApi } from '@earendil-works/pi-ai/api/openai-completions.lazy';
3

4
const ollamaModel: Model<'openai-completions'> = {
5
  id: 'llama-3.1-8b',
6
  name: 'Llama 3.1 8B (Ollama)',
7
  api: 'openai-completions',
8
  provider: 'ollama',
9
  baseUrl: 'http://localhost:11434/v1',
10
  reasoning: false,
11
  input: ['text'],
12
  cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 },
13
  contextWindow: 128000,
14
  maxTokens: 32000,
15
  // Ollama 这类服务不认 developer role 和 reasoning_effort
16
  compat: { supportsDeveloperRole: false, supportsReasoningEffort: false },
17
};
18

19
const ollama = createProvider({
20
  id: 'ollama',
21
  // 免密钥服务也要声明 auth：resolve 返回空配置即视为已配置
22
  auth: { apiKey: { name: 'Ollama', resolve: async () => ({ auth: {} }) } },
23
  models: [ollamaModel],
24
  api: openAICompletionsApi(),
25
});
26

27
const models = createModels();
28
models.setProvider(ollama);

compat 挂在模型上（Model.compat），按模型所属 API 分型。OpenAI 兼容层的开关最多，有 24 个：maxTokensField 选字段名、requiresThinkingAsText 把 thinking 降级成文本、thinkingFormat 在 11 种思考参数格式之间切换（openai、openrouter、deepseek、together、zai、qwen、chat-template 等）。Anthropic 和 Bedrock 也各有小开关组——比如生成目录里 Anthropic 模型自带的 compat: { supportsStrictTools: true }（见「目录是生成的，SDK 是懒加载的」一节）。

不设置时，按 provider ID 和 baseUrl 对十来家已知厂商探测（xAI、Cerebras、DeepSeek、Together、Moonshot、OpenRouter、Cloudflare 等）；部分设置时，没指定的字段沿用探测结果或固定默认值。所以接自建网关、LiteLLM、vLLM 这类端点，只需覆盖实际有差异的几项。

每次请求回来的 usage.cost 已经按目录里的定价算好了钱，输入、输出、cache 读写、总计分开统计。配合流式选项里的 telemetryContext 可以接 pi 自己的 telemetry 包——那包的设计之前也拆过：

三层结构：Provider、Models、API 实现#

用法看完了，回头看设计：pi-ai 怎么用三层结构切开前面那组差异。

Provider 是运行时单元。它拥有三样东西：模型目录、鉴权逻辑、stream 行为。接口的核心部分：

1
export interface Provider<TApi extends Api = Api> {
2
    readonly id: string;
3
    readonly baseUrl?: string;
4
    // 必需：哪怕是免密钥的本地服务也要声明 auth 语义
5
    readonly auth: ProviderAuth;
6
    // 同步返回当前已知模型列表，不许抛异常
7
    getModels(): readonly Model<TApi>[];
8
    // 动态 provider 才有：拉最新模型列表
9
    refreshModels?(context: RefreshModelsContext): Promise<void>;
10
    stream<T extends TApi>(model: Model<T>, context: Context, options?: ApiStreamOptions<T>): AssistantMessageEventStream;
11
    streamSimple(model: Model<TApi>, context: Context, options?: SimpleStreamOptions): AssistantMessageEventStream;
12
}

Models 是集合。注册若干 provider，负责同步查模型、解析鉴权、把请求路由给拥有该模型的 provider。complete() 没有任何独立逻辑，就是 stream().result()：

1
async complete(model, context, options) {
2
    return this.stream(model, context, options).result();
3
}

API 实现是线协议。10 个：anthropic-messages、openai-responses、openai-completions、openai-codex-responses、azure-openai-responses、google-generative-ai、google-vertex、mistral-conversations、bedrock-converse-stream、pi-messages。混合 API 的 provider 按 model.api 逐模型分发——GitHub Copilot 一家就同时挂三种协议（anthropic-messages 10 个模型、openai-responses 14 个、openai-completions 8 个），OpenCode Zen 同理。

分层的收益在组合性：新增一家 OpenAI 兼容厂商，不用写任何协议代码，给一个模型列表加 baseUrl 就完事——上一节 Ollama 的例子已经演示过。

错误不抛出：一套事件协议#

这一节回答：40 家厂商各不相同的流式回复，怎么统一成一套事件。

所有 provider 的所有流，都收敛到 12 种事件：

最有特色的设计是请求失败不 throw。网络错误、鉴权失败、abort、工具参数校验失败，全部走 error 事件；最终消息（await s.result()）里带着 stopReason: "error"、errorMessage、已收到的部分内容和部分 token 用量。调用方不需要 try/catch 包流，统一在事件循环里处理。

流的实现也朴素——一个 queue 加一组等待中的 resolver，done / error 事件结算 result() 的 promise：

1
// dist/utils/event-stream.js，节选
2
push(event) {
3
    if (this.done) return;
4
    if (this.isComplete(event)) {          // done 或 error
5
        this.done = true;
6
        this.resolveFinalResult(this.extractResult(event));
7
    }
8
    const waiter = this.waiting.shift();
9
    if (waiter) waiter({ value: event, done: false });
10
    else this.queue.push(event);
11
}

两个使用上的坑，README 自己都标了：

1. 不同内容块的事件不保证连续。一个上游 chunk 可能同时带文本和工具调用的增量，事件会交错。必须用 contentIndex 关联增量和块，不能假设一个块的 *_start/*_delta/*_end 不被别人打断。
2. toolcall_delta 里的 arguments 是尽力解析的半成品。库里用 partial-json 包边收边解析，字段可能缺、字符串可能半截。做渐进 UI 可以，执行业务逻辑必须等 toolcall_end，再过一遍 validateToolCall。

Context 是纯数据，可以跨厂商交接#

这一节回答：会话中途换一家厂商，上下文怎么不丢。

Context 只有三个字段：systemPrompt、messages、tools。全部可 JSON 序列化，持久化会话就是 JSON.stringify，换机器、换进程、换模型都行。

「换模型」是重点。凡是要把 Context 翻译成厂商格式的 API 实现——anthropic-messages、openai-completions、openai-responses 系、google 系、mistral、bedrock——都会先过一遍 transformMessages（dist/api/transform-messages.js，186 行，是全文最值得读的函数）。唯一的例外是 pi-messages：它本身就是 pi 的协议，Context 原样 POST 给后端，翻译是服务端的事。0.84.2 里 transformMessages 按顺序做这些事：

1. 把 null 内容归一化成空数组，兜住手搓的历史消息
2. 目标模型不支持视觉时，把图片降级成占位文本 (image omitted: model does not support images)
3. 跨模型的 thinking 块转成普通文本块（同模型则保留带签名的原始块，供回放；redacted 的加密 thinking 直接丢弃）。注意 README 说会加 <thinking> 标签，0.84.2 的代码已经不加标签了，就是纯文本
4. 跨模型的 tool call 剥掉 thoughtSignature，并按目标 API 归一化 ID。给 Anthropic 的归一化就一行：id.replace(/[^a-zA-Z0-9_-]/g, "_").slice(0, 64)
5. 整条跳过 stopReason 为 error / aborted 的 assistant 消息——残缺的回合不回放，让模型从上一次有效状态重来，避免 OpenAI 的 “reasoning without following item” 这类报错
6. 给没有结果的孤儿 tool call 补合成的错误结果（"No result provided"），满足各家「tool call 必须有结果」的协议要求

效果是：Claude 思考到一半切 GPT-5 再切 Gemini，工具调用链不断，上下文不丢。这也回答了「为什么 Model 和 Context 必须是纯数据」——只有纯数据才能在 provider 之间自由流动。

目录是生成的，SDK 是懒加载的#

这一节回答：1267 个模型的元数据和 4 个重型 SDK，怎么不拖慢你的构建。

模型元数据不是手维护的。构建期脚本（官方仓库 scripts/generate-models.ts，约 3000 行）以 models.dev 为主数据源，NVIDIA NIM、OpenRouter、Vercel AI Gateway 等几家直接拉各自的 /models 接口，再叠加一批手工修正；产物 hydrate 成 providers/data/<id>.json，再由 <id>.models.ts 从 JSON 的 key 导出字面量类型——所以 getBuiltinModel('openai', 'gpt-4o-mini') 有完整的自动补全和返回类型。Model 是纯数据（id、定价、contextWindow、是否支持图像/推理），没有方法，JSON.stringify 直接序列化。

一条真实记录长这样（anthropic.json 里的 Claude Haiku 4.5）：

1
"claude-haiku-4-5": {
2
  "id": "claude-haiku-4-5",
3
  "name": "Claude Haiku 4.5 (latest)",
4
  "api": "anthropic-messages",
5
  "provider": "anthropic",
6
  "baseUrl": "https://api.anthropic.com",
7
  "reasoning": true,
8
  "input": ["text", "image"],
9
  "cost": { "input": 1, "output": 5, "cacheRead": 0.1, "cacheWrite": 1.25 },
10
  "contextWindow": 200000,
11
  "maxTokens": 64000,
12
  "compat": { "supportsStrictTools": true }
13
}

SDK 加载走的是 lazyApi 包装。provider 工厂 import 的不是 @anthropic-ai/sdk，而是一个 .lazy 模块；首次请求到达时才动态 import 真 SDK。核心代码很短：

1
// dist/api/lazy.js，节选
2
export function lazyApi(load, capabilities) {
3
    const api = {
4
        stream: (model, context, options) =>
5
            lazyStream(model, async () => (await load()).stream(model, context, options)),
6
        streamSimple: (model, context, options) =>
7
            lazyStream(model, async () => (await load()).streamSimple(model, context, options)),
8
    };
9
    return api;
10
}

lazyStream 同步返回一个事件流，把异步的模块加载藏在流后面——加载失败也不会 throw，而是终止流并推一个 error 事件，和前面「错误不抛出」一节的约定一致。

配合打包器代码分割，tree shaking 规则很干净：

• 主入口 @earendil-works/pi-ai 不含任何内置目录和 SDK
• providers/<p> 只拉这一家的目录和 lazy 包装
• providers/all 是全量重入口，只在真要 40 家全注册时用

鉴权属于 provider#

这一节回答：三种鉴权方式（key、OAuth、云凭证链）怎么收进一个接口。

鉴权不是全局注册表，而是每个 provider 自己的 auth 字段。以 Anthropic 为例，解析顺序直接从源码能看到：

1
// dist/providers/anthropic.js，节选
2
resolve: async ({ ctx, credential, signal }) => {
3
    // 1. 存过的 credential 永远优先
4
    if (credential?.key) {
5
        return { auth: { apiKey: credential.key }, source: "stored credential" };
6
    }
7
    // 2. ANTHROPIC_AUTH_TOKEN 走 Bearer 头
8
    const authToken = await ctx.env(ANTHROPIC_AUTH_TOKEN_ENV);
9
    if (authToken) {
10
        return { auth: { headers: { Authorization: `Bearer ${authToken}` } }, source: ANTHROPIC_AUTH_TOKEN_ENV };
11
    }
12
    // 3. 再依次看 ANTHROPIC_OAUTH_TOKEN、ANTHROPIC_API_KEY
13
    // ...
14
    return undefined;  // 未配置
15
}

持久化凭证走 CredentialStore，契约刻意做得很小：read、list（只返回 { providerId, type } 元数据，不碰 secret）、modify（唯一写路径，串行化的 read-modify-write）、delete。两个值得抄的决策：

• OAuth 刷新在 modify 里跑。token 刷新持有存储锁，并发请求、甚至多进程不会重复刷新同一个轮换过的 token。
• 存过的 credential 拥有它的 provider。只要存了，环境变量就不再被咨询；刷新失败也不会静默回落到 env key。避免「线上偷偷用了开发机的 key」这类事故。

不发请求也能检查配置状态：await models.getAuth(model) 返回 headers 和来源标签（"ANTHROPIC_API_KEY"、"OAuth"、"stored credential"），做状态栏 UI 直接够用。

请求头的合并顺序也是固定的，后者覆盖前者：

1
provider auth → model.headers → options.headers → transformHeaders → 发给 provider

订阅制 provider（Anthropic、OpenAI Codex、GitHub Copilot）的 OAuth 登录内置了完整流程，命令行直接可用，凭证写到当前目录的 auth.json：

1
npx @earendil-works/pi-ai login anthropic

工具、思考等级与 Simple API#

这一节回答：四家厂商四种 thinking 参数，怎么变成一个选项。

工具用 TypeBox 定义，schema 本身就是可序列化的 JSON：

1
import { Type, StringEnum, type Tool } from '@earendil-works/pi-ai';
2

3
const weatherTool: Tool = {
4
  name: 'get_weather',
5
  description: 'Get current weather for a location',
6
  parameters: Type.Object({
7
    location: Type.String({ description: 'City name or coordinates' }),
8
    // 兼容 Google：不要用 Type.Enum，它生成的 anyOf/const 模式 Google 不收
9
    units: StringEnum(['celsius', 'fahrenheit'], { default: 'celsius' })
10
  })
11
};

流式 API 有两层，对应两种心智负担：

• stream / complete：吃所属 API 的完整选项（Anthropic 的 thinkingBudgetTokens、OpenAI 的 reasoningEffort），动态查出来的模型用 hasApi(model, 'anthropic-messages') 收窄后获得完整类型
• streamSimple / completeSimple：统一六档 reasoning: 'minimal' | 'low' | 'medium' | 'high' | 'xhigh' | 'max'，由库里映射成各家参数

token 系 provider 的默认思考预算在 simple-options.js 里：

xhigh / max 是模型级 opt-in（由模型的 thinkingLevelMap 声明，null 表示该档不支持），预算计算时按 high 处理。模型级的 thinkingBudgets 选项可以整体覆盖这张表。

同一文件里还有个安静的保护：maxTokens 会被钳到 contextWindow − 上下文估算 tokens − 4096 安全边距内，防止把上下文撑爆。

深入：精读 openai-completions（可跳过）#

这一节是源码精读，不影响使用，可以跳过。前面 demo 和真实模型骨架用的 stream / complete，底层走的就是这类 API 实现。

transformMessages 只是预处理。真正把 Context 变成厂商 wire 格式的代码，住在每个 API 实现里。以 openai-completions 为例——源码在仓库的 packages/ai/src/api/openai-completions.ts（npm 包里只有编译产物 dist/），约 1670 行。读这类实现带三个问题就够了：输入怎么转、输出怎么转、厂商差异怎么抹平。

先看主干。stream 函数同步返回一个事件流，所有工作塞进 async IIFE，try/catch 兜住全部异常转成 error 事件——「错误不抛出」就是在这里落地的：

1
// src/api/openai-completions.ts，骨架精简
2
export const stream = (model, context, options) => {
3
  const stream = new AssistantMessageEventStream();
4
  (async () => {
5
    const output: AssistantMessage = { role: "assistant", content: [], stopReason: "pending", /* ... */ };
6
    try {
7
      // 建 client、buildParams、发请求
8
      for await (const chunk of openaiStream) { /* 逐 chunk 翻译成事件 */ }
9
      stream.push({ type: "done", reason: output.stopReason, message: output });
10
      stream.end();
11
    } catch (error) {
12
      output.stopReason = options?.signal?.aborted ? "aborted" : "error";
13
      output.errorMessage = formatProviderError(normalizeProviderError(error));
14
      stream.push({ type: "error", reason: output.stopReason, error: output });
15
      stream.end();
16
    }
17
  })();
18
  return stream;
19
};

output 从 pending 空壳开始，在循环里被逐渐填满；每个事件携带的 partial 都是它。

1
if (context.systemPrompt) {
2
  const useDeveloperRole = model.reasoning && compat.supportsDeveloperRole;
3
  const role = useDeveloperRole ? "developer" : "system";
4
  params.push({ role, content: sanitizeSurrogates(context.systemPrompt) });
5
}

1
// Always send assistant content as a plain string (OpenAI Chat Completions
2
// API standard format). Sending as an array of {type:"text", text:"..."}
3
// objects is non-standard and causes some models (e.g. DeepSeek V3.2 via
4
// NVIDIA NIM) to mirror the content-block structure literally in their output

1
if (toolCall.function?.arguments) {
2
  delta = toolCall.function.arguments;
3
  block.partialArgs = (block.partialArgs ?? "") + toolCall.function.arguments;
4
  block.arguments = parseStreamingJson(block.partialArgs);
5
}

1
function getCompat(model) {
2
  const detected = detectCompat(model); // 按 provider ID + baseUrl 匹配十来家已知厂商
3
  if (!model.compat) return detected;
4
  return {
5
    supportsStore: model.compat.supportsStore ?? detected.supportsStore,
6
    supportsDeveloperRole: model.compat.supportsDeveloperRole ?? detected.supportsDeveloperRole,
7
    // ...24 个字段全是同一个模式
8
  };
9
}

什么时候用它，什么时候不用#

适合的场景：写 agent 或任何多轮工具循环；需要在多家模型间切换、对比、兜底；需要统一的 token / 成本统计；测试想要确定性（faux provider）；想在浏览器里直连模型（Bedrock 和 OAuth 登录除外，这两样是 Node only）。

不适合的场景：

• 只要对一家厂商发一次 completion——官方 SDK 更直接，这个库的价值在多模型和 agent 循环
• 要 embedding、语音、视频——目录只收工具调用模型，这些模态不在范围内
• 要图像生成——有独立的 ImagesModels API，但 0.84.2 只有 OpenRouter 一家接入
• 老代码的全局 API（getModel()、stream() 直调）已冻结在 @earendil-works/pi-ai/compat，未来版本会删，新项目直接用 createModels() + provider 工厂

回到开头的问题：统一 40 家 LLM 靠的不是一个巨大的适配层，而是明确一个事实「厂商很多、协议很少」，然后把鉴权、目录、流行为收进 Provider，把路由和凭证解析收进 Models，把 10 条线协议做成可懒加载、可单测的实现。每层都可以单独替换——这也是 pi 全家桶一贯的形状。