PiMono 源码精读PiMono Deep Dive

pi-ai Adapter Pattern 概念扫pi-ai Adapter Pattern Overview

源码：sources/pi-mono/packages/ai/src/ | 约 40 个文件 | 概念扫（看骨架，不逐行）Source: sources/pi-mono/packages/ai/src/ | ~40 files | Overview (architecture, not line-by-line)

这是什么？为什么要看它？What Is This? Why Read It?

我们已经精读过 agent-core —— 那是 agent 的引擎，核心就是一个 while 循环：调 LLM → 检查工具调用 → 执行 → 结果喂回 → 再调 LLM。

We've already done a deep read of agent-core — the agent's engine, whose core is a while loop: call LLM → check for tool calls → execute → feed results back → call LLM again.

但"调 LLM"这三个字背后，具体发生了什么？

But what actually happens behind the words "call LLM"?

引擎里有一个回调函数叫 streamFn。每次循环要调 LLM 时，引擎就喊一声 streamFn(messages, tools, systemPrompt) —— "帮我把这些内容发给 LLM，然后给我一个统一格式的回复流"。

Inside the engine there's a callback called streamFn. Every time the loop needs to call the LLM, the engine shouts streamFn(messages, tools, systemPrompt) — "send this to the LLM and give me back a unified response stream."

pi-ai 就是 streamFn 背后干活的那一层。

pi-ai is the layer that does the work behind streamFn.

问题是：世界上有 27 家 LLM 供应商（Anthropic、OpenAI、Google、DeepSeek、Groq……），每家的 API 格式都不一样。pi-ai 的工作就是当"翻译官"：

The problem: there are 27 LLM providers in the world (Anthropic, OpenAI, Google, DeepSeek, Groq...), each with a different API format. pi-ai's job is to be the "translator":

所以你现在读的是：agent 引擎调 LLM 时，底下那根管子是怎么接的。不需要逐行读，看懂架构模式就够。

So what you're reading now is: how the pipe underneath is connected when the agent engine calls the LLM. No need for line-by-line reading — understanding the architecture pattern is enough.

完整架构：从引擎到 LLM 供应商的调用链Full Architecture: The Call Chain from Engine to LLM Provider

model 参数从哪来？Where Does the model Parameter Come From?

model 不是 pi-ai 自己决定的，而是更上层给的。在 Pi 这个产品里，用户可以在界面上选模型（比如选 Claude Opus），管理层（agent.ts）把用户的选择存为一个 Model 对象，每次调引擎时传进去。如果你自己做产品，也可以在系统里直接写死"用 DeepSeek" —— 用户完全不知道。引擎不关心 model 是谁给的，它只管用。

model is not decided by pi-ai itself — it's given from above. In Pi the product, users can pick a model in the UI (e.g. Claude Opus), the management layer (agent.ts) stores the choice as a Model object and passes it to the engine each loop. If you build your own product, you could also hardcode "use DeepSeek" — the user would never know. The engine doesn't care who provides the model; it just uses it.

model 从哪来？ ──────────────── 用户在界面选模型（如"Claude Opus"）/ 产品方写死配置 │ │ 用户选的只是一个名字， │ 管理层要去模型注册表查出完整信息 ▼ ┌─────────────────────────────────────────────────────────────────┐ │ 管理层（agent.ts） │ │ │ │ 拿到用户选择后，去模型注册表查出完整的 Model 对象 │ │ 设好 streamFn，每次跑循环时把 model + streamFn 交给引擎 │ │ │ │ ┌──────────────────────────────┐ │ │ │ 模型注册表 │ │ │ │ (models.ts + models.generated) │ │ │ │ │ │ │ │ 按"供应商 + 模型名"查询， │ │ │ │ 返回一个 Model 对象： │ │ │ │ │ │ │ │ { │ │ │ │ id: "claude-opus-4-6" │ │ │ │ provider: "anthropic" │ │ │ │ api: "anthropic-messages" │ ← 这个字段决定走哪条路 │ │ │ baseUrl: "https://..." │ │ │ │ cost: { input: 15, ... } │ │ │ │ contextWindow: 200000 │ │ │ │ } │ │ │ └──────────────────────────────┘ │ └────────────────────────────┬────────────────────────────────────┘ │ 传入 model + streamFn ▼
Where does model come from? ────────────────────────── User selects in UI (e.g. "Claude Opus") / hardcoded by product │ │ The user only picks a name; │ the management layer looks up full info from registry ▼ ┌─────────────────────────────────────────────────────────────────┐ │ Management Layer (agent.ts) │ │ │ │ After getting user's choice, looks up Model object from │ │ model registry. Sets up streamFn, passes model + streamFn │ │ to engine each loop. │ │ │ │ ┌──────────────────────────────┐ │ │ │ Model Registry │ │ │ │ (models.ts + models.generated)│ │ │ │ │ │ │ │ Query by "provider + model │ │ │ │ name", returns a Model object:│ │ │ │ │ │ │ │ { │ │ │ │ id: "claude-opus-4-6" │ │ │ │ provider: "anthropic" │ │ │ │ api: "anthropic-messages" │ ← this field decides the route│ │ │ baseUrl: "https://..." │ │ │ │ cost: { input: 15, ... } │ │ │ │ contextWindow: 200000 │ │ │ │ } │ │ │ └──────────────────────────────┘ │ └────────────────────────────┬────────────────────────────────────┘ │ passes model + streamFn ▼
┌─────────────────────────────────────────────────────────────────┐ │ 引擎（agent-loop.ts） │ │ │ │ 职责：跑 while 循环。 │ │ 每轮调一次 streamFn(model, context, options) │ │ 它不知道也不关心底下是哪家 LLM 供应商。 │ │ │ │ 传出：model + context + options │ │ 期望拿回：事件流（LLM 的回复） │ └──────────────┬──────────────────────────────────┬───────────────┘ │ ① ▲ ④ │ 调用 streamFn │ 事件流返回 ▼ │ ┌──────────────────────────────────────────────────────────────────┐ │ 路由（stream.ts，59 行） │ │ │ │ 职责：中转站。自己不干活，做两件事： │ │ 1. 读 model.api → 去 API 注册表查到对应的调用函数 │ │ 2. 用查到的函数调 provider，把结果原样返回给引擎 │ │ │ │ ┌─────────────────────┐ │ │ │ API 注册表 │ │ │ │ (api-registry.ts) │ │ │ │ │ │ │ │ API 名字 → 调用函数 │ │ │ │ 共 9 条 │ │ │ └─────────────────────┘ │ └──────────────┬──────────────────────────────────┬───────────────┘ │ ② ▲ ③ │ 调用查到的函数 │ 事件流返回 ▼ │ ┌──────────────────────────────────────────────────────────────────┐ │ Provider（providers/*.ts）—— 唯一真正干活的层 │ │ │ │ 收到：model + context + options │ │ 干活： │ │ context（统一格式）→ 翻译成供应商能懂的格式 → 发 HTTP 请求 │ │ HTTP 响应 → 翻译回统一格式 → 作为事件流返回 │ │ 返回：事件流 │ │ │ │ 每家供应商一个文件： │ │ anthropic.ts（1191 行）/ openai-completions.ts / google.ts / … │ └──────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────┐ │ Engine (agent-loop.ts) │ │ │ │ Job: run the while loop. │ │ Calls streamFn(model, context, options) each turn. │ │ It doesn't know or care which LLM provider is underneath. │ │ │ │ Sends out: model + context + options │ │ Expects back: event stream (LLM's response) │ └──────────────┬──────────────────────────────────┬───────────────┘ │ ① ▲ ④ │ calls streamFn │ event stream returns ▼ │ ┌──────────────────────────────────────────────────────────────────┐ │ Router (stream.ts, 59 lines) │ │ │ │ Job: relay station. Does no real work, just two things: │ │ 1. Reads model.api → looks up matching function from API registry│ │ 2. Calls the provider with that function, returns results to engine│ │ │ │ ┌─────────────────────┐ │ │ │ API Registry │ │ │ │ (api-registry.ts) │ │ │ │ │ │ │ │ API name → function │ │ │ │ 9 entries total │ │ │ └─────────────────────┘ │ └──────────────┬──────────────────────────────────┬───────────────┘ │ ② ▲ ③ │ calls the matched function │ event stream returns ▼ │ ┌──────────────────────────────────────────────────────────────────┐ │ Provider (providers/*.ts) — the only layer that does real work │ │ │ │ Receives: model + context + options │ │ Does: │ │ context (unified format) → translate to provider format → HTTP│ │ HTTP response → translate back to unified format → stream │ │ Returns: event stream │ │ │ │ One file per provider: │ │ anthropic.ts (1191 lines) / openai-completions.ts / google.ts / …│ └──────────────────────────────────────────────────────────────────┘

两张注册表的区别：

The difference between the two registries:

完整流程：

Complete flow:

  1. 引擎调路由：streamFn(model, context, options)
  2. Engine calls router: streamFn(model, context, options)
  3. 路由查 API 注册表后调 provider：providerFn(model, context, options)
  4. Router looks up API registry then calls provider: providerFn(model, context, options)
  5. Provider 干完活，事件流返回给路由
  6. Provider does the work, event stream returns to router
  7. 路由把事件流原样返回给引擎
  8. Router passes event stream unchanged back to engine
为什么不直接在引擎里调 Anthropic SDK？
那样引擎就被"焊死"在 Anthropic 上了。引擎的逻辑（循环、工具调用、停止判断）和"跟谁通信"是两件不同的事 —— 分开之后，换供应商只需要换底下的 provider 文件，引擎一个字都不用改。 Why not call the Anthropic SDK directly from the engine?
That would "weld" the engine to Anthropic. The engine's logic (looping, tool calls, stop conditions) and "who to talk to" are two separate concerns — by separating them, switching providers only requires swapping the provider file underneath. The engine doesn't change a single line.

Provider ≠ API：为什么 27 个供应商只需要 9 套代码Provider ≠ API: Why 27 Providers Need Only 9 Code Modules 通用模式Universal Pattern

这是理解整个 adapter 架构的关键区分。

This is the key distinction for understanding the entire adapter architecture.

模型注册表（models.ts） API 注册表（api-registry.ts） 记录每家供应商的模型信息 记录每种协议的调用函数 ───────────────────── ──────────────────────── Anthropic ─────────────────────────────→ anthropic-messages OpenAI ────────────────────────────────→ openai-completions DeepSeek ──────────────────┐ Groq ──────────────────────┤ xAI ───────────────────────┤ Cerebras ──────────────────┤ OpenRouter ────────────────┼──────────→ openai-completions（共用！） Fireworks ─────────────────┤ HuggingFace ───────────────┤ 小米 ──────────────────────┘ Google ────────────────────────────────→ google-generative-ai Google Vertex ─────────────────────────→ google-vertex Amazon Bedrock ────────────────────────→ bedrock-converse-stream Mistral ───────────────────────────────→ mistral-conversations Azure OpenAI ──────────────────────────→ azure-openai-responses OpenAI Codex ──────────────────────────→ openai-codex-responses OpenAI Responses ──────────────────────→ openai-responses 27 个 Provider 9 种 API
Model Registry (models.ts) API Registry (api-registry.ts) Records each provider's model info Records each protocol's call function ───────────────────────────── ──────────────────────────────── Anthropic ─────────────────────────────→ anthropic-messages OpenAI ────────────────────────────────→ openai-completions DeepSeek ──────────────────┐ Groq ──────────────────────┤ xAI ───────────────────────┤ Cerebras ──────────────────┤ OpenRouter ────────────────┼──────────→ openai-completions (shared!) Fireworks ─────────────────┤ HuggingFace ───────────────┤ Xiaomi ────────────────────┘ Google ────────────────────────────────→ google-generative-ai Google Vertex ─────────────────────────→ google-vertex Amazon Bedrock ────────────────────────→ bedrock-converse-stream Mistral ───────────────────────────────→ mistral-conversations Azure OpenAI ──────────────────────────→ azure-openai-responses OpenAI Codex ──────────────────────────→ openai-codex-responses OpenAI Responses ──────────────────────→ openai-responses 27 Providers 9 APIs
为什么这么设计？
因为很多小公司（DeepSeek、Groq、xAI……）直接照抄了 OpenAI 的接口协议。它们是不同的公司（provider 不同），但说的是同一种语言（API 相同）。

所以 Pi 把问题分成两层：
• 第一层：Provider + Model ID → 一个 Model 对象（里面有 api 字段，告诉你这个 provider 说什么语言）
• 第二层：API 名字 → 一个调用函数（知道怎么和这种语言通信）

结果：添加一个新 provider（比如小米），如果它用的是 OpenAI 协议，不需要写任何新的调用代码 —— 只需要在模型注册表里加一条记录。 Why design it this way?
Because many smaller companies (DeepSeek, Groq, xAI...) directly copied OpenAI's API protocol. They are different companies (different providers), but speak the same language (same API).

So Pi splits the problem into two layers:
• Layer 1: Provider + Model ID → a Model object (with an api field that tells you what language this provider speaks)
• Layer 2: API name → a call function (knows how to communicate in that language)

Result: adding a new provider (e.g. Xiaomi), if it uses the OpenAI protocol, requires zero new code — just add one entry to the model registry.
类比：想象你开一家翻译公司，要跟 27 个国家的客户打交道。
你发现这 27 个国家只用了 9 种语言 —— 很多小国说的是大国的语言。
所以你只需要 9 个翻译员（每种 API 一个 provider 实现），就能服务 27 个客户。
你维护的是一本"客户 → 语言"的对照表（model registry），和一本"语言 → 翻译员"的对照表（api registry）。 Analogy: Imagine you run a translation company serving 27 countries.
You discover these 27 countries only use 9 languages — many smaller countries speak a larger country's language.
So you only need 9 translators (one provider implementation per API), and you can serve all 27 clients.
You maintain a "client → language" lookup table (model registry) and a "language → translator" lookup table (api registry).

通用 vs Pi 特有Universal vs Pi-Specific

机制Mechanism 通用 / 特有Universal / Specific 说明Notes
Provider ≠ API 的两层映射Two-layer mapping: Provider ≠ API 通用模式Universal 任何需要对接多个 LLM 的系统都会面对这个问题。LangChain、LiteLLM 等也做了类似设计。Any system connecting to multiple LLMs faces this problem. LangChain, LiteLLM, etc. use similar designs.
统一的 StreamFunction 合同Unified StreamFunction contract 通用模式Universal 输入 (model, context) → 输出事件流。合同的具体字段可以不同，但"统一合同"这个思路是通用的。Input (model, context) → output event stream. The specific fields may differ, but the "unified contract" approach is universal.
注册表 + 查找Registry + lookup 通用模式Universal 经典的 Registry Pattern（注册表模式），在很多领域都用。Classic Registry Pattern, used across many domains.
懒加载 + ||= 单例Lazy loading + ||= singleton Pi 优化Pi Optimization 不是必须的，但对启动速度有帮助。如果只支持 2-3 个 provider，直接全部加载也行。Not required, but helps startup speed. If you only support 2-3 providers, loading all at once is fine.
27 个 Provider 归纳为 9 个 API27 Providers consolidated into 9 APIs Pi 具体数据Pi Specific Data 具体数量是 Pi 的选择。小型项目可能只需要 2-3 个 provider。The specific numbers are Pi's choice. Small projects may only need 2-3 providers.
stream vs streamSimple 双版本stream vs streamSimple dual version Pi 特有Pi-Specific 为了同时满足"引擎层用通用选项"和"高级用户用 provider 特有选项"的需求。简单系统可以只要一个版本。To serve both "engine uses generic options" and "advanced users use provider-specific options." Simple systems can have just one version.
models.generated.ts 自动生成models.generated.ts auto-generated Pi 工程实践Pi Engineering 27 个 provider 的模型数据量大，手动维护容易出错，所以用代码生成。少量 provider 可以手写。With 27 providers, model data is massive and error-prone to maintain manually, hence code generation. Few providers can be hand-written.

思考题 + 答案Questions + Answers

Q1：streamFn 回调是在哪里被设定的？最终走到了 pi-ai 的哪个函数？

A：在管理层 agent.ts 里设定（this.streamFn = options.streamFn ?? streamSimple）。默认值就是 pi-ai 的 streamSimple 函数。管理层把它和 model 一起传给引擎，引擎每轮循环调用它，最终经过路由（stream.ts）走到具体的 provider 文件。 Q1: Where is the streamFn callback set up? Which pi-ai function does it ultimately reach?

A: It's set in the management layer agent.ts (this.streamFn = options.streamFn ?? streamSimple). The default is pi-ai's streamSimple function. The management layer passes it along with model to the engine, the engine calls it each loop, and it ultimately reaches the specific provider file through the router (stream.ts).
Q2：新增一个兼容 OpenAI 协议的 provider（如 Moonshot AI），需要改哪些文件？

A：只需要在模型注册表（models.generated.ts）里加一条记录，写明这家公司的名字、API 地址、价格、以及 api: "openai-completions"。API 注册表里已经有 "openai-completions" 的记录了，路由会自动走到已有的 OpenAI provider 文件。不需要写任何新代码 —— 只加一条数据。 Q2: To add a new provider compatible with the OpenAI protocol (e.g. Moonshot AI), which files need changing?

A: Only add one entry to the model registry (models.generated.ts), specifying the company name, API URL, pricing, and api: "openai-completions". The API registry already has a "openai-completions" entry, so the router will automatically route to the existing OpenAI provider file. No new code needed — just one data entry.
Q3：pi-ai 和 agent-core 是什么关系？

A：agent-core（引擎）不直接依赖 pi-ai。它们通过 streamFn 回调连接 —— 管理层（agent.ts）把 pi-ai 的函数作为回调注入给引擎。引擎只知道"我有一个函数能帮我调 LLM"，不知道这个函数来自 pi-ai。这就是为什么同一个引擎可以接任何 LLM 调用层，不被绑死。 Q3: What is the relationship between pi-ai and agent-core?

A: agent-core (the engine) does not directly depend on pi-ai. They connect through the streamFn callback — the management layer (agent.ts) injects pi-ai's function as a callback into the engine. The engine only knows "I have a function that can call the LLM for me," not that it comes from pi-ai. This is why the same engine can plug into any LLM call layer without being locked in.