返回课程首页

第 8 章

并行工具、取消和失败恢复

处理并发执行、中止、错误和停止条件。

源码基线：Pi v0.82.0 · commit 518855d

桌面 Agent 现在能展示工具进度。新的实际需求是：

用户让 Agent 同时检查多个文件；可以随时停止；一个文件读取失败时，其他结果仍然保留。

这会把上一章的事件 UI 推向真正的并发运行。我们需要处理：

  • 顺序与并行执行；
  • 协作式取消；
  • 工具异常；
  • 模型是否以及怎样重试；
  • 执行前后的 Hook；
  • 完成一轮后的优雅停止。

1. 默认不是顺序执行

ToolExecutionMode 只有两个值：

type ToolExecutionMode =
  | "sequential"
  | "parallel";

Agent 的默认值是 "parallel"。

如果一条 Assistant Message 同时请求读取三个文件，默认行为是：

flowchart LR
    A["Assistant Message"] --> C1["read file A"]
    A --> C2["read file B"]
    A --> C3["read file C"]
    C1 --> R["Tool Results"]
    C2 --> R
    C3 --> R

但并行不是“拿到 Tool Call 后立刻全部无序执行”。源码分成两个阶段：

  1. 按 Assistant 源顺序逐个发出 start，并完成 preflight；
  2. 让所有通过 preflight 的调用并发执行。

Preflight 包括工具查找、参数准备、Schema 验证和 beforeToolCall。

2. 全局模式与单工具模式

可以在 Agent 上设置全局策略：

const agent = new Agent({
  initialState: {
    model,
    tools,
  },
  streamFn: models.streamSimple.bind(models),
  toolExecution: "parallel",
});

单个工具也可以声明：

const editFileTool: AgentTool<
  typeof EditFileParameters,
  EditDetails
> = {
  name: "edit_file",
  label: "编辑文件",
  description: "修改一个文件",
  parameters: EditFileParameters,
  executionMode: "sequential",
  async execute(...) {
    // ...
  },
};

只要当前批次中任意 Tool Call 对应的工具声明了 "sequential"，整批就顺序执行。

flowchart TD
    B["一批 ToolCall"] --> G{"全局模式是 sequential?"}
    G -->|是| S["整批顺序执行"]
    G -->|否| P{"任一目标工具要求 sequential?"}
    P -->|是| S
    P -->|否| A["通过 preflight 的调用并发执行"]

适合强制顺序的工具包括：

  • 修改同一份文件；
  • 依赖共享事务；
  • 操作前后顺序有业务含义；
  • 底层资源不支持并发访问。

只读且互不依赖的文件查询更适合并行。

3. 并行不改变 Transcript 顺序

并行模式中：

  • tool_execution_end 按最终化完成顺序发出；
  • ToolResult Message 按原 Tool Call 顺序生成；
  • turn_end.toolResults 也按原顺序。
sequenceDiagram
    participant A as Agent
    participant F1 as File A
    participant F2 as File B
    participant UI

    A->>F1: execute call-1
    A->>F2: execute call-2
    F2-->>UI: end call-2
    F1-->>UI: end call-1
    A-->>UI: ToolResult call-1
    A-->>UI: ToolResult call-2

这样 UI 可以即时展示谁先完成，同时模型 Context 和 Session 保持确定的源顺序。

4. 并发 UI 状态怎样管理

沿用上一章按 toolCallId 建立的 Map：

type ConcurrentToolState = {
  byId: Record<string, ToolRunView>;
  activeIds: string[];
};

不要只保存一个：

currentTool?: ToolRunView;

因为它会被后启动的调用覆盖。

AgentState.pendingToolCalls 本身就是 ReadonlySet<string>，也说明运行时允许同时存在多个工具。

UI 可以派生：

const activeCount = agent.state.pendingToolCalls.size;

但历史卡片、进度文字和最终结果仍应放在独立 View Model 中。

5. Stop 按钮调用什么

Agent 为每次运行创建一个 AbortController。停止按钮调用：

stopButton.onclick = () => {
  agent.abort();
};

agent.abort() 只做：

this.activeRun?.abortController.abort();

没有运行时调用是安全的空操作。

同一个 Signal 会传给：

  • streamFn 请求；
  • Agent Event 订阅者；
  • transformContext；
  • beforeToolCall；
  • AgentTool.execute；
  • afterToolCall；
  • prepareNextTurn 包装。

6. 取消是协作式的

AbortSignal 不会强制杀死任意 JavaScript Promise。

工具必须主动观察：

async execute(_id, args, signal, onUpdate) {
  if (signal?.aborted) {
    throw new Error("操作已取消");
  }

  const files = [];

  for (const path of args.paths) {
    if (signal?.aborted) {
      throw new Error("操作已取消");
    }

    files.push(await readOneFile(path));

    onUpdate?.({
      content: [
        {
          type: "text",
          text: `已读取 ${files.length} 个文件`,
        },
      ],
      details: {
        completed: files.length,
        total: args.paths.length,
      },
    });
  }

  return {
    content: [
      {
        type: "text",
        text: JSON.stringify({ files }),
      },
    ],
    details: { files },
  };
}

若底层 API 接受 Signal，应继续传下去：

await fetch(url, { signal });

如果工具忽略 Signal，Agent 不能凭空终止它。agent.abort() 只是发出取消意图。

7. Abort 在不同阶段的表现

模型生成阶段

Signal 传给 streamFn。Provider 识别取消后通常以 stopReason: "aborted" 的错误消息结束，Loop 发出 turn_end 和 agent_end。

Tool preflight 阶段

beforeToolCall 返回后，Loop 检查 Signal。若已取消，生成 "Operation aborted" 错误结果，不调用工具。

Tool execute 阶段

Signal 交给工具。工具是否及时结束取决于自身和底层 API 是否协作。

顺序批次

一个调用最终化后，如果 Signal 已取消，顺序循环停止准备后续 Tool Call。

并行批次

已经进入执行的多个工具都会收到同一 Signal。它们可能以不同速度结束。

如果取消发生在串行 preflight 中，循环会在当前调用 finalization 后停止为后续 Tool Call 发出 start/result；此前已经准备好的执行闭包仍会带着同一个、此时已 aborted 的 Signal 运行，所以工具依然必须主动检查 Signal。

因此 UI 不应在点击停止时立刻把全部卡片标成“已取消”。更准确的状态是：

cancel-requested

等每个 tool_execution_end 或最终 Agent 生命周期事件到达后，再落定结果。

8. 单个工具抛错会发生什么

executePreparedToolCall() 捕获工具异常：

try {
  const result = await tool.execute(...);
  return { result, isError: false };
} catch (error) {
  return {
    result: createErrorToolResult(message),
    isError: true,
  };
}

在没有 afterToolCall 覆盖的默认路径中，随后产生：

  • tool_execution_end(isError: true)；
  • ToolResultMessage(isError: true)；
  • 若这批最终结果没有全部要求 terminate，Loop 保留继续条件；除非 shouldStopAfterTurn 等后续停止机制结束运行，才会进入下一次模型请求。
flowchart LR
    E["tool.execute throws"] --> ER["错误 AgentToolResult"]
    ER --> AF["afterToolCall 可覆盖结果"]
    AF --> END["tool_execution_end"]
    END --> TR["最终 ToolResultMessage"]
    TR --> T{"整批最终结果全 terminate?"}
    T -->|否| G["经过 Turn 后停止条件"]
    G -->|继续| M["模型决定怎样继续"]
    G -->|停止| S
    T -->|是| S["跳过自动下一次模型请求"]

afterToolCall 可以覆盖 content、details、usage、isError 和 terminate。因此 Tool 抛错只说明进入 Hook 前是 error outcome，最终 isError 甚至可以被改成 false。

所以单工具异常通常不会直接让 Agent Loop reject，但也不能无条件断言最终一定是错误状态或一定继续下一 Turn。

但是错误消息默认来自 error.message。工具应先记录本地诊断，再抛出可安全发送给模型的简洁错误。

9. Agent 会不会自动 Retry

要区分三种 Retry：

9.1 模型自己重新发出 Tool Call

错误 Tool Result 会进入 Context。下一 Turn 中，模型可能：

  • 修正参数后再次调用；
  • 换一个工具；
  • 告诉用户失败；
  • 放弃任务。

这是模型决策，不是 Agent 保证的重试。

9.2 工具内部重试瞬时错误

工具可以对明确可重试的本地错误做有限重试：

async function withTransientRetry<T>(
  operation: () => Promise<T>,
  signal?: AbortSignal,
) {
  let lastError: unknown;

  for (let attempt = 1; attempt <= 3; attempt += 1) {
    if (signal?.aborted) {
      throw new Error("操作已取消");
    }

    try {
      return await operation();
    } catch (error) {
      lastError = error;
      if (!isTransient(error) || attempt === 3) {
        throw error;
      }
    }
  }

  throw lastError;
}

这是课程应用策略，不是 Pi Agent 内置工具重试器。

9.3 Provider 请求重试

SimpleStreamOptions 继承的 StreamOptions 还有 maxRetries、maxRetryDelayMs 等请求选项，是否支持由具体 Provider/SDK 路径决定。它与 Tool Retry 是不同层。

当前高层 AgentOptions 直接公开并转发 maxRetryDelayMs，但没有公开 maxRetries 字段。maxRetries 可用于低层 AgentLoopConfig 或直接模型 Stream 调用；不要在高层 Agent 示例中假设能直接传入。

10. beforeToolCall：执行前拦截

Agent 可以注册：

const agent = new Agent({
  // ...
  beforeToolCall: async (
    { toolCall, args },
    signal,
  ) => {
    if (signal?.aborted) {
      return {
        block: true,
        reason: "操作已取消",
      };
    }

    if (
      toolCall.name === "read_many_files" &&
      (args as { paths: string[] }).paths.length > 20
    ) {
      return {
        block: true,
        reason: "一次最多读取 20 个文件",
      };
    }
  },
});

Hook 运行时：

  • Tool 已找到；
  • prepareArguments 已执行；
  • 参数已通过 Schema 验证；
  • execute 尚未开始。

返回 { block: true } 会生成错误 Tool Result。

一个容易忽略的源码事实：Hook 收到的是验证后的可变对象。如果 Hook 原地修改 args，Loop 不会再次验证，修改后的值会直接进入 execute。

因此安全 Hook 应把参数当作只读；如确需转换，优先使用 prepareArguments，并确保返回值接受后续 Schema 验证。

execute 与 afterToolCall 收到的是验证后的 args；tool_execution_start 和 tool_execution_update 事件中的 args 仍来自原始 toolCall.arguments。UI 不能假定事件预览与 execute 实参在经过准备或 Hook 变异后仍完全相同。

11. afterToolCall：最终结果后处理

已经执行的工具会进入：

afterToolCall: async ({
  toolCall,
  result,
  isError,
}) => {
  if (isError) {
    return {
      content: [
        {
          type: "text",
          text: "文件读取失败，请检查权限或文件状态。",
        },
      ],
      details: {
        category: "file-read-failed",
      },
    };
  }

  return {
    details: {
      ...result.details,
      audited: true,
    },
  };
},

覆盖是逐字段替换，不是深度合并。当前实现使用空值合并，因此返回非 null、非 undefined 的字段值才会替换：

  • 提供 content 就替换整个 Content 数组；
  • 提供 details 就替换整个 Details；
  • usage、isError、terminate 同理；
  • 未提供的字段保留原结果。

示例若要给 Details 增加字段，必须像上面一样显式展开原值。

Hook 自己抛错时，Loop 会用 Hook 的错误替换成新的错误 Tool Result。

Immediate outcome 不会运行 afterToolCall，例如工具不存在、验证失败、before 阻止和执行前取消。

12. shouldStopAfterTurn：优雅停止

低层 AgentLoopConfig 支持：

shouldStopAfterTurn: async ({
  message,
  toolResults,
  context,
}) => {
  return shouldPauseBeforeMoreWork(
    message,
    toolResults,
    context,
  );
},

它发生在：

  1. Assistant 和本轮全部工具执行完成；
  2. Tool Result 已加入 Context；
  3. turn_end 已发出；
  4. prepareNextTurn 已有机会更新快照；
  5. 本轮结束后的下一次 Steering 轮询和 Follow-up 轮询之前。

返回 true 会发出 agent_end 并退出。

runLoop() 在进入主循环前可能已经执行过一次初始 Steering 轮询。shouldStopAfterTurn 不会撤销这次初始读取；它阻止的是当前 Turn 之后继续轮询和启动后续工作。

它不会：

  • 取消当前 Provider Stream；
  • 中止正在运行的工具；
  • 修改 Assistant Message 的 stopReason；
  • 消费 Follow-up 队列。

因此它适合“完成这一轮后暂停”，不是紧急停止按钮。

13. 失败恢复的推荐分层

flowchart TD
    F["发生失败"] --> K{"哪一层?"}
    K -->|Provider 请求| PR["Provider/SDK 有限 Retry\n或显示模型错误"]
    K -->|可识别瞬时工具错误| TR["工具内部有限 Retry"]
    K -->|参数/权限/业务错误| MR["错误 Tool Result 给模型"]
    K -->|用户停止| AB["传播 AbortSignal"]
    MR --> N["模型决定修正、解释或退出"]

推荐原则：

  • 只重试明确瞬时且幂等的操作；
  • 写入、支付、发送消息等有副作用操作不能盲目重试；
  • 每次重试都观察 AbortSignal；
  • 对模型保留真实的成功/失败语义；
  • 对用户展示可理解状态，对日志保留脱敏诊断。

14. 并发工具的状态机

一个工具卡片可以使用：

stateDiagram-v2
    [*] --> Preparing: tool_execution_start
    Preparing --> Running: tool_execution_update
    Preparing --> Done: tool_execution_end success
    Preparing --> Error: tool_execution_end error
    Running --> Running: tool_execution_update
    Running --> Done: tool_execution_end success
    Running --> Error: tool_execution_end error
    Preparing --> CancelRequested: user clicks stop
    Running --> CancelRequested: user clicks stop
    CancelRequested --> Done: tool still completes
    CancelRequested --> Error: tool ends as error/aborted

“请求取消”与“已经取消”必须分开，因为协作式取消存在时间差，工具也可能在看到 Signal 前恰好完成。

15. 源码调用链复盘

flowchart TD
    TC["ToolCall batch"] --> MODE{"sequential or parallel"}
    MODE --> PRE["start + preflight"]
    PRE --> B["beforeToolCall"]
    B -->|block| IE["immediate error"]
    B -->|allow| EX["execute(signal, onUpdate)"]
    EX -->|throw| ER["error result"]
    EX -->|return| OK["tool result"]
    ER --> AF["afterToolCall"]
    OK --> AF
    AF --> END["tool_execution_end"]
    IE --> END
    END --> MSG["ToolResultMessage"]
    MSG --> STOP{"shouldStopAfterTurn?"}

注意：图中 immediate error 不经过 afterToolCall；已经进入 execute 的成功或失败结果才会进入该 Hook。

16. 本章小结

  • Agent 默认并行执行同一 Assistant Message 中的 Tool Call。
  • 全局 sequential 或批次中任一工具要求 sequential，都会让整批顺序执行。
  • 并行 end 事件按最终化完成顺序，持久 Tool Result 按源顺序。
  • 并发 UI 必须按 Tool Call ID 管理多项状态。
  • agent.abort() 触发本次运行的 AbortSignal，不会强制杀死 Promise。
  • Provider、Hook 和 Tool 都需要协作观察 Signal。
  • 单工具异常先变成错误结果，after Hook 仍可覆盖最终状态；通常不会直接崩溃整个 Loop。
  • Agent 不保证自动重试同一个 Tool Call；模型可能在下一 Turn 自行修正。
  • 工具内部 Retry 是应用策略，应限制次数、判断幂等性并支持取消。
  • beforeToolCall 在验证后、执行前运行，可以阻止调用。
  • before Hook 原地修改 args 后不会再次验证。
  • afterToolCall 对执行结果做逐字段替换，不做深度合并。
  • Immediate outcome 不运行 after Hook。
  • shouldStopAfterTurn 在完整 Turn 后优雅停止，不等于紧急 Abort。
  • UI 应区分 cancel requested 与最终取消结果。

下一章会在这些 Hook 之上建立危险操作审批、项目信任和受保护路径。

17. 自测

  1. 默认并行模式是否意味着 preflight 也完全并发？
  2. 一批调用中只有一个工具声明 sequential，会怎样执行？
  3. 为什么 end 事件顺序不能用来持久化 Transcript 顺序？
  4. agent.abort() 能否强制终止一个忽略 Signal 的 Promise？
  5. 单个工具抛错后，Loop 通常如何继续？
  6. Agent 是否保证自动重试同一个失败 Tool Call？
  7. before Hook 收到的参数是否已验证？修改后会不会再次验证？
  8. after Hook 返回新的 details 时会不会与旧值深度合并？
  9. 哪些 immediate outcome 不经过 after Hook？
  10. shouldStopAfterTurn 与 Stop 按钮有什么本质区别？

本章源码依据

  • packages/agent/src/types.ts
  • packages/agent/src/agent.ts
  • packages/agent/src/agent-loop.ts
  • packages/agent/README.md
  • packages/agent/test/agent-loop.test.ts
  • packages/agent/test/agent.test.ts
  • packages/ai/src/types.ts
  • packages/ai/src/models.ts