第 6 章：subagent 与编排（以及何时别用）
把 PR 审查拆给多个并行 subagent，并认清那些让单 agent 反而更优的失败模式。
小满 · 分身回廊
活多到小满一个人扛不动。今天它学会派出分身。
草稿章节。跑通格式用的第一版，正式索引前会再打磨。
本章目标
你会把第 1 部分里那个单一的 pr-reviewer 拆成一个编排器（orchestrator）加三个 subagent：安全车道、性能车道、测试车道。编排器负责切分一个大 diff，把切片 fan-out 出去，给每条车道设 token 预算，再把三份局部审查合并成一份去重后的报告。然后你用同一个 PR 分别跑单 agent 版和多 agent 版，把成本和质量并排摆出来，并套用一条写死的判断规则，决定 fan-out 到底值不值。本章真正要给你的不是「一套多 agent 系统」，而是一个判断：对某个具体的 PR，你能判断自己到底需不需要它。
微软的多 agent 课程列出了几个真实的好处：专精（每个 agent 只做一件事，而不是变成一个什么都管不好的全才）、可扩展（加一条车道比把一个 prompt 塞到爆要容易）、容错（一条车道挂了不至于整次运行报废）。但这些好处每一条都有对应的代价，而大多数 PR 配不上这笔交易。
前置准备
- 完成第 5 章：一个带明确范围、最小权限、可组合的
pr-reviewerskill。 - 一个足够大的 PR，让不同关注点能真正分开（设想 800+ 改动行，同时碰到鉴权、一段热点循环和一套测试）。30 行的 PR 没什么可 fan-out 的。
- 每次运行的 token 可观测性，好让你把各车道成本加总。参见你所用厂商的用量日志，详见官方文档。
动手做
1. 先判断：这个 PR 真的需要 fan-out 吗？
在写任何编排逻辑之前，先用下面这条规则对照这个 PR。只有当工作确实相互独立、且体量大到协调税相对并行收益可以忽略时，fan-out 才划算。如果两条车道会读同一批文件、对同一批行做推理，它们就不是独立的，你会花双倍的钱拿到两份重叠的意见。
仅当以下全部成立才用 FAN-OUT：
- diff 规模 > 约 600 改动行，或 > 约 8 个跨不同关注点的文件
- 各车道碰的文件基本不相交（安全 != 性能 != 测试）
- 单 agent 跑一次已经撞到上下文上限或超时
- 某车道的发现不依赖于另一车道的发现
否则用一个 agent。一个 agent 是默认值，不是退路。2. 每个 subagent 一份狭窄职责
每个 subagent 拿到一份单一职责、它需要的那一小片 diff，别的都不给。最怕的就是重叠：如果安全车道和测试车道都检查同一个文件，你付两次钱，还得在合并时调和两份意见。在车道 prompt 里明确告诉它该忽略什么。
# 安全车道
角色：只审查所给 diff 切片里的安全缺陷。
范围：files = [auth/*, api/handlers/*]
盯：注入、越权、密钥泄露、不安全的反序列化。
忽略：风格、性能、测试覆盖。那些归别的 agent 管。
输出：JSON 列表 {file, line, severity, finding, suggested_fix}
# 性能车道
角色：只审查性能缺陷。
范围：files = [core/engine/*, db/queries/*]
盯：N+1 查询、无界循环、热点路径上的阻塞 I/O。
忽略：安全、风格。输出：同一 JSON schema。
# 测试车道
角色：只审查改动是否被充分测试。
范围：files = [**/*_test.*, 缺少对应测试的 src/**]
盯：未测分支、缺失的边界用例、被删掉的断言。
输出：同一 JSON schema。3. 把每条车道定义成一个 subagent，让 lead agent 去 fan-out
在 Claude Agent SDK 里你不用自己写「并发调度器」。你把每条车道声明成一个 AgentDefinition，挂进 options.agents；再写一份 lead（编排器）的 system prompt，把唯一的工具收成 Task。lead agent 读到一个大 diff 后，会用 Task 工具把各车道并行派出去，每个 subagent 跑在自己隔离的上下文窗口里，各自只拿到声明范围内的文件。description 字段很关键：lead 正是靠它决定何时派哪条车道。把 max_turns 当作每条车道的护栏，迷糊的车道到上限就停。
import anyio
from claude_agent_sdk import (
query, ClaudeAgentOptions, AgentDefinition,
AssistantMessage, TextBlock, ResultMessage,
)
LANES = {
"security-lane": AgentDefinition(
description="审查 diff 的安全缺陷时用它。只碰 auth 与 api/handlers 下的文件。",
prompt="你只审查安全缺陷：注入、越权、密钥泄露、不安全反序列化。"
"忽略风格、性能、测试。只输出 JSON 列表 "
"{file, line, severity, finding, suggested_fix}。",
tools=["Read", "Grep"], model="sonnet",
),
"perf-lane": AgentDefinition(
description="审查 diff 的性能缺陷时用它。只碰 core/engine 与 db/queries 下的文件。",
prompt="你只审查性能缺陷：N+1 查询、无界循环、热点路径上的阻塞 I/O。"
"忽略安全、风格。输出同一 JSON schema。",
tools=["Read", "Grep"], model="sonnet",
),
"tests-lane": AgentDefinition(
description="审查改动是否被充分测试时用它。只碰测试文件与缺测试的 src。",
prompt="你只审查测试覆盖：未测分支、缺失边界用例、被删的断言。"
"输出同一 JSON schema。",
tools=["Read", "Grep"], model="sonnet",
),
}
LEAD = """你是 PR 审查编排器。你唯一的工具是 Task。
你绝不亲自审查代码。按文件路径把 diff 切给互不相交的车道，
并行派出 security-lane / perf-lane / tests-lane，
再把它们返回的 JSON 发现原样汇总。绝不让两条车道读同一批文件。"""
options = ClaudeAgentOptions(
system_prompt=LEAD,
agents=LANES,
allowed_tools=["Task"], # lead 只能派分身，自己不读不写
max_turns=8,
cwd="./repo",
)
async def main():
async for message in query(
prompt="审查 src/payments/ 这个大 diff，按车道并行 fan-out。",
options=options,
):
if isinstance(message, AssistantMessage):
for block in message.content:
if isinstance(block, TextBlock):
print(block.text)
elif isinstance(message, ResultMessage):
print(f"用量：{message.usage} 成本：${message.total_cost_usd or 0:.4f}")
anyio.run(main)import { query } from "@anthropic-ai/claude-agent-sdk";
const LANES = {
"security-lane": {
description: "审查 diff 的安全缺陷时用它。只碰 auth 与 api/handlers 下的文件。",
prompt:
"你只审查安全缺陷：注入、越权、密钥泄露、不安全反序列化。" +
"忽略风格、性能、测试。只输出 JSON 列表 " +
"{file, line, severity, finding, suggested_fix}。",
tools: ["Read", "Grep"],
model: "sonnet",
},
"perf-lane": {
description: "审查 diff 的性能缺陷时用它。只碰 core/engine 与 db/queries 下的文件。",
prompt:
"你只审查性能缺陷：N+1 查询、无界循环、热点路径上的阻塞 I/O。" +
"忽略安全、风格。输出同一 JSON schema。",
tools: ["Read", "Grep"],
model: "sonnet",
},
"tests-lane": {
description: "审查改动是否被充分测试时用它。只碰测试文件与缺测试的 src。",
prompt:
"你只审查测试覆盖：未测分支、缺失边界用例、被删的断言。" +
"输出同一 JSON schema。",
tools: ["Read", "Grep"],
model: "sonnet",
},
};
const LEAD = `你是 PR 审查编排器。你唯一的工具是 Task。
你绝不亲自审查代码。按文件路径把 diff 切给互不相交的车道，
并行派出 security-lane / perf-lane / tests-lane，
再把它们返回的 JSON 发现原样汇总。绝不让两条车道读同一批文件。`;
const q = query({
prompt: "审查 src/payments/ 这个大 diff，按车道并行 fan-out。",
options: {
systemPrompt: LEAD,
agents: LANES,
allowedTools: ["Task"], // lead 只能派分身，自己不读不写
maxTurns: 8,
cwd: "./repo",
},
});
for await (const message of q) {
if (message.type === "assistant") {
for (const block of message.message.content) {
if (block.type === "text") console.log(block.text);
}
} else if (message.type === "result") {
console.log("用量与成本：", message.usage, message.total_cost_usd);
}
}4. 慎重合并：去重并调和冲突
合并这一步，正是多 agent 系统最容易出问题的地方。subagent 从没看过彼此的推理，于是会产出重复（两条车道因不同理由标记同一行），有时还会矛盾（性能想加缓存，安全把缓存标成数据泄露途径）。这一步不是 SDK 调用，而是你跑在各车道结构化发现之上的普通代码。因为车道吐的是 JSON 而非散文，合并大体上可以走机械逻辑，只在真正冲突时才调一次模型。下面这个函数接收各车道汇总来的 findings 列表，去重、调和、按严重度排序。
def merge(findings):
by_loc = group_by(findings, key=lambda f: (f.file, f.line))
report = []
for loc, group in by_loc.items():
if len(group) == 1:
report.append(group[0])
else:
# 同一行、多条车道：保留最高严重度，
# 若建议相互矛盾，升级到编排器模型裁决
if conflicting_fixes(group):
report.append(resolve_conflict(loc, group)) # 1 次小 LLM 调用
else:
report.append(highest_severity(group))
return sorted(report, key=lambda f: f["severity"], reverse=True)function merge(findings) {
const byLoc = groupBy(findings, (f) => `${f.file}:${f.line}`);
const report = [];
for (const group of Object.values(byLoc)) {
if (group.length === 1) {
report.push(group[0]);
} else {
// 同一行、多条车道：保留最高严重度，
// 若建议相互矛盾，升级到编排器模型裁决
if (conflictingFixes(group)) {
report.push(resolveConflict(group)); // 1 次小 LLM 调用
} else {
report.push(highestSeverity(group));
}
}
}
return report.sort((a, b) => severityRank(b) - severityRank(a));
}5. 在同一个 PR 上与单 agent 对比
用第 5 章那个单 agent reviewer 跑完全相同的 PR。两边都记录总 token、墙钟延迟和发现集合。现在你能回答唯一要紧的问题：fan-out 是真的找出了单 agent 漏掉的真实缺陷，还是只是多花三到四倍 token，产出一份更吵的同款审查？
单 agent 多 agent（3 车道 + 合并）
token ~38k ~140k (~3.7 倍)
延迟 ~22s ~31s (并行了，但合并多一跳)
真发现 9 11 (+2 真问题，都在性能车道)
重复 0 4 (合并时调和掉)
结论：fan-out 这次值，只是因为性能车道在一条单 agent 草草扫过的
热点路径上挖得更深。换个更小的 PR，就不值了。习得「派分身，也敢不派」小满能把一个大 diff 切成几条不相交的车道，并行派给几个 subagent 各管一摊。更要紧的是，它会先用那条规则问一句：这个 PR 到底值不值得分？多数时候答案是不分。
如何验证
- 确认每个 subagent 守在自己车道：grep 它的输出，看有没有越出声明范围的发现。一旦泄漏，说明你的「忽略」指令太弱。
- 把所有车道加编排器那次合并调用的 token 加总，摆在单 agent 那次旁边。如果多 agent 连质量都没打平，那笔多花的钱就是纯浪费。
- 检查合并后的报告里有没有残留重复或没调和的矛盾。任意一种都说明你的薄弱环节是合并步骤（第 4 步），而非车道本身。
- 故意杀掉一条车道（让它返回错误）。这次运行仍应产出一份局部报告。如果它直接崩了，你其实没买到容错。
习得「把分身的账算清」你现在能把多 agent 那次的总 token 和单 agent 那次摆在一起，再故意杀掉一条车道看它会不会降级成局部报告，由此判断 fan-out 是真省了事，还是只多花三四倍钱买了一份更吵的同款审查。
原理
每个 subagent 跑在自己的上下文窗口里，于是一条只读 auth 文件的车道，绝不会花 token 去推理测试套件。这就是「专精」背后真正起作用的机制：不是模型在每条车道里变聪明了，而是每条车道的上下文更小、更干净、更相关。第 3 章讲的上下文工程，本来是在一个 agent 内部做，这里把它搬到了多个进程之间。fan-out 还把对一个巨大 diff 的串行阅读，变成对各切片的并发阅读，这就是切片真正不相交时延迟收益的来源。
但这一切都不免费。编排器和合并是纯开销：单 agent 本来不用付的 token 和一次往返。你还失去了单 agent 免费带来的横切推理，就是那种注意到一处鉴权改动、于是顺手把相关的一条查询看得更狠的联动。微软的课程说得很直白：多 agent 只在那种能「分解成专精子任务」的「复杂任务」上才用得上。一个中等 PR 不是那种任务。
小结
当工作确实相互独立、diff 大到协调开销只是误差项、且单 agent 已经撞到上下文或延迟墙时，subagent 才有帮助。当关注点重叠（你为冗余意见付双倍）、合并时丢上下文（重复和矛盾溜进报告）、或 token 成本膨胀盖过质量收益时，它反而坑你。默认用一个 agent；只有当第 1 步的判断规则说「行」、且第 5 步的对比证明它划算时，才上多个。下一章在 OpenAI Codex 里重建同一个 PR reviewer，提炼出跨厂商可迁移的东西。
常见坑
- 默认上多 agent。 最常见的错误，是觉得 fan-out 显得「高级」就上它。多数 PR 并不需要；协调开销通常盖过并行收益。一个 agent 是默认值，不是退路。
- 车道重叠。 两个 subagent 读同一批文件，你付两次钱，还白捡一个调和难题。按不相交的路径切分，并写明确的「忽略」指令。
- 合并时丢上下文。 subagent 之间从不共享推理。让它们输出结构化发现，并慎重调和冲突，否则重复和矛盾会漏进报告。
- 没预算导致成本膨胀。 一条犯迷糊的车道会打转，把你的预算抽干。给每条车道设 token 上限，为合并留余量，并设一个全局天花板。
- 图上画着容错，实际没有。 如果一条挂掉的车道把整次运行带崩，你就只付了多 agent 的成本，却没拿到它最主要的韧性好处。让编排器降级成一份局部报告。
小满第一次派分身替它跑腿，却也第一次因为「该不该派」判断失误，把一件简单的活搞复杂了。它学会了一件更难的事：何时别分身。分身回廊，亮了。
刚点亮 分身回廊 · 地图已点亮 7 / 16
来源
- Anthropic Agent Skills 官方文档 · official
- 多 agent 设计模式（Microsoft AI Agents for Beginners 第 8 课） · official