2026-08-24 20 分钟阅读阅读加载中... 次访客加载中... 人

Pi 架构深度拆解：如何用现代 TypeScript 构建高可控的 AI Agent 与 Coding Harness

很多 Agent 框架在 Demo 阶段看起来很美好，一旦进到真实工程——长上下文膨胀、工具流式截断、状态回滚与权限门禁——就会迅速失控。开源 Agent 套件 Pi（earendil-works/pi）采用极度解耦的分层设计，从底层的流式 Partial JSON 解析、状态化 Agent Loop，到 Harness 层的分支压缩与 Subagent 派发。本文从一手源码与实战角度，拆解如何用 Pi 搭建高可控的智能体系统。

#AI Agent#Pi#Agent Framework#TypeScript#源码拆解

在构建复杂智能体（如 Coding Agent、自动化测试 Agent、DevOps 运维 Agent）时，许多团队最初都会尝试 LangChain、CrewAI 或 AutoGen。但当项目从“跑通 Demo”走向“真实交付”，很快就会撞上一堵硬墙：

框架黑盒与状态失控：框架把 Prompt 组装、Tool Calling 调度和内部状态层层深埋，遇到模型幻觉或者工具执行报错时，很难在中途做精确干预。
上下文膨胀与窗口爆仓：长会话进行到二三十轮后，Token 消耗急剧上升，缺乏确定性的上下文剪枝与分支回滚策略。
UI 与推理协议耦合：界面需要的展示消息（如折叠的工具日志、UI 弹窗、进度卡片）与大模型 API 严格要求的 [user, assistant, tool_result] 消息格式混杂在一起，稍有不慎就导致模型报错。
流式体验割裂：无法在模型生成参数的瞬间进行 Partial JSON 增量解析，导致终端或前端界面无法实时展示工具调用的参数流。

开源 Agent 套件 Pi（GitHub: earendil-works/pi，包含 @earendil-works/pi-ai、@earendil-works/pi-agent-core、@earendil-works/pi-coding-agent、@earendil-works/pi-tui）给出了一个极为硬核且克制的解法：拒绝黑盒魔法，通过纯粹的 TypeScript 分层架构，提供高透明度、完全可控且可嵌入的 Agent 基础设施。

本文将结合 Pi 的核心源码与工程实践，深度拆解其四层分层架构、Agent Loop 状态机、生命周期拦截钩子、会话分支压缩以及多 Agent 派发机制，并给出完整的实战代码指南。

一、Pi 的四层解耦架构全景

Pi 并不是一个单一的单体 CLI，而是一个高度模块化的 Monorepo。它将智能体系统的职责严格划分为四个独立层次：

每个分层都有非常清晰的输入与输出边界，下层完全不依赖上层：

1. 统一模型网关（`@earendil-works/pi-ai`）

职责：抹平所有主流模型厂商（Anthropic, OpenAI, Google Gemini, Ollama, DeepSeek, Bedrock, Groq, OpenRouter 等）的接口差异。
核心能力：
- 流式 Partial JSON 解析：在模型吐出 token 的同时流式解析工具调用参数，不用等整段 JSON 生成完毕就能提取字段。
- Thinking / 思考流抽象：统一不同模型的推理模式（如 Claude 的 thinkingBudget、DeepSeek 的 <think> 标签），输出结构化的 thinking_delta。
- 精确 Token 与成本计量：实时统计每轮对话的 input、output、cacheRead、cacheWrite 以及按美分计价的成本，支持在中途无缝将上下文切给另一个模型（Cross-Provider Handoff）。

2. 状态化运行时与事件循环（`@earendil-works/pi-agent-core`）

职责：纯净的 ReAct 状态机与事件驱动执行器，无任何文件系统或 UI 绑定，可运行在 Node.js、浏览器甚至 Cloudflare Workers 中。
核心能力：
- 消息解耦（AgentMessage vs Message）：内部维护富文本与 UI 状态，在调用 LLM 边界处通过 convertToLlm() 进行纯化。
- 细粒度生命周期钩子：beforeToolCall（权限拦截/参数修改）、afterToolCall（结果脱敏/提前终止）、shouldStopAfterTurn、prepareNextTurn。
- 全链路事件流：向上层抛出结构化的 AgentEvent（20+ 种细粒度事件）。

3. Harness 与控制面（`@earendil-works/pi-coding-agent`）

职责：为 Coding 与复杂任务提供生产级脚手架与环境治理。
核心能力：
- 树状会话与分支导航（SessionManager）：会话不是一条单向链表，而是一棵可分叉、可切换、可回滚的树（类似 Git 分支）。
- 滑动窗口自动压缩（Compaction）：当 Context 达到阈值时，自动触发结构化摘要，同时精准追踪已修改/已读取文件的累计变更。
- Extensions 扩展机制：允许开发者通过 TypeScript 编写插件，动态注册工具、添加 Slash 命令、持久化会话数据与定制 TUI 交互。
- Skills 规范：遵循 agentskills.io 标准，支持按需发现、渐进注入与外部脚本挂载。

4. 交互与通信层（`@earendil-works/pi-tui` & RPC Mode）

职责：人机交互与外部系统集成。
核心能力：
- 终端无闪烁渲染：pi-tui 采用类似 React 的虚拟 DOM 差分算法（Differential rendering），保证高频流式输出下的丝滑体验。
- Headless RPC 协议：基于标准 stdin/stdout 的 JSONL 协议，支持把 Agent 作为后端内核无缝嵌入到 VSCode 插件、Web 平台或自动化测试流水线中。

二、核心机制拆解：Agent Loop 与消息隔离

要理解 Pi 为什么足够稳健，关键在于看清它的核心循环 agentLoop 是如何流转数据的。

1. 消息隔离：为什么必须区分 `AgentMessage` 与 `Message`？

在大部分初学者写的 Agent 中，传递给 LLM 的历史记录往往直接用作前端渲染数据。一旦你想在会话里插入“正在运行单测”、“用户点击了确认按钮”、“环境检查通过”等提示信息，就会污染 LLM 的上下文，导致 OpenAI / Anthropic 的 API 抛出 Invalid message role 错误。

Pi 在设计上做了绝对的隔离：

UI/业务层 (AgentMessage[]) 
  ──> transformContext() (裁剪、RAG注入、压缩)
  ──> convertToLlm() (过滤自定义UI角色，转为纯标准Message[])
  ──> 发送给大模型 API

在 @earendil-works/pi-agent-core 的实现中：

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

这意味着你可以在 AgentMessage 中任意扩展业务消息（例如 role: "system_status", role: "qa_verification"），只要在 convertToLlm 钩子中决定是将其丢弃、还是转译为特定的 user prompt，模型端看到的永远是标准、合规的上下文。

2. 生命周期拦截：给 Agent 装上“安全刹车片”

在自主智能体执行复杂任务时，最危险的就是模型调用了具有破坏性的工具（如 rm -rf、修改数据库、向生产集群推送代码）。

Pi 在 agent-loop.ts 中通过 beforeToolCall 和 afterToolCall 提供了严格的确定性拦截机制：

// 源码摘录自 packages/agent/src/types.ts
export interface BeforeToolCallResult {
  block?: boolean;      // 设为 true 则直接阻断工具执行
  reason?: string;     // 阻断原因，作为错误结果回传给模型
  terminate?: boolean;  // 是否在阻断后提前结束整个 Agent 会话
}

export interface AfterToolCallResult {
  content?: (TextContent | ImageContent)[]; // 覆写/脱敏工具返回值
  details?: unknown;
  isError?: boolean;
  terminate?: boolean;  // 是否立即终止后续步骤
}

当 beforeToolCall 返回 { block: true, reason: "用户拒绝了删除操作" } 时，Agent Loop 不会崩溃，而是会构造一个 isError: true 的 toolResult 喂回给模型，让模型能够理解失败原因并尝试其他路径。

三、实战演练：从零到一构建高可控 Agent

接下来，我们通过四个层次的递进代码，演示如何使用 Pi 构建不同复杂度的智能体。

Level 1: 极简流式 Agent（基于 `@earendil-works/pi-agent-core`）

首先安装核心包：

npm install @earendil-works/pi-ai @earendil-works/pi-agent-core @sinclair/typebox

编写一个具备天气查询工具的最小 Agent：

import { Agent } from "@earendil-works/pi-agent-core";
import { createModels } from "@earendil-works/pi-ai";
import { anthropicProvider } from "@earendil-works/pi-ai/providers/anthropic";
import { Type } from "@sinclair/typebox";

// 1. 初始化模型管理器并配置 Provider
const models = createModels();
models.setProvider(anthropicProvider());
const model = models.getModel("anthropic", "claude-sonnet-4-6");
if (!model) throw new Error("Model not found");

// 2. 定义强类型工具
const weatherTool = {
  name: "get_weather",
  description: "查询指定城市的天气信息",
  parameters: Type.Object({
    city: Type.String({ description: "城市名称，如 Beijing, Shanghai" }),
    unit: Type.Optional(Type.Union([Type.Literal("celsius"), Type.Literal("fahrenheit")])),
  }),
  execute: async ({ city, unit }) => {
    // 模拟工具真实执行
    return {
      content: [{ type: "text", text: `${city} 当前气温 24°C，晴空万里，风力 2 级。` }],
      details: { humidity: 45, aqi: 28 },
    };
  },
};

// 3. 实例化 Agent
const agent = new Agent({
  initialState: {
    systemPrompt: "你是一个专业的出行助手。在调用工具前后给出简明扼要的说明。",
    model,
    tools: [weatherTool],
  },
  streamFn: models.streamSimple.bind(models),
});

// 4. 订阅全链路事件流
agent.subscribe((event) => {
  switch (event.type) {
    case "turn_start":
      console.log("\n[Turn Start]");
      break;
    case "message_update":
      if (event.assistantMessageEvent.type === "text_delta") {
        process.stdout.write(event.assistantMessageEvent.delta);
      } else if (event.assistantMessageEvent.type === "thinking_delta") {
        process.stdout.write(`\x1b[33m${event.assistantMessageEvent.delta}\x1b[0m`);
      }
      break;
    case "tool_execution_start":
      console.log(`\n⚙️ 正在执行工具: ${event.toolCall.name}(${JSON.stringify(event.toolCall.arguments)})`);
      break;
    case "tool_execution_end":
      console.log(`✅ 工具执行完成`);
      break;
  }
});

// 5. 触发对话
await agent.prompt("北京现在的天气怎么样？我穿短袖合适吗？");

Level 2: 注入权限门禁与动态上下文治理

在实际业务中，我们往往需要拦截某些敏感工具，并在长对话中动态裁剪历史：

import { Agent, type BeforeToolCallContext, type BeforeToolCallResult } from "@earendil-works/pi-agent-core";

const dangerousBashTool = {
  name: "bash",
  description: "在宿主机执行终端命令",
  parameters: Type.Object({ command: Type.String() }),
  execute: async ({ command }) => {
    // 执行 shell...
    return { content: [{ type: "text", text: `Output of ${command}` }] };
  },
};

const agent = new Agent({
  initialState: {
    systemPrompt: "你是一个自动化运维助手。",
    model,
    tools: [dangerousBashTool],
  },
  streamFn: models.streamSimple.bind(models),

  // 1. 工具调用前的权限防线
  beforeToolCall: async ({ toolCall, args }: BeforeToolCallContext): Promise<BeforeToolCallResult> => {
    if (toolCall.name === "bash") {
      const cmd = (args as { command: string }).command;
      if (cmd.includes("rm -rf") || cmd.includes("drop database")) {
        console.warn(`[Security Alert] 阻断高危命令: ${cmd}`);
        return {
          block: true,
          reason: `安全策略拦截：禁止执行破坏性命令 '${cmd}'`,
        };
      }
    }
    return {};
  },

  // 2. 工具返回结果脱敏与后处理
  afterToolCall: async ({ toolCall, result }) => {
    // 对敏感输出中的 Token 或密码进行正则脱敏
    if (toolCall.name === "bash") {
      const sanitized = result.content.map((item) => {
        if (item.type === "text") {
          return { type: "text" as const, text: item.text.replace(/ghp_[a-zA-Z0-9]{36}/g, "[REDACTED_TOKEN]") };
        }
        return item;
      });
      return { content: sanitized };
    }
  },

  // 3. 上下文滑动窗口治理
  transformContext: async (messages) => {
    // 保留最近 15 轮对话，避免长会话导致上下文爆炸
    if (messages.length > 30) {
      const systemOrFirst = messages.slice(0, 2);
      const recent = messages.slice(-20);
      return [...systemOrFirst, ...recent];
    }
    return messages;
  },
});

Level 3: 基于 `@earendil-works/pi-coding-agent` SDK 构建完整 Harness

如果你需要构建一个包含完整会话持久化、树状分支回滚、自动上下文压缩（Compaction）和技能扩展（Skills）的 Coding Agent，可以直接使用上层的 Coding Harness SDK：

import { createAgentSession, ModelRuntime, SessionManager, SettingsManager } from "@earendil-works/pi-coding-agent";

// 1. 初始化模型运行时与认证存储
const modelRuntime = await ModelRuntime.create({
  authPath: "./auth.json",
  modelsPath: "./models.json",
});

// 2. 配置会话管理器（支持持久化到本地目录）
const sessionManager = SessionManager.create("./my-agent-session");

// 3. 创建具备全功能特性的 Agent Session
const { session } = await createAgentSession({
  cwd: process.cwd(),
  tools: ["read", "write", "edit", "bash", "grep", "find"], // 内置工具
  sessionManager,
  settingsManager: SettingsManager.inMemory({
    compaction: {
      enabled: true,
      reserveTokens: 16384, // 预留 16k Token 给模型输出
    },
  }),
});

session.subscribe((event) => {
  if (event.type === "message_update" && event.assistantMessageEvent.type === "text_delta") {
    process.stdout.write(event.assistantMessageEvent.delta);
  }
});

// 4. 执行多步骤开发任务
await session.prompt("请检查当前目录下的 package.json，找出所有未锁版本的依赖并修复");

Level 4: 编写 Extensions 插件扩展

Pi 的 Extension 机制极其灵活。它允许你通过 TypeScript 模块接入生命周期，实现工具注入、UI 交互确认甚至自定义命令：

// .pi/extensions/git-guard.ts
import type { ExtensionAPI, ExtensionContext } from "@earendil-works/pi-coding-agent";

export default function gitGuardExtension(pi: ExtensionAPI) {
  // 1. 注册自定义 Slash 命令
  pi.registerCommand({
    name: "snapshot",
    description: "为当前工作区创建 Git 临时检查点",
    handler: async (args, ctx: ExtensionContext) => {
      ctx.ui.notify("正在创建本地快照...");
      // 执行快照存储逻辑...
      pi.appendEntry("snapshot-state", { timestamp: Date.now() });
      ctx.ui.notify("快照创建完成！");
    },
  });

  // 2. 注册动态业务工具
  pi.registerTool({
    name: "query_db_schema",
    description: "查询数据库表的 DDL 定义",
    parameters: {
      type: "object",
      properties: { tableName: { type: "string" } },
      required: ["tableName"],
    },
    execute: async ({ tableName }) => {
      return {
        content: [{ type: "text", text: `CREATE TABLE ${tableName} (id BIGINT PRIMARY KEY);` }],
      };
    },
  });

  // 3. 监听会话分支切换事件
  pi.on("session_branch_switch", async (event) => {
    console.log(`[Extension] 用户切换了会话分支: ${event.branchId}`);
  });
}

四、进阶模式：多智能体编排与 Subagents

在长程开发任务中，让单个 Agent 处理所有工作会迅速撑爆上下文。Pi 官方提供了 pi-subagents 模式，通过多进程启动相互独立的 pi 实例，实现任务的解耦与分发：

                    ┌─────────────────────────┐
                    │      Parent Agent       │
                    │  (规划、决策、汇总协调)    │
                    └────────────┬────────────┘
                                 │ 派发任务 (JSON/RPC)
         ┌───────────────────────┼───────────────────────┐
         ▼                       ▼                       ▼
┌──────────────────┐   ┌──────────────────┐   ┌──────────────────┐
│ Subagent 1 (测试) │   │ Subagent 2 (文档) │   │ Subagent 3 (重构) │
│ 独立上下文 (4k)   │   │ 独立上下文 (6k)   │   │ 独立上下文 (8k)   │
└──────────────────┘   └──────────────────┘   └──────────────────┘

Subagent 机制的核心优势在于：

上下文彻底隔离：子 Agent 在独立的沙盒进程中执行，读写了数十个文件产生的数万 Token 不会污染父 Agent 的主上下文。
结构化结果回传：子 Agent 完成任务后，仅将精炼的执行总结与产物元数据以 JSON 格式回传给父 Agent。
支持三种派发模式：
- Single：{ agent: "tester", task: "为 auth.ts 编写测试用例" }
- Parallel：并发派发多个互不依赖的任务（如同时检查 5 个模块的代码异味）。
- Chain：链式传递，后一个 Subagent 将前一个 Subagent 的输出作为输入。

五、设计取舍与工程对比

我们将 Pi 与当前业内主流的 Agent 构建方式进行横向对比：

维度	Pi (`@earendil-works/pi`)	LangChain / LangGraph	Claude Code (原生)	AutoGen / CrewAI
设计哲学	极简、模块化、高透明、全可控	重型抽象、层层封装	单体 CLI、面向终端交付	多 Agent 角色扮演、高层抽象
语言与生态	现代 TypeScript / Node.js	Python / TypeScript	TypeScript (闭源/半打包)	Python
消息隔离	强隔离 (`AgentMessage` vs `Message`)	较弱 (混在 State / BaseMessage)	内部专有格式	依赖框架消息字典
嵌入灵活性	极高 (可拆解使用 ai / agent / harness)	中等 (绑定其生态组件)	低 (主要作为独立 CLI)	中等 (适合独立运行)
会话状态管理	树状分支 (`/tree`) + 结构化剪枝	Checkpointer 状态机	本地文件会话	内存 Conversation
流式工具调用	支持 Partial JSON 增量解析	取决于底层适配器	原生流式	较弱
生产调试难度	极低 (全事件流、明确的 Hook)	较高 (链条长、栈深)	中等	较高 (Prompt 隐藏较深)

适用场景建议

强烈推荐使用 Pi 的场景：
- 需要自研专用的 Coding Agent 或 IDE 插件。
- 需要构建移动端 APP 自动化测试 Agent、Web 爬虫与自动化运维平台。
- 对上下文 Token 成本、数据安全脱敏、工具权限拦截有严苛要求的企业级系统。
不一定适合 Pi 的场景：
- 纯 Python 算法团队或纯学术研究型原型验证。
- 只需要单次 Prompt + RAG 检索的简单问答应用。

六、总结

Pi 的设计体现了当前智能体工程落地的一个重要趋势：从“神奇的黑盒框架”回归到“清晰的系统工程”。

它没有试图用抽象的名词去掩盖 LLM 的不确定性，而是老老实实做好了四件事：

用 @earendil-works/pi-ai 做好多模型流式与工具调用的底层标准化；
用 @earendil-works/pi-agent-core 建立纯净、可观察、具备安全拦截的 Turn 状态机；
用 AgentMessage 与 Message 的隔离机制，彻底解耦 UI 业务展示与模型推理协议；
用树状会话与结构化 Compaction，为长程任务提供了扎实的上下文治理能力。

如果你正在考虑用 TypeScript 搭建自己的 AI Agent，不妨深入翻阅一下 Pi 的源码仓库，它的架构设计与代码质量绝对值得每一位智能体工程师细细品味。

参考链接与资源

Pi 官方代码仓：earendil-works/pi (GitHub)
Agent Skills 开放标准：agentskills.io
Pi 扩展与 SDK 示例：packages/coding-agent/examples/