Pi Agent 的简单上手与实践

PI 是一个用于构建 AI Agent 的 TypeScript 开源项目。项目结构为一个由多个包组成的 monorepo，核心的几个包功能如下：pi-ai 负责跨多个提供商的 LLM 通信，pi-agent-core 在之上加入带工具调用的 agent 循环，pi-coding-agent提供完整的编码 agent（内置工具、会话持久化、可扩展性），而 pi-tui 用于构建 CLI 界面的终端 UI。

本指南逐层讲解，动手逐步构建出一个功能完整的编码助手，具备终端 UI、会话持久化和自定义工具。旨在了解每一层的作用和如何组合使用它们。

技术栈

┌─────────────────────────────────────────┐
│  Your Application                       │
│  (a CLI tool, a Slack bot, a web UI)    │
├────────────────────┬────────────────────┤
│  pi-coding-agent   │  pi-tui            │
│  Sessions, tools,  │  Terminal UI,      │
│  extensions        │  markdown, editor  │
├────────────────────┴────────────────────┤
│  pi-agent-core                          │
│  Agent loop, tool execution, events     │
├─────────────────────────────────────────┤
│  pi-ai                                  │
│  Streaming, models, multi-provider LLM  │
└─────────────────────────────────────────┘

每一层都增加一项能力。按需使用，用多少取多少。

• pi-ai —— 通过一个统一的接口调用任何 LLM。Anthropic、OpenAI、Google、Bedrock、Mistral、Groq、xAI、OpenRouter、Ollama 等。流式输出、补全、工具定义、成本统计。
• pi-agent-core —— 把 pi-ai 包装成 agent 循环。你定义工具，agent 调用 LLM、执行工具、把结果回传，然后重复，直到完成。
• pi-coding-agent —— 完整的 agent 运行时。内置文件工具（read、write、edit、bash、powershell、grep、find、ls）、JSONL 会话持久化、上下文压缩、skills 以及扩展系统。
• pi-tui —— 带差分渲染的终端 UI 库。Markdown 显示、带自动补全的多行编辑器、加载动画，以及无闪烁的屏幕更新。

环境要求

• Node.js 22.19+
• 至少一个提供商的 API key（DeepSeek、OpenAI、Google 等）

安装

mkdir pi-agent && cd pi-agent
npm init -y
npm install @earendil-works/pi-ai @earendil-works/pi-agent-core @earendil-works/pi-coding-agent @earendil-works/pi-tui chalk
npm install -D typescript @types/node tsx

设置你的 API key：

export DEEPSEEK_API_KEY=sk-ant-...
# or
export OPENAI_API_KEY=sk-...

第一层：pi-ai

你的第一个 LLM 调用

创建 basics.ts：

import { builtinModels } from "@earendil-works/pi-ai/providers/all";

async function main() {
  const models = builtinModels();
    const model = models.getModel("deepseek", "deepseek-v4-flash")!;

  const response = await models.complete(model, {
    systemPrompt: "You are a helpful assistant.",
    messages: [
      { role: "user", content: "What is the capital of France?", timestamp: Date.now() }
    ],
  });

  // response 是一个 AssistantMessage
  for (const block of response.content) {
    if (block.type === "text") {
      console.log(block.text);
    }
  }

  console.log(`\nTokens: ${response.usage.totalTokens}`);
  console.log(`Stop reason: ${response.stopReason}`);
}

main();

执行：

npx tsx basics.ts

builtinModels() 会创建一个由 pi 内置目录支持的模型集合。models.getModel(provider, id) 查找指定的模型，models.complete(model, context) 发送消息，并在模型结束时返回完整的 AssistantMessage。

响应有一个 .content 数组，包含类型化的块 —— text、thinking 或 toolCall —— 还有用于 token 计数的 .usage，以及说明模型为什么停止的 .stopReason（"stop"、"toolUse"、"length"、"error"、"aborted"、"deferred"）。

流式输出

complete 会等待完整响应。要实时输出，请使用 streamSimple：

import { builtinModels } from "@earendil-works/pi-ai/providers/all";

async function main() {
  const models = builtinModels();
  const model = models.getModel("deepseek", "deepseek-v4-flash")!;

  const stream = models.streamSimple(model, {
    systemPrompt: "You are a helpful assistant.",
    messages: [
      { role: "user", content: "Explain how TCP works in 3 sentences.", timestamp: Date.now() }
    ],
  });

  for await (const event of stream) {
    switch (event.type) {
      case "text_delta":
        process.stdout.write(event.delta);
        break;
      case "done":
        console.log(`\n\nTokens: ${event.message.usage.totalTokens}`);
        break;
      case "error":
        console.error("Error:", event.error.errorMessage);
        break;
    }
  }
}

main();

每个提供商都有自己的流式格式 —— Anthropic、OpenAI 和 Google 的实现各不相同。streamSimple 把它们规范化为一组统一的事件：start、text_start、text_delta、text_end、thinking_start/delta/end、toolcall_start/delta/end、done 和 error。流式处理器只需写一次，就能用于任何提供商。大多数场景下，你只需要关心 text_delta（文本块）和 done（最终消息）。

你也可以直接等待最终消息：

const stream = models.streamSimple(model, context);
const finalMessage = await stream.result(); // AssistantMessage

切换提供商

只需修改 models.getModel 调用即可切换提供商。其余代码保持不变。

// 只需修改这一行——其余代码保持不变
const model = models.getModel("deepseek", "deepseek-v4-flash")!;
// const model = models.getModel("openai", "gpt-4o")!;
// const model = models.getModel("google", "gemini-2.5-pro")!;
// const model = models.getModel("groq", "llama-3.3-70b-versatile")!;

const stream = models.streamSimple(model, context);

每个提供商都需要在环境变量中设置自己的 API key（DEEPSEEK_API_KEY、OPENAI_API_KEY、GEMINI_API_KEY、GROQ_API_KEY 等）。

你也可以为自托管端点定义自定义模型。先定义模型：

import type { Model } from "@earendil-works/pi-ai";

const localModel: Model<"openai-completions"> = {
  id: "llama3.1:8b",
  name: "llama3.1:8b",
  api: "openai-completions",
  provider: "ollama",
  baseUrl: "http://localhost:11434/v1",
  reasoning: false,
  input: ["text"],
  cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 },
  contextWindow: 128000,
  maxTokens: 8192,
};

然后通过一个 provider 把它注册到模型集合上：

import { createModels, createProvider } from "@earendil-works/pi-ai";
import { openAICompletionsApi } from "@earendil-works/pi-ai/api/openai-completions.lazy";

const models = createModels();

models.setProvider(
  createProvider({
    id: "ollama",
    name: "Ollama",
    baseUrl: "http://localhost:11434/v1",
    api: openAICompletionsApi(),
    auth: {
      apiKey: {
        name: "Ollama API key",
        resolve: async () => ({ auth: { apiKey: "ollama" } }),
      },
    },
    models: [localModel],
  }),
);

const model = models.getModel("ollama", "llama3.1:8b")!;
const response = await models.complete(model, {
  systemPrompt: "You are a helpful assistant.",
  messages: [{ role: "user", content: "Hello!", timestamp: Date.now() }],
});

在底层，pi-ai 使用各提供商的官方 SDK（OpenAI SDK、Anthropic SDK 等）。api 字段决定由哪个 SDK 处理请求 —— "openai-completions" 会走 OpenAI SDK，因此它适用于任何与 OpenAI 兼容的端点（Ollama、vLLM、Mistral 等）。

API key 会按提供商名称从环境变量中自动解析（OPENAI_API_KEY、DEEPSEEK_API_KEY 等），并传给 SDK 完成认证。Ollama 不需要认证，所以上面的示例提供了一个固定的本地 key。

思考等级

支持扩展思考的模型（DeepSeek、Claude、o3、Gemini 2.5）可以通过 reasoning 选项启用：

const stream = models.streamSimple(model, context, {
  reasoning: "high", // "minimal" | "low" | "medium" | "high" | "xhigh" | "max"
});

启用后，流中会同时发出 thinking_delta 事件和 text_delta 事件。

第二层：pi-agent-core

pi-ai 让你能和 LLM 对话。pi-agent-core 让 LLM 能反过来“动手” —— 通过工具。

Agent 类运行标准的 agent 循环：把消息发给 LLM，执行它产生的任何工具调用，把结果回传，然后重复，直到模型停下来。

定义工具

工具使用 TypeBox schema 做类型安全的参数定义：

import { Type } from "@earendil-works/pi-ai";
import type { AgentTool } from "@earendil-works/pi-agent-core";

const weatherParams = Type.Object({
  city: Type.String({ description: "City name" }),
});

const weatherTool: AgentTool<typeof weatherParams> = {
  name: "get_weather",
  label: "Weather",
  description: "Get the current weather for a city",
  parameters: weatherParams,
  execute: async (toolCallId, params, signal, onUpdate) => {
    // params 的类型为：{ city: string }
    const temp = Math.round(Math.random() * 30);
    return {
      content: [{ type: "text", text: `${params.city}: ${temp}C, partly cloudy` }],
      details: { temp, city: params.city },
    };
  },
};

把 schema 定义为独立变量，并作为泛型参数传给 AgentTool<typeof schema> —— 这样 TypeScript 就能获得所需的类型信息，在 execute 内正确推断出 params。

每个工具都有：

• name —— LLM 调用它时使用的标识符
• label —— 人类可读的显示名称
• description —— 告诉 LLM 何时以及如何使用该工具
• parameters —— TypeBox schema；执行前用 AJV 校验
• execute —— 当 LLM 调用该工具时运行；返回 content（回传给 LLM）和 details（给你的 UI，不会发给 LLM）

onUpdate 回调让你在执行过程中流式输出部分结果 —— 对 bash 命令这类长时间运行的工具很有用。

工具失败应该通过在 execute 中抛出异常来表示。agent 循环会捕获错误，并把它记录为一条错误工具结果，这样 LLM 就能看到这次调用失败了。

创建 agent

把上面的天气工具和一个模型、一个流式函数接起来。我们会在后续章节中加入事件处理、提示词和一个完整可运行的示例。

import { Agent } from "@earendil-works/pi-agent-core";
import { builtinModels } from "@earendil-works/pi-ai/providers/all";

const models = builtinModels();
const model = models.getModel("deepseek", "deepseek-v4-flash")!;

const agent = new Agent({
  initialState: {
    systemPrompt: "You are a helpful assistant with access to tools.",
    model,
    tools: [weatherTool],
    thinkingLevel: "off",
  },
  streamFn: (model, context, options) => models.streamSimple(model, context, options),
});

Agent 接收一个 initialState（系统提示词、模型、工具、思考等级）和一个 streamFn —— 真正调用 LLM 的函数。传入一个包装了 models.streamSimple 的函数，就能把 agent 连接到模型所指定的任何提供商。

事件流

订阅事件来观察 agent 在做什么：

agent.subscribe((event) => {
  switch (event.type) {
    case "agent_start":
      console.log("Agent started");
      break;

    case "message_update":
      // 来自 LLM 的流式文本
      if (event.assistantMessageEvent.type === "text_delta") {
        process.stdout.write(event.assistantMessageEvent.delta);
      }
      break;

    case "tool_execution_start":
      console.log(`\nTool: ${event.toolName}(${JSON.stringify(event.args)})`);
      break;

    case "tool_execution_end":
      console.log(`Result: ${event.isError ? "ERROR" : "OK"}`);
      break;

    case "agent_end":
      console.log("\nAgent finished");
      break;
  }
});

完整事件列表：agent_start、agent_end、turn_start、turn_end、message_start、message_update、message_end、tool_execution_start、tool_execution_update、tool_execution_end。

运行 agent

await agent.prompt("What's the weather in Tokyo and London?");

就是这样。agent 会：

1. 把你的消息发给 LLM
2. LLM 用一条 assistant 消息回复，其中可能包含零个或多个工具调用（本例：同时在同一条消息里为东京和伦敦调用 get_weather）
3. agent 执行该消息里的所有工具调用，再把所有结果一并回传
4. LLM 生成最终的文本回复

默认情况下，同一条 assistant 消息里的多个工具调用会并发执行（toolExecution: "parallel"）。如需严格串行，可在 agent 上设置 toolExecution: "sequential"。

你不需要自己写这个循环，agent 会处理。

完整示例

下面是一个带两个工具的完整可用 agent：

import { Agent } from "@earendil-works/pi-agent-core";
import { builtinModels } from "@earendil-works/pi-ai/providers/all";
import { Type } from "@earendil-works/pi-ai";
import type { AgentTool } from "@earendil-works/pi-agent-core";
import * as fs from "fs";

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

const readFileTool: AgentTool<typeof readFileParams> = {
  name: "read_file",
  label: "Read File",
  description: "Read the contents of a file",
  parameters: readFileParams,
  execute: async (_id, params) => {
    const content = fs.readFileSync(params.path, "utf-8");
    return {
      content: [{ type: "text", text: content }],
      details: {},
    };
  },
};

const listFilesParams = Type.Object({
  path: Type.String({ description: "Directory path", default: "." }),
});

const listFilesTool: AgentTool<typeof listFilesParams> = {
  name: "list_files",
  label: "List Files",
  description: "List files in a directory",
  parameters: listFilesParams,
  execute: async (_id, params) => {
    const files = fs.readdirSync(params.path);
    return {
      content: [{ type: "text", text: files.join("\n") }],
      details: { count: files.length },
    };
  },
};

async function main() {
  const models = builtinModels();
  const model = models.getModel("deepseek", "deepseek-v4-flash")!;

  const agent = new Agent({
    initialState: {
      systemPrompt: "You can read files and list directories. Be concise.",
      model,
      tools: [readFileTool, listFilesTool],
      thinkingLevel: "off",
    },
    streamFn: (model, context, options) => models.streamSimple(model, context, options),
  });

  agent.subscribe((event) => {
    if (event.type === "message_update" && event.assistantMessageEvent.type === "text_delta") {
      process.stdout.write(event.assistantMessageEvent.delta);
    }
    if (event.type === "tool_execution_start") {
      console.log(`\n[${event.toolName}] ${JSON.stringify(event.args)}`);
    }
  });

  await agent.prompt("What files are in the current directory? Read the package.json if it exists.");
  console.log();
}

main();

引导与后续消息

当 agent 正在工作时，如果你想重新引导它：

// 打断：当前工具执行完后送达。
// 剩余待执行的工具会被跳过。
agent.steer({
  role: "user",
  content: "Actually, skip that and read tsconfig.json instead.",
  timestamp: Date.now(),
});

// 后续：排队到 agent 自然结束后执行。
// 不会打断当前工作。
agent.followUp({
  role: "user",
  content: "Now summarize what you found.",
  timestamp: Date.now(),
});

steer 会打断 —— 跳过剩余的工具并注入你的消息。followUp 会等待 —— 把消息排队到 agent 自然结束之后。用 steering 处理实时用户消息（用户在 agent 工作时输入），用 follow-up 做程序化串联。

状态管理

你可以随时通过修改 agent.state 改变 agent 的配置：

agent.state.model = models.getModel("openai", "gpt-4o")!;   // 会话中途切换提供商
agent.state.thinkingLevel = "high";                          // 启用扩展思考
agent.state.systemPrompt = "New instructions.";              // 更新系统提示词
agent.state.tools = [...newTools];                           // 替换工具集
agent.state.messages = trimmedMessages;                      // 替换对话历史

agent 会在下一轮中应用这些更改。

第三层：pi-coding-agent

pi-agent-core 给你循环。pi-coding-agent 给你生产级的 agent：内置工具、会话持久化和可扩展性。它构建在 pi-agent-core 之上 —— 当你使用 pi-coding-agent 时，底层已经包含了 pi-agent-core。

大多数用户应该从这里开始；只有当你需要一个不使用内置编码工具或会话系统的自定义 agent 时，才直接降到 pi-agent-core。

内置工具

pi-coding-agent 有 8 个内置工具。其中 4 个默认启用，另有 4 个默认关闭、需要手动开启：

默认工具（启用）：

附加工具（手动开启）：

通过给 createAgentSession 传一个工具名允许列表来选择工具：

const { session } = await createAgentSession({
  model,
  tools: ["read", "bash", "grep"], // 要启用的内置工具
  sessionManager: SessionManager.inMemory(),
});

createAgentSession

createAgentSession 把一切都接起来 —— 模型、工具、会话持久化、设置：

import { createAgentSession, SessionManager } from "@earendil-works/pi-coding-agent";
import { builtinModels } from "@earendil-works/pi-ai/providers/all";

async function main() {
  const models = builtinModels();
  const model = models.getModel("deepseek", "deepseek-v4-flash")!;

  const { session } = await createAgentSession({
    model,
    thinkingLevel: "off",
    sessionManager: SessionManager.inMemory(),
  });

  session.subscribe((event) => {
    if (event.type === "message_update" && event.assistantMessageEvent.type === "text_delta") {
      process.stdout.write(event.assistantMessageEvent.delta);
    }
    if (event.type === "tool_execution_start") {
      console.log(`\n[${event.toolName}]`);
    }
  });

  await session.prompt("What files are in the current directory? Summarize the package.json.");
  console.log();

  session.dispose();
}

main();

这是一个能用的编码 agent。它能读取你的文件、运行命令、修改代码、写新文件。SessionManager.inMemory() 表示会话只存在内存中，进程退出后即消失。

会话持久化

要持久化会话，把 SessionManager 指向一个文件：

import * as path from "path";

const sessionFile = path.join(process.cwd(), ".sessions", "my-session.jsonl");
const sessionManager = SessionManager.open(sessionFile);

const { session } = await createAgentSession({
  model,
  sessionManager,
});

会话以 JSONL 文件存储，并带有树形结构 —— 每个条目都有 id 和 parentId。这实现了分支：你可以导航到对话中的任意历史节点，并从那里继续，而不会丢失历史记录。

SessionManager 有几个静态工厂方法。根据你的使用场景选择其一，传给 createAgentSession：

// 选项 1：内存模式（临时，不写入磁盘）
const sessionManager = SessionManager.inMemory();

// 选项 2：在默认会话目录新建持久会话
const sessionManager = SessionManager.create(process.cwd());

// 选项 3：打开指定的会话文件
const sessionManager = SessionManager.open("/path/to/session.jsonl");

// 选项 4：继续最近的会话（如果不存在则新建）
const sessionManager = SessionManager.continueRecent(process.cwd());

// 然后把你选定的那个传进去：
const { session } = await createAgentSession({ model, sessionManager });

你也可以列出一个目录下已有的会话：

const sessions = await SessionManager.list(process.cwd());

拿到 SessionManager 后，你很少需要直接调用它的方法 —— createAgentSession 已经处理了大部分接线工作。但如果你在构建自定义的会话逻辑（例如多通道路由），以下这些是关键方法：

// 从 JSONL 文件重建对话。
// 当需要在 agent 会话之外查看或展示当前对话时使用
// （例如在 Web UI 中展示历史记录）。
const { messages, thinkingLevel, model } = sessionManager.buildSessionContext();

// 获取当前分支的最后一条条目。
// 用于检查最近一条消息是什么，
// 或获取一个条目 ID 以便从这里分叉。
const leaf = sessionManager.getLeafEntry();

// 从某个指定位置分叉对话。
// entryId 之后的所有内容都会被舍弃（但仍保留在文件中）。
// agent 会在下一次提示时从该位置继续。
// 用于“从这里重试”的流程。
sessionManager.branch(entryId);

// 手动向会话记录追加一条消息。
// createAgentSession 会在 prompt() 期间自动这样做，
// 但你可以用它来程序化地注入消息——
// 例如添加系统通知或定时任务触发的提示。
sessionManager.appendMessage(message);

// 获取会话的完整树结构。
// 每个节点都有 children，因此你可以渲染一个分支选择器，
// 或让用户浏览对话历史。
const tree = sessionManager.getTree();

一种常见做法是每个对话线程使用一个会话文件 —— 例如 ~/.myapp/agents/<agentId>/sessions/<sessionId>.jsonl —— 因此每个对话都是独立的、崩溃安全的（JSONL 是追加写入；崩溃时最多损失一行）。

使用工具工厂

createAgentSession 会把启用的内置工具绑定到它的 cwd，因此会话会自然地在你提供的工作目录上操作：

const { session } = await createAgentSession({
  cwd: "/path/to/workspace",
  tools: ["read", "bash", "edit", "write"],
});

当你需要直接拿到工具数组时 —— 比如用于 pi-agent-core 里的自定义 Agent，或者手工组装服务层 —— 使用这些工厂函数：

import {
  createCodingTools,
  createReadOnlyTools,
  createReadTool,
  createBashTool,
  createGrepTool,
} from "@earendil-works/pi-coding-agent";

// 创建限定在工作区内的预设工具组
const customCodingTools = createCodingTools("/path/to/workspace");       // [read, bash, edit, write]
const customReadOnlyTools = createReadOnlyTools("/path/to/workspace");   // [read, grep, find, ls]

// 或单独创建每个工具——每个内置工具都有一个对应的工厂函数
const customRead = createReadTool("/path/to/workspace");
const customBash = createBashTool("/path/to/workspace");
const customGrep = createGrepTool("/path/to/workspace");

每个工厂都接受一个可选的 operations 对象来覆盖底层 I/O —— 如果你想在 Docker 容器内、通过 SSH、或对着虚拟文件系统运行工具，这会很有用：

// 从远程服务器而不是本地磁盘读取文件
const remoteRead = createReadTool("/workspace", {
  operations: {
    readFile: async (path) => fetchFileFromRemote(path),
    access: async (path) => checkRemoteFileExists(path),
  },
});

// 在 Docker 沙箱而不是宿主机上执行命令
const sandboxedBash = createBashTool("/workspace", {
  operations: {
    exec: async (command, cwd, opts) => runInDockerContainer(command, cwd, opts),
  },
});

你可以用这些工厂为每个 agent 创建限定在工作区内的工具，再给它们套上额外的中间件 —— 权限检查、read 工具的图片归一化，以及 Claude Code 参数兼容别名（file_path → path、old_string → oldText）。

自定义工具与内置工具并存

内置工具覆盖了文件操作和 shell 命令。

其他任何需求 —— 部署、调用 API、查询数据库 —— 都可以定义你自己的工具，并通过 customTools 传入。它们会和默认工具一起可用：

import { createAgentSession, SessionManager, defineTool } from "@earendil-works/pi-coding-agent";
import { Type } from "@earendil-works/pi-ai";

const deployParams = Type.Object({
  environment: Type.String({ description: "Target environment", default: "staging" }),
});

const deployTool = defineTool({
  name: "deploy",
  label: "Deploy",
  description: "Deploy the application to production",
  parameters: deployParams,
  execute: async (_id, params, signal, onUpdate) => {
    onUpdate?.({
      content: [{ type: "text", text: `Deploying to ${params.environment}...` }],
      details: {},
    });

    // 在这里写自定义逻辑——调用 API、运行脚本、触发 CI 流水线等
    await new Promise((resolve) => setTimeout(resolve, 2000));

    return {
      content: [{ type: "text", text: `Deployed to ${params.environment} successfully.` }],
      details: { environment: params.environment, timestamp: Date.now() },
    };
  },
});

const { session } = await createAgentSession({
  model,
  customTools: [deployTool],
  sessionManager: SessionManager.inMemory(),
});

现在 agent 同时拥有 read、write、edit、bash 和 deploy。

压缩（Compaction）

长对话会超出模型的上下文窗口。pi-coding-agent 通过压缩来处理 —— 总结旧消息，保留最近的消息：

import { estimateTokens } from "@earendil-works/pi-coding-agent";

// 检查对话占用了多少 token
const totalTokens = session.messages.reduce(
  (sum, msg) => sum + estimateTokens(msg),
  0
);

// 手动触发压缩——可选字符串用于指导摘要应保留什么
if (totalTokens > 100_000) {
  await session.compact("Preserve all file paths and code changes.");
}

默认情况下，createAgentSession 会启用自动压缩 —— 当上下文接近模型的窗口上限时自动触发。完整的消息历史仍保留在 JSONL 文件中；只有内存中的上下文会被压缩。

扩展（Extensions）

工具让 LLM 做事情。扩展让你在 LLM 不知情的情况下修改 agent 的行为方式。它们挂接在 agent 循环中触发的生命周期事件上：消息发送给 LLM 之前、压缩运行之前、工具被调用时、会话启动时。LLM 永远不会在其上下文中看到扩展；它们在幕后运行。

你可以在这里放这类逻辑：裁剪旧的工具结果让上下文窗口保持专注、用自定义摘要流水线替换默认压缩、按权限拦截工具调用、或根据对话当前状态注入额外上下文。

扩展是一个 TypeScript 模块，导出一个接收 ExtensionAPI 的函数：

import type { ExtensionAPI } from "@earendil-works/pi-coding-agent";

export default function myExtension(api: ExtensionAPI): void {
  // 在每次 LLM 调用前触发。允许你重写消息数组。
  api.on("context", (event, ctx) => {
    const pruned = event.messages.filter((msg) => {
      // 丢弃超过 10 条消息之前的超大工具结果
      if (msg.role === "toolResult" && event.messages.indexOf(msg) < event.messages.length - 10) {
        const text = msg.content.map((c) => (c.type === "text" ? c.text : "")).join("");
        if (text.length > 5000) return false;
      }
      return true;
    });
    return { messages: pruned };
  });

  // 用你自己的摘要逻辑替换默认压缩
  api.on("session_before_compact", async (event, ctx) => {
    const summary = await myCustomSummarize(event.preparation.messagesToSummarize);
    return {
      compaction: {
        summary,
        firstKeptEntryId: event.preparation.firstKeptEntryId,
        tokensBefore: event.preparation.tokensBefore,
      },
    };
  });

  // 注册一个面向用户的命令（不是 LLM 工具）
  api.registerCommand("stats", {
    description: "Show session statistics",
    handler: async (_args, ctx) => {
      const usage = ctx.getContextUsage();
      console.log(
        `Messages: ${ctx.sessionManager.getEntries().length}, ` +
        `Context: ${usage?.tokens ?? "?"}/${usage?.contextWindow ?? "?"} tokens`,
      );
    },
  });
}

常用的扩展事件：session_start / session_shutdown（初始化与清理）、before_agent_start（注入上下文或调整提示词）、tool_call（把关/拦截工具调用）、context（每次 LLM 调用前重写消息）、session_before_compact（自定义摘要），以及 turn_start / turn_end / agent_settled（响应 agent 进度）。

常见模式：在 tool_call 里把关危险命令（permission-gate.ts、protected-paths.ts），用 before_agent_start 注入上下文（prompt-customizer.ts、claude-rules.ts），在 session_before_compact 里自定义压缩（custom-compaction.ts），用 user_bash 包装 ! 命令（ssh.ts）。

生产环境可以用扩展做上下文裁剪（静默裁剪超大的工具结果以节省 token）和压缩保护（用多阶段流水线替换 pi 的默认摘要，保留文件操作历史和工具失败数据）。

构建真实可用的助手

下面是一个把三层组合起来的完整示例：一个代码库助手，能读你的项目、回答问题、修改代码，并在重启后仍记得对话。

创建 assistant.ts：

import {
  createAgentSession,
  SessionManager,
  estimateTokens,
  type AgentSession,
} from "@earendil-works/pi-coding-agent";
import { builtinModels } from "@earendil-works/pi-ai/providers/all";
import { Type } from "@earendil-works/pi-ai";
import type { AgentTool } from "@earendil-works/pi-agent-core";
import * as path from "path";
import * as fs from "fs";
import * as readline from "readline";

// --- 自定义工具：搜索网页 ---
const webSearchParams = Type.Object({
  query: Type.String({ description: "Search query" }),
});

const webSearchTool: AgentTool<typeof webSearchParams> = {
  name: "web_search",
  label: "Web Search",
  description: "Search the web for documentation, error messages, or general information",
  parameters: webSearchParams,
  execute: async (_id, params) => {
    // 生产环境中应调用搜索 API（Brave、Serper 等）
    return {
      content: [{ type: "text", text: `[Search results for: "${params.query}" would appear here]` }],
      details: { query: params.query },
    };
  },
};

// --- 会话持久化 ---
const sessionDir = path.join(process.cwd(), ".sessions");
fs.mkdirSync(sessionDir, { recursive: true });

const sessionFile = path.join(sessionDir, "assistant.jsonl");
const sessionManager = SessionManager.open(sessionFile);

// --- 创建 agent 会话 ---
async function createAssistant(): Promise<AgentSession> {
  const models = builtinModels();
  const model = models.getModel("anthropic", "claude-opus-4-5")!;

  const { session } = await createAgentSession({
    model,
    thinkingLevel: "off",
    sessionManager,
    customTools: [webSearchTool],
  });

  return session;
}

// --- 事件处理器 ---
function attachEventHandlers(session: AgentSession) {
  session.subscribe((event) => {
    switch (event.type) {
      case "message_update":
        if (event.assistantMessageEvent.type === "text_delta") {
          process.stdout.write(event.assistantMessageEvent.delta);
        }
        break;

      case "tool_execution_start":
        console.log(`\n  [${event.toolName}] ${summarizeArgs(event.args)}`);
        break;

      case "tool_execution_end":
        if (event.isError) {
          console.log(`  ERROR`);
        }
        break;

      case "compaction_start":
        console.log("\n  [compacting context...]");
        break;

      case "agent_end":
        console.log();
        break;
    }
  });
}

function summarizeArgs(args: any): string {
  if (args?.path) return args.path;
  if (args?.command) return args.command.slice(0, 60);
  if (args?.query) return `"${args.query}"`;
  if (args?.pattern) return args.pattern;
  return JSON.stringify(args).slice(0, 60);
}

// --- REPL ---
async function main() {
  let session = await createAssistant();
  attachEventHandlers(session);

  const tokenCount = session.messages.reduce((sum, msg) => sum + estimateTokens(msg), 0);

  console.log("PI Assistant");
  console.log(`  Model: ${session.model?.id}`);
  console.log(`  Session: ${sessionFile}`);
  console.log(`  History: ${session.messages.length} messages, ~${tokenCount} tokens`);
  console.log(`  Tools: ${session.getActiveToolNames().join(", ")}`);
  console.log(`  Type "exit" to quit, "new" to reset session\n`);

  const rl = readline.createInterface({ input: process.stdin, output: process.stdout });

  const ask = () => {
    rl.question("You: ", async (input) => {
      const trimmed = input.trim();

      if (trimmed === "exit") {
        session.dispose();
        rl.close();
        return;
      }

      if (trimmed === "new") {
        session.dispose();
        session = await createAssistant();
        attachEventHandlers(session);
        console.log("Session reset.\n");
        ask();
        return;
      }

      if (!trimmed) {
        ask();
        return;
      }

      try {
        await session.prompt(trimmed);
      } catch (err: any) {
        console.error(`Error: ${err.message}`);
      }

      ask();
    });
  };

  ask();
}

main();

执行：

npx tsx assistant.ts

这大约 120 行代码就能给你一个持久化的编码助手。它能读文件、运行命令、修改代码、搜索网络，并且跨重启记住你的对话。即使经过压缩，JSONL 文件中的会话树仍保留完整历史。

一次会话看起来像这样：

PI Assistant
  Model: claude-opus-4-5
  Session: /your/project/.sessions/assistant.jsonl
  History: 0 messages, ~0 tokens
  Tools: read, bash, edit, write, web_search

You: What does this project do? Look at the README and main entry point.

  [read] README.md
  [read] src/index.ts

This is a TypeScript library that...

You: Find all TODO comments in the source code.

  [bash] grep -rn "TODO" src/

Found 3 TODOs:
- src/auth.ts:42 - TODO: add token refresh
- src/api.ts:18 - TODO: handle rate limits
- src/index.ts:7 - TODO: add graceful shutdown

You: Fix the token refresh TODO. Implement a proper refresh flow.

  [read] src/auth.ts
  [edit] src/auth.ts

Done. Added a `refreshToken()` function that...

为生产环境适配

生产环境在此模式之上还会加上以下这些层：

多提供商认证。 与其使用单一的 ANTHROPIC_API_KEY，你可以用 ModelRuntime 管理跨提供商的凭据并支持 OAuth 流程：

import { ModelRuntime, createAgentSession } from "@earendil-works/pi-coding-agent";
import * as path from "path";

const modelRuntime = await ModelRuntime.create({
  authPath: path.join(agentDir, "auth.json"),
  modelsPath: path.join(agentDir, "models.json"),
});

const { session } = await createAgentSession({
  modelRuntime,
  model: modelRuntime.getModel("ollama", "llama3.1:8b"),
  // ...
});

ModelRuntime 从 auth.json 文件读取凭据 —— 一个以提供商名称为键的扁平对象，每个值要么是 API key，要么是 OAuth 凭据：

{
  "anthropic": { "type": "api_key", "key": "sk-ant-..." },
  "openai": { "type": "api_key", "key": "sk-..." },
  "devin": { "type": "api_key", "key": "cog_..." },
  "github-copilot": {
    "type": "oauth",
    "refresh": "gho_xxxxxxxxxxxx",
    "access": "ghu_yyyyyyyyyyyy",
    "expires": 1700000000000
  }
}

key 字段可以是字面值、环境变量引用（${OPENAI_API_KEY}），或以 ! 开头的 shell 命令（例如用于 1Password 的 "!op read 'op://vault/openai/key'"）。OAuth token 过期时会自动刷新。

ModelRuntime 还会读取 models.json 文件，其中定义了自定义提供商和模型。这样就能添加 pi 内置之外的自托管模型或提供商：

{
  "providers": {
    "ollama": {
      "baseUrl": "http://localhost:11434/v1",
      "api": "openai-completions",
      "apiKey": "ollama",
      "models": [
        { "id": "llama3.1:8b" },
        { "id": "qwen2.5-coder:7b" }
      ]
    },
    "my-company-api": {
      "baseUrl": "https://llm.internal.company.com/v1",
      "api": "openai-completions",
      "apiKey": "COMPANY_LLM_KEY",
      "authHeader": true,
      "models": [
        { "id": "internal-model-v2" }
      ]
    }
  }
}

这里定义的模型会和内置目录一起出现。modelRuntime.getModel("ollama", "llama3.1:8b") 返回一个类型完整的 Model，可以直接传给 createAgentSession。

提供商请求头。 每个 AgentSession 都通过它的 ModelRuntime 进行流式通信，由后者组装提供商认证和请求头。要给提供商请求添加额外的请求头，可以在扩展里钩住 before_provider_headers，在原处修改请求头：

api.on("before_provider_headers", (event) => {
  event.headers["X-Title"] = "My App";
  event.headers["HTTP-Referer"] = "https://myapp.com";
});

你也可以在 models.json 中某个提供商的 headers 字段下设置静态请求头。

工具定制。 默认内置工具在会话的 cwd 上操作：

const { session } = await createAgentSession({
  cwd: workspace,
  tools: ["read", "bash", "edit", "write"],
});

在多用户产品中，可以把每个 agent 会话锁定在特定的工作区目录，使用户无法读写其项目之外的内容。要更细粒度地控制，可以用工厂重建各个工具：

import {
  createReadTool,
  createWriteTool,
  createEditTool,
  createBashTool,
} from "@earendil-works/pi-coding-agent";

const tools = [
  createReadTool(workspace),
  createBashTool(workspace),
  createWriteTool(workspace),
  createEditTool(workspace),
];

事件路由。 agent 运行时会产生事件 —— 文本 token 一个个流入、工具调用开始和结束、agent 完成本轮。在终端应用里，你只需把它们打印到 stdout。但如果你是替通过 Telegram、Discord 或 Slack 聊天的用户运行 agent，就需要把这些事件翻译成各平台特定的消息。session.subscribe() 为每个事件提供回调，由你决定如何处理：

session.subscribe((event) => {
  switch (event.type) {
    case "message_update":
      if (event.assistantMessageEvent.type === "text_delta") {
        // token 逐个到达——先缓冲，然后作为一条消息发送
        messageBuffer.append(event.assistantMessageEvent.delta);
      }
      break;

    case "tool_execution_start":
      // 向频道发送工具调用通知
      channel.sendNotification(`Running ${event.toolName}...`);
      break;

    case "agent_end":
      // 刷新剩余缓冲的文本
      messageBuffer.flush();
      break;
  }
});

添加终端 UI

assistant.ts 示例用 readline 做输入 —— 它能用，但没有 Markdown 渲染、没有自动补全，流式输出也只是裸的 process.stdout.write。pi-tui 用一个真正的终端 UI 替换掉这一切：带语法高亮的 Markdown、带斜杠命令和文件路径自动补全的编辑器、加载动画，以及无闪烁的差分渲染。

下面是同一个助手升级到 pi-tui 的版本。创建 assistant-tui.ts：

import {
  createAgentSession,
  SessionManager,
  estimateTokens,
  type AgentSession,
} from "@earendil-works/pi-coding-agent";
import { builtinModels } from "@earendil-works/pi-ai/providers/all";
import { Type } from "@earendil-works/pi-ai";
import type { AgentTool } from "@earendil-works/pi-agent-core";
import {
  TuiMainScreen,
  ProcessTerminal,
  Editor,
  Markdown,
  Text,
  Loader,
  CombinedAutocompleteProvider,
} from "@earendil-works/pi-tui";
import type { EditorTheme, MarkdownTheme } from "@earendil-works/pi-tui";
import chalk from "chalk";
import * as path from "path";
import * as fs from "fs";

// --- 主题样式 ---
const markdownTheme: MarkdownTheme = {
  heading: (s) => chalk.bold.cyan(s),
  link: (s) => chalk.blue(s),
  linkUrl: (s) => chalk.dim(s),
  code: (s) => chalk.yellow(s),
  codeBlock: (s) => chalk.green(s),
  codeBlockBorder: (s) => chalk.dim(s),
  quote: (s) => chalk.italic(s),
  quoteBorder: (s) => chalk.dim(s),
  hr: (s) => chalk.dim(s),
  listBullet: (s) => chalk.cyan(s),
  bold: (s) => chalk.bold(s),
  italic: (s) => chalk.italic(s),
  strikethrough: (s) => chalk.strikethrough(s),
  underline: (s) => chalk.underline(s),
};

const editorTheme: EditorTheme = {
  borderColor: (s) => chalk.dim(s),
  selectList: {
    selectedPrefix: (s) => chalk.blue(s),
    selectedText: (s) => chalk.bold(s),
    description: (s) => chalk.dim(s),
    scrollInfo: (s) => chalk.dim(s),
    noMatch: (s) => chalk.dim(s),
  },
};

// --- 自定义工具 ---
const webSearchParams = Type.Object({
  query: Type.String({ description: "Search query" }),
});

const webSearchTool: AgentTool<typeof webSearchParams> = {
  name: "web_search",
  label: "Web Search",
  description: "Search the web for documentation, error messages, or general information",
  parameters: webSearchParams,
  execute: async (_id, params) => ({
    content: [{ type: "text", text: `[Search results for: "${params.query}" would appear here]` }],
    details: { query: params.query },
  }),
};

// --- 会话持久化 ---
const sessionDir = path.join(process.cwd(), ".sessions");
fs.mkdirSync(sessionDir, { recursive: true });
const sessionFile = path.join(sessionDir, "assistant.jsonl");
const sessionManager = SessionManager.open(sessionFile);

// --- TUI 设置 ---
const tui = new TuiMainScreen(new ProcessTerminal());

tui.addChild(new Text(chalk.bold("PI Assistant") + chalk.dim(" (Ctrl+C to exit)\n")));

const editor = new Editor(tui, editorTheme);
editor.setAutocompleteProvider(
  new CombinedAutocompleteProvider(
    [
      { name: "new", description: "Reset the session" },
      { name: "exit", description: "Quit the assistant" },
    ],
    process.cwd(),
  ),
);
tui.addChild(editor);
tui.setFocus(editor);

async function main() {
  const models = builtinModels();
  const model = models.getModel("anthropic", "claude-opus-4-5")!;

  let session: AgentSession;

  // 会话信息是一个专用组件，原地更新。
  const infoText = new Text("");
  tui.children.splice(tui.children.length - 1, 0, infoText);

  const refreshSessionInfo = (currentSession: AgentSession) => {
    const tokenCount = currentSession.messages.reduce((sum, msg) => sum + estimateTokens(msg), 0);
    infoText.setText(
      chalk.dim(`  Model: ${model.id}\n`) +
      chalk.dim(`  Session: ${sessionFile}\n`) +
      chalk.dim(`  History: ${currentSession.messages.length} messages, ~${tokenCount} tokens\n`) +
      chalk.dim(`  Tools: ${currentSession.getActiveToolNames().join(", ")}\n`),
    );
    tui.requestRender();
  };

  const clearChat = () => {
    // 保留标题和信息文本；编辑器是最后一个子组件。
    const editorIndex = tui.children.indexOf(editor);
    tui.children.splice(2, editorIndex - 2);
  };

  const attachSessionEvents = (currentSession: AgentSession) => {
    refreshSessionInfo(currentSession);

    // 流式状态
    let streamingMarkdown: Markdown | null = null;
    let streamingText = "";
    let loader: Loader | null = null;

    currentSession.subscribe((event) => {
      const children = tui.children;

      switch (event.type) {
        case "agent_start":
          editor.disableSubmit = true;
          loader = new Loader(tui, (s) => chalk.cyan(s), (s) => chalk.dim(s), "Thinking...");
          children.splice(children.length - 1, 0, loader);
          tui.requestRender();
          break;

        case "message_update":
          if (event.assistantMessageEvent.type === "text_delta") {
            // 收到第一段文本时移除加载动画
            if (loader) {
              tui.removeChild(loader);
              loader = null;
            }
            // 创建或更新流式 markdown 组件
            streamingText += event.assistantMessageEvent.delta;
            if (!streamingMarkdown) {
              streamingMarkdown = new Markdown(streamingText, 1, 0, markdownTheme);
              children.splice(children.length - 1, 0, streamingMarkdown);
            } else {
              streamingMarkdown.setText(streamingText);
            }
            tui.requestRender();
          }
          break;

        case "tool_execution_start": {
          if (loader) {
            tui.removeChild(loader);
            loader = null;
          }
          const args = event.args?.path || event.args?.command?.slice(0, 60) || event.args?.query || "";
          const toolMsg = new Text(chalk.dim(`  [${event.toolName}] ${args}`));
          children.splice(children.length - 1, 0, toolMsg);
          tui.requestRender();
          break;
        }

        case "agent_end":
          if (loader) {
            tui.removeChild(loader);
            loader = null;
          }
          streamingMarkdown = null;
          streamingText = "";
          editor.disableSubmit = false;
          tui.requestRender();
          break;
      }
    });
  };

  const { session: initialSession } = await createAgentSession({
    model,
    thinkingLevel: "off",
    sessionManager,
    customTools: [webSearchTool],
  });
  session = initialSession;
  attachSessionEvents(session);

  // 处理输入提交
  editor.onSubmit = async (value: string) => {
    if (editor.disableSubmit) return;
    const trimmed = value.trim();
    if (!trimmed) return;

    if (trimmed === "/exit") {
      session.dispose();
      tui.stop();
      process.exit(0);
    }

    if (trimmed === "/new") {
      session.dispose();
      const { session: freshSession } = await createAgentSession({
        model,
        thinkingLevel: "off",
        sessionManager,
        customTools: [webSearchTool],
      });
      session = freshSession;
      clearChat();
      attachSessionEvents(session);
      return;
    }

    // 将用户消息添加到聊天
    const userMsg = new Markdown(value, 1, 0, markdownTheme, (s) => chalk.bold(s));
    tui.children.splice(tui.children.length - 1, 0, userMsg);
    tui.requestRender();

    // 发送给 agent
    try {
      await session.prompt(trimmed);
    } catch (err: any) {
      tui.children.splice(tui.children.length - 1, 0, new Text(chalk.red(`Error: ${err.message}`)));
      editor.disableSubmit = false;
      tui.requestRender();
    }
  };

  tui.start();
}

main();

执行：

npx tsx assistant-tui.ts

相比 readline 版本的优势：

• Markdown 渲染。 agent 回复以带语法高亮的代码块、粗体、斜体、列表和链接渲染出来 —— 而不是把原始文本直接倒到 stdout。
• 通过 setText 流式更新。 每当 token 到达，我们把它追加到字符串，并调用 streamingMarkdown.setText()。TUI 的差分渲染器只更新变化的行 —— 不闪烁，也不清屏。
• 带自动补全的编辑器。 输入 / 会弹出斜杠命令下拉列表。按 Tab 补全文件路径。用 Shift+Enter 换行输入多行内容。
• 加载动画。 Loader 组件在 agent 思考时显示动画加载条，文本开始流出时自动移除。
• 无需手动管理光标。 TUI 处理终端状态、光标定位和清理。事件处理器里不再到处散布 process.stdout.write 调用。

架构是一样的 —— createAgentSession + session.subscribe() + session.prompt()。唯一区别在于如何渲染事件：不是写 stdout，而是在 TUI 的组件树里添加和更新 Markdown、Text 和 Loader 组件。

最后

本指南涵盖了构建终端 agent 所需的几个核心的包。pi monorepo 还提供了几个相关的的包：