oh-my-pi 全景代码解析（二）：pi-ai 统一 LLM 抽象层

系列导读：本文深入 Pi 架构的最底层——pi-ai 包。这个包负责将所有 LLM Provider 的差异封装在一个统一的流式接口之后，是 Pi 能够支持 15+ Provider、数百种模型、并实现 mid-session 模型切换的技术根基。

---

一、为什么需要统一抽象层

在 AI Agent 开发中，每个 LLM Provider 都有独特的 API 格式、认证方式、流式协议、工具调用格式和上下文管理机制。

Provider 差异全景

graph LR
    subgraph "OpenAI"
        O1[Chat Completions API]
        O2[tool_calls array]
        O3[content string]
        O4[SSE streaming]
    end

    subgraph "Anthropic"
        A1[Messages API]
        A2[content blocks array]
        A3[thinking blocks]
        A4[SSE streaming]
    end

    subgraph "Google"
        G1[Generative Language API]
        G2[functionCalls array]
        G3[contents array]
        G4[Custom streaming]
    end

    subgraph "Ollama"
        L1[Local API]
        L2[Custom format]
        L3[No tool standard]
    end

    O1 -->|Unified| U[pi-ai Unified Layer]
    A1 -->|Unified| U
    G1 -->|Unified| U
    L1 -->|Unified| U

    U -->|Stream Events| Agent[Agent Core]

如果没有统一抽象，Agent 核心逻辑将与特定 Provider 深度耦合。切换模型？重写请求构建逻辑。支持新 Provider？复制粘贴并修改大量代码。mid-session 模型切换？几乎不可能。

---

二、核心类型系统：通用语言的设计

2.1 Model 类型：能力声明

interface Model<Api extends string = string> {
    id: string;                    // "gpt-5.1-codex"
    api: Api;                      // "openai" | "anthropic" | ...
    provider: string;
    reasoning?: boolean;           // Capability declaration
    input: ("text" | "image" | "audio")[];
    cost: {
        input: number;             // $ per 1K tokens
        output: number;
        cacheRead?: number;
        cacheWrite?: number;
    };
    contextWindow: number;
    maxTokens: number;
}

编译时类型安全：

const openaiModel: Model<"openai"> = {
    id: "gpt-5.1-codex",
    api: "openai",  // Valid
};

const badModel: Model<"unknown"> = {
    api: "unknown",  // TypeScript Error
};

2.2 Context 类型：对话上下文的统一表示

$\text{Context} = \langle \text{SystemPrompt}, \text{Messages}, \text{Tools}, \text{Config} \rangle$

classDiagram
    class Context {
        +string systemPrompt
        +Message[] messages
        +Tool[] tools
        +number temperature
        +number maxTokens
        +ToolChoice toolChoice
    }

    class Message {
        +Role role
        +string content
        +number timestamp
    }

    class UserMessage {
        +role: "user"
        +content: string | ContentPart[]
    }

    class AssistantMessage {
        +role: "assistant"
        +content: string
        +thinking?: string
        +tool_calls?: ToolCall[]
    }

    class ToolMessage {
        +role: "tool"
        +content: string
        +tool_call_id: string
    }

    Context --> Message
    Message <|-- UserMessage
    Message <|-- AssistantMessage
    Message <|-- ToolMessage

2.3 工具调用统一契约

interface ToolCall {
    id: string;                    // Unique identifier
    type: "function";
    function: {
        name: string;
        arguments: string;         // JSON string
    };
}

---

三、Provider 注册机制：插件化架构

3.1 运行时注册表

graph TB
    subgraph "Registry"
        R[Map string, ApiProvider]
    end

    subgraph "Built-in Providers"
        P1[OpenAI]
        P2[Anthropic]
        P3[Google]
        P4[Azure]
        P5[Ollama]
        P6[OpenRouter]
    end

    subgraph "Extension Providers"
        E1[Custom API]
        E2[Local Model]
    end

    P1 -->|register| R
    P2 -->|register| R
    P3 -->|register| R
    P4 -->|register| R
    P5 -->|register| R
    P6 -->|register| R
    E1 -->|register| R
    E2 -->|register| R

    R -->|getProvider| Consumer[Agent Core]

核心注册代码：

const registry = new Map<string, ApiProvider>();

export function registerApiProvider(
    api: string, 
    provider: ApiProvider
): void {
    if (registry.has(api)) {
        console.warn(`Overwriting provider: ${api}`);
    }
    registry.set(api, provider);
}

export function getProvider(api: string): ApiProvider {
    const p = registry.get(api);
    if (!p) throw new Error(
        `Unknown API: "${api}". ` +
        `Available: ${[...registry.keys()].join(", ")}`
    );
    return p;
}

3.2 四步集成模式

\text{New Provider} \xrightarrow{1} \text{Implement Streams} \xrightarrow{2} \text{Type Export} \xrightarrow{3} \text{Register} \xrightarrow{4} \text{Consume}

---

四、流式事件模型：统一的事件流抽象

4.1 事件类型设计

stateDiagram-v2
    [*] --> TextDelta
    [*] --> ThinkingDelta
    [*] --> ToolCallStart

    TextDelta --> TextDelta
    TextDelta --> ToolCallStart
    TextDelta --> Usage
    TextDelta --> Done

    ToolCallStart --> ToolCallDelta
    ToolCallDelta --> ToolCallDelta
    ToolCallDelta --> ToolCallComplete

    ToolCallComplete --> TextDelta
    ToolCallComplete --> Usage

    ThinkingDelta --> TextDelta
    ThinkingDelta --> Done

    Usage --> Done

    Done --> [*]

    Error --> [*]

4.2 AssistantMessageEventStream 实现

基于 Web Streams API 的 ReadableStream：

class AssistantMessageEventStream 
    implements AsyncIterable<AssistantMessageEvent> {

    private readable: ReadableStream<AssistantMessageEvent>;

    constructor(source: UnderlyingSource) {
        this.readable = new ReadableStream(source);
    }

    [Symbol.asyncIterator](): AsyncIterator<AssistantMessageEvent> {
        const reader = this.readable.getReader();
        return {
            async next() {
                const { done, value } = await reader.read();
                return done 
                    ? { done: true, value: undefined }
                    : { done: false, value };
            }
        };
    }

    onTextDelta(handler: (delta: string) => void): void {
        this.on("text_delta", e => {
            if (e.type === "text_delta") handler(e.delta);
        });
    }
}

4.3 OpenAI Provider 流式转换

sequenceDiagram
    participant Agent as Agent Core
    participant OAI as OpenAI Provider
    participant API as OpenAI API

    Agent->>OAI: streamSimple(model, context)
    OAI->>API: POST /v1/chat/completions
    API-->>OAI: SSE: data: {delta: "Hello"}
    OAI-->>Agent: {type: "text_delta", delta: "Hello"}
    API-->>OAI: SSE: data: {tool_calls: [...]}
    OAI-->>Agent: {type: "tool_call_start", ...}
    API-->>OAI: SSE: data: [DONE]
    OAI-->>Agent: {type: "done", message: {...}}
    OAI-->>Agent: {type: "usage", inputTokens: 1200, ...}

---

五、跨 Provider 上下文交接

5.1 问题分析

当从 Anthropic 切换到 OpenAI 时，以下差异需要处理：

graph LR
    subgraph "Anthropic Context"
        A1[thinking blocks]
        A2[tool_use blocks]
        A3[tool_result blocks]
        A4[system param]
    end

    subgraph "Conversion"
        C1[thinking tags]
        C2[tool_calls array]
        C3[tool messages]
        C4[system role message]
    end

    subgraph "OpenAI Context"
        O1[thinking...thinking]
        O2[tool_calls]
        O3[role: tool]
        O4[role: system]
    end

    A1 --> C1 --> O1
    A2 --> C2 --> O2
    A3 --> C3 --> O3
    A4 --> C4 --> O4

Mario Zechner 的坦诚评价：

"Context handoff between providers was a feature pi-ai was designed for from the start… This can only be a best-effort thing."

5.2 标准化中间表示

function convertContextForProvider(
    context: Context,
    source: string,
    target: string
): Context {
    if (source === target) return context;

    return {
        ...context,
        messages: context.messages.map(msg => {
            if (msg.role === "assistant" && msg.thinking && target !== "anthropic") {
                return {
                    ...msg,
                    content: `<thinking>\n${msg.thinking}\n</thinking>\n\n${msg.content}`,
                    thinking: undefined,
                };
            }
            return msg;
        })
    };
}

---

六、高级功能：缓存、重试、并发控制

6.1 提示缓存（Prompt Caching）

Anthropic 和 OpenAI 支持提示缓存，显著降低长上下文成本：

$\text{Cost}_{\text{cache}} = \text{Cost}_{\text{input}} \times \alpha_{\text{cache}}$

其中 $\alpha_{\text{cache}} \approx 0.1$ 表示缓存读取成本约为正常输入成本的 10%。

6.2 自动重试与指数退避

graph TD
    A[Request] --> B{Success?}
    B -->|Yes| C[Return Result]
    B -->|No| D{Retryable?}
    D -->|No| E[Throw Error]
    D -->|Yes| F{Attempt less than Max?}
    F -->|No| E
    F -->|Yes| G[Wait: 2^attempt x 1s + jitter]
    G --> A

6.3 背压控制（Backpressure）

for await (const event of stream) {
    // Consumer slower than producer:
    // ReadableStream auto-pauses reading
    await renderToUI(event);  // Slow operation
}

---

七、设计原则总结

mindmap
  root((pi-ai 设计原则))
    抽象层即通用语言
      类型系统表达所有变体
      非简单适配器映射
    流式是默认
      非流式是特例
      事件流为核心契约
    Provider 差异封装
      能力声明字段
      上层可做出决策
    尽力而为兼容性
      信息传递大于格式保真
      thinking to tags
    零拷贝解析
      TextDecoder streaming
      避免中间字符串

pi-ai 的设计为我们提供了可复用的架构原则：好的抽象层定义一套能够表达所有底层变体的类型系统，而非简单地将 A 的 API 映射到 B 的 API。

在下一篇中，我们将上升到 pi-agent-core，剖析 Agent 的核心循环、权限门设计、工具调用生命周期、以及状态机的实现细节。

---

本系列基于 oh-my-pi v2026.07 和 Pi Agent v0.66+ 的公开源码与文档编写。