正在加载

关于pi-ai的所思 · Sleepy

pi-ai：统一 AI SDK 层的设计与使用

1. pi-ai 是什么？

pi-ai 可以理解成一个 统一的 AI SDK / LLM Runtime。

我们平时调用不同厂商的模型，通常需要使用各自提供的 SDK，例如：

openai
@anthropic-ai/sdk
@google/genai

不同 SDK 的请求参数、Message 格式、Tool Calling、Reasoning、Streaming Event 等都有差异。

例如：

// OpenAI
openai.responses.create(...)
 
// Anthropic
anthropic.messages.create(...)

如果业务代码直接依赖这些 SDK，那么切换模型厂商时，就需要处理大量厂商特有逻辑。

pi-ai 在这些厂商 SDK 上增加了一层抽象：

Mermaid

正在准备图形，源码仍可阅读。

flowchart TD
    A[业务 / Agent] --> B[pi-ai]
    B --> C[Provider / API Adapter]
    C --> D[OpenAI SDK]
    C --> E[Anthropic SDK]
    C --> F[Google SDK]
    D --> G[OpenAI API]
    E --> H[Anthropic API]
    F --> I[Gemini API]

它采用了类似 依赖倒置 + Adapter 的设计。

高层代码不直接依赖 OpenAI、Anthropic，而是依赖 pi-ai 定义的统一抽象，例如：

代码

Model
Context
Message
Tool
Stream Event

具体使用哪个厂商，则由 pi-ai 在下面完成适配。

因此可以简单理解成：

OpenAI SDK 解决“如何调用 OpenAI”，Anthropic SDK 解决“如何调用 Anthropic”，而 pi-ai 解决“如何用同一种方式调用不同厂商”。

2. pi-ai 主要负责什么？

pi-ai 主要负责：

代码

模型管理
Provider 管理
认证 / API Key
统一 Message
统一 Tool Calling
统一 Reasoning
统一 Streaming Event
Usage / Token / Cost
厂商协议转换

例如： OpenAI的格式

const model = models.getModel(
  "openai",
  "gpt-4o-mini"
);
 
const response = await models.complete(
  model,
  context
);

如果换成 Anthropic：

const model = models.getModel(
  "anthropic",
  "claude-sonnet-4-5"
);
 
const response = await models.complete(
  model,
  context
);

上层代码的调用方式基本不变。

pi-ai 内部根据 model.provider 和 model.api，找到对应 Provider 和 API Adapter，再调用真正的厂商 SDK。

而且：

Provider 不等于协议。

例如一些 Provider 可以共享 OpenAI-compatible 协议实现：

Mermaid

正在准备图形，源码仍可阅读。

flowchart LR
    A[OpenAI Provider] --> B[openai-responses]
    C[Anthropic Provider] --> D[anthropic-messages]

    E[OpenRouter Provider] --> F[openai-compatible]
    G[Groq Provider] --> F
    H[xAI Provider] --> F

这样就不需要每个 OpenAI-compatible Provider 都重新实现一套请求和 Streaming 解析逻辑。

3. pi-ai 不负责什么？

pi-ai 本身并不是 Agent，也不是 Session 管理器。

它不会负责：

代码

长期保存聊天历史
Session 持久化
决定保留哪些历史消息
上下文压缩
执行 Tool
Agent 循环
业务状态管理
...

调用模型时，需要由上层把当前模型应该看到的 Context 传给它：

const context = {
  systemPrompt: "You are a helpful assistant.",
 
  messages: [
    {
      role: "user",
      content: "你好",
      timestamp: Date.now()
    }
  ]
};

所以可以把它理解成一个比较接近无状态的调用：

代码

response = LLM(model, context, options)

状态由上层维护，pi-ai 负责模型调用。

职责关系可以理解为：

Mermaid

正在准备图形，源码仍可阅读。

flowchart TD
    A[业务 / Agent] --> B[维护 Session / Messages / 状态]
    B --> C[构造 Context]
    C --> D[pi-ai]
    D --> E[调用模型]
    E --> F[返回 AssistantMessage / Stream]

4. 常用接口

最核心的入口是：

createModels()

它创建一个模型运行时容器。

例如只注册 OpenAI：

import { createModels } from "@earendil-works/pi-ai";
import { openaiProvider } from "@earendil-works/pi-ai/providers/openai";
 
const models = createModels();
 
models.setProvider(
  openaiProvider()
);

如果想直接注册所有内置 Provider：

import {
  builtinModels
} from "@earendil-works/pi-ai/providers/all";
 
const models = builtinModels();

之后可以查询模型：

const model = models.getModel(
  "openai",
  "gpt-4o-mini"
);
 
const allModels = models.getModels();
 
const anthropicModels =
  models.getModels("anthropic");

可以把 Models 简单理解成：

Mermaid

正在准备图形，源码仍可阅读。

flowchart LR
    A[Models] --> B[Provider Registry]
    A --> C[Model Registry]
    A --> D[Auth Manager]
    A --> E[Request Router]

除此之外，比较常用的是：

models.stream(...)
models.complete(...)
 
models.streamSimple(...)
models.completeSimple(...)

其中 Simple 版本主要用于提供更加 Provider-neutral 的参数。

例如统一指定：

{
  reasoning: "high"
}

上层只表达：

我想要 high reasoning。

至于底层 OpenAI、Anthropic、Google 分别应该转换成什么参数，由对应 Adapter 处理。

5. 流式与非流式

所谓流式和非流式，最终模型表达的内容本身没有本质区别，区别主要在数据到达客户端的方式。

非流式

非流式会等待模型把整次 Response 生成完成，再一次性返回最终结果。

const response = await models.complete(
  model,
  context
);

例如最终得到：

{
  role: "assistant",
  content: [
    {
      type: "text",
      text: "你好，我可以帮你解决问题。"
    }
  ]
}

调用过程：

Mermaid

正在准备图形，源码仍可阅读。

sequenceDiagram
    participant App as 应用
    participant AI as pi-ai / LLM

    App->>AI: 请求
    Note over AI: 模型持续生成内容
    Note over AI: 模型持续生成内容
    Note over AI: 生成完成
    AI-->>App: 一次性返回完整 Response

优点是使用简单。

缺点是如果模型生成时间很长，用户在最终 Response 返回之前看不到内容。

流式

流式调用：

const stream = models.stream(
  model,
  context
);
 
for await (const event of stream) {
  console.log(event);
}

模型生成一点，服务端就返回一点。

例如模型最终输出：

代码

你好，我可以帮你解决问题。

流式过程中可能收到：

代码

text_delta: "你好"
text_delta: "，"
text_delta: "我可以"
text_delta: "帮你解决"
text_delta: "问题。"

前端不断 append：

let text = "";
 
for await (const event of stream) {
  if (event.type === "text_delta") {
    text += event.delta;
  }
}

用户看到的过程类似：

Mermaid

正在准备图形，源码仍可阅读。

sequenceDiagram
    participant App as 应用
    participant AI as pi-ai / LLM

    App->>AI: 请求
    AI-->>App: text_delta "你好"
    AI-->>App: text_delta "，"
    AI-->>App: text_delta "我可以"
    AI-->>App: text_delta "帮你解决"
    AI-->>App: text_delta "问题。"
    AI-->>App: done

这就是 AI 产品常见的“打字机效果”。

6. pi-ai 的流式事件

不同厂商原始 Streaming Event 完全不同。

OpenAI 可能返回：

代码

response.output_text.delta
response.function_call_arguments.delta

Anthropic 可能返回：

代码

content_block_delta
text_delta
input_json_delta

pi-ai 把它们统一成自己的事件：

代码

start
 
thinking_start
thinking_delta
thinking_end
 
text_start
text_delta
text_end
 
toolcall_start
toolcall_delta
toolcall_end
 
done
error

于是上层只需要处理 pi-ai 的事件：

const stream = models.stream(
  model,
  context
);
 
for await (const event of stream) {
  switch (event.type) {
    case "text_delta":
      process.stdout.write(event.delta);
      break;
 
    case "thinking_delta"

整个适配过程可以理解成：

Mermaid

正在准备图形，源码仍可阅读。

flowchart TD
    A[OpenAI Streaming Event] --> D[pi-ai Adapter]
    B[Anthropic Streaming Event] --> D
    C[Google Streaming Event] --> D

    D --> E[统一 Stream Event]

    E --> F[text_delta]
    E --> G[thinking_delta]
    E --> H[toolcall_delta]
    E --> I[done]
    E --> J[error]

因此上层不需要判断底层到底是 OpenAI 还是 Anthropic。

7. Tool Call 也可以流式返回

Tool Call 本质也是模型生成内容的一部分，因此同样可以 Streaming。

假设模型决定调用：

get_weather({
  city: "Shanghai"
});

底层厂商真实 Streaming 时，参数可能并不是一次性返回。

第一段：

代码

{"city

第二段：

代码

":"Shang

第三段：

代码

hai"}

最终才能得到：

json

{
  "city": "Shanghai"
}

整体过程：

Mermaid

正在准备图形，源码仍可阅读。

sequenceDiagram
    participant LLM as 模型
    participant Adapter as pi-ai
    participant App as 上层应用

    LLM-->>Adapter: tool call start: get_weather
    Adapter-->>App: toolcall_start

    LLM-->>Adapter: {"city
    Adapter-->>App: toolcall_delta

    LLM-->>Adapter: ":"Shang
    Adapter-->>App: toolcall_delta

    LLM-->>Adapter: hai"}
    Adapter-->>App: toolcall_delta

    Note over Adapter: 拼接并解析完整 arguments

    Adapter-->>App: toolcall_end

所以底层 AI SDK 收到的 Tool arguments 本质上可能只是 partial JSON。

不过使用 pi-ai 时，它已经帮我们处理了这一层。

我们可以观察中间过程：

case "toolcall_delta": {
  const partial =
    event.partial.content[
      event.contentIndex
    ];
 
  console.log(partial);
 
  break;
}

等 Tool Call 完成：

case "toolcall_end": {
  console.log(
    event.toolCall.name
  );
 
  console.log(
    event.toolCall.arguments
  );
 
  break;
}

此时：

event.toolCall.arguments

已经是完整参数。

因此应该区分：

Mermaid

正在准备图形，源码仍可阅读。

flowchart TD
    A[厂商 Streaming API] --> B[arguments delta]
    B --> C[arguments delta]
    C --> D[arguments delta]
    D --> E[pi-ai 聚合 / 解析]
    E --> F[完整 ToolCall]
    F --> G[toolcall_end]

简单来说：

厂商底层 Streaming 中，Tool arguments 确实可能是碎片。

pi-ai 会负责聚合和统一，上层可以通过 toolcall_delta 观察过程，通过 toolcall_end 获得完整 Tool Call。

8. Streaming 最后仍然可以拿到完整结果

流式并不意味着只能自己拼最终消息。

例如：

const stream = models.stream(
  model,
  context
);
 
for await (const event of stream) {
  if (event.type === "text_delta") {
    process.stdout.write(event.delta);
  }
}
 
const message =
  await stream.result

stream.result() 最终仍然可以得到完整的 Assistant Message：

{
  role: "assistant",
 
  content: [
    {
      type: "text",
      text: "完整回答"
    }
  ],
 
  usage: {
    ...
  }
}

所以二者的关系可以理解成：

Mermaid

正在准备图形，源码仍可阅读。

flowchart LR
    A[非流式 Request] --> B[等待生成完成]
    B --> C[Final Response]

流式则是：

Mermaid

正在准备图形，源码仍可阅读。

flowchart LR
    A[流式 Request] --> B[Event]
    B --> C[Event]
    C --> D[Event]
    D --> E[Event]
    E --> F[Final Response]

因此：

非流式直接拿最终状态，流式可以在最终状态产生的过程中持续消费增量事件。

9. Tool Call 并不等于 Tool Execution

这里还需要区分两个概念：

模型决定调用工具。

和：

程序真正执行工具。

完全是两回事。

pi-ai 负责告诉你：

{
  type: "toolCall",
  name: "get_weather",
  arguments: {
    city: "Shanghai"
  }
}

但是：

getWeather(...)

需要由上层程序自己执行。

执行完成以后，再把 Tool Result 放回 Context：

context.messages.push({
  role: "toolResult",
 
  toolCallId: call.id,
 
  toolName: call.name,
 
  content: [
    {
      type: "text",
      text: "Shanghai: 28°C"
    }
  ],
 
  isError: false,
 
  timestamp: Date.now()
});

之后再次请求模型：

const response =
  await models.complete(
    model,
    context
  );

整个过程就是一个典型的 Agent Loop：

Mermaid

正在准备图形，源码仍可阅读。

flowchart TD
    A[用户问题] --> B[调用 LLM]
    B --> C{模型是否产生 Tool Call?}

    C -- 否 --> H[返回最终回答]

    C -- 是 --> D[应用执行 Tool]
    D --> E[得到 Tool Result]
    E --> F[Tool Result 放回 Context]
    F --> G[再次调用 LLM]
    G --> C

这个循环属于 Agent 层的职责，而不是 pi-ai 本身。

10. 总结

整个架构可以简化为：

Mermaid

正在准备图形，源码仍可阅读。

flowchart TD
    A[业务 / Agent] --> B[pi-ai]

    B --> C[统一 Model]
    B --> D[统一 Context / Message]
    B --> E[统一 Tool]
    B --> F[统一 Streaming Event]

    C --> G[Provider / API Adapter]
    D --> G
    E --> G
    F --> G

    G --> H[OpenAI SDK]
    G --> I[Anthropic SDK]
    G --> J[Google SDK]

    H --> K[OpenAI API]
    I --> L[Anthropic API]
    J --> M[Gemini API]

因此可以给 pi-ai 一个比较准确的定义：

pi-ai = 一个无状态、多 Provider 的统一 AI SDK / LLM Runtime。

它最主要解决的是：

屏蔽不同 AI 厂商 SDK 和 API 的差异，让上层只面对一种模型调用方式。

流式与非流式则可以简单理解为：

非流式返回最终状态，流式返回生成最终状态过程中的一系列增量事件。

而 Tool Call 同样可以 Streaming：

底层厂商返回的参数可能是 partial JSON，pi-ai 会负责聚合并统一成完整的 Tool Call。

最后把几个层次放在一起：

Mermaid

正在准备图形，源码仍可阅读。

flowchart TD
    A[业务 / Coding Agent] --> B[Agent Runtime]
    B --> C[pi-ai]
    C --> D[OpenAI / Anthropic / Google SDK]
    D --> E[厂商 API]

可以分别理解为：

业务 / Agent：维护状态、Session、历史消息，决定下一步做什么。
pi-ai：统一调用模型，屏蔽不同 Provider 的差异。
官方 AI SDK：负责调用某一家厂商的 API。
厂商 API：真正运行模型并产生 Response。