第 7 讲工具定义、校验与执行策略

§7.0学习目标

§7.1AgentTool：一个工具的完整定义

在第 6 讲里，工具执行只是循环里的一个黑盒。本讲把这个黑盒拆开。Pi 中工具的最小完整定义是 AgentTool——先用一句话类比：它像一个带 JSON Schema 参数说明书的 SPI 接口实现，模型看到的是「说明书」（name / description / parameters），运行时持有的是「实现」（execute），两者通过 schema 校验绑定参数。

export interface AgentTool<TParameters extends object = object, 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) => AgentToolParams<TParameters>;
  /** JSON Schema of `structuredContent` in successful results. */
  outputSchema?: object;
    execute: (
    toolCallId: string,
    params: AgentToolParams<TParameters>,
    signal?: AbortSignal,
    onUpdate?: AgentToolUpdateCallback<TDetails>,
  ) => Promise<AgentToolResult<TDetails>>;
  /** Recovery policy for an effect whose durable intent exists but whose outcome is unknown. */
  replay?: "never" | "safe";
  /** Per-tool execution mode override. If omitted, the default execution mode applies. */
  executionMode?: ToolExecutionMode;
}

它继承自 pi-ai 的 Tool<TParameters>，因此天然携带模型可见的三件套：name（调用名）、description（给模型看的功能说明）、parameters（参数的 JSON Schema）。Pi 在其上加了运行时专属字段：label 只给 UI 显示、模型看不到；prepareArguments 是校验前的旧参数兼容垫片；outputSchema 声明结构化输出；replay 是持久化运行时的恢复策略标记；executionMode 覆盖全局执行模式。

execute 的四个参数值得逐个记住：toolCallId 用于对齐 toolResult；params 是校验并矫正之后的强类型参数（类型来自 Static<TParameters>）；signal 是中止信号，长任务必须响应；onUpdate 用于流式上报部分结果。（coding-agent 扩展层 pi.registerTool() 的 execute 还有第五个 ctx 上下文参数，提供 cwd 等宿主信息，见官方 extensions.md 的 Custom Tools 示例。）

execute 返回什么：AgentToolResult 六个字段

export interface AgentToolResult<T = JsonValue | undefined> {
  /** Text or image content returned to the model. */
  content: (TextContent | ImageContent)[];
  /** Arbitrary structured details for logs or UI rendering. */
  details: T;
  /** Machine-readable result matching the tool's `outputSchema` … Not sent to the model. */
  structuredContent?: JsonValue;
  /** Usage from the final tool execution itself, if available. */
  usage?: Usage;
  /** Report a failure without throwing. … `details` and `structuredContent` are kept. */
  isError?: boolean;
  /** Hint that the agent should stop after the current tool batch. */
  terminate?: boolean;
}

报错有两条路：throw 或 返回 isError: true。区别在载荷保留：throw 的错误消息会被包装成 createErrorToolResult(message)（content 为错误文本、details 为空对象），原 details / structuredContent 全部丢失；返回 isError: true 则模型同样看到错误标记，但 UI 与脚本仍能拿到结构化数据。官方建议：校验类失败可以 throw；「执行了但结论是失败、且失败细节有价值」时用 isError。

§7.2参数校验：pi-schema 与宽松矫正

模型输出的工具参数只是「一段尽力而为的 JSON」。不同模型、不同 provider 会把数字写成 "42"、把布尔写成 "true"、把数组只给单个值、或在可选属性里塞 null。如果校验器只做严格检查，这些本可挽救的调用都会失败，白白烧掉一轮模型往返。Pi 的解法在 @earendil-works/pi-ai 的 validateToolArguments：先宽松矫正（coerce），再严格校验。

// One compiled validator per schema object: WeakMap 随工具生命周期回收缓存
const validatorCache = new WeakMap<object, CoercingValidator>();

function getValidator(schema: Tool["parameters"]): CoercingValidator {
  const validator = compile(schema, { formats, coerce: true, unresolvedRefs: "reject" });
  validatorCache.set(schema, validator);
  return validator;
}

export function validateToolArguments(tool: Tool, toolCall: ToolCall): any {
  if (toolCall.argumentsError) {
    throw new Error(`Invalid arguments for tool "${toolCall.name}": ${toolCall.argumentsError}`);
  }
  // A copy, so tools that modify their arguments never modify the tool call in the transcript.
  const result = getValidator(tool.parameters).parse(structuredClone(toolCall.arguments));
  if (result.ok) return result.value;

  const errors = result.errors.map((e) => `  - ${formatValidationPath(e.path)}: ${e.message}`).join("\n");
  throw new Error(`Validation failed for tool "${toolCall.name}":\n${errors}\n\nReceived arguments:\n${JSON.stringify(toolCall.arguments, null, 2)}`);
}

四个实现细节：其一，校验器按 schema 对象引用缓存在 WeakMap，工具是长生命周期对象，schema 被回收时缓存随之释放；其二，编译选项带 formats（format 真正参与检查）、coerce: true（启用 parse() 矫正）、unresolvedRefs: "reject"（MCP 服务器悬空 $ref 只让该位置的值失败，而不是整个工具每次调用都崩）；其三，入参先 structuredClone，工具内改参数不会污染 transcript 里的原始 toolCall；其四，失败时抛出的错误消息会被包成错误 toolResult 返回给模型，让模型自我修正。

宽松矫正的三趟规则

pi-schema 的 compile(schema, { coerce: true }) 提供 coerce() 与 parse()。parse(value) = 先矫正再校验，返回 { ok: true, value } 或 { ok: false, errors }。矫正按固定顺序跑三趟：

1. 可选属性丢 null（Optional nulls）：schema 不接受 null 的可选属性里出现 null，直接丢弃该属性，视同未传。
2. 类型转换（Conversions，移植自 TypeBox 1.3.27 的 Value.Convert）：按 JSON Schema 关键字驱动——number 接受数字字符串（"" 转 0）、"true"/"false"、布尔转 1/0；integer 在此基础上截断（"3.5" 与 3.5 都变 3）；boolean 接受任意大小写的 "true"/"false"、"1"/"0"、1/0；array 把单个值包成单元素数组，再逐项转换；object 按 properties / patternProperties / additionalProperties 递归。
3. 类型数组的原始矫正（Ajv coerceTypes 风格）：处理第二趟没碰的类型数组（如 ["integer", "null"]），数值来自数字字符串与布尔，布尔来自 "true"/"false" 与 1/0 等。

三趟有优先级：第二、三趟冲突时第二趟赢（必填 string 收到 null 变 "null" 而不是 ""）。矫正不改原值——只复制变化的路径，无变化时原样返回。

TypeBox schema：能编译，但 JavaScript 专属特性不生效

仓库测试 agent-tool-typebox.test.ts 专门验证了一件事：pi-agent-core 不依赖 TypeBox，但 AgentTool 的参数类型可以从 TypeBox 构建的 schema 推导——Static<typeof parameters> 读的是 TypeBox 的类型形状（~kind: "Object" 等），execute 的 params 因此获得精确的 { command: string; mode?: "fast" | "deep" }：

const parameters: TObject<{ command: TString; mode: TOptional<TUnion<[TLiteral<"fast">, TLiteral<"deep">]>> }> = {
  "~kind": "Object", type: "object",
  properties: {
    command: { "~kind": "String", type: "string" },
    mode: { "~kind": "Union", anyOf: [/* "fast" | "deep" */], "~optional": true },
  },
  required: ["command"],
};

const tool: AgentTool<typeof parameters> = {
  name: "bash", label: "Bash", description: "Runs a command", parameters,
  async execute(_toolCallId, params) {
    expectTypeOf(params).toEqualTypeOf<Params>();   // { command: string; mode?: "fast" | "deep" }
    return { content: [], details: undefined };
  },
};

但边界必须清楚：TypeBox 1.x 的 schema 本身就是 JSON Schema，所以 pi-schema 原样编译；TypeBox 的 JavaScript 专属扩展——refinements、codecs、Function/Symbol 等类型——不是 JSON Schema，不参与校验，也不会出现在发给模型的声明里。Static 对这些类型（Intersect、Cyclic、Ref 等）也只能给出 unknown。换句话说：想让模型「知道」的约束必须写成 JSON Schema 关键字；refine 只约束 TypeScript 编译期。

校验错误的格式：给模型看的修复说明书

失败时错误被格式化成「工具名 + 逐字段错误清单 + 收到的原始参数」。错误路径由 pi-schema 的 JSON Pointer（如 /edits/0/oldText）转成模型更易读的点分路径（edits.0.oldText）。真实测试断言的完整格式如下：

expect(() => validateToolArguments(tool, toolCall)).toThrow(
  'Validation failed for tool "echo":\n  - value: expected number, got string\n\nReceived arguments:\n{\n  "value": "many"\n}',
);
// 缺必填属性时：'  - value: missing'（错误定位在属性本身）

这个错误不会让循环崩溃：prepareToolCall 捕获异常后返回 immediate 错误结果，包装成一条 isError: true 的 toolResult 回填——模型读到字段级错误清单，通常下一轮就会重新发参数。同理，若模型流式输出的参数 JSON 根本无法解析，ToolCall.argumentsError 会记录原因（此时 arguments 为 {}），校验入口直接抛出 Invalid arguments for tool …；若 assistant 消息因输出 token 上限被截断，整批 toolCall 都会走 failToolCallsFromTruncatedMessage，不执行、只产出「参数可能被截断，请重新发起调用」的错误结果。

§7.3执行策略：sequential 与 parallel

一条 assistant 消息可以携带多个 toolCall。怎么执行这「一批」调用？Pi 的答案写在 ToolExecutionMode 的注释里，默认 parallel。先给结论再读源码：

• sequential：每个 toolCall 依次走完「预检 → 执行 → 终检 → 发事件 → 回填消息」，下一个才开始。
• parallel：所有 toolCall 串行预检（发 tool_execution_start、校验参数、跑 beforeToolCall），放行的调用并发执行；每个工具 finalize 后立刻按完成顺序发 tool_execution_end，而 toolResult 消息产物稍后按 assistant 源顺序统一发出。

为什么必须分离？两个原因。第一，事件流面向 UI：慢工具不该挡住快工具的结果上屏，所以 tool_execution_end 在每个工具 finalize 时即时发出。第二，持久化序面向 provider 协议：OpenAI / Anthropic 等协议要求 toolResult 能与 assistant 消息里的 toolCall 一一配对，乱序回填轻则报错、重则破坏缓存前缀与上下文一致性。实现上很优雅：Promise.all 对「返回 promise 的闭包数组」求值，结果数组天然保持源顺序，随后按这个顺序生成消息。

const hasSequentialToolCall = toolCalls.some(
  (tc) => currentContext.tools?.find((t) => t.name === tc.name)?.executionMode === "sequential",
);
if (config.toolExecution === "sequential" || hasSequentialToolCall) {
  return executeToolCallsSequential(currentContext, assistantMessage, toolCalls, config, signal, emit);
}
return executeToolCallsParallel(currentContext, assistantMessage, toolCalls, config, signal, emit);

// executeToolCallsParallel 收尾（节选）：保序的 Promise.all，再按源顺序生成 toolResult 消息
const orderedFinalizedCalls = await Promise.all(
  finalizedCalls.map((entry) => (typeof entry === "function" ? entry() : Promise.resolve(entry))),
);
const messages: ToolResultMessage[] = [];
for (const finalized of orderedFinalizedCalls) {
  const toolResultMessage = createToolResultMessage(finalized);
  await emitToolResultMessage(toolResultMessage, emit);
  messages.push(toolResultMessage);
}

per-tool executionMode：一颗老鼠屎坏一锅粥（故意的）

AgentTool.executionMode 可以覆盖全局配置，但语义是整批降级：批内只要有任何一个 toolCall 命中的工具声明了 executionMode: "sequential"，整批就走串行（上段代码的 hasSequentialToolCall）。测试名说得很直白——"should force sequential execution when one of multiple tools has executionMode=sequential"。为什么不做「混排」？因为并行批次里如果一部分工具彼此有顺序依赖，调度器无法只对「涉及 sequential 工具的子集」排序而不引入死锁风险；一刀切降级到串行，把顺序问题交给最简单的全序解决。对文件变更类工具，coding-agent 还提供 withFileMutationQueue() 按目标文件路径排队（官方 Custom Tools 文档明确：并行是默认，改文件的整个「读-改-写」窗口都要进队列，否则两个工具同轮编辑同一文件会互相覆盖）。

§7.4钩子：beforeToolCall 与 afterToolCall

两个钩子把一次工具调用包成「进站检查 → 执行 → 出站改写」的 AOP 切面，定义在 AgentLoopConfig 上。类比：beforeToolCall 是命令总线的入站 Filter（可拒绝请求），afterToolCall 是出站 ResponseBodyAdvice（可改写响应）。

beforeToolCall：预检与拦截

export interface BeforeToolCallResult {
  block?: boolean;      // true 则不执行，循环产出错误 toolResult
  reason?: string;     // 错误结果中的文本；缺省用默认 blocked 消息
  terminate?: boolean;  // 拦截同时参与批终止投票
}

beforeToolCall?: (context: BeforeToolCallContext, signal?: AbortSignal)
  => Promise<BeforeToolCallResult | undefined>;

// BeforeToolCallContext = { assistantMessage, toolCall, args, context }
// 注意：args 已经是校验矫正后的参数；hook 自己负责响应 signal

时序上它排在参数校验之后（所以拿到的 args 是干净参数），且在 execute 之前。返回 { block: true, reason: "Blocked by policy" } 时：execute 不被调用、不产生 afterToolCall、循环产出一条 isError: true、content 为 reason 的 toolResult。真实测试验证了三件事：executed 保持 false、LLM 只被调用一次（不再追问）、toolResult 内容恰为 "Blocked by policy"。注意「校验失败」与「工具不存在」根本不会到达钩子——它们在更早的阶段已被拦截（runToolCall 测试断言：hookCalls 只含通过校验的调用）。

afterToolCall：字段级替换，绝不深合并

afterToolCall 在 execute 结束后、tool_execution_end 与消息事件发出之前运行，用于脱敏、改写结果文本、修补 usage。它的合并语义是字段级整体替换，源码里一行注释说得很重：「There is no deep merge for content, details, or usage」：

const results = [
  { content: redacted },                                  // 只换 content
  { structuredContent: { value: "replaced" } },           // 只换结构化输出
  { content: redacted, structuredContent: { value: "both" } }, // 两个都换
  { details: { note: "kept" } },                          // 不碰 content
];
// 原结果 structuredContent = { value: "original" }
expect(seen).toEqual([
  undefined,                        // 只换 content → structuredContent 被丢弃
  { value: "replaced" },
  { value: "both" },
  { value: "original" },               // 未提及 → 保留
]);

两个容易忽略的边界：其一，afterToolCall 自己抛异常时，整个结果被替换为错误 toolResult（isError: true），钩子必须自律——钩子抛错是有意设计的兜底路径，不要在钩子里做「顺便 throw 一下」的调试残留。其二，结果上的 durationMs 用单调时钟只度量 execute() 本身，不含两个钩子；被 block 的调用没有 durationMs（测试断言 expect(blocked).not.toHaveProperty("durationMs")，而放行调用的耗时 ≥ 25ms 且 < 100ms，恰好卡在钩子 sleep 100ms 之外）。

最后是给「工具调工具」准备的 runToolCall(toolCall, options)：它让一次手动发起的调用走完与模型发起完全相同的五步（参数垫片 → 校验 → beforeToolCall → execute → afterToolCall），但不发任何事件、不追加消息。这样嵌套调用无法绕过权限钩子——coding-agent 的 codemode 脚本调用工具时就靠它保证 beforeToolCall 的许可检查同样生效。它对工具失败永不 reject：未知工具、校验失败、被拦截、抛错都统一返回 isError: true 的 outcome。

§7.5terminate 批语义与 onUpdate 流式输出

terminate：整批一致才提前收工

第 6 讲提过循环退出的几种方式；这里补上工具侧的退出通道。execute 结果、被 block 的 beforeToolCall、afterToolCall 改写，三处都可以置 terminate: true，但它的语义是「提示」而非「命令」——判定函数只有一行：

function shouldTerminateToolBatch(finalizedCalls: FinalizedToolCallOutcome[]): boolean {
  return finalizedCalls.length > 0 && finalizedCalls.every((f) => f.result.terminate === true);
}

为什么要求整批全部 terminate？回想 parallel 模式：批内调用并发执行，如果「第一个完成且 terminate 的工具」就能终止循环，那么终止与否将取决于完成顺序——同样的工具组合，今天慢一点结果就不同，语义不可复现；而且此时其他工具可能已经产生副作用，强行丢弃它们的 toolResult 会破坏「每个 toolCall 必须有对应 toolResult」的 provider 契约。整批一致规则把终止判定推迟到批完全结算后：全票通过才把 hasMoreToolCalls 置 false、不再发起下一轮模型请求（hasMoreToolCalls = !executedToolBatch.terminate）；有任何一票反对，循环照常继续。测试覆盖了三种情形：单个 terminate 工具（LLM 只调 1 次）、beforeToolCall block + terminate（execute 未运行、仍只 1 次 LLM 调用）、混合批次（一个 terminate 一个不 terminate，循环继续到第 2 次调用）。

onUpdate：长任务的流式进度

长命令（构建、测试、批量搜索）如果只能等最终结果，UI 会长时间静默。onUpdate 回调让 execute 期间随时上报部分结果，循环把它转成 tool_execution_update 事件：

| { 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; durationMs?: number }

部分结果的类型与最终结果相同，官方约定「每个部分结果也要带 content 与一个 details（可为 undefined）」。实现上有三道保险：

1. 作用域限定：回调只在当前这次 execute() 内有效。源码用 acceptingUpdates 标志——工具 promise 一旦 settle（无论成功失败）立即置 false，之后的调用被直接忽略（if (!acceptingUpdates) return;）。
2. 不阻塞、不丢序：更新事件先收集进 updateEvents 数组，execute 返回后 await Promise.all(updateEvents) 再产出最终结果，保证 UI 先收到全部进度、再收到 tool_execution_end。
3. 迟到调用安全：测试专门模拟「工具保存了 onUpdate 引用、在 prompt 结束后才调用」，断言事件总数不变且无 unhandledRejection；并行批次中一个工具已结算而另一个还在跑时，迟到更新同样被忽略。

典型用法（来自官方 Custom Tools 示例）：跑命令前先 onUpdate?.({ content: [{ type: "text", text: "Working..." }], details: {} })，然后逐段把输出追加为部分结果；UI 的工具渲染器据此实时滚动显示。

§7.6工具状态变更与声明：toToolDeclaration

工具不是启动后就一成不变：coding-agent 里 pi.setActiveTools()、扩展注册新工具、tool-search 延迟暴露工具，都会中途改变可用工具集。模型怎么知道？Pi 的答案不是把「当前工具全量列表」放在每次请求的顶层字段里，而是把工具声明差异作为 transcript 事件记录在 system 消息上——这就是第 4/5 讲提过的 toolsAdded / toolsRemoved 字段。

三个工具函数构成闭环（都在 packages/ai/src/utils/transcript.ts）：

• getCurrentTools(messages)：按顺序回放所有 system 消息的 toolsRemoved / toolsAdded 增量，用 Map 折叠出「当前工具集」。
• toToolDeclaration(tool)：把运行时工具裁剪成模型可见声明——只保留 name / description / parameters（JSON 往返深拷贝）/ constrainedSampling。execute、label、renderCall 等「可执行与展示字段」全部剥掉，且 JSON 序列化会顺带去掉 TypeBox 的不可枚举标记与 undefined 字段。
• getToolStateChanges(previous, current)：比较两个完整工具状态。同名但声明变化（declarationsEqual 用序列化字符串精确比较）计为一次「移除 + 新增」，而不是原地更新。

export function toToolDeclaration(tool: Tool): Tool {
  return {
    name: tool.name,
    description: tool.description,
    parameters: JSON.parse(JSON.stringify(tool.parameters)),
    ...(tool.constrainedSampling === undefined ? {} : { constrainedSampling: tool.constrainedSampling }),
  };
}

export function getToolStateChanges(previous, current): ToolStateChanges {
  return {
    toolsAdded: current.filter((tool) => /* 新增或声明变化 */).map(toToolDeclaration),
    toolsRemoved: previous.filter((tool) => /* 移除或声明变化 */).map((tool) => ({ name: tool.name })),
  };
}

循环侧的接入点是每次模型请求前的 declareToolChanges(context, pendingMessages)：拿 getCurrentTools(历史 + 待发消息) 与 (context.tools ?? []).map(toToolDeclaration) 比较；无差异时不写任何东西（保持缓存前缀逐字节稳定），有差异时把 toolsAdded / toolsRemoved 写进待发送的 system 消息（已有待发 system 消息就更新它的工具字段，否则追加一条 content: "" 的声明消息）。AgentState.tools 的 setter 只做浅拷贝数组，赋值本身不发事件——声明发生在下一次请求边界。

这与第 5 讲 provider 适配层呼应：Anthropic 等支持会话中途 system 消息的 provider 就地携带增量；不支持的 provider 由 resolveTranscriptTools(messages, support) 决定折叠策略——"none"（顶层字段始终是当前全集）、"additions"（只能加不能删，遇到删除就把折叠点推进到最后一次非增量变更）、"changes"（所有声明就地携带，同名后声明覆盖前声明）。对缓存的影响也随之不同：删除工具这种「非增量变更」在 additions 型传输上会导致一次前缀失效，这是工具集频繁切换有真实 token 成本的原因。

§7.7动手实践

在第 6 讲的小 agent 基础上完成四个递进任务（参考 test/utils/calculate.ts 与 get-current-time.ts 的真实写法）：

1. 写一个结构化输出工具

   参数用 pi-schema 的 obj() 构建，声明 outputSchema，execute 返回 structuredContent 与给模型看的 content 摘要；参照 calculateTool 的 AgentTool<typeof schema, undefined> 泛型写法，让 Static<typeof schema> 自动推导 params 类型。

2. 写一个流式长任务工具

   execute 中循环调用 onUpdate?.({ content: [{ type: "text", text: "进度 …" }], details: { step } })，并在每次迭代检查 signal?.aborted 提前返回；订阅事件流验证 tool_execution_update 先于 tool_execution_end。

3. 加两道钩子

   beforeToolCall 拦截危险参数（如路径包含 .. 时返回 { block: true, reason: "…" }）；afterToolCall 把结果文本脱敏后连同 structuredContent 一起返回，观察不返回 structuredContent 时它被丢弃的行为。

4. 对比执行模式

   给两个耗时不同的工具（一个 300ms、一个 50ms），分别在 toolExecution: "parallel" 与 "sequential"（以及给慢工具加 executionMode: "sequential"）下收集事件，验证完成序与持久化序的差异和「单点 sequential 整批降级」。

§7.8常见误区