ARTICLE · 1192141

02. pi 源码解读:packages/agent 概览

发布时间：2026-10-16 11:27:24 最近访问：2026-10-16 11:27:26

02. pi 源码解读：packages/agent 概览

文档内容基于 earendil-works/pi 的 main 分支 9fba660cf 提交进行说明

runLoop

packages/agent/src/agent-loop.ts 提供了 runAgentLoop 和 runAgentLoopContinue 两个方法，二者的主体实现为 runLoop:

async function runLoop(  // 包含了历史消息和可用的工具集合  initialContext: AgentContext,  // 进入本次 Loop 的新消息  // runAgentLoop 是通过新输入触发的 Loop 流程，该参数会传值  // runAgentLoopContinue 用于继续未完成的会话，只需要使用 AgentContext 中的历史消息即可  newMessages: AgentMessage[],  // Agent Loop 配置参数，定义了很多的扩展点  initialConfig: AgentLoopConfig,  signal: AbortSignal | undefined,  // AgentEvent 消费者  emit: AgentEventSink,  // LLM 调用逻辑封装  streamFunction: StreamFn,): Promise<void> {  // 略}

AgentEvent: Agent Loop 中的事件

其定义在 packages/agent/src/types.ts 中。

/** * Events emitted by the Agent for UI updates. * * `agent_end` is the last event emitted for a run, but awaited `Agent.subscribe()` * listeners for that event are still part of run settlement. The agent becomes * idle only after those listeners finish. */export type AgentEvent =  // Agent lifecycle  | { type: "agent_start" }  | { type: "agent_end"; messages: AgentMessage[] }  // Turn lifecycle - a turn is one assistant response + any tool calls/results  | { type: "turn_start" }  | { type: "turn_end"; message: AgentMessage; toolResults: ToolResultMessage[] }  // Message lifecycle - emitted for system, user, assistant, and toolResult messages  | { type: "message_start"; message: AgentMessage }  // Only emitted for assistant messages during streaming  | { type: "message_update"; message: AgentMessage; assistantMessageEvent: AssistantMessageEvent }  | { type: "message_end"; message: AgentMessage }  // Tool execution lifecycle  | { type: "tool_execution_start"; toolCallId: string; toolName: string; args: any }  | { type: "tool_execution_update"; toolCallId: string; toolName: string; args: any; partialResult: any }  | { type: "tool_execution_end"; toolCallId: string; toolName: string; result: any; isError: boolean };

事件产生的基本流程（以 runAgentLoop 为例）如下图所示：

agent-event

• 每个 Agent Loop 构成一个 agent_start/agent_end 对
• Agent Loop 流程中每一轮（turn）构成一个 turn_start/turn_end 对
• 经用户输入转换的提示词、模型返回的内容、工具调用结果的封装都会触发一个 message_start/message_end 对，模型返回的内容还可能额外触发 message_update
• 每一个工具调用必然构成一个 tool_execution_start/tool_execution_end 对，部分工具调用还可能涉及 tool_execution_update

工具调用

工具调用 Hook

Agent Loop 流程中需要执行工具调用，pi 为此提供了 Hook 回调。详见 packages/agent/src/types.ts 中定义的 AgentLoopConfig:

• beforeToolCall: 工具执行前的回调，用于检查是否阻塞该工具的执行
• afterToolCall: 工具执行完成后的回调，返回值中的字段会替换原始结果

export interface AgentLoopConfig extends SimpleStreamOptions {  // 前略  /**   * Called before a tool is executed, after arguments have been validated.   *   * Return `{ block: true }` to prevent execution. The loop emits an error tool result instead.   * A blocked result can also set `terminate: true` to participate in the batch early-termination rule.   * The hook receives the agent abort signal and is responsible for honoring it.   */  beforeToolCall?: (context: BeforeToolCallContext, signal?: AbortSignal) => Promise<BeforeToolCallResult | undefined>;  /**   * Called after a tool finishes executing, before `tool_execution_end` and tool-result message events are emitted.   *   * Return an `AfterToolCallResult` to override parts of the executed tool result:   * - `content` replaces the full content array   * - `details` replaces the full details payload   * - `isError` replaces the error flag   * - `usage` replaces the tool result usage   * - `terminate` replaces the early-termination hint   *   * Any omitted fields keep their original values. No deep merge is performed.   * The hook receives the agent abort signal and is responsible for honoring it.   */  afterToolCall?: (context: AfterToolCallContext, signal?: AbortSignal) => Promise<AfterToolCallResult | undefined>;}

动态工具集合

pi 在 Agent Loop 流程中发送给模型的工具集合不是固定不变的，每次模型调用前都需要与当前可用的工具集合对齐。

pi 的动态工具集合依赖 AgentContext 和 SystemMessage 两个接口，定义如下:

// packages/agent/src/types.ts/** Context snapshot passed into the low-level agent loop. */export interface AgentContext {  /** Transcript visible to the model. */  // 已经发送给模型的消息  messages: AgentMessage[];  /** Tools available for execution in this run. */  // 当前状态下可用的工具集合，这个值会发生变化，比如  //   - 全量工具太耗 token，用 tool_search 来临时添加  //   - MCP Server就绪与否会导致可用的工具集合发生变化  //   - Plan Mode、Agent Mode等不同模式的预定义工具不一样，切换模式会导致工具变化  tools?: AgentTool<any>[];}

// packages/ai/src/types.ts/** * System instructions and tool declarations at one point in the transcript. * * The leading system message is the system prompt. Later system messages change it: * `content` adds instructions from that point on, `sections` replace or remove named * prompt sections, and `toolsAdded`/`toolsRemoved` change the tool set. Replaying * every system message in order yields the current prompt and tools. Providers that * accept system messages mid-conversation send each one in place; other providers * rebuild the leading system message from the replayed state. */export interface SystemMessage {  role: "system";  /** Instruction text. On the leading message this is the base prompt; later, additional instructions. */  content: string | TextContent[];  /**   * Named, ordered prompt sections rendered verbatim after `content`. The leading message   * declares them; later messages replace sections by name, and `null` removes one. Keep   * each section self-delimiting (a tag, a heading) so the model can relate an update to   * the original. Avoid integer-like names; JSON objects reorder those.   */  sections?: Record<string, string | null>;  /** Complete definitions of tools that become available at this point. */  // 有哪些工具需要添加  toolsAdded?: Tool[];  /** Tools that stop being available at this point. */  // 有哪些工具需要移除  toolsRemoved?: ToolReference[];  timestamp: number; // Unix timestamp in milliseconds}

每次执行模型调用时，传入的工具集合必须与当前状态下可用的工具集合（即 AgentContext.tools）精确对齐。一个直观的设计是将 AgentContext.tools 转换为 JSON Schema 后直接传入 LLM 接口的顶层 tools 参数即可，但是 pi 设计了工具的 diff，通过 SystemMessage 做了持久化。有如下原因:

• 一般情况下，会将固定不变的参数放在提示词最前面，以达到最高的缓存命中率。而 tool JSON Schema 就是这么一个参数。当工具发生变化，如果直接修改之前的提示词，则会导致缓存失效。有模型支持在消息流中原位追加工具变更来规避此现象。pi 通过 SystemMessage.toolsAdded、SystemMessage.toolsRemoved 统一记录工具变更，作为与模型无关的事实源；针对缓存友好的模型，由下游 adapter 将这份日志转换为增量格式（如 Anthropic 的原生 tool_addition/tool_removal 块）。
• AgentContext.tools 是一个临时变量，而 SystemMessage 作为对话消息会被持久化，将工具变化记录在其中即可一并持久化。
• 有一个统一的 SystemMessage 记录工具的变化，下游只需要对接 SystemMessage 即可。

在 pi 的设计中，工具集合的变化流程如下：

当存在未发送给模型的新消息 pendingMessages 时，每次请求前需对账工具集合是否发生变化。

• 倒序搜索 pendingMessages 最后一条 SystemMessage 消息，该消息将存储工具的新增/移除记录；若不存在，则新建一条 SystemMessage 插入到第一个非 system 消息之前。
• 将历史消息 AgentContext.messages 与整个 pendingMessages 合并重放，计算出一个工具集合。重放时仅目标 SystemMessage 的工具字段被清零，其余尚未发送的 SystemMessage 的工具字段原样保留参与计算。
• 再与当前可用的工具集合 AgentContext.tools 做对比（同名重定义视为移除+新增），计算出工具变更情况。
• 将变更情况写入目标 SystemMessage 的 toolsRemoved/toolsAdded 字段（覆盖其原有工具字段）。
• 下游将消息发送给模型时，根据模型特性将这份日志转换为三种格式之一：

• 重放出全量工具集合，传入模型接口的顶层 tools 参数
• Anthropic 原生 tool_addition/tool_removal 块（不破坏前缀缓存）
• 原位透传 SystemMessage（支持对话中途 system message 的模型）

扩展点

pi 在 Agent Loop 流程实现中预定了很多的扩展点（如上述工具调用 Hook），具体可见下图（AgentEvent 只保留了 agent_start/agent_end 和 turn_start/turn_end）：

agent-loop-hooks.png

Agent 类

Agent 类（详见 packages/agent/src/agent.ts）在 runAgentLoop 和 runAgentLoopContinue 基础上补齐需要维护状态的部分，如：

• 单次运行的互斥、中止与等待（activeRun）
• 持有当前的历史消息（state.messages）和可执行的工具集合（state.tools）
• 事件分发
• steering 和 follow-up 消息队列

Agent 类对外暴露的 prompt 和 continue 方法分别对应 runAgentLoop 和 runAgentLoopContinue。

activeRun

Agent 使用了 activeRun?: ActiveRun 字段来承载互斥、完成通知、取消句柄三个功能。

type ActiveRun = {  promise: Promise<void>;  resolve: () => void;  abortController: AbortController;};

Agent 的 prompt 和 continue 方法都是包裹在 runWithLifecycle 中执行的，在 runWithLifecycle 代码中大致可以观察到上述三个功能的实现：

private async runWithLifecycle(executor: (signal: AbortSignal) => Promise<void>): Promise<void> {  if (this.activeRun) {    // 初始状态下 activeRun = undefined，此逻辑限制 Agent 同一时刻只能存在一个 run    throw new Error("Agent is already processing.");  }  // 此对象还会传入 Agent Loop 流程以控制取消  const abortController = new AbortController();  let resolvePromise = () => {};  const promise = new Promise<void>((resolve) => {    resolvePromise = resolve;  });  // 设置 activeRun，activeRun 不为 undefined 时说明 Agent 实例即将进入 Agent Loop 流程  this.activeRun = { promise, resolve: resolvePromise, abortController };  // 其它的状态设置  this._state.isStreaming = true;  this._state.streamingMessage = undefined;  this._state.errorMessage = undefined;  try {    // Agent Loop 流程    await executor(abortController.signal);  } catch (error) {    await this.handleRunFailure(error, abortController.signal.aborted);  } finally {    // 会执行 this.activeRun?.resolve(); this.activeRun = undefined;     // activeRun = undefined 说明 Agent 实例已经完成了 Agent Loop 流程    // activeRun.resolve() 执行后，await activeRun.promise 不会阻塞    // 外部如果获取到了 activeRun.promise，则可以凭此判断 Agent Loop 逻辑已经完成    this.finishRun();  }}

AgentEvent 事件处理

processEvents 方法中包含了对 AgentEvent 事件的处理逻辑。简单概括一下，就是将事件流转换为内部状态 _state，然后调用监听器 listeners 继续处理事件。

/** * Reduce internal state for a loop event, then await listeners. * * `agent_end` only means no further loop events will be emitted. The run is * considered idle later, after all awaited listeners for `agent_end` finish * and `finishRun()` clears runtime-owned state. */private async processEvents(event: AgentEvent): Promise<void> {  switch (event.type) {    case "message_start":      this._state.streamingMessage = event.message;      break;    case "message_update":      this._state.streamingMessage = event.message;      break;    case "message_end":      this._state.streamingMessage = undefined;      this._state.messages.push(event.message);      break;    case "tool_execution_start": {      const pendingToolCalls = new Set(this._state.pendingToolCalls);      pendingToolCalls.add(event.toolCallId);      this._state.pendingToolCalls = pendingToolCalls;      break;    }    case "tool_execution_end": {      const pendingToolCalls = new Set(this._state.pendingToolCalls);      pendingToolCalls.delete(event.toolCallId);      this._state.pendingToolCalls = pendingToolCalls;      break;    }    case "turn_end":      if (event.message.role === "assistant" && event.message.errorMessage) {        this._state.errorMessage = event.message.errorMessage;      }      break;    case "agent_end":      this._state.streamingMessage = undefined;      break;  }  const signal = this.activeRun?.abortController.signal;  if (!signal) {    throw new Error("Agent listener invoked outside active run");  }  for (const listener of this.listeners) {    await listener(event, signal);  }}

• _state 使用 streamingMessage 记录正在接收的消息：事件为 message_start / message_update 时更新为最新内容，事件为 message_end / agent_end 时重置为 undefined。
• _state 使用 messages 记录所有的历史消息，当事件为 message_end 时将完整消息追加进来。
• _state 使用 pendingToolCalls 记录执行中的工具调用集合，当事件为 tool_execution_start / tool_execution_end 时做对应增删操作。

steeringQueue 和 followUpQueue

steeringQueue 和 followUpQueue 是两个用于暂存消息的队列。Agent 限制了同一时刻只能存在一个进入 Agent Loop 流程的 run，如果当前 run 处理时间太长，新消息不能等到它完成后再提交，就需要借助这两个队列。

steering（转向）消息用于在 agent 工作过程中调整方向，follow-up（后续）消息则等待 agent 完成当前工作后再处理。二者的写入入口是 Agent 暴露的 steer 和 followUp 方法；若 run 进行期间调用 prompt，Agent 会直接抛出错误，并提示改用这两个方法排队消息。

两个队列的消费时机由扩展点 getSteeringMessages 和 getFollowUpMessages 暴露给 Agent Loop（Agent 在构造 AgentLoopConfig 时，将队列的 drain() 封装为这两个回调，注入位置可参见前述扩展点一节的示意图）：

• getSteeringMessages：steering 消息在当前 turn 的工具调用全部执行完毕后、下一次 LLM 调用前注入上下文，不会打断正在执行的工具，run 也不会中断。Agent Loop 会在三个时机轮询该扩展点：run 开始时、prepareNextTurn（可能耗时较长的准备工作，如上下文压缩）执行完毕后、每个 turn 结束后。
• getFollowUpMessages：follow-up 消息仅在 Agent Loop 即将停止时（没有更多工具调用、也没有 steering 消息）才被消费；一旦取到消息，外层循环会将其作为新的 pendingMessages 再执行一轮，从而延长当前 run。若 run 已经结束，后续调用 continue 时也会先排空这两个队列，作为新一轮 Agent Loop 流程的输入。

此外，两个队列都支持 all 与 one-at-a-time（默认）两种消费模式：前者在每个消费点注入全部排队消息，后者只注入最旧的一条，其余留待后续消费点。

代码示例

import { readFileSync } from "node:fs";import { createModels, Type } from "@earendil-works/pi-ai";import { zaiCodingCnProvider } from "@earendil-works/pi-ai/providers/zai-coding-cn";import { runAgentLoop } from "@earendil-works/pi-agent-core";import type {  AgentContext,  AgentEvent,  AgentLoopConfig,  AgentMessage,  AgentTool,  Message,} from "@earendil-works/pi-agent-core";// 使用智谱coding planconst models = createModels();models.setProvider(zaiCodingCnProvider());const model = models.getModel("zai-coding-cn", "glm-5.3-flash");// 加个工具方便查看效果const weatherTool: AgentTool = {  name: "get_weather",  label: "查询天气",  description: "获取某个地点的当前天气",  parameters: Type.Object({    location: Type.String({ description: "城市名称" }),  }),  execute: async (_toolCallId, params) => {    return {      content: [{ type: "text", text: `${params.location}：晴，26°C，东南风 2 级，湿度 45%` }],    };  },};// 准备 AgentContext，包含初始消息和工具列表const context: AgentContext = {  messages: [{ role: "system", content: "你是一个有用的助手，会用中文回答用户问题。", timestamp: Date.now() }],  tools: [weatherTool],};// 准备 AgentLoopConfig，这里只加上 model 和 convertToLlm,其他使用默认值const isLlmMessage = (m: AgentMessage): m is Message =>  m.role === "system" || m.role === "user" || m.role === "assistant" || m.role === "toolResult";const config: AgentLoopConfig = {  model,  // 每次调用 LLM 前,把 AgentMessage[] 转换为 provider 兼容的 Message[]。  convertToLlm: (messages) => messages.filter(isLlmMessage),};function preview(value: unknown): string {  const json = JSON.stringify(value) ?? "undefined";  return json.length > 120 ? `${json.slice(0, 120)}…` : json;}function stopReasonOf(message: AgentMessage): string {  return message.role === "assistant" ? message.stopReason : "unknown";}// AgentEvent 消费者function printEvent(event: AgentEvent): void {  switch (event.type) {    case "agent_start":      console.log("[agent_start]");      break;    case "agent_end":      console.log(`[agent_end] ${event.messages.length} messages: ${event.messages.map((m) => m.role).join(" -> ")}`);      break;    case "turn_start":      console.log("[turn_start]");      break;    case "turn_end":      console.log(        `[turn_end] stopReason=${stopReasonOf(event.message)} toolResults=${event.toolResults.length} (${event.toolResults.map((r) => r.toolName).join(", ")})`,      );      break;    case "message_start":      console.log(`[message_start] role=${event.message.role}`);      break;    case "message_end":      console.log(        `[message_end] role=${event.message.role} stopReason=${stopReasonOf(event.message)}` +          (event.message.role === "assistant"            ? ` tokens(in/out)=${event.message.usage.input}/${event.message.usage.output}`            : ""),      );      break;    case "message_update": {      const e = event.assistantMessageEvent;      switch (e.type) {        case "text_start":          console.log("[message_update][text_start]");          break;        case "text_delta":          // delta 追加到原内容后面，不换行不加前缀          process.stdout.write(e.delta);          break;        case "text_end":          console.log("[message_update][text_end]");          break;        case "thinking_start":          console.log("[message_update][thinking_start]");          break;        case "thinking_delta":          // delta 追加到原内容后面，不换行不加前缀          process.stdout.write(e.delta);          break;        case "thinking_end":          console.log("[message_update][thinking_end]");          break;        case "toolcall_start":          console.log(`[message_update][toolcall_start] index=${e.contentIndex}`);          break;        case "toolcall_delta":          // delta 追加到原内容后面，不换行不加前缀          process.stdout.write(e.delta);          break;        case "toolcall_end":          console.log(`[message_update][toolcall_end] ${e.toolCall.name}(${preview(e.toolCall.arguments)})`);          break;        case "done":          console.log(`[message_update][done] reason=${e.reason}`);          break;        case "error":          console.log(`[message_update][error] reason=${e.reason} errorMessage=${e.error.errorMessage}`);          break;      }      break;    }    case "tool_execution_start":      console.log(`[tool_execution_start] ${event.toolName}(${preview(event.args)}) toolCallId=${event.toolCallId}`);      break;    case "tool_execution_update":      console.log(`[tool_execution_update] ${event.toolName} partialResult=${preview(event.partialResult)}`);      break;    case "tool_execution_end":      console.log(        `[tool_execution_end] ${event.toolName} isError=${event.isError} result=${preview(event.result)} toolCallId=${event.toolCallId}`,      );      break;  }}console.log("=== runAgentLoop ===");const auth = JSON.parse(readFileSync(process.env.USERPROFILE + "/.pi/agent/auth.json", "utf8"));const apiKey: string = auth["zai-coding-cn"].key;const newMessages = await runAgentLoop(  [{ role: "user", content: "上海和北京现在天气怎么样？", timestamp: Date.now() }],  context,  config,  printEvent,  new AbortController().signal,  (model, ctx, options) => models.streamSimple(model, ctx, { ...options, apiKey }),);console.log("=== output ===");for (const message of newMessages) {  if (typeof message.content === "string") {    console.log(`[${message.role}] ${message.content}`);  } else {    for (const block of message.content) {      if (block.type === "text") {        console.log(`[${message.role}] ${block.text}`);      } else if (block.type === "thinking") {        console.log(`[${message.role}] [thinking] ${block.thinking}`);      } else if (block.type === "toolCall") {        console.log(`[${message.role}] [toolCall] ${block.name}(${JSON.stringify(block.arguments)})`);      } else {        console.log(`[${message.role}] [${block.type}] ${preview(block)}`);      }    }  }}

上述代码执行输入如下图所示:

example-output