TS·08 实战总结:读懂 pi 的一个真实工具与 agent 主循环

前八篇是零件。我们一个一个地拆过:值与函数、interface 与 type、联合与可辨识联合、窄化与类型守卫、泛型、类与模块、异步与流式。每一篇都在一个孤立的语法点上打转,你可能会想:这些零件拼起来到底长什么样?

这篇是整机。我们不再介绍新语法——而是把零件拼成一台能跑的机器,然后用它去读 pi 的真实代码。目标很具体:读完这篇,你能独立打开 pi 的仓库,读懂它的一个内置工具是怎么定义的、agent 主循环是怎么把"模型吐出的 token"变成"真正执行的工具调用"的。我们会走两条主线:第一条是「一个工具从 schema 到处理器」的静态骨架(第 1、2 章),第二条是「主循环把一切串起来」的动态流程(第 3 章),最后第 4 章给你一张自学地图,让你合上这篇之后能自己继续读下去。

一句话预告全篇的主角:一份 schema,活两遍。它编译期被 Static<typeof schema> 投影成静态类型(给你的代码看),运行时又被 validateToolArguments 当校验器(给模型的输出把关)。你会在第 1 章看它派生类型,在第 3 章看它校验数据——同一个值,两个世界。抓住这条线,pi 的工具体系就不再是黑箱。

每一章仍走 5 步螺旋。但这次 Step 4 引用的都是 pi 里你真的会读到的代码,Step 2 的手写 demo 只是那段真实代码的最小骨架。读的时候不妨对照着 clone 里的文件一起看。

1. 一个工具是怎么定义的:schema → 类型 → 处理器

1.1 直觉

pi 里的"工具"(tool)就是模型能调用的一个函数——读文件、跑 bash、算数。但模型是通过 JSON 调用它的,所以工具必须回答两个问题:参数长什么样(给模型看的运行时 schema),以及参数在 TS 里是什么类型(给你的代码看的静态类型)。pi 的做法是:先用 typebox 写一份运行时 schema(相当于 Python 的 Pydantic model),再用 Static<typeof schema> 从这份 schema 派生出静态类型(见泛型),最后把 schema 和一个 execute 函数(见值与函数)装进一个对象(见对象形状)。一份 schema,两个用途——这是 pi 全仓库的定式。

1.2 最小 demo

// 教学示例 — 非生产代码
import { type Static, Type } from "typebox";

// 1) 运行时 schema:像 Pydantic,一个真实存在的对象
const greetSchema = Type.Object({
  name: Type.String({ description: "谁要被问候" }),
});

// 2) 从 schema 派生静态类型 —— 不用手写 { name: string }
type GreetParams = Static<typeof greetSchema>;

// 3) schema + 处理器 装进一个工具对象
const greetTool = {
  name: "greet",
  parameters: greetSchema,
  execute: async (args: GreetParams) => {
    return `Hello, ${args.name}!`; // args.name 是 string,编译器知道
  },
};

注意第 2 步:GreetParams 不是手写的,它是编译器从 greetSchema 算出来的 { name: string }。你改 schema,类型自动跟着变——这就是"单一真理来源"。

1.3 正式化

拆开 demo 里的三个新写法:

Python 对照:typebox 之于 TS,约等于 Pydantic 之于 Python——都是"运行时能校验的数据模型"。差别是:Pydantic 里 class M(BaseModel) 同时给你运行时校验和静态类型;typebox 把两者拆开,schema 管运行时,Static<typeof schema> 单独派生静态类型。

1.4 代码引用

这是 pi 里最干净的一个端到端例子——一个计算器工具,schema、派生类型、工具对象三件套齐全:

pi/packages/agent/test/utils/calculate.ts:L18-L32 — schema → Static 派生类型 → 工具对象 三件套

const calculateSchema = Type.Object({
	expression: Type.String({ description: "The mathematical expression to evaluate" }),
});

type CalculateParams = Static<typeof calculateSchema>;

export const calculateTool: AgentTool<typeof calculateSchema, undefined> = {
	label: "Calculator",
	name: "calculate",
	description: "Evaluate mathematical expressions",
	parameters: calculateSchema,
	execute: async (_toolCallId: string, args: CalculateParams) => {
		return calculate(args.expression);
	},
};

对照本节:calculateSchema 就是 demo 的第 1 步(Type.Object);type CalculateParams = Static<typeof calculateSchema> 就是第 2 步(派生类型,而非手写 { expression: string });calculateTool 就是第 3 步(schema 装进 parameters,处理器装进 execute)。execute 里的 args: CalculateParams 让 args.expression 被编译器认作 string——所以 calculate(args.expression) 这行不会报类型错。第一个参数 _toolCallId 前面的下划线是 pi 的惯例:表示"这个参数签名要求有,但本函数用不上",下划线让 linter 闭嘴,也给读者一个信号。

真实的内置工具会更花哨一点,但骨架一模一样。看 read 工具的 schema,它用 Type.Optional(...) 标出可选参数:

pi/packages/coding-agent/src/core/tools/read.ts:L20-L26 — 带 Type.Optional 的 schema 与派生类型

const readSchema = Type.Object({
	path: Type.String({ description: "Path to the file to read (relative or absolute)" }),
	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" })),
});

export type ReadToolInput = Static<typeof readSchema>;

Type.Optional(Type.Number(...)) 派生出的类型是 offset?: number——即对象形状里讲过的可选属性。同一套 Static<typeof ...> 定式,Type.Optional 自动变成 ?:。你不需要在别处再声明一遍"offset 可以不传":schema 里写了 Type.Optional,静态类型、运行时校验、给模型的 JSON schema 三处同时知道它可选。这就是"单一真理来源"落到实处的样子——改一处,三处同步,没有第二个地方能和它不一致。

1.5 洞察

2. AgentTool 契约与 execute 的类型

2.1 直觉

第 1 章的 calculateTool 标了个类型注解:AgentTool<typeof calculateSchema, undefined>。这个 AgentTool 就是 pi 给所有工具定的契约(见接口契约):任何工具都必须有 name、description、parameters,还必须有一个 execute 函数。它是个泛型接口(见泛型),两个类型参数分别是"参数 schema 的类型"和"结果里 details 的类型"。它还继承了一个更基础的 Tool 接口。读懂这个契约,你就读懂了 pi 里每一个工具的"外形"——read、bash、edit、calculate,外形全都一样,只是 schema 和 execute 的内容不同。契约的价值就在这:主循环不需要认识任何具体工具,只要对方满足 AgentTool,它就能被查找、校验、执行。

2.2 最小 demo

// 教学示例 — 非生产代码
import { type Static, type TSchema } from "typebox";

// 基础契约:任何工具都有 name 和 parameters
interface Tool<TParameters extends TSchema = TSchema> {
  name: string;
  parameters: TParameters;
}

// 扩展契约:再加一个 execute;两个泛型参数
interface MyTool<TParams extends TSchema, TDetails> extends Tool<TParams> {
  execute: (
    toolCallId: string,
    params: Static<TParams>,        // 参数类型从 schema 派生
    signal?: AbortSignal,            // 可选:用于取消
  ) => Promise<{ details: TDetails }>;
}

extends Tool<TParams> 表示 MyTool 在 Tool 的基础上追加字段——继承接口。execute 的第二个参数 Static<TParams> 又用了第 1 章的派生定式,只是这次 schema 的类型是泛型参数 TParams。

2.3 正式化

逐个拆解契约里的类型机制:

2.4 代码引用

先看基础契约 Tool——极简,只三个字段,是所有工具的最小公分母:

pi/packages/ai/src/types.ts:L440-L444 — Tool 基础接口:带约束+默认值的泛型接口

export interface Tool<TParameters extends TSchema = TSchema> {
	name: string;
	description: string;
	parameters: TParameters;
}

再看 AgentTool 如何在它之上追加 label、execute 等,构成 agent 运行时用的完整契约:

pi/packages/agent/src/types.ts:L373-L396 — AgentTool 继承 Tool,execute 的完整签名

export interface AgentTool<TParameters extends TSchema = TSchema, TDetails = any> extends Tool<TParameters> {
	/** Human-readable label for UI display. */
	label: string;
	/**
	 * Optional compatibility shim for raw tool-call arguments before schema validation.
	 * Must return an object that matches `TParameters`.
	 */
	prepareArguments?: (args: unknown) => Static<TParameters>;
	/** Execute the tool call. Throw on failure instead of encoding errors in `content`. */
	execute: (
		toolCallId: string,
		params: Static<TParameters>,
		signal?: AbortSignal,
		onUpdate?: AgentToolUpdateCallback<TDetails>,
	) => Promise<AgentToolResult<TDetails>>;
	// ... executionMode?: ToolExecutionMode
}

对照本节:extends Tool<TParameters> 就是 demo 里的接口继承;两个泛型参数 TParameters(约束到 TSchema)和 TDetails 对应 demo 的 TParams/TDetails;execute 的 params: Static<TParameters> 正是第 1 章派生定式的复用——工具作者写 schema,execute 的参数类型就自动对上。signal?: AbortSignal 是可选的取消钩子。还有个 prepareArguments?,参数类型是 unknown(见窄化)——因为它拿的是模型吐出的、还没校验的原始参数,只能是 unknown,得先校验/收窄才能用。

2.5 洞察

3. 主循环:流式消费事件 + 分派工具 + 校验参数

3.1 直觉

工具是零件,主循环是发动机。一轮 agent 交互的链条是:模型吐 token → 变成类型化事件流 → 攒成工具调用 → 校验参数 → 执行工具 → 把结果回填进对话 → 再问模型。这条链上,前八篇的语法几乎全用上了:for await 消费异步流、switch(event.type) 对可辨识联合窄化、schema 校验模型给的参数、Promise.all 并行执行工具。这一章我们把这条链走通,你会看到零件是怎么咬合的。可以把它当成"总复习":每读到一处,回想它对应前面哪一篇,你会发现主循环没有一处是新东西,全是拼装。

3.2 最小 demo

// 教学示例 — 非生产代码
type ModelEvent =
  | { type: "text_delta"; text: string }
  | { type: "toolcall_end"; name: string; args: unknown }
  | { type: "done" };

async function runTurn(stream: AsyncIterable<ModelEvent>) {
  const calls: { name: string; args: unknown }[] = [];
  // 1) 流式消费事件,按 type 窄化
  for await (const event of stream) {
    switch (event.type) {
      case "text_delta":
        process.stdout.write(event.text); // 这里 event 被窄化成有 text
        break;
      case "toolcall_end":
        calls.push({ name: event.name, args: event.args });
        break;
      case "done":
        break;
    }
  }
  // 2) 并行执行攒下来的工具调用
  const results = await Promise.all(calls.map((c) => runTool(c.name, c.args)));
  return results; // 回填给下一轮
}

这台"玩具发动机"就是 pi 主循环的骨架:for await + switch 把无类型的 token 流整理成有类型的事件,toolcall_end 攒进 calls,最后 Promise.all 并行跑。真实的 pi 只是每一步都更严谨。

3.3 正式化

3.4 代码引用

第一段:流式消费。 pi 主循环里,response 是模型返回的事件流,for await + switch 把它整理成生命周期事件:

pi/packages/agent/src/agent-loop.ts:L319-L346 — for await 消费事件流,switch 按 type 窄化并转发

	for await (const event of response) {
		switch (event.type) {
			case "start":
				partialMessage = event.partial;
				context.messages.push(partialMessage);
				addedPartial = true;
				await emit({ type: "message_start", message: { ...partialMessage } });
				break;

			case "text_start":
			case "text_delta":
			case "text_end":
			case "thinking_start":
			case "thinking_delta":
			case "thinking_end":
			case "toolcall_start":
			case "toolcall_delta":
			case "toolcall_end":
				if (partialMessage) {
					partialMessage = event.partial;
					context.messages[context.messages.length - 1] = partialMessage;
					await emit({
						type: "message_update",
						assistantMessageEvent: event,
						message: { ...partialMessage },
					});
				}
				break;

对照本节:for await (const event of response) 就是 demo 的流式消费;switch (event.type) 对 AssistantMessageEvent 这个可辨识联合窄化。注意多个 case 叠在一起(fall-through)——text_*/thinking_*/toolcall_* 走同一段处理,把 event.partial 更新进消息,再 emit 出去给上层订阅。start 单独一个 case,因为它要新建消息而非更新。

第二段:校验参数。 模型攒出一个工具调用后,主循环先按名字找到工具,再用 schema 校验模型给的参数:

pi/packages/agent/src/agent-loop.ts:L619-L620 — 找到工具后,用 schema 校验模型给的原始参数

		const preparedToolCall = prepareToolCallArguments(tool, toolCall);
		const validatedArgs = validateToolArguments(tool, preparedToolCall);

validateToolArguments 就是那道信任边界的关卡,它拿工具的 typebox schema 去校验/强转模型给的 JSON:

pi/packages/ai/src/utils/validation.ts:L278-L310 — 用 typebox schema 校验模型参数,失败即 throw

export function validateToolArguments(tool: Tool, toolCall: ToolCall): any {
	const args = structuredClone(toolCall.arguments);
	Value.Convert(tool.parameters, args);

	const validator = getValidator(tool.parameters);
	// ...
	if (validator.Check(args)) {
		return args;
	}

	const errors =
		validator
			.Errors(args)
			.map((error) => `  - ${formatValidationPath(error)}: ${error.message}`)
			.join("\n") || "Unknown validation error";

	const errorMessage = `Validation failed for tool "${toolCall.name}":\n${errors}\n\nReceived arguments:\n${JSON.stringify(toolCall.arguments, null, 2)}`;

	throw new Error(errorMessage);
}

对照本节:tool.parameters 就是第 1 章那份运行时 schema(现在派上用场了——它运行时真实存在,才能拿来校验);validator.Check(args) 通过就返回窄化后的参数,不通过就攒出可读的错误信息并 throw(呼应第 2 章"靠 throw 报错")。toolCall.arguments 进来时是模型给的原始 JSON(unknown 级别的信任),出去时才是结构合法、可当 Static<TParameters> 用的数据。

第三段:并行执行。 校验通过的工具调用被攒成一批,Promise.all 并行跑:

pi/packages/agent/src/agent-loop.ts:L542-L544 — 一批工具调用并行执行,结果按原顺序返回

	const orderedFinalizedCalls = await Promise.all(
		finalizedCalls.map((entry) => (typeof entry === "function" ? entry() : Promise.resolve(entry))),
	);

对照本节:这正是 demo 里的 Promise.all(calls.map(...))。区别是 pi 里 finalizedCalls 混着两种成员——已经算好的直接结果、和"还没跑的函数"(用 typeof entry === "function" 窄化后调用 entry() 才产生 promise)。全部完成后得到有序结果,再转成 ToolResultMessage 回填进对话,循环进入下一轮向模型提问。整条链就此闭合。

3.5 洞察

4. 收尾:你现在能读 pi 了

4.1 直觉

读一个陌生的大仓库,最忌讳从第一个文件线性往下读。正确姿势是由已知锚点顺藤摸瓜:你现在有两个锚点——「工具三件套」和「主循环链条」。接下来的自学法就是:选一个你好奇的工具(比如 bash 或 edit),用 grep 定位它的 schema,顺着 execute 读它干了什么,再回到主循环看它何时被调用。这一节不教新语法,教你怎么继续自己读。

4.2 最小 demo

在 clone 里定位任意一个工具,三条命令就够:

// 教学示例 — 这是 shell 命令,不是 TS;演示如何在 clone 里定位一个工具
// 1) 找所有工具的 schema 定义(它们都叫 xxxSchema = Type.Object)
//    grep -rn "Schema = Type.Object" packages/coding-agent/src/core/tools/
//
// 2) 锁定某个工具后,看它的工具对象长啥样(名字、execute)
//    grep -n "name:\|execute" packages/coding-agent/src/core/tools/bash.ts
//
// 3) 看主循环在哪调用工具、怎么校验
//    grep -rn "validateToolArguments\|Promise.all" packages/agent/src/agent-loop.ts

第 1 条把你带到"参数长什么样",第 2 条带到"它干什么",第 3 条带回"主循环怎么用它"。三步覆盖一个工具的完整生命。

4.3 正式化

pi 五个包各管一段,速查如下(读代码前先定位到对的包,能省一半力气):

规律:契约在 ai/agent,实现在 coding-agent。 想读"工具怎么定义"去 agent/types.ts;想读"read 工具具体怎么读文件"去 coding-agent/.../tools/read.ts。这个分层不是偶然:agent 包不该知道任何具体工具的存在,它只认 AgentTool 契约;具体工具住在 coding-agent 里,启动时被"注入"进 agent 上下文的 tools 数组。所以你会看到主循环用 tools?.find((t) => t.name === toolCall.name) 按名字查找工具——它从不 import 任何一个具体工具。

4.4 代码引用

每个包的 src/index.ts 是它的入口地图——barrel 文件,一眼看清这个包对外导出什么。agent 包的入口开头就把主循环和 harness 摊在你面前:

pi/packages/agent/src/index.ts:L1-L13 — agent 包的 barrel:一眼看清导出了什么

// Core Agent
export * from "./agent.ts";
// Loop functions
export * from "./agent-loop.ts";
export * from "./harness/agent-harness.ts";
export {
	type BranchPreparation,
	type BranchSummaryDetails,
	type CollectEntriesResult,
	collectEntriesForBranchSummary,
	generateBranchSummary,
	prepareBranchEntries,
} from "./harness/compaction/branch-summarization.ts";

对照本节:export * from "./agent-loop.ts" 告诉你"主循环从这里对外暴露"——第 3 章那段代码就住在这个文件里(见模块的 barrel 惯用法)。读一个陌生包,先读它的 index.ts:export * 是"整份文件都导出",具名 export { ... } 是"只挑这几个",注释(// Core Agent、// Loop functions)就是作者亲手画的分区地图。顺着地图点开对应文件,比从头硬读高效得多。

4.5 洞察

讨论 / Comments

评论托管在本仓库的 GitHub Discussions, 需 GitHub 账号。