多个 agent 怎么协作

这是Pi《源码精读》的第九集，多个agent怎么协作，这次的例子，改自包里的第二十一个示例（21-subagent-foreground.ts），我们让主agent说出三个数字，他没自己答，而是把活交给了一个工具，叫subagent，接着另一个agent回答，2、3 和 5。这个示例会先看环境里有没有 OPENAI_API_KEY：有就走 OpenAI，没有就用 faux，一个按剧本回话的假模型。我们这次把它 unset 掉，所以下面看到的是脚本写好的回答，不是真实模型现场发挥。

说到以后，主agent转述给我们：帮手说，2、3 和 5。回答问题的那一个就叫做agent，它其实就是一个普通的对话，工具新开了一本记录本，把任务写进去，让它自己去问模型，子agent答完了，工具把答案交回主agent，作为这次调用的结果。

图 9-1 · 子 agent 是工具新开的普通对话，答案再作为工具结果交回主对话

看代码，工具里新建一个对话，交给它任务，等它回答。还记得开头的问题吗，主 agent 被停了，帮手怎么办？答案藏在新开对话的那一刻，那时同时记下了一样东西，它归谁管，我们把它叫做归属。例子里子对话的记录上写着，归主对话里的那次工具调用管。

源码对照 · packages/durable/test/examples/21-subagent-foreground.ts:54

        const child = await api.commit(async (tx) => {
          // Ownership records the child: a rerun of this call finds it instead of creating another.
          const existing = (await tx.scanConversations({ ownerTaskId: api.taskId }, 1)).items[0];
          if (existing !== undefined) return existing.id;
          // Starts as a copy of this conversation's agent: model, thinking level, cwd, extensions, tools.
          const created = await tx.createConversation({ ownership: { kind: "task", taskId: api.taskId } });
          // Without this extension, the child is not offered this tool.
          await configure(tx, created.id, { extensions: { remove: [Subagent] } });
          return created.id;
        }, callContext);

这里先按 ownerTaskId 找有没有现成的子对话，找不到才 createConversation，并把当前工具任务的 id 写进 ownership。所以“新建”和“认领”是同一个动作的两面：子对话从第一刻起就带着主人，重跑时也先按这个主人找。创建子对话时，副本还包括父对话的 maxTurns 上限；任务拥有的子对话和 fork 都继承它。

看记录本身，对话里有一个 owner，里面是 { conversationId, taskId }，写着它归哪个对话里的哪个任务；存进 SQLite 时，表里拆成两列，owner_conversation_id 和 owner_task_id。这条归属和新对话在同一次存档里写下，之后没有改它的路径。

图 9-2 · 子对话创建时同时写下 owner，SQLite 里落成两列

归属有什么用？为了看清这一点，我们在同样的例子上把子agent的回答拖慢，就在它还在生成时，只停主对话。结果主对话这一轮没有回答，子对话里问模型的那个任务也停了。没有人直接去停子对话，它是顺着归属被找到的。

看代码，按下停止，就从这个对话出发，顺着归属一路往下找，在同一次存档里，把找到的前台活都标上停止。

源码对照 · packages/durable/src/harness/scheduler.ts:654

  async abortConversation(conversationId: ConversationId, background: boolean, context: Context): Promise<void> {
    const reached = await this.#session.commitWith(async (tx) => {
      // …
      const reached: TaskId[] = [];
      const walk = this.#walkDown(conversationNode(conversationId), { crossBackground: background });
      for (const { node, live } of walk) {
        if (live === undefined || (node.background && !background)) continue;
        reached.push(live.id);
        if (!live.abortRequested || live.abortReason !== undefined) tx.setTask(withAbortMark(live));
      }
      // …
      return reached;
    }, context);
    // …
    await this.waitForIdle(conversationId, context);
  }

#walkDown 不只走任务，也会经过任务名下的对话，所以子对话不是特殊分支，只是树上的下一层。标记提交后，正在跑的任务会被 abort signal 打断，走自己的停止流程。

图 9-3 · 停止主对话后，调度器顺着归属树把子对话里的生成任务一起停掉

示例 21 用 SQLite，数据库在当前目录的 pi-durable-subagent-data/session.sqlite；下面的 abort 探针也用这个库，ownership 的写法不变。同一个库再次运行会继续往后编号；要复现干净结果，把 cwd 换到另一个空目录，再用本文件的绝对路径运行。杀进程恢复测试另开一个空库，不和 abort 探针混用。

存储实况 · SQLite（示例 21 的 abort 变体，本次运行的 session.sqlite）

SELECT id, record ->> '$.owner' AS owner, owner_conversation_id, owner_task_id
FROM conversations
ORDER BY id;

id	owner	owner_conversation_id	owner_task_id
1	NULL	NULL	NULL
13	{“conversationId”:1,“taskId”:12}	1	12

SELECT id, conversation_id, record ->> '$.kind' AS kind, status,
       abort_requested, record ->> '$.state.outcome.status' AS outcome
FROM tasks
ORDER BY id;

id	conversation_id	kind	status	abort_requested	outcome
9	1	pi.generation	terminal	1	aborted
12	1	pi.tool	terminal	1	aborted
21	13	pi.generation	terminal	1	aborted

看这条链：1 是主对话，12 是主对话里 subagent 工具调用对应的 pi.tool 任务，13 是子对话，21 是子对话里问模型的 pi.generation 任务。21 自己没有 owner 字段，它用 conversation_id=13 挂在子对话上；13 再用 owner_task_id=12 挂在工具任务上。归属不是每个任务都写一个 owner 任务：会话拥有的任务只记 conversation_id，任务归任务时记录里才有 owner；子对话再记 owner 任务，一段一段接起来。这次两条 input 也都停在 unanswered，原因 aborted；主对话没答案，子对话的结果也没人收。

归属还有第二个用处：等空闲。waitForIdle 也从主对话往下走，只要还有一个非 background 的任务没有结束，就不算空闲。子对话里的 21 还在生成，主对话就闲不下来。这不是另外写的一套判断，用的还是同一棵归属树（scheduler.ts:2594）。

图 9-4 · waitForIdle 走同一棵归属树，子树里还有前台任务就不算空闲

这里说的是前台子agent。也可以专门让子agent单独干活，主对话的普通停止就不带它。示例 22（22-subagent-background.ts）是后台版：subagent 工具有 spawn、send、stop、status 四个动作，子对话挂在一个 background 的 anchor 任务下；主对话的 Esc 和 waitForIdle 走到 background 就停，要连它一起停，得用 abort(context, { background: true })，或者直接对那个子对话调用 abort。它还会在进程重启后接着把答案送回主对话。这里不展开。

图 9-5 · 普通停止和 waitForIdle 到 background 边界就停，background:true 才跨过去

为什么要记下归谁管？假如没有这一栏，按下停止，只能停住主对话自己的活，顺着找也找不到子agent，它就会接着问模型，接着花钱，也没人收它的结果。所以这个包的做法是：任务必须有归属，这个模式里的子对话必须带 owner；根对话或独立对话可以 ownerless。按一下停止，前台下面的活会跟着停；反过来，一个任务要等自己名下的前台活结束，才会真正完成。

例子里的这个工具，就算崩溃后重跑，也能按归属把原来那个subagent找回来。前提是它把 replay 标成了 safe：恢复时工具任务会重跑，先按 ownerTaskId 找到原来的子对话，再用同一个 requestId 取回原来那次提交，不会新开一个。要是标成 unsafe，这次调用不会重跑：首次中断标成 interrupted，连续被切断到 settings.restartLimit 次则标成 restart_limit；两种失败都会带停还在跑的子活。afterTool 也会在 Harness 写入的失败结果上运行，结果码由 diagnostics 区分。

图 9-6 · 崩溃重跑时 safe 按 owner 和 requestId 找回，unsafe 直接标记中断

上面的 abort 表来自示例 21 的库；恢复测试另开一个空库，让子对话 13 在生成时退出，再重开。重开时，在途的模型请求按一次失败计，按 settings.retry 重试；重试成功，任务 21 完成，主对话拿到 The subagent says: 2, 3, and 5.；重试耗尽，任务 21 失败，原因是 interrupted。恢复后的库里还是只有 1 和 13 两个对话，13 的 owner_task_id 还是 12，没有出现第二个子对话。

图 9-7 · 杀进程重开后仍然只有对话 1 和 13，没有新建子对话；重试成功时任务 21 完成

这一集记住一句话，子agent是工具新开的一个对话，它记着自己归哪个任务管；前台子agent就是这样，主agent停，它也跟着停。到这里，路线图上的九个问题都讲完了，以后遇到具体问题，就打开对应的文件，细节交给大模型帮你读。