第 1 课：packages/ai —— 和 40 家大模型说同一种普通话

源码位置：packages/ai/src/ 核心文件：types.ts（904 行，词汇表）、api/*（翻译官）、providers/*（厂商名片）、utils/event-stream.ts（事件流）

1. 它解决什么问题

每家模型厂商的 HTTP 接口格式都不一样。如果应用层直接对接每一家：

不用 pi-ai 的世界（N × M 爆炸）:

  你的应用 ──── 各写一套 ────▶ Anthropic 特有格式
  你的应用 ──── 各写一套 ────▶ OpenAI 特有格式
  你的应用 ──── 各写一套 ────▶ Google 特有格式
  ... 40 家厂商 = 40 套胶水代码，换一家改一次

用 pi-ai 的世界（N + M）:

  coding-agent ─┐
  (只说一种话)   ├─▶ pi-ai 统一格式 ─▶ 翻译官1: anthropic-messages
                │            │      ├─▶ Anthropic
  未来任何 app ─┘            │      ├─▶ OpenAI
                             │      ├─▶ Google
                     翻译官2: openai-completions ─▶ DeepSeek/Groq/...

核心思想：把「应用说什么」和「厂商听什么」彻底分开。 应用只学一种格式（第 2 节），每家厂商的方言由一个「翻译官」文件单独处理。

目录结构（packages/ai/src/）：

src/
├── types.ts       ★ 词汇表：所有核心类型，904 行
├── api/           ★ 9 个「翻译官」：一种 API 方言一个文件
│     anthropic-messages.ts   (1575 行)
│     openai-completions.ts   (1610 行)
│     google-generative-ai.ts ( 436 行) ...
├── providers/     ★ 40 个「厂商名片」，每个只有十几行（如 deepseek.ts 15 行）
├── models.ts        Models 类：总入口，注册/查找厂商
├── auth/            登录方式：API key、OAuth 扫码...
└── utils/           工具箱：事件流、重试、代理...

反差：翻译官上千行，厂商名片十几行。难活全在"翻译"， 接入新厂商几乎免费——这个思路后面反复出现。

2. 概念一：Message —— 对话的词汇表

types.ts:650 一行定义骨架：

export type Message = SystemMessage | UserMessage | AssistantMessage | ToolResultMessage;
//                    ^^^^^^^^^^^^^ 第 552 行  ^^^^^^^^^^ 第 570 行  ^^^^^^^^^^^^^^^ 第 576 行  ^^^^^^^^^^^^^^^^^^ 第 631 行

联合类型：一条消息只能是四种角色之一，像戏剧里的四种演员：

一次对话就是这四种消息排成的列表（历史记录），顺序不能乱：

messages: [
  ① system      "你是编程助手，可用工具: read, bash..."
  ② user        "帮我看看 package.json"
  ③ assistant   content: [ ToolCall{ name:"read", arguments:{path:"package.json"} } ]
  ④ toolResult  content: [ Text{ text:"{ \"name\": \"pi-monorepo\", ... }" } ]
  ⑤ assistant   content: [ Text{ text:"这个项目叫 pi-monorepo..." } ]   ← 最终回答
]

每条消息的 content 不是一整根字符串，而是内容块的数组（像乐高一块块拼）：

// types.ts:420  普通文字块
export interface TextContent    { type: "text";  text: string; ... }

// types.ts:426  思考块（推理模型会先"打草稿"）
export interface ThinkingContent{ type: "thinking"; thinking: string; ... }

// types.ts:436  图片块
export interface ImageContent   { type: "image"; data: string; mimeType: string; }  // base64

// types.ts:442  工具调用块（模型说"我要用工具"）
export interface ToolCall {
  type: "toolCall";
  id: string;                 // 调用的唯一编号，结果要凭它认领
  name: string;               // 工具名，如 "bash"
  arguments: JsonObject;      // 参数，如 { command: "npm test" }
}

一个 assistant 消息可以混合多种块：

AssistantMessage.content = [
  ThinkingContent("用户想让我先看看目录..."),   ← 思考
  TextContent("我来查看目录结构。"),            ← 说话
  ToolCall{ name:"bash", arguments:{...} },   ← 同时发起工具调用
]

两个"记账"字段：

// types.ts:457
export interface Usage {
  input: number; output: number;          // 输入/输出 token 数
  cacheRead: number; cacheWrite: number;  // 缓存命中部分（省钱的关键）
  cost: { input: number; output: number; ... total: number };  // 直接算好花了多少钱
}

// types.ts:480  回合结束的原因
export type StopReason = "pending" | "stop" | "length" | "toolUse" | "error" | "aborted" | "deferred";
//                                            ▲ 用完长度上限  ▲ 模型想用工具

3. 概念二：Model —— 模型的身份证

types.ts:845，一个 Model 对象描述"某个具体模型怎么找到、多贵、多能装"：

export interface Model<TApi extends Api> extends BaseModel<TApi> {
  id: "claude-sonnet-4-5"    // 型号
  name: "Claude Sonnet 4.5"  // 展示名
  api: "anthropic-messages"  // ★ 说哪种方言 → 决定用哪个翻译官
  provider: "anthropic"      // 哪家厂商
  baseUrl: "https://api.anthropic.com"  // 服务器地址
  contextWindow: 200000      // 肚子多大（能装多少 token）
  maxTokens: 64000           // 一次最多吐多少
  cost: { input: 3, output: 15, ... }   // 单价（美元/百万 token）
  reasoning: true            // 会不会"思考"
}

api 字段是整个设计的枢纽：像机场的登机口编号，拿着它去 api/ 目录找对应的翻译官。

关键统计（types.ts:88-131 的 KnownProvider vs types.ts:62-72 的 KnownApi）： 40 家厂商，只有 9 种方言。 大量厂商直接"蹭"别人的接口——DeepSeek 兼容 OpenAI 格式。

真实例子 providers/deepseek.ts（全文 15 行）：

export function deepseekProvider(): Provider<"openai-completions"> {
  return createProvider({
    id: "deepseek",
    baseUrl: "https://api.deepseek.com",
    auth: { apiKey: envApiKeyAuth("DeepSeek API key", ["DEEPSEEK_API_KEY"]) },
    models: Object.values(DEEPSEEK_MODELS),
    api: openAICompletionsApi(),   // ← 直接复用 OpenAI 翻译官！零翻译成本
  });
}

厂商 × 方言的映射（节选）:

  Anthropic ──────────────▶ anthropic-messages      (自家方言)
  OpenAI ─────────────────▶ openai-responses        (自家新方言)
  DeepSeek/Groq/xAI/Cerebras ▶ openai-completions   (蹭 OpenAI 旧方言)
  Google/Gemini ───────────▶ google-generative-ai   (自家方言)
  Azure ───────────────────▶ azure-openai-responses (OpenAI 的企业版)
        40 个名片文件           9 个翻译官文件

4. 概念三：适配器契约 —— 翻译官的"上岗证"

9 个翻译官文件有一模一样的签名。types.ts:396 定义的合同：

export type StreamFunction<TApi, TOptions> = (
  model: Model<TApi>,              // 用哪个模型（身份证）
  context: TranscriptContext,      // 说什么（消息列表）
  options?: TOptions,              // 可选配置（温度、超时、密钥...）
) => AssistantMessageEventStream;   // 返回：一个事件流

Anthropic 翻译官的真实上岗现场，api/anthropic-messages.ts:571：

export const stream: StreamFunction<"anthropic-messages", AnthropicOptions> = (
  model: Model<"anthropic-messages">,
  context: TranscriptContext,
  options?: AnthropicOptions,
): AssistantMessageEventStream => {
  const stream = new AssistantMessageEventStream();   // ① 备好一个事件流
  ...
  (async () => {
    const output: AssistantMessage = { role: "assistant", content: [], ... stopReason: "pending" };
    // ② 把统一格式翻译成 Anthropic 的 HTTP 请求，发出去
    // ③ 边收 HTTP 流，边往 stream 里 push 事件
  })();
  return stream;
};

翻译官干三件事：翻译请求 → 发 HTTP → 把厂商的回复流"同声传译"成统一事件。 1575 行里绝大部分是处理 Anthropic 的怪癖（思考块签名、缓存标记、报错格式），主干就这三步。

5. 概念四：流式事件 —— 字是怎么一个一个飞回来的

ChatGPT 里文字逐渐冒出来，靠 HTTP 流式响应（SSE）。 pi 把任何厂商的流统一翻译成一套事件，定义在 types.ts:812：

export type AssistantMessageEvent =
  | { type: "start";          partial: AssistantMessage }                    // 回合开始
  | { type: "thinking_start"; contentIndex: number; partial: AssistantMessage }
  | { type: "thinking_delta"; delta: string; ... }     // 思考文字飘来一块
  | { type: "text_start";     ... }                    // 开始说正文
  | { type: "text_delta";     delta: string; ... }     // ★ 正文飘来一小段
  | { type: "text_end";       content: string; ... }   // 正文说完了
  | { type: "toolcall_start"; ... }
  | { type: "toolcall_delta"; delta: string; ... }     // 工具参数 JSON 一点点飘来
  | { type: "toolcall_end";   toolCall: ToolCall; ... }
  | { type: "done";  reason: "stop"|"toolUse"|...; message: AssistantMessage }  // 大团圆
  | { type: "error"; reason: "aborted"|"error"; error: AssistantMessage };

一次完整回复的事件时间线（假设模型思考后调用了 bash）：

时间 ──────────────────────────────────────────────────────────────────▶

 start ─ thinking_start ─ thinking_delta ─ ... ─ thinking_end
                                                        └─ text_start ─ text_delta("我先") ─ text_delta("跑个测试") ─ text_end
                                                                                                    └─ toolcall_start ─ toolcall_delta('{"command"') ─ ... ─ toolcall_end
                                                                                                                                                      └─ done (reason:"toolUse")
 每个事件都带 partial: 当时的完整消息快照 ── 界面拿它直接画，不用自己拼

事件靠 utils/event-stream.ts:26 的 EventStream 类送达，本质是生产者-消费者队列：

翻译官(生产者)                      EventStream                    界面/代理循环(消费者)
    │                                  │                                │
    │ push(text_delta "我先")  ────────▶ [队列: ①]                      │
    │ push(text_delta "跑个")  ────────▶ [队列: ①②]                     │
    │                                  │          for await (event of stream) ─▶ 消费① 画屏幕
    │ push(text_delta "测试")  ────────▶ [队列: ②③]                     │
    │                                  │                    消费②③
    │ push(done) ─────────────────────▶ [队列: ④] ── resolve(result) ──▶ result() 的 Promise 也兑现

核心方法（event-stream.ts:43-58 和 72-84）：

push(event: T): void {                      // 生产者调用
  if (this.isComplete(event)) {             // done/error 事件 → 整个流结束
    this.done = true;
    this.resolveFinalResult(this.extractResult(event));  // result() 兑现
  }
  const waiter = this.waiting.dequeue();
  if (waiter) waiter({ value: event, done: false });  // 有人在等 → 立刻给他
  else this.queue.enqueue(event);           // 没人等 → 先存队列，防止丢事件
}

async *[Symbol.asyncIterator]() {           // 消费者写 for await 就能用
  while (true) {
    if (this.queue.length > 0) yield this.queue.dequeue()!;  // 队列有货先吐货
    else if (this.done) return;                               // 货尽流停
    else {
      const result = await new Promise(r => this.waiting.enqueue(r));  // 睡觉等货
      if (result.done) return;
      yield result.value;
    }
  }
}

调用方的代码非常干净（伪代码）：

const stream = model.stream(context);
for await (const event of stream) {     // 一行消费所有事件
  if (event.type === "text_delta") 画到屏幕上(event.delta);
}
const final = await stream.result();    // 顺便拿到完整最终消息

一个类，两种用法：for await 逐个看事件（界面用），或 result() 只要最终结果（后台用）。

6. 一个精妙的小设计：骗不过去的类型（品牌类型）

types.ts:783-794：

declare const transcriptContextBrand: unique symbol;   // 一个"不存在的记号"

export type TranscriptContext = {
  messages: Message[];
  readonly [transcriptContextBrand]: true;   // 携带这个记号的才算数
};

问题：stream() 收到的消息列表必须先经过 normalizeContext() 整理 （把散落的 systemPrompt、tools 折叠进首条 system 消息）。 万一开发者把没整理的原始 Context 直接塞给翻译官怎么办？

解法：TranscriptContext 强制携带一个运行时不存在、只在类型检查时存在的记号字段。 只有 normalizeContext() 的返回值带这个记号：

normalizeContext(ctx) ──▶ TranscriptContext ✅ 类型对上，能进翻译官
raw Context ────────────▶ 缺记号 ❌ 编译直接报错，运行时错误根本没机会发生

行话叫 branded type（品牌类型），思想是"让非法状态无法被表示出来"—— 靠编译器把关，而不是靠人肉小心。

7. 本课总结

            ┌─────────────────────── 你写的调用代码 ───────────────────────┐
            │  models.stream({ systemPrompt, messages, tools })            │
            └──────────────────────────────┬──────────────────────────────┘
                                           ▼
                    ┌────────── Context ──────────┐
                    │ systemPrompt / messages / tools │   ← 2 四种 Message + 内容块
                    └──────────────┬──────────────┘
                                   ▼ normalizeContext() 折叠
                          TranscriptContext（带品牌记号，6）
                                   ▼
        ┌─────────── 按 model.api 找翻译官（4 契约）──────────────┐
        │  api/anthropic-messages.ts │ api/openai-completions.ts │ ...
        └──────────────────────────────┬───────────────────────┘
                                       ▼ 翻译成厂商 HTTP，发出去
                        事件流 (5)：start → *_delta → done
                                       ▼
                     AssistantMessage（含 usage 记账 + stopReason）

要点清单：

1. 消息 = 4 种角色之一；内容 = 块数组（text / thinking / image / toolCall）。
2. Model.api 字段决定用哪个翻译官；40 家厂商只对应 9 种方言。
3. 每个翻译官实现同一个 StreamFunction 签名，返回统一的事件流。
4. 事件流 = 生产者-消费者队列；界面 for await 消费，后台 result() 等终局。
5. 品牌类型让"没整理的上下文"在编译期就被拦下。

下一课：packages/agent —— 5 个文件实现的 AgentLoop 主循环。