Agent 实现方法论 From Scratch

多智能体 v09

多智能体 v09

从 ch14 到 ch21，我们的产品始终是一个 Agent：一个主循环、一条对话历史、一个上下文窗口、一套工具。单 Agent 简单、可预测、容易调试，但上下文窗口是硬上限（ch09 讲过滑动窗口和压缩，可压缩改变不了窗口本身），工具列表越膨胀，模型选错和走神的机会越多。这一章我们把「一个 Agent 干所有事」升级成「一个 orchestrator 分派、多个 worker 并行」：多智能体 v09。

多智能体的核心思想用一句话概括：subagent = 独立的 loop 实例。我们 ch07 起就手写的 runAgent，再实例化一个、跑独立上下文，就是一个 subagent。这一章不引入任何框架，还是那个 loop，只是多了一个「委派」的注入点。

本章目标

读完本章并做完配套练习后，你应该能够：

概念与动机：单 Agent 的瓶颈

先看单 Agent 会怎么「撑不住」

假设产品只有一个 Agent，用户让它「帮我把这三个城市的天气查了，顺便把两个数字算一下，再汇总」。模型在一条对话里依次调用 get_weather ×3、add ×2，最后写汇总，这没问题，五个工具调用而已。

但把场景放大：用户让一个 Agent 做一次「调研报告」，需要查 20 个数据源、读 10 份文档、跑 5 个计算、写 3 个章节。单 Agent 的问题开始显现：

  1. 上下文窗口是硬上限。20 个工具调用的原始结果塞进一条历史，很快顶到窗口顶。ch09 的压缩/截断只是延后，不是解决；
  2. 上下文污染（context pollution）。任务 A 的调试过程、失败的尝试、无关的中间输出，全堆在历史里，任务 B 的模型被迫「隔着垃圾看题」。Philipp Schmid 在《The Rise of Subagents》里把这一点列为单 Agent 最核心的失效模式：单个大 Agent 的上下文窗口和工具列表「变得杂乱、越来越不可靠」，而 subagent 拥有自己隔离的上下文窗口（LangChain 博客对 Deep Agents 的讨论 与 learn-claude-code s06_subagent 都引用了这个观点）；
  3. 工具列表膨胀。工具越多，模型选错工具的概率越高，每个请求把全部工具的 schema 塞给模型的 token 成本也越高（ch08 的工具注册表到 ch21 的 MCP，工具只会越来越多）；
  4. 权限边界模糊。一个 Agent 拿到全部工具，意味着它同时拿到「读文件」「删文件」「发消息」，破坏半径等于最大权限（ch18 审批门管的是「要不要执行」，管不了「Agent 手里为什么有这么多工具」）。

什么时候才需要多个 Agent

多 Agent 不是免费午餐，它是复杂度税。判断标准三条，至少满足一条才值得：

标准说明反例（不需要多 Agent）
任务可拆分成互不依赖的子任务子任务之间没有强串行依赖，拆开各自能独立完成一步到位的计算、单轮问答
子任务需要不同的上下文/知识/工具权限查天气的 subagent 只需要 get_weather，不需要 delete_file全部子任务共用同一套工具和知识
并行能显著缩短延迟3 个独立查询串行 3 秒，并行 1 秒子任务本来就很快，编排往返反而更慢

第 18 章讲过的「审批疲劳」在这里有一个镜像问题：审批疲劳是「人的注意力有限」，上下文污染是「模型的注意力有限」。单 Agent 把注意力花在无关历史上是浪费，多 Agent 用「隔离」把注意力切回正题，但隔离本身要花钱（每个 subagent 有自己的 system prompt、自己的历史、结果还要回填汇总）。

subagent 隔离委派

核心模型：subagent = 独立的 loop 实例

我们在 ch07 手写了 runAgent：think → act → observe 的 while 循环，注入 callLlm、tools、onEvent，跑一条消息历史。subagent 不需要任何新机制，把 runAgent 再调用一次，喂一份全新的消息历史、一份裁剪过的工具集、一个独立的迭代预算、一个自己的 sessionId，它就是 subagent。同一个函数，两个实例，互不相识。

这就是本章最重要的那句话：多 Agent 不是新架构，是同一个 loop 的多份实例 + 一层委派编排。

四个隔离维度

隔离维度主 Agentsubagent（工厂创建）
消息上下文会话历史（含所有工具结果）全新 [system(SUBAGENT_PROMPT), user(task)]，主对话不泄漏进去，两个 subagent 互不可见
工具集全部注册工具 + delegate 元工具白名单：只复制允许的工具（默认 get_weather/add），够不着主 Agent 的其他工具
预算maxIterations（主循环保险丝）subagentMaxIterations，独立的保险丝，与主循环互不共享
会话/事件主 sessionId + 主事件流sub_N sessionId + 独立的 onEvent 观察者，不进主会话的 SSE 流

隔离是设计出来的，不是默认得到的。主循环的 runAgent(initialMessages, ...) 本来就接收「初始消息」参数，如果委派时把主对话整个塞进去，隔离就破了。我们的工厂刻意只接受 { task, context? }：subagent 从 [system, user(task)] 开始，构造性隔离，没有入口，自然串不了。

委派时序：delegate 工具 → spawn → 结果回填

主循环新增一个注入点 spawnSubagent：注入后，模型的工具列表里多出一个 delegate 元工具（名字和 schema 由主循环拥有，跑起来做什么由注入的工厂决定）。模型决定委派时调用它，主循环拦截这次调用（不经过普通工具执行器），await 工厂去跑一个完整的独立 loop，然后把 subagent 的终答作为普通的 tool_result 回填。

sequenceDiagram
  participant LLM as mock LLM（主）
  participant MAIN as lib/loop.mjs（主循环）
  participant FACT as lib/orchestrate.mjs（委派层）
  participant SUB as runAgent（subagent 实例）
  participant SUBLLM as mock LLM（subagent 的模型）

  LLM->>MAIN: tool_calls: [delegate({task, context}), delegate({task, context})]
  MAIN->>MAIN: 拦截 delegate（不进 executeTool）
  MAIN->>FACT: await spawnSubagent({task})
  FACT->>FACT: 建全新消息历史 + 白名单工具 + 独立预算
  FACT->>SUB: runAgent(subMessages, {白名单, maxIterations, sessionId:'sub_1'})
  SUB->>SUBLLM: 调模型（只看到白名单工具）
  SUBLLM-->>SUB: tool_calls: [get_weather(...)]
  SUB->>SUB: 执行工具 → tool 消息 → 再调模型
  SUBLLM-->>SUB: 终答文本
  SUB-->>FACT: { ok:true, result: 终答 }
  FACT-->>MAIN: { ok:true, result }
  MAIN-->>MAIN: 结果回填：tool_result（subagent 终答 = 工具观察）
  MAIN->>LLM: 下一轮：带着全部 tool_result 汇总

注意两个「不知道」：

  1. 主循环不知道 subagent 内部怎么跑。它只知道工厂的签名 ({ task, context? }) => Promise<{ ok, result?, error?, iterations? }>。上下文怎么建、白名单怎么裁、预算多少，都在 lib/orchestrate.mjs 里，主循环一个字都不管（和 ch18 的 getApproval 是同一个注入模式）；
  2. subagent 的事件不进主会话的 SSE 流。subagent 自己跑的是同一个 runAgent，照样发 text_delta/tool_call_start/tool_result/done，但它们带自己的 sessionId（sub_1、sub_2…），由委派层的 onEvent 观察者消费。主会话只看到两件事：tool_call_start（delegate）和 tool_result（结果回收）。协议层面零新增事件类型，多智能体没有破坏 ch15 定稿的契约，只是把它用在了多个实例上。

结果回收：subagent 的终答 = 主 Agent 的一条 tool 消息

「回收」就是回填：subagent 的终答是字符串，主循环把它塞进主对话的 tool 消息。主 Agent 下一轮能看到「delegate 返回了 上海：晴，27°C…」，就像看到一个普通工具的执行结果，主 Agent 不需要知道这个结果来自一个完整的 subagent。隔离的反面是：结果必须足够好地穿过边界（终答字符串），细节留在 subagent 那边（完整历史只在 subagent 自己的 session 里）。

orchestrator-worker 与 handoff

多 Agent 的编排模式不止一种。业界最常提的两个是：

orchestrator-worker（我们实现的）

中央 orchestrator 拆任务 → 派给多个 worker 并行 → 汇总。控制权始终在 orchestrator 手里，worker 是「用完即弃」的执行单元：

flowchart LR
  USER["用户"] --> ORCH["orchestrator 主 Agent<br/>拆任务 · 委派 · 汇总"]

  subgraph workers["worker subagents（并行，互相隔离）"]
    W1["subagent 1<br/>白名单工具 A<br/>预算 N 次迭代"]
    W2["subagent 2<br/>白名单工具 B<br/>预算 N 次迭代"]
    W3["subagent 3<br/>白名单工具 A<br/>预算 N 次迭代"]
  end

  ORCH -->|delegate×3 一批| W1
  ORCH -->|delegate×3 一批| W2
  ORCH -->|delegate×3 一批| W3
  W1 -->|tool_result 回填| ORCH
  W2 -->|tool_result 回填| ORCH
  W3 -->|tool_result 回填| ORCH
  ORCH -->|汇总| USER

这个模式的优势是中央控制：所有 worker 的目标、边界、权限都由 orchestrator 决定；worker 之间不需要互相理解。代价是编排往返：每多一层委派就多一轮「主→子→回填→汇总」的 token 和延迟。OpenAI Agents SDK 的编排文档 把 orchestrator-worker（也叫「agents as tools」）列为主流模式之一，worker 对 orchestrator 而言就是一个普通工具。

handoff（对照，我们不做完整版）

handoff = 控制权整体移交：一个 Agent 干完自己那部分后，把对话状态整个交给下一个 Agent，自己退出。OpenAI Agents SDK 里 handoff 就是一个工具，模型调用它，执行就切换到目标 Agent，最新对话状态原样转移，下一个 Agent 从上一个的上下文续着干（OpenAI · Agent orchestration and handoffs）。典型场景是串行流水线：分诊 Agent → 处理 Agent → 质检 Agent，链式接力（Enison · Multi-Agent Design Patterns 对两种模式的对比）。

handoff 与 orchestrator-worker 的关键区别：

orchestrator-workerhandoff
控制权始终在 orchestrator随移交转移
子任务关系并行、互不依赖串行、有依赖
状态worker 隔离，只回传结果整体转移给下一个
典型形态「查 3 个城市 → 汇总」「客服→技术→结算」流水线

我们这一章只做委派 + 结果回收（orchestrator-worker 的简化版），不做完整 handoff。理由很实际：handoff 的「状态整体转移」意味着下一个 Agent 要继承上一个的全部上下文，这恰恰是我们刚用「隔离」解决的问题。简化版里 subagent 的终答作为 tool_result 回填，主 Agent 保留了「汇总」的判断权，并没有把整条对话交给别人。要支持真正的 handoff，可以在协议上加一种「交接」事件（把当前消息历史整体传给下一个实例），这是留给你的延伸练习。

共享状态与并发

并发：Promise.all 一批调用

主循环的 act 阶段（ch06 的「并行工具调用」在这一章变成了刚需）：模型在一轮里发出多个 delegate 调用，主循环把它们并发跑起来——Promise.all，所有 subagent 先启动，再等最慢的那个完成。结果按调用顺序回填（Promise.all 保序），主 Agent 看到的 tool 消息永远有序，不管三个 subagent 实际谁先谁后。

一个必须守住的边界：approve 级工具永远不进并发批。ch18 的审批门是「逐个暂停、逐个审计」的，如果一批 delete_file 并行执行、审批混在一起，就说不清「人批准的是哪个文件」。所以主循环把一批调用分成两组：需要审批的（顺序走审批门，ch18 语义不变）与其余（auto/suggest/delegate，并发执行）。权限语义优先于并发收益。

共享状态：隔离优先，结果回填是唯一共享通道

多 Agent 的「共享状态」是个两难：共享多了，上下文污染又回来了；共享少了，worker 之间没法协作。我们的取舍是隔离优先：worker 之间零共享（各自的历史互不可见），主 Agent 通过 tool_result 做最小共享（只传回终答字符串）。要传更多信息？把信息写进 task/context 参数，或者让 subagent 把结果落盘、主 Agent 再读，显式、可审计，不做隐式共享的全局上下文。learn-claude-code 的 subagent 章把这个原则概括为「subagent 返回摘要给父 Agent，不返回原始上下文」（s06_subagent · 深入 CC 源码）。

多智能体的代价

多 Agent 的每一层编排都是成本，四笔账必须心里有数：

  1. token 翻倍。每个 subagent 有自己的 system prompt + 消息历史 + 工具 schema；subagent 的终答还要作为 tool_result 再回填给主 Agent，主 Agent 汇总时又读一遍。同一个事实被「子任务输入 + 子任务输出 + 汇总」各写了一次。子任务越多，重复开销越大；
  2. 延迟。编排有往返：主 Agent 发 delegate（1 次模型调用）→ subagent 完整跑（≥2 次）→ 回填 → 主 Agent 再汇总（1 次）。即使 worker 并行，主 Agent 也必须等最慢的 worker，慢牛拖垮整队；
  3. 错误传播与级联失败。subagent 失败 → tool_result 带错误 → 主 Agent 读到的是一段错误文本，它可能把错误当真结果汇总进答案。错误从 subagent 传到主 Agent 再传到用户，每一层都可能被「洗白」；
  4. 死循环与预算失控。subagent 递归委派（subagent 再 delegate 一个 subagent）会指数爆炸；一个 subagent 忘记收尾会在自己的预算里空转。所以：委派只限一层（subagent 的 loop 不再注入工厂）、每个 subagent 有独立预算（耗尽即失败返回）。

手写实现：把委派层装进产品

一、主循环：注入 spawnSubagent 工厂（lib/loop.mjs）

改动集中在三处。第一处：runAgent 的选项多一个注入点 spawnSubagent；注入后，模型工具列表追加 delegate 元工具：

// 模型看到的工具列表 = 注册表工具 + （注入了工厂时的）delegate 元工具
const toolList = spawnSubagent ? [...modelTools(tools), DELEGATE_TOOL] : modelTools(tools);

DELEGATE_TOOL 只有 name/description/schema，名字和 schema 归主循环，跑起来做什么归工厂：

const DELEGATE_TOOL = {
  name: 'delegate',
  description: 'Assign a self-contained subtask to a fresh subagent. '
    + 'The subagent runs in its own context with its own tool whitelist and budget, '
    + 'and its final answer comes back as this tool result.',
  schema: {
    type: 'object',
    properties: {
      task: { type: 'string', description: 'The self-contained subtask for the subagent.' },
      context: { type: 'string', description: 'Optional background facts or constraints.' },
    },
    required: ['task'],
  },
};

第二处：act 阶段把一批调用分成两组：需要审批的（顺序走 ch18 审批门）与其余（auto/suggest/delegate，Promise.all 并发），结果按调用顺序回填：

const parallel = [];
const sequential = [];
for (const call of response.toolCalls) {
  if (riskOf(tools, call.name) === 'approve') sequential.push(call);
  else parallel.push(call);
}

// 并发组：先全部宣告（tool_call_start + suggest 提示），再 Promise.all，
// 结果按调用顺序回填
const outcomes = await Promise.all(parallel.map((call) => runOne(call, tools, spawnSubagent)));
// ...顺序组逐个走审批门（ch18 原样）...
// 最后统一按 response.toolCalls 顺序 appendObservation（一个 call 一条 tool 消息）

第三处：delegate 调用的拦截与执行，它不经过 executeTool，而是走注入的工厂，失败也被 catch 成数据（部分失败不炸掉整批）：

function runOne(call, tools, spawnSubagent) {
  if (call.name === 'delegate' && spawnSubagent) {
    return Promise.resolve(spawnSubagent(call.arguments)).catch((err) => ({
      ok: false,
      error: err instanceof Error ? err.message : String(err),
    }));
  }
  return executeTool(tools, call.name, call.arguments);
}

二、委派层：工厂决定隔离的一切（lib/orchestrate.mjs）

lib/orchestrate.mjs 是新增的 core 模块，隔离的三件事都在这里决定，主循环不碰：

// 白名单：从主注册表复制允许的工具到全新注册表（spec + impl 原样复制）
export function buildSubagentRegistry(mainTools, allowNames) {
  const sub = createToolRegistry();
  for (const { spec, impl } of mainTools.list()) {
    if (allowNames.includes(spec.name)) sub.register(spec, impl);
  }
  return sub;
}

// 工厂：每次委派 = 一次独立的 runAgent
export function buildSubagent({ callLlm, tools, allowTools, maxIterations, sessionId, onEvent }) {
  const subTools = buildSubagentRegistry(tools, allowTools);
  let next = 1;
  return async function spawnSubagent({ task, context } = {}) {
    // 构造性隔离：subagent 只从 [system, user(task)] 开始
    const subMessages = [
      { role: 'system', content: SUBAGENT_PROMPT },
      { role: 'user', content: context ? `${task}\n\nContext: ${context}` : task },
    ];
    const result = await runAgent(subMessages, {
      callLlm,
      tools: subTools,       // 白名单
      sessionId: `${sessionId}_${next++}`,
      maxIterations,          // subagent 自己的预算
      onEvent,               // subagent 事件走观察者，不进主会话流
    });
    // 预算截断：没跑完 = 失败可见，半成品不当作真话回填
    if (result.budgetExceeded) {
      return { ok: false, error: `subagent budget exceeded after ${result.iterations} iterations`, iterations: result.iterations };
    }
    return { ok: true, result: result.finalText, iterations: result.iterations };
  };
}

主循环为这个改动付出的代价只是一个 budgetExceeded 字段：正常结束返回 false，保险丝熔断返回 true，subagent 用它「失败可见」，主 Agent 还是原来的熔断兜底。

三、演示：模型驱动委派 + 程序化拆分委派

demo 用确定性 mock 驱动。主会话问「请帮我查一下上海、北京、伦敦三个城市的天气，可以并行去查，最后汇总给我」，mock 主模型在一轮里发出 3 个 delegate 调用（一个批次），主循环并发跑 3 个 subagent。主会话的 SSE 事件流（协议视角，只有 delegate 的开始与结果）：

[1] event: tool_call_start
    data: {"sessionId":"s-main","type":"tool_call_start","callId":"call_delegate_上海","name":"delegate","args":{"task":"查询上海的天气","context":"用户想知道上海现在的天气"}}
[2] event: tool_call_start
    data: {"sessionId":"s-main","type":"tool_call_start","callId":"call_delegate_北京","name":"delegate","args":{"task":"查询北京的天气","context":"用户想知道北京现在的天气"}}
[3] event: tool_call_start
    data: {"sessionId":"s-main","type":"tool_call_start","callId":"call_delegate_伦敦","name":"delegate","args":{"task":"查询伦敦的天气","context":"用户想知道伦敦现在的天气"}}
[4] event: tool_result
    data: {"sessionId":"s-main","type":"tool_result","callId":"call_delegate_上海","ok":true,"result":"上海：晴，27°C，微风，紫外线中等"}
[5] event: tool_result
    data: {"sessionId":"s-main","type":"tool_result","callId":"call_delegate_北京","ok":true,"result":"北京：多云，24°C，西北风 3 级"}
[6] event: tool_result
    data: {"sessionId":"s-main","type":"tool_result","callId":"call_delegate_伦敦","ok":true,"result":"伦敦：小雨，15°C，体感偏凉"}
[7] event: text_delta
    data: {"sessionId":"s-main","type":"text_delta","delta":"天气汇总（3 个城市）：\n1. 上海：晴，27°C，微风，紫外线中等\n2. 北京：多云，24°C，西北风 3 级\n3. 伦敦：小雨，15°C，体感偏凉"}
[8] event: done
    data: {"sessionId":"s-main","type":"done","usage":{"inputTokens":42,"outputTokens":24}}

三个 subagent 的内部事件（各自的 text_delta/tool_call_start/tool_result/done）带着自己的 sessionId（sub_1/sub_2/sub_3），由委派层的观察者打印，主会话流保持干净。并发是可见的：三个 subagent 在同一个批次里启动，主循环 Promise.all 等最慢的那个（mock 给每个城市加了不同的延迟，让交错看得见），全部完成才统一回填。

「程序化拆分委派」是另一种口味：不靠模型拆任务，用规则函数 splitTasks 把一份多行任务按长度预算切块，每块 spawn 一个 subagent，aggregateResults 汇总：

splitTasks(mission, 12) -> ["1. 查询上海的天气","2. 查询北京的天气","3. 查询伦敦的天气","4. 查询纽约的天气","5. 查询迪拜的天气"]
subagents 5/5 ok
- [ok] worker_1: 上海：晴，27°C，微风，紫外线中等
- [ok] worker_2: 北京：多云，24°C，西北风 3 级
- [ok] worker_3: 伦敦：小雨，15°C，体感偏凉
- [ok] worker_4: 纽约：晴，20°C
- [ok] worker_5: 迪拜：晴，20°C

模型驱动（模型决定怎么拆、拆几个）和代码驱动（规则决定怎么拆）是 orchestrator-worker 的两种实现口味，练习里两者都做。

四、预算截断演示

一个只有 1 次迭代预算的 subagent 干不完「查天气」这种两轮任务（工具调用 + 终答），工厂返回失败可见的结果：

tight budget outcome: {"ok":false,"error":"subagent budget exceeded after 1 iterations","iterations":1}

主 Agent 读到的是一条普通工具错误——它知道「这个子任务没跑完」，可以重试、换招、或如实告诉用户。

常见坑与失败模式

坑一：上下文串扰（隔离被破坏）。 委派时把主对话整个塞给 subagent（或者让 subagent 共享主会话的历史引用），任务 A 的调试过程污染任务 B，隔离名存实亡。隔离是构造性的：subagent 只接受 { task, context? }，历史从 [system, user(task)] 重新开始。判断标准一句话：把主历史删掉，subagent 还能不能独立跑？ 能，才是隔离。

坑二：级联失败（一个 worker 打翻整批）。 三个 delegate 调用并发，其中一个 subagent 的模型调用抛错，如果这个 reject 直接冒泡，Promise.all 整体失败，另外两个本来成功的 worker 结果全丢。解法（本章实现）：per-task catch，每个委派调用单独 catch 成 { ok:false, error }，主 Agent 读到错误数据后自行恢复；失败是一个 worker 的，不是整个批次的。

坑三：预算失控（subagent 没有自己的预算）。 让 subagent 共享主循环的 maxIterations，或者干脆不设上限，一个卡住的 subagent 会无限空转，把整个批次拖死。subagent 必须有独立的预算（subagentMaxIterations），耗尽即失败返回。注意「独立」二字：主循环熔断 ≠ subagent 熔断，两个保险丝各管各的。

坑四：死循环（递归委派）。 subagent 也拿到 delegate 工具、再委派 subagent……指数爆炸，谁也拦不住。本章委派只限一层：工厂给 subagent 的 runAgent 不注入 spawnSubagent，subagent 的模型列表里根本没有 delegate——想递归也递归不了。

坑五：半成品被当真话（预算截断缺位）。 subagent 没跑完，主循环按老逻辑「回退到最后一句 assistant 文本」，半截答案被当成完成结果回填、汇总、发给用户。subagent 语境下这是最危险的默认值：所以工厂检查 budgetExceeded，没跑完就失败可见（错误是数据，主 Agent 能读）。

坑六：subagent 事件混进主会话流。 把 subagent 的 text_delta 原样塞进主会话的 SSE 缓冲，前端渲染出一堆「子任务中间过程」，主会话协议被污染。subagent 事件带自己的 sessionId，走委派层的观察者（真产品可把 subagent 会话接入 ch15 的事件缓冲按 subagentId 分桶，前端订阅子任务进度，多会话机制复用，协议不新增）。

小结

下一章（ch23）是收尾：配置与部署、v01→v10 的架构演进回顾、全书技术点全景回顾。先把本章练习做完：亲手实现任务拆分、白名单隔离执行、并发委派与汇总这三层，你会体会到「多 Agent 不是新框架，是同一个 loop 的多份实例」。

延伸阅读

完成阅读，去做练习 →