PI Agent 核心架构解析

整个系统的核心其实就三层，每层只做一件事。
本篇涵盖了 90% 的设计意图剖析，剩下的 10% 是细节。
本文聚焦于项目：PI agent 的核心架构进行阅读学习。

一、为什么要分三层？

先说结论：这三层不是拍脑袋分的，是为了让每一层都能独立替换。

想象一下：

明天要支持一个新的 LLM 提供商？只改第一层。
想换一套工具（比如从文件操作换成数据库操作）？只改第三层。
想换一个 Agent 编排策略？只改第二层。

这就是分层的好处。每一层都是"可插拔"的。

层	包名	一句话职责
底层	`packages/ai`	跟 LLM 说话
中层	`packages/agent`	编排工具调用循环
上层	`packages/coding-agent`	具体工具实现 + 业务逻辑

二、第一层：`pi-ai` — LLM 通信层

2.1 这一层解决什么问题？

不同的 LLM 提供商（Anthropic、OpenAI、Google）API 长得不一样：

消息格式不同
流式响应格式不同
工具调用格式不同

这一层把它们统一成一套接口，让上层不用关心"对面是 Claude 还是 GPT"。

2.2 核心类型

Message — LLM 能理解的消息

// packages/ai/src/types.ts
type Message = UserMessage | AssistantMessage | ToolResultMessage;

注意：只有三种角色。这是 LLM 世界的"通用语言"。

Context — 发给 LLM 的完整上下文

// packages/ai/src/types.ts
interface Context {
  systemPrompt: string;      // 系统提示词
  messages: Message[];       // 对话历史
  tools: Tool[];             // 可用工具定义
}

Tool — 工具的 JSON Schema 定义

// packages/ai/src/types.ts
interface Tool<TParameters> {
  name: string;
  description: string;
  parameters: TSchema;       // TypeBox schema
}

这只是"告诉 LLM 有哪些工具可用"，不包含执行逻辑。执行在第二层。

Model — 模型元数据

// packages/ai/src/types.ts
interface Model<TApi> {
  id: string;
  name: string;
  provider: string;          // "anthropic" | "openai" | "google" ...
  api: TApi;
  reasoning: boolean;        // 是否支持 thinking/reasoning
  input: string[];           // ["text", "image"]
  cost: { input: number; output: number; ... };
  contextWindow: number;
  maxTokens: number;
}

2.3 亮点设计：`StreamFn` 类型

// packages/agent/src/types.ts:28
type StreamFn = (
  model: Model<Api>,
  context: Context,
  options?: SimpleStreamOptions,
) => AssistantMessageEventStream | Promise<AssistantMessageEventStream>;

为什么要这样设计？

这是一个"工厂函数"类型。第一层提供 Models.streamSimple() 实现这个类型，第二层调用它。

好处：

解耦：第二层不直接依赖具体的 LLM 调用代码
可测试：可以传一个 mock 的 StreamFn 进去
可替换：想换 LLM 实现？换一个 StreamFn 就行

三、第二层：`pi-agent-core` — Agent 编排层

3.1 这一层解决什么问题？

第一层只能"调一次 LLM"。但 Agent 需要：

调 LLM
拿到 toolCall
执行工具
把结果塞回上下文
再调 LLM
... 直到没有 toolCall

这就是循环。第二层就是干这个的。

3.2 核心类型：`AgentMessage` vs `Message`

这是整个设计里最精妙的地方之一。

// packages/agent/src/types.ts:319
type AgentMessage = Message | CustomAgentMessages[keyof CustomAgentMessages];

为什么要区分？

因为现实应用中，对话历史里不只有 user/assistant/toolResult。你可能有：

UI 通知消息（"正在搜索..."）
状态消息（"工具执行中"）
自定义消息（"代码审查结果"）

这些消息LLM 不需要看到，但UI 需要展示。

所以：

AgentMessage = 所有消息（LLM 能看的 + 只能 UI 看的）
Message = LLM 能看的消息

转换管道：

// packages/agent/src/agent.ts:32
function defaultConvertToLlm(messages: AgentMessage[]): Message[] {
  return messages.filter(
    (message) => message.role === "user" || message.role === "assistant" || message.role === "toolResult",
  );
}

默认实现很简单：只保留 LLM 能理解的三种角色。

但你可以自定义：

// packages/agent/src/types.ts:173
convertToLlm: (messages: AgentMessage[]) => Message[] | Promise<Message[]>;

比如把"代码审查结果"转成 user 消息发给 LLM。

好处：

关注点分离：UI 和 LLM 看到的东西不一样
灵活性：可以动态控制发给 LLM 的内容
向后兼容：老代码不用改，新消息类型自动被过滤

3.3 双重循环 — 核心编排逻辑

源码位置：packages/agent/src/agent-loop.ts:155 的 runLoop() 函数。

这是整个系统的心脏，理解了这个就理解了 Agent 的工作原理。

外层循环 (while true):
  内层循环 (while hasMoreToolCalls || pendingMessages):
    1. 注入 steering 消息（如果有）
    2. 调用 LLM → 流式获取 assistant 响应
    3. 如果响应包含 toolCall → 执行工具 → 把结果加入上下文 → 继续内层循环
    4. 如果没有 toolCall → 退出内层循环
  检查 follow-up 队列：
    有消息 → 设为 pending → 继续外层循环
    无消息 → break，结束

为什么要双重循环？

因为两种消息的语义完全不同：

类型	语义	检查时机	效果
Steering	"别做了，改做这个"	每个工具执行后	注入新消息，打断当前方向
Follow-up	"做完了再做这个"	Agent 即将停止时	追加消息，继续循环

Steering 的例子：

用户输入："帮我写一个排序函数"
Agent 开始写...
用户中途输入："等等，用 TypeScript 写"
→ 这条消息是 steering，会立即注入，打断当前方向

Follow-up 的例子：

用户输入："帮我写一个排序函数，然后写个测试"
Agent 写完排序函数...
→ "写个测试" 是 follow-up，等排序完成后才执行

好处：

精确控制：不同场景用不同机制
避免混乱：steering 不会在错误时机被处理
符合直觉：用户预期"打断"和"追加"是两回事

3.4 关键函数：`streamAssistantResponse()`

源码位置：packages/agent/src/agent-loop.ts:281

这是 AgentMessage → Message 转换发生的地方：

async function streamAssistantResponse(...) {
  // 1. 可选：裁剪/注入上下文
  let messages = context.messages;
  if (config.transformContext) {
    messages = await config.transformContext(messages, signal);
  }

  // 2. AgentMessage[] → Message[]（只保留 LLM 能理解的）
  const llmMessages = await config.convertToLlm(messages);

  // 3. 构建 LLM 上下文
  const llmContext: Context = {
    systemPrompt: context.systemPrompt,
    messages: llmMessages,
    tools: context.tools,
  };

  // 4. 调用 LLM（流式）
  const response = await streamFunction(config.model, llmContext, options);

  // 5. 处理流式事件
  for await (const event of response) {
    switch (event.type) {
      case "start": ...
      case "text_delta": ...
      case "toolcall_delta": ...
      case "done": ...
    }
  }
}

亮点：transformContext 钩子

// packages/agent/src/types.ts:195
transformContext?: (messages: AgentMessage[], signal?: AbortSignal) => Promise<AgentMessage[]>;

在 convertToLlm 之前执行，用于：

上下文裁剪（token 太多时删掉旧消息）
注入外部上下文（比如从数据库查相关信息）

好处：

解耦：上下文管理和消息转换分开
灵活：可以动态控制上下文大小
安全：不会意外把敏感信息发给 LLM

3.5 工具执行的生命周期

源码位置：packages/agent/src/agent-loop.ts:411 的 executeToolCalls()

每个工具执行经过三个阶段：

1. prepare（准备）
   - 查找工具定义
   - 验证参数（TypeBox schema）
   - 调用 beforeToolCall 钩子（可以 block）

2. execute（执行）
   - 调用 tool.execute()
   - 支持流式更新（onUpdate 回调）

3. finalize（收尾）
   - 调用 afterToolCall 钩子（可以修改结果）

亮点设计：beforeToolCall 和 afterToolCall

// packages/agent/src/types.ts:271
beforeToolCall?: (context: BeforeToolCallContext, signal?: AbortSignal) 
  => Promise<BeforeToolCallResult | undefined>;

// packages/agent/src/types.ts:286
afterToolCall?: (context: AfterToolCallContext, signal?: AbortSignal) 
  => Promise<AfterToolCallResult | undefined>;

beforeToolCall 可以：

阻止工具执行（返回 { block: true }）
用于权限检查、安全审计

afterToolCall 可以：

修改工具结果（替换 content、details）
标记为错误（isError: true）
提示终止（terminate: true）

好处：

横切关注点：权限、日志、监控不用写在每个工具里
灵活性：可以动态控制工具行为
安全性：可以在执行前拦截危险操作

3.6 工具执行模式：并行 vs 串行

// packages/agent/src/types.ts:42
type ToolExecutionMode = "sequential" | "parallel";

默认是 parallel：

先顺序 prepare（验证参数）
然后并发执行所有工具
按完成顺序发射 tool_execution_end
按原始顺序发射 tool-result 消息

为什么默认并行？

因为大多数工具调用是独立的（比如同时读两个文件），并行可以显著减少等待时间。

但如果工具有副作用（比如写文件），可以设置为 sequential。

3.7 `Agent` 类 — 有状态的封装

源码位置：packages/agent/src/agent.ts:171

agentLoop 是"无状态"的函数，每次调用都要传完整的 context。

Agent 类是"有状态"的封装，拥有：

消息历史 messages[]
工具列表 tools[]
Steering 队列 / Follow-up 队列
事件订阅 subscribe()
状态管理（isStreaming、pendingToolCalls 等）

对外 API 很简洁：

// 开始新对话
await agent.prompt("帮我写一个排序函数");

// 运行中注入消息（打断当前方向）
agent.steer({ role: "user", content: "等等，用 TypeScript 写" });

// 完成后追加消息
agent.followUp({ role: "user", content: "再写个测试" });

// 中止
agent.abort();

// 订阅事件
agent.subscribe((event) => {
  if (event.type === "message_end") {
    console.log("收到消息:", event.message);
  }
});

亮点：事件驱动架构

// packages/agent/src/types.ts:422
type AgentEvent =
  | { type: "agent_start" }
  | { type: "agent_end"; messages: AgentMessage[] }
  | { type: "turn_start" }
  | { type: "turn_end"; message: AgentMessage; toolResults: ToolResultMessage[] }
  | { type: "message_start"; message: AgentMessage }
  | { type: "message_update"; message: AgentMessage; assistantMessageEvent: AssistantMessageEvent }
  | { type: "message_end"; message: AgentMessage }
  | { 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 };

所有状态变化都通过事件通知，UI 只需要订阅事件即可。

好处：

解耦：核心逻辑和 UI 完全分离
可测试：可以 mock 事件流
可扩展：新 UI（Web、CLI、TUI）只需要订阅相同的事件

四、第三层：`coding-agent` — 四个核心编码工具

源码位置：packages/coding-agent/src/core/tools/

createCodingTools() 返回四个核心工具：

// packages/coding-agent/src/core/tools/index.ts:168
export function createCodingTools(cwd: string, options?: ToolsOptions): Tool[] {
  return [
    createReadTool(cwd, options?.read),
    createBashTool(cwd, options?.bash),
    createEditTool(cwd, options?.edit),
    createWriteTool(cwd, options?.write),
  ];
}

4.1 `read` — 读文件

源码位置：packages/coding-agent/src/core/tools/read.ts

参数：

const readSchema = Type.Object({
  path: Type.String({ description: "Path to the file to read" }),
  offset: Type.Optional(Type.Number({ description: "Line number to start reading from (1-indexed)" })),
  limit: Type.Optional(Type.Number({ description: "Maximum number of lines to read" })),
});

亮点设计：

支持文本和图片
- 文本文件：返回文本内容
- 图片文件（jpg/png/gif/webp/bmp）：返回 base64 编码的图片

智能截断

// packages/coding-agent/src/core/tools/truncate.ts
const DEFAULT_MAX_LINES = 2000;
const DEFAULT_MAX_BYTES = 100 * 1024; // 100KB

先到先停。截断后会告诉 LLM 用 offset 继续读。

可插拔的操作

// packages/coding-agent/src/core/tools/read.ts:43
export interface ReadOperations {
  readFile: (absolutePath: string) => Promise<Buffer>;
  access: (absolutePath: string) => Promise<void>;
  detectImageMimeType?: (absolutePath: string) => Promise<string | null | undefined>;
}

默认用本地文件系统，但可以替换成远程读取（比如 SSH）。

好处：

安全：不会一次性读入超大文件
灵活：可以分页读取大文件
可扩展：可以轻松支持远程文件系统

4.2 `write` — 写文件

源码位置：packages/coding-agent/src/core/tools/write.ts

参数：

const writeSchema = Type.Object({
  path: Type.String({ description: "Path to the file to write" }),
  content: Type.String({ description: "Content to write to the file" }),
});

亮点设计：

自动创建父目录

await ops.mkdir(dir);  // recursive: true

文件修改队列

return withFileMutationQueue(absolutePath, async () => {
  // 写文件操作
});

串行化同一文件的修改，避免并发冲突。

Abort 信号处理

const throwIfAborted = (): void => {
  if (signal?.aborted) throw new Error("Operation aborted");
};
throwIfAborted();
await ops.mkdir(dir);
throwIfAborted();
await ops.writeFile(absolutePath, content);

每个异步操作后检查 abort 状态，及时中止。

好处：

安全：并发写同一文件不会冲突
响应式：用户中止时能及时停止
易用：不用手动创建目录

4.3 `edit` — 精确编辑

源码位置：packages/coding-agent/src/core/tools/edit.ts

参数：

const editSchema = Type.Object({
  path: Type.String({ description: "Path to the file to edit" }),
  edits: Type.Array(Type.Object({
    oldText: Type.String({ description: "Exact text for one targeted replacement" }),
    newText: Type.String({ description: "Replacement text" }),
  })),
});

亮点设计：

精确文本替换，不是正则
- oldText 必须在文件中唯一匹配
- 避免正则的复杂性和意外匹配

支持一次多处替换

edits: [
  { oldText: "function foo()", newText: "function bar()" },
  { oldText: "const x = 1", newText: "const x = 2" },
]

所有 oldText 都匹配原始文件，不是增量匹配。

自动处理编码问题

// packages/coding-agent/src/core/tools/edit-diff.ts
const { bom, text: content } = stripBom(rawContent);
const originalEnding = detectLineEnding(content);
const normalizedContent = normalizeToLF(content);
// ... 编辑操作 ...
const finalContent = bom + restoreLineEndings(newContent, originalEnding);

剥离 BOM（Byte Order Mark）
统一换行符为 LF
编辑完成后恢复原始换行符

生成 diff 和 patch

const diffResult = generateDiffString(baseContent, newContent);
const patch = generateUnifiedPatch(path, baseContent, newContent);

用于 UI 展示和版本控制。

好处：

精确：不会误改其他地方
高效：一次调用改多处
安全：自动处理编码问题
可追溯：生成标准 diff/patch

4.4 `bash` — 执行命令

源码位置：packages/coding-agent/src/core/tools/bash.ts

参数：

const bashSchema = Type.Object({
  command: Type.String({ description: "Bash command to execute" }),
  timeout: Type.Optional(Type.Number({ description: "Timeout in seconds" })),
});

亮点设计：

流式输出

child.stdout?.on("data", onData);
child.stderr?.on("data", onData);

实时收集 stdout 和 stderr。

智能截断
- 保留最后 N 行（默认 2000 行）
- 或保留最后 N KB（默认 100KB）
- 先到先停
- 完整输出存到临时文件

进程树管理

// packages/coding-agent/src/utils/shell.ts
export function killProcessTree(pid: number): void {
  // 杀掉整个进程树，不只是父进程
}

避免留下僵尸进程。

Abort 信号处理

const onAbort = () => {
  if (child.pid) killProcessTree(child.pid);
};
if (signal) {
  if (signal.aborted) onAbort();
  else signal.addEventListener("abort", onAbort, { once: true });
}

会话环境变量

// packages/coding-agent/src/core/tools/bash.ts:171
if (exposeSessionEnvironment && ctx) {
  env.PI_SESSION_ID = ctx.sessionManager.getSessionId();
  env.PI_PROVIDER = model.provider;
  env.PI_MODEL = model.id;
}

命令可以获取当前会话信息。

好处：

安全：超时自动杀掉，不会卡死
响应式：用户中止时能及时停止
可追溯：完整输出保存到临时文件
信息丰富：命令可以获取会话上下文

五、数据流全景

最后，让我们把整个数据流串起来：

用户输入: "帮我写一个排序函数"
  │
  ▼
Agent.prompt("帮我写一个排序函数")
  │
  ▼
agentLoop 双重循环
  │
  ├─► convertToLlm(messages)
  │     把 AgentMessage[] 转成 Message[]
  │     （过滤掉 LLM 不需要看到的消息）
  │
  ├─► streamFn(model, context)
  │     调用 LLM API（流式）
  │     LLM 返回: "好的，我来写一个排序函数..."
  │              + toolCall: { name: "write", arguments: { path: "sort.js", content: "..." } }
  │
  ├─► executeToolCalls()
  │     │
  │     ├─► prepare: 查找 write 工具定义，验证参数
  │     ├─► beforeToolCall: 权限检查（可选）
  │     ├─► execute: 调用 write.execute()，写文件
  │     └─► afterToolCall: 日志记录（可选）
  │
  ├─► 工具结果加入上下文
  │     messages.push(toolResultMessage)
  │
  ├─► 再调 LLM
  │     LLM 返回: "已经写好了排序函数"
  │              （没有 toolCall）
  │
  ▼
Agent 停止，等待用户输入

六、总结

PI Agent 的核心设计哲学：

分层清晰：每层只做一件事，可独立替换
关注点分离：LLM 看到的和 UI 看到的不一样
事件驱动：所有状态变化通过事件通知，UI 完全解耦
可插拔：工具、LLM 提供商、上下文管理都可以替换
安全优先：超时控制、abort 信号、权限检查、文件修改队列

理解了这些，就理解了整个系统的骨架。剩下的细节，看代码就明白了。

写于 2026-07-30

如果你有疑问，欢迎随时讨论。

目录CONTENT

PI Agent 核心架构剖析学习

PI Agent 核心架构解析

一、为什么要分三层？

二、第一层：`pi-ai` — LLM 通信层

2.1 这一层解决什么问题？

2.2 核心类型

2.3 亮点设计：`StreamFn` 类型

三、第二层：`pi-agent-core` — Agent 编排层

3.1 这一层解决什么问题？

3.2 核心类型：`AgentMessage` vs `Message`

3.3 双重循环 — 核心编排逻辑

3.4 关键函数：`streamAssistantResponse()`

3.5 工具执行的生命周期

3.6 工具执行模式：并行 vs 串行

3.7 `Agent` 类 — 有状态的封装

四、第三层：`coding-agent` — 四个核心编码工具

4.1 `read` — 读文件

4.2 `write` — 写文件

4.3 `edit` — 精确编辑

4.4 `bash` — 执行命令

五、数据流全景

六、总结

评论区

PI Agent 核心架构剖析学习

PI Agent 核心架构解析

一、为什么要分三层？

二、第一层：pi-ai — LLM 通信层

2.1 这一层解决什么问题？

2.2 核心类型

2.3 亮点设计：StreamFn 类型

三、第二层：pi-agent-core — Agent 编排层

3.1 这一层解决什么问题？

3.2 核心类型：AgentMessage vs Message

3.3 双重循环 — 核心编排逻辑

3.4 关键函数：streamAssistantResponse()

3.5 工具执行的生命周期

3.6 工具执行模式：并行 vs 串行

3.7 Agent 类 — 有状态的封装

四、第三层：coding-agent — 四个核心编码工具

4.1 read — 读文件

4.2 write — 写文件

4.3 edit — 精确编辑

4.4 bash — 执行命令

五、数据流全景

六、总结

评论区

二、第一层：`pi-ai` — LLM 通信层

2.3 亮点设计：`StreamFn` 类型

三、第二层：`pi-agent-core` — Agent 编排层

3.2 核心类型：`AgentMessage` vs `Message`

3.4 关键函数：`streamAssistantResponse()`

3.7 `Agent` 类 — 有状态的封装

四、第三层：`coding-agent` — 四个核心编码工具

4.1 `read` — 读文件

4.2 `write` — 写文件

4.3 `edit` — 精确编辑

4.4 `bash` — 执行命令