大话 Pi

Appearance

🐣 小a的拆解任务

入职第一天,老z给小a派了第一个任务:"pi-ai 这个包,是整个工坊的地基。你给我把它拆明白——它怎么用一套接口,盖住 近 40 家厂商的?"

小a打开 packages/ai/src/providers/ 一数,近 40 家厂商。又打开 types.ts,好家伙,32KB。

"这……从哪下手?"

老z拍拍它肩膀:"别被吓到。这套抽象的核心,其实就一个函数 + 一个接口。 抓住这俩,其他的都是它们的展开。"

从这一章起,我们大量贴真实源码。每段代码都来自 pi 仓库,标了文件路径和行号,你可以对照着看。

17.1 先抓核心:一个工厂 + 一个接口 ​

老z带着小a直奔要害。pi-ai 的统一抽象,核心就两样东西:

  1. Provider 接口:定义"一个厂商长什么样"
  2. createProvider 工厂:把"配置"变成符合接口的对象

抓住这俩,近 40 家厂商的秘密就解开了。

17.2 Provider 接口:一个厂商的"标准画像" ​

我们先看 Provider 接口——它定义了"一个厂商必须提供什么"。

📄 源码证据 packages/ai/src/models.ts:75(真实完整接口)

typescript
export interface Provider<TApi extends Api = Api> {
  readonly id: string;          // 厂商唯一标识,如 "anthropic"
  readonly name: string;        // 显示名,如 "Anthropic"
  readonly baseUrl?: string;    // API 地址,如 "https://api.anthropic.com"
  readonly headers?: ProviderHeaders;  // 额外的请求头
  readonly auth: ProviderAuth;  // 鉴权方式(怎么证明身份)

  // 当前已知的模型列表(同步返回)
  getModels(): readonly Model<TApi>[];

  // 动态厂商专用:刷新模型列表(从厂商 API 拉最新的)
  refreshModels?(context: RefreshModelsContext): Promise<void>;

  // 可选:过滤模型(如按凭据过滤掉不可用的)
  filterModels?(models: readonly Model<TApi>[], credential: Credential | undefined): readonly Model<TApi>[];

  // 完整流式:暴露厂商所有细节(思考过程、原始事件)
  stream<T extends TApi>(model: Model<T>, context: Context, options?: ApiStreamOptions<T>): AssistantMessageEventStream;

  // 简化流式:抹平思考等厂商差异,统一事件流
  streamSimple(model: Model<TApi>, context: Context, options?: SimpleStreamOptions): AssistantMessageEventStream;
}

🧙 老z拆解这个接口:

一个 Provider 要回答这几个问题:

  • 你是谁?(id / name)
  • 上哪找你?(baseUrl)
  • 怎么证明身份?(auth)
  • 你有哪些模型?(getModels())
  • 怎么调用你?(通过 api 提供的 stream 方法)

只要一个厂商能回答这五个问题,它就能被纳入 pi 的统一体系。 这就是"标准画像"的威力——不管你多大牌(OpenAI)还是多小众,都套进同一个模子。

注释里的重要细节 ​

注意 auth 字段的注释:

📄 注释原文:Required: at least one of apiKey/oauth. Every provider has auth semantics — even providers with only ambient credentials...

翻译:每个 provider 都必须有 auth——哪怕它只用环境变量、不显式登录,也要提供 auth(其 resolve() 会报告"我有没有配置好")。pi 不允许"不知道自己鉴权状态"的 provider。 这是 fail loud 原则的体现(附录 E.6)。

17.3 createProvider 工厂:把配置变成对象 ​

光有接口不够,得有个东西"造"出符合接口的对象。这就是 createProvider。

📄 源码证据 packages/ai/src/models.ts:556(真实签名)

typescript
export function createProvider<TApi extends Api = Api>(
  input: CreateProviderOptions<TApi>   // 一份"配置单"
): Provider<TApi>                      // 返回符合接口的 Provider

CreateProviderOptions 就是"配置单"的类型定义(真实字段):

📄 源码证据 models.ts:533

typescript
interface CreateProviderOptions<TApi extends Api = Api> {
  id: string;                                // 厂商唯一标识
  name?: string;                             // 显示名
  baseUrl?: string;                          // API 地址
  headers?: ProviderHeaders;                 // 额外请求头
  auth: ProviderAuth;                        // 鉴权方式
  models: readonly Model<TApi>[];            // 模型列表
  fetchModels?: ...;                         // 可选:动态拉取模型
  filterModels?: ...;                        // 可选:按凭据过滤
  api: ProviderStreams | Partial<Record<TApi, ProviderStreams>>;  // 说话方式
}

这段代码是本章最重要的一段,老z带小a分三步读——每步只看一个职责,别一锅炖。

第一步:维护两张模型表(静态 + 动态) ​

意图:模型会变。pi 维护两张表——静态的(构建期抓的快照)和动态的(运行期可刷新)。

📄 源码证据 packages/ai/src/models.ts —— createProvider 内部(精简)

typescript
const baselineModels = input.models;            // 静态表(构建期生成)
let dynamicModels: readonly Model<TApi>[] = []; // 动态表(运行期可刷新)

const currentModels = (): readonly Model<TApi>[] => {
  const merged = [...baselineModels];
  for (const model of dynamicModels) {
    const index = merged.findIndex((entry) => entry.id === model.id);
    if (index >= 0) merged[index] = model;  // 同 id → 覆盖
    else merged.push(model);                 // 新 id → 追加
  }
  return merged;
};

🧙 currentModels() 的合并逻辑——动态表覆盖同 id 的静态项。

好处:既能离线用(只看静态),又能在线更新(动态覆盖)。一个厂商刚上线了新模型,运行时刷新 dynamicModels,currentModels() 就能返回最新列表,静态表不用改。

第二步:按协议派发到正确的"说话方式" ​

意图:不同模型用不同 API 协议(OpenAI 用 openai-completions,Anthropic 用 anthropic-messages)。要根据 model.api 找到对应的流实现。

📄 源码证据(接上)

typescript
// 根据 model.api 找到对应的流实现
const single = typeof (input.api as ProviderStreams).stream === "function"
  ? (input.api as ProviderStreams) : undefined;
const byApi = single ? undefined : (input.api as Partial<Record<string, ProviderStreams>>);
const apiFor = (model: Model<Api>): ProviderStreams | undefined =>
  single ?? byApi?.[model.api];

// 派发:把调用转到正确的流实现
const dispatch = (model, run) => {
  const streams = apiFor(model);
  if (!streams) {
    return lazyStream(model, async () => {
      throw new ModelsError("stream",
        `Provider ${input.id} has no API implementation for "${model.api}"`);
    });
  }
  return run(streams);
};

🧙 apiFor(model) 干什么——给一个模型,返回它该用的流实现。

  • DeepSeek 的模型 api 是 openai-completions → 用 openAICompletionsApi() 那套
  • Anthropic 的模型 api 是 anthropic-messages → 用 anthropicMessagesApi() 那套

找不到怎么办? dispatch 会抛 ModelsError —— fail loud,不静默失败(附录 E 的原则)。这避免了"模型配错了却偷偷跑出诡异结果"。

第三步:打包成 Provider 对象 ​

意图:把上面两步封装成一个符合 Provider 接口的对象返回。外部只看到统一接口,内部复杂性(合并、派发)全被闭包吃掉。

📄 源码证据(接上)

typescript
return {
  id: input.id,
  name: input.name ?? input.id,
  baseUrl: input.baseUrl,
  headers: input.headers,
  auth: input.auth,
  getModels: currentModels,   // 对外只暴露"当前模型表"
  refreshModels: /* 动态刷新逻辑 */,
  // stream/streamSimple 通过 dispatch 实现
};

🐣 小a恍然大悟:"所以 createProvider 是个闭包工厂——它返回的对象,内部状态(baselineModels、dynamicModels)被闭包封住,外部只能通过 getModels() 访问?"

🧙 "对!这就是闭包的威力——封装。 不用 class,用闭包照样实现『私有状态 + 公开方法』。pi 这里选了函数式风格,不是 OOP。三步合起来:维护模型表 → 按协议派发 → 打包成统一对象,这就是把『一个厂商的配置』变成『一个可用的 Provider』的全部秘密。"

17.4 对比两个真实 provider:最简 vs 最复杂 ​

抽象讲完了,看真实的 provider 长什么样。老z挑了两个极端对比——最简单的 DeepSeek,和最复杂的 Anthropic。

📄 先看 Anthropic 的真实"配置单"(packages/ai/src/providers/anthropic.ts:38):

typescript
export function anthropicProvider(): Provider<"anthropic-messages"> {
  return createProvider({
    id: "anthropic",
    name: "Anthropic",
    baseUrl: "https://api.anthropic.com",
    auth: {
      apiKey: anthropicApiKeyAuth(),               // API Key 责任链
      oauth: lazyOAuth({ name: "Anthropic (Claude Pro/Max)", load: loadAnthropicOAuth }),
    },
    models: Object.values(ANTHROPIC_MODELS),       // 内置模型清单
    api: anthropicMessagesApi(),                   // 自有协议实现
  });
}

对照配置单读一遍:id(anthropic)、name、baseUrl、auth(apiKey + oauth 两种鉴权)、models(模型列表)、api(它自己的 anthropic-messages 协议)。"一个厂商 = 一份配置单"在真实源码里就是字面意思。

最简的:DeepSeek(15 行) ​

📄 源码证据 packages/ai/src/providers/deepseek.ts(全 15 行)

typescript
import { openAICompletionsApi } from "../api/openai-completions.lazy.ts";
import { envApiKeyAuth } from "../auth/helpers.ts";
import { createProvider, type Provider } from "../models.ts";
import { DEEPSEEK_MODELS } from "./deepseek.models.ts";

export function deepseekProvider(): Provider<"openai-completions"> {
  return createProvider({
    id: "deepseek",
    name: "DeepSeek",
    baseUrl: "https://api.deepseek.com",
    auth: { apiKey: envApiKeyAuth("DeepSeek API key", ["DEEPSEEK_API_KEY"]) },
    models: Object.values(DEEPSEEK_MODELS),
    api: openAICompletionsApi(),  // 复用 OpenAI 协议!
  });
}

🧙 为什么 DeepSeek 这么短?

因为它兼容 OpenAI 协议——api: openAICompletionsApi() 直接复用了 OpenAI 的实现。DeepSeek 不用自己造一套 API 适配,白嫖 OpenAI 的。

15 行里,真正"DeepSeek 特有"的只有:id、name、baseUrl、auth(环境变量名)、models(它的模型列表)。差异被抽象吃掉了,剩下的是纯配置。

最复杂的:Anthropic(自带 OAuth 责任链) ​

再看 Anthropic——它有自己的协议、自己的 OAuth、复杂得多的鉴权:

📄 源码证据 packages/ai/src/providers/anthropic.ts(节选鉴权部分)

typescript
function anthropicApiKeyAuth(): ApiKeyAuth {
  return {
    resolve: async ({ ctx, credential }) => {
      // 1️⃣ 先用存档凭据
      if (credential?.key) {
        return { auth: { apiKey: credential.key }, source: "stored credential" };
      }

      // 2️⃣ 找 AUTH_TOKEN(Bearer 方式)
      const authToken = await ctx.env(ANTHROPIC_AUTH_TOKEN_ENV);
      if (authToken) {
        return {
          auth: { headers: { Authorization: `Bearer ${authToken}` } },
          source: ANTHROPIC_AUTH_TOKEN_ENV,
        };
      }

      // 3️⃣ 找 OAuth token / API key(两个环境变量都试)
      for (const envVar of [ANTHROPIC_OAUTH_TOKEN_ENV, ANTHROPIC_API_KEY_ENV]) {
        const apiKey = await ctx.env(envVar);
        if (apiKey) return { auth: { apiKey }, source: envVar };
      }

      // 4️⃣ 都没有 → 显式返回 undefined(不偷偷降级)
      return undefined;
    },
  };
}

🧙 这段代码展示了"责任链模式":

Anthropic 找一个 key,要依次试四条路:

  1. 存档凭据(用户之前登录存的)
  2. AUTH_TOKEN 环境变量(Bearer 方式)
  3. OAUTH_TOKEN 环境变量
  4. API_KEY 环境变量

一路往下试,任一条成功就返回,全失败就显式返回 undefined(告诉调用方"我没找到",而不是偷偷用空值继续)。

对比 DeepSeek:DeepSeek 的 auth 是 envApiKeyAuth("...", ["DEEPSEEK_API_KEY"])——一行搞定,只找一个环境变量。Anthropic 复杂 4 倍,因为它支持多种登录方式。

对比表:抽象吃掉了什么 ​

维度DeepSeekAnthropic
协议复用 OpenAI(openAICompletionsApi)自有(anthropic-messages)
鉴权单环境变量OAuth + 4 路 env 责任链
行数15 行约 50 行
用到的模式FactoryFactory + Chain of Responsibility

🐣 小a顿悟:"所以加一个新厂商,我只要写一个 xxxProvider() 函数,填好配置,根本不用动 createProvider?"

🧙 "对! 这就是开闭原则——对扩展开放(加新 provider),对修改封闭(不动 createProvider)。pi 能支持 近 40 家厂商,核心就靠这个。"

17.5 统一调用入口:上层怎么用 ​

有了 provider,上层怎么调?老z带小a看统一调用入口。

ProviderStreams:每个 provider 必须实现的两个方法 ​

📄 源码证据 packages/ai/src/types.ts:236

typescript
export interface ProviderStreams {
  // 完整流式(暴露所有细节)
  stream(model, context, options?: StreamOptions): AssistantMessageEventStream;
  // 简化流式(屏蔽思考等差异,第 2 章讲的"统一接口")
  streamSimple(model, context, options?: SimpleStreamOptions): AssistantMessageEventStream;
}

🧙 两个方法的区别:

  • stream:完整版,暴露厂商的所有细节(包括思考过程、原始事件)
  • streamSimple:简化版,把"思考"等厂商差异抹平,给你统一的事件流

回忆第 2 章——"思考强度"各厂商格式不同(10 种 thinkingFormat)。streamSimple 就是那个"抹平差异"的层:你用 streamSimple,就不用管底层是哪家、思考格式是啥。

上层的调用流程 ​

上层代码:
  models.streamSimple(model, context, {
    messages,     // 统一消息格式
    tools,        // 统一工具格式
    reasoning,    // 思考强度(第 2 章)
  })
        ↓
  models 找到 model 对应的 provider
        ↓
  provider.streamSimple(...)  ← 翻译成厂商格式,发请求
        ↓
  收到厂商的流式响应
        ↓
  翻译回统一格式,吐给上层

🧙 关键:上层只调 streamSimple,不用知道 provider 是谁。 这就是第 3 章讲的"统一接口"——管你哪家,都是一套调用。

17.6 消息格式互转:最难的"翻译" ​

统一接口最大的难点,是消息格式互转。各家的消息结构天差地别(第 3 章的巴别塔表),pi 怎么把它们统一?

这部分代码在 api/ 目录下,核心是 transform-messages.ts。老z不展开全部细节(那是几千行),只讲清它在干什么:

🧙 transform-messages 的职责:

上层的统一消息              厂商的原生消息
(user/assistant/toolResult)  →  (OpenAI 的 messages / Google 的 contents / ...)
                             ←
                         (反向:厂商响应翻译回统一格式)
  • 正向:你给统一格式 → 翻译成厂商要的(比如把 toolResult 转成 OpenAI 的 tool 角色消息)
  • 反向:厂商返回 → 翻译回统一格式(比如把 Anthropic 的 content_block 翻译成统一的 assistant 事件)

这是统一抽象里代码量最大的部分——因为每家方言都得写翻译规则。但好在,这些规则是声明式的、可维护的,加新厂商只加新规则,不改核心。

🐣 小a感叹:"所以统一接口的『干净』,是靠底下大量翻译代码换来的?"

🧙 "对。这就是抽象的代价——上层越简单,下层越复杂。 pi-ai 让你一行 streamSimple 调任何厂商,背后是几千行翻译代码。抽象把复杂度从『调用方』搬到了『抽象内部』。"

17.7 鉴权解耦:provider 不直接读环境变量 ​

回头看 Anthropic 那段 auth 代码,有个精妙的设计值得专门讲——provider 不直接读环境变量,而是声明式描述"我需要什么"。

🧙 传统做法 vs pi 的做法:

传统(直接读):

typescript
const apiKey = process.env.ANTHROPIC_API_KEY;  // 直接读

问题:测试时难 mock,且 provider 耦合到环境。

pi 的做法(声明式 + context):

typescript
const apiKey = await ctx.env(ANTHROPIC_API_KEY_ENV);  // 通过 context 读

ctx.env() 是个抽象层——它可能读环境变量,也可能读配置文件,也可能在测试时返回假值。provider 不关心"环境变量从哪来",只声明"我需要这个变量"。

好处:可测试(测试时注入假 context)、可配置(来源可换)、解耦。

17.8 代价:抽象不是免费的 ​

老z不让小a只看好处。pi-ai 的这套抽象,代价是什么?

🧙 代价一:差异黑洞

看 types.ts 里 AnthropicMessagesCompat、OpenAIResponsesCompat、BedrockCompat 一堆子接口——光 thinkingFormat 就有 10 种格式。每家电厂的怪癖都被塞进配置。抽象的上层干净,但抽象内部是个不断膨胀的"差异黑洞"。

📄 源码证据 types.ts:540 的 thinkingFormat 枚举

typescript
thinkingFormat?:
  | "openai"      // 用 reasoning_effort
  | "openrouter"  // 用 reasoning: { effort }
  | "deepseek"    // 用 thinking: { type } + reasoning_effort
  | "together"    // 用 reasoning: { enabled }
  | "zai"         // 用 thinking: { type }
  | "qwen"        // 用 enable_thinking: boolean
  | "chat-template"
  | "qwen-chat-template"
  | "string-thinking"
  | "ant-ling";

🧙 代价二:性能损耗

每次调用都要经过"翻译"——统一格式 → 厂商格式 → 发请求 → 响应 → 翻译回统一格式。这有 CPU 开销(虽然相比网络 IO 可忽略)。

🧙 代价三:学习曲线

新开发者看 pi-ai,要理解 Provider、Api、Compat、Stream、SimpleStream 一堆概念,门槛不低。

🐣 小a记下:"所以统一接口的收益(可替换、可复用)是用复杂度换的。如果你只用一家厂商,这套抽象其实是过度设计——直接用官方 SDK 更简单。"

🧙 "完全正确!这就是 Part III 要讲的——你的需求决定该不该上这套抽象。"

本章小结 ​

🐣 小a的第十七课

┌──────── pi-ai 的统一抽象 ────────┐
│                                  │
│  • 核心:一个工厂 + 一个接口     │
│    createProvider + Provider     │
│                                  │
│  • Provider = 厂商的标准画像     │
│    id/name/baseUrl/auth/models   │
│    + stream/streamSimple         │
│                                  │
│  • createProvider = 闭包工厂     │
│    合并静态+动态模型             │
│    按 model.api 派发到流实现     │
│                                  │
│  • 加新厂商 = 写配置单           │
│    (DeepSeek 15 行,            │
│     Anthropic 约 50 行带责任链)    │
│    → 开闭原则                    │
│                                  │
│  • 统一入口:streamSimple        │
│    抹平思考等厂商差异            │
│                                  │
│  • 代价:差异黑洞 + 学习曲线     │
└──────────────────────────────────┘

关键认知:pi-ai 的统一抽象,本质是"用工厂模式 + 配置化"把厂商差异吃掉。上层干净,下层复杂。加厂商容易,但维护这套抽象要持续投入。

课后实验 ​

  1. 读一个 provider:打开 packages/ai/src/providers/ 下任意一个厂商(比如 zai.ts 或 moonshotai.ts),对照 DeepSeek 的 15 行,看它的配置单有什么不同。
  2. 找责任链:在 anthropic.ts 里,数数 auth 的 resolve 函数试了几条路。画一张"找 key 责任链"的流程图。
  3. 感受差异黑洞:打开 types.ts,搜 Compat,数数有多少个厂商特定的兼容接口。体会"抽象内部有多复杂"。

下一章:pi-ai 解决了"怎么调 LLM"。但光会调还不够——怎么让模型会调工具、会循环?这是 pi-agent-core 干的事。 → 第 18 章 · pi-agent-core 源码