学习和整理 Pi 的 LLM 模块设计

2026-07-06 updated 2026-07-30 13 min

PI 这个项目中，对于 LLM 模块的设计，尤其是多供应商的设计是比较出色的，还有它的模型配置文件的字段设计和对于上下文的理解都很不错，在设计层面是考虑到上下文的重要性的。

和我文章最后提到的开源项目：「大模型应用开发 - 上下文工程与运行空间实践指南」中定位是一样的，Context 在目前的大模型应用工程中依旧很重要，并且在优化的阶段，是 harness 工程调整的原则之一

Pi 非常值得我们花时间去整理和学习，这也是 kimi-code 和 openclaw 的构建基础框架之一

一、LLM 模块的总设计

PI 的编码 Agent 是支持多种供应商和模型的，并且在会话中途是可以更换模型和供应商的，如何协调不同的协议和消息格式是 PI 的核心模块 pi-ai 的设计初心，我们可以一起来从全局的角度来理解

PI 定义自己内部通用的消息类型，用户发送的任何消息都会先转换为这种通用的格式，作为消息流通的核心
用户在输入的时候，会传递供应商和模型进来，根据这个我们就可以得到用户想要调用的具体的协议，那么通用定义就会先经过转换为相关协议的类型，然后调用模型，模型回复也会被转换为通用的类型，下一次用户输入的时候，切换了模型，依旧进行目标协议的转换

重点理解就是：无论怎么做，无论有多少模型和供应商，无论有多少种 api 协议，只要关注自己内部定义好的通用格式就可以啦，到真正使用的时候，进行临时转换

接下来我们一起看看这个内部通用定义的东西是什么？长什么样子？

二、LLM 模块内部通用定义

在一个成熟的 Agent 的 Loop 中，消息会有不同的角色，角色可以让消息列表有流程循环的状态，常见的消息角色：user、assistant、toolResult，下面是一个完整的 Context 对象的例子

const context: Context = {
    systemPrompt: "你是...",
    tools: [],
    messages: [
      { role: "user", content: "北京今天天气怎么样？"},
      {
        role: "assistant",
        content: [
          { type: "thinking", thinking: "...", thinkingSignature:"eyJhbGciOi..." },
          { type: "text", text: "让我帮你查一下。" },
          { type: "toolCall", id: "call_01ABC", name:"get_weather", arguments: { city: "北京" } },
        ],
      },
      {
        role: "toolResult", toolCallId: "call_01ABC", toolName:"get_weather",
        content: [{ type: "text", text: "晴，25°C，东北风 3 级"}],
      },
      {
        role: "assistant",
        content: [{ type: "text", text: "北京今天晴朗，25°C，适合外出！"}],
      },
      {
        role: "user",
        content: [
          { type: "text", text: "这张图里的数学题帮我算一下：" },
          { type: "image", data: "iVBORw0KGgo...", mimeType:"image/png" },
        ],
      },
    ],
  };

内部流通的是 Context，并且你可以向 Context 中增加一些数据状态，文章《停止将聊天历史用作智能体的状态存储》中有详细的介绍，感兴趣的开发者可以去仔细阅读一下，我感觉重点一句是：

你的应用拥有结构化状态：当前用户、选中项目、流程位置、数据库数据。而 LLM 只有一维消息数组。两者持续产生偏差，最终只能由你来调和。

所以内部使用 Context 对象来作为通用消息格式，并且可以增加一些字段未来可以用来表示状态和承担控制流的职责，

当要将上下文注入给相应的模型的时候，这个时候可以进行临时的格式转换以适配对应的模型 api 协议，这样未来增加任何供应商和模型的时候，只要增加对应的转换函数就可以，其他的都不用改动，所以 Context 是所有消息的源头

三、消息的数据流动

接下来我们一起来看看 Context 是如何转换为对应模型的消息格式的，设计层面非常有意思，很值得学习

🎃 通用消息处理函数：一般这里会编写异常工具消息的处理和错误/中止消息的过滤，让即将传入给模型的消息看起来健康一些

NOTE
异常工具消息表示的是：在 message 变量中，工具的调用时需要成双成对的，不能单一出现，不然调用会报错的

这里可以理解成为复用函数，共同逻辑编写的地方

🎃 消息转换函数：这个就非常好理解啦，就是将 Context 里面的消息转换为对应的供应商的格式，具体的格式大家可以去看官方文档

四、具体的 LLM 协议类的实现

具体的 API 协议类的实现，例如：anthropic 的实现，openai-completions 的实现，google 的实现，核心就是五个方法，其他的都是辅助函数

参数检测：根据传入的模型和供应商自动检测哪些参数是被支持，是可以使用的
工具定义转换：将 tools 转换为目标 LLM 协议支持的格式
消息格式转换：将 Context 中的通用格式转换为目标的 LLM 协议支持的格式
parseChunkUsage：从模型输出的结果中解析出来对应的 Token，输入、缓存、输出，总计
streamXXXCompletions：核心执行方法，里面会执行上面四个封装好对应逻辑的函数，同时调用相应的 SDK 完成调用任务

五、核心工具类

这个工具类 EventStream 写的非常好，我想完整的把代码展示出来：

export class EventStream<T, R = T> implements AsyncIterable<T> {
    private queue: T[] = [];
    private waiting: ((value: IteratorResult<T>) => void)[] = [];
    private done = false;
    private finalResultPromise: Promise<R>;
    private resolveFinalResult!: (result: R) => void;

    constructor(
        private isComplete: (event: T) => boolean,
        private extractResult: (event: T) => R,
    ) {
        this.finalResultPromise = new Promise((resolve) => {
            this.resolveFinalResult = resolve;
        });
    }

    push(event: T): void {
        if (this.done) return;
        if (this.isComplete(event)) {
            this.done = true;
            this.resolveFinalResult(this.extractResult(event));
        }
        const waiter = this.waiting.shift();
        if (waiter) {
            waiter({ value: event, done: false });
        } else {
            this.queue.push(event);
        }
    }

    end(result?: R): void {
        this.done = true;
        if (result !== undefined) {
            this.resolveFinalResult(result);
        }
        while (this.waiting.length > 0) {
            const waiter = this.waiting.shift()!;
            waiter({ value: undefined as any, done: true });
        }
    }

    async *[Symbol.asyncIterator](): AsyncIterator<T> {
        while (true) {
            if (this.queue.length > 0) {
                yield this.queue.shift()!;
            } else if (this.done) {
                return;
            } else {
                const result = await new Promise<IteratorResult<T>>(
                  (resolve) => this.waiting.push(resolve)
                );
                if (result.done) return;
                yield result.value;
            }
        }
    }

    result(): Promise<R> {
        return this.finalResultPromise;
    }
}

queue 和 waiting 的设计：一个是从生产端角度设计的，如果消费端没有准备好，那么使用 queue 队列保存住，一个是从消费端角度设计的，如果生产端在推送数据的时候，消费端已经准备好，那么直接推送给消费端
result 方法和 Generator 函数：一个是异步生成器，消费端可以 await 的方式流式取数据，一个是堵塞住，等待模型输出完成之后，返回给对应的变量

六、事件流和调用链路

6.1、Agent 执行事件流

一次完整的 Agent 事件流，里面会存在 Agent 生命周期，turn 执行，消息输入，模型执行，工具执行，终止和错误这些状态存在，梳理清楚这些状态会非常有益于我们做执行日志记录和前端的执行显示

🌴在理解 Agent 事件流中，一定要记住**：一次 Agent 交互，会存在多轮模型的调用（也就是多个 turn），每一轮模型的调用都有模型的回复，模型的回复中如果有工具调用执行，那么这一轮还会有工具执行的流程存在**

6.2、Agent 执行链路

在 LLM 模块的总设计思路下，Agent 执行调用链路的完整情况如下，从用户输入到要触发的 LLM 模块的核心地方

总结下来就是两点：

根据模型定义好的 api 变量的值，来选择要实例化哪一个 LLM 模块协议类
根据选择好的 LLM 模块协议类，进行参数的转换和消息格式的转换，让模型调用可以成功执行

Agent LLM 架构设计