Harness Engineering

第16章 多 Agent 协调模式

作者 杨艺韬 · 8,747 字

"The strength of the team is each individual member. The strength of each member is the team." — Phil Jackson

本章要点

  • 多 Agent 的两大核心价值：专业化分工 + 上下文隔离——缺一不可
  • 五种协调拓扑：Coordinator（树）/ Pipeline（链）/ Swarm（对等/网状）/ Debate（网）/ Hierarchical（多层树）
  • 上下文隔离是多 Agent 的一级工程价值，而非副产品
  • 成本控制：多 Agent 会放大 API 调用、上下文复制和协调开销，必须有前置预算机制
  • 失败传播：子 Agent 失败模式会相互叠加，必须有显式降级策略
  • 不要过度设计：单 Agent 够用就不要上多 Agent——复杂度是有成本的

16.1 为什么需要多 Agent：单 Agent 的两个天花板

单 Agent 架构在工程实践中会撞上两堵墙。

第一堵墙：上下文窗口有限。

即使模型上下文窗口已经很大，在真实的大型代码库任务中依然可能不够用。典型场景：用户要求"重构整个认证模块"——涉及大量文件、跨层依赖、测试日志和历史决策——单个 Agent 的上下文很容易被无关中间过程填满。

更隐蔽的问题是上下文退化（Context Degradation）：上下文越长，模型越容易被中间结果、工具日志和局部细节牵引，早期目标和全局约束会被稀释。多 Agent 的价值之一，就是把探索过程隔离在子 Agent 内，只把结论传回主 Agent。

第二堵墙：专业化困难。

用一个 system prompt 让模型同时擅长：

  • 代码架构设计（需要宏观视野）
  • 具体编码实现（需要语法精度）
  • Code Review（需要批判性思维）
  • 测试编写（需要边界思维）
  • 文档撰写（需要用户视角）

这在工程上很容易变成角色混杂：提示词每加入一个角色的指导，模型需要同时权衡的行为模式就更多，冲突也更多。多 Agent 把这种混杂拆成多个专用上下文。

graph TD
    subgraph Single["❌ 单 Agent 的困境"]
        SA["一个 Agent<br/>一个上下文窗口<br/>一个 system prompt"]
        SA --> P1["读 30 个文件"]
        SA --> P2["写代码"]
        SA --> P3["写测试"]
        SA --> P4["Code Review"]
        SA --> P5["写文档"]
        P1 & P2 & P3 & P4 & P5 --> OOM["💥 上下文溢出<br/>+ 角色精分<br/>+ 注意力稀释"]
    end
    subgraph Multi["✅ 多 Agent 的解法"]
        MA["Coordinator<br/>只保留全局计划"]
        MA -->|"派发任务 A"| A1["研究 Agent<br/>独立上下文 + 专用 prompt"]
        MA -->|"派发任务 B"| A2["编码 Agent<br/>独立上下文 + 专用 prompt"]
        MA -->|"派发任务 C"| A3["测试 Agent<br/>独立上下文 + 专用 prompt"]
        A1 -->|"仅返回结论"| MA
        A2 -->|"仅返回结论"| MA
        A3 -->|"仅返回结论"| MA
    end
    style OOM fill:#fee2e2,stroke:#ef4444,stroke-width:2px
    style MA fill:#dbeafe,stroke:#3b82f6,stroke-width:2px
    style SA fill:#fef3c7,stroke:#f59e0b

多 Agent 的解法思路：把大问题拆成子问题，每个子 Agent 拥有独立的上下文窗口 + 专门的 system prompt，各司其职。

但这不是银弹——复杂度是有成本的。多 Agent 引入了新的工程挑战：协调、通信、错误传播、成本控制。本章的目标，就是把这些挑战讲清楚。

16.2 五种协调拓扑的系统对比

多 Agent 的核心是拓扑结构——Agent 之间的组织形式决定了系统的行为。工业界有五种主流拓扑：

graph TB
    subgraph T1["1. Coordinator（树形）"]
        C1((Main)) --> C2((Sub A))
        C1 --> C3((Sub B))
        C1 --> C4((Sub C))
    end
    subgraph T2["2. Pipeline（链式）"]
        P1((A1)) --> P2((A2)) --> P3((A3)) --> P4((A4))
    end
    subgraph T3["3. Swarm（对等）"]
        S1((A)) <--> S2((B))
        S2 <--> S3((C))
        S3 <--> S1
    end
    subgraph T4["4. Debate（辩论）"]
        D1((正方))
        D2((反方))
        D3((裁判))
        D1 -.->|"观点"| D3
        D2 -.->|"观点"| D3
        D3 --> D4((结论))
    end
    subgraph T5["5. Hierarchical（分层）"]
        H1((CEO)) --> H2((PM))
        H2 --> H3((DevA))
        H2 --> H4((DevB))
        H1 --> H5((QA Lead))
        H5 --> H6((QA1))
    end
    style C1 fill:#dbeafe,stroke:#3b82f6
    style H1 fill:#fef3c7,stroke:#f59e0b
    style P1 fill:#dcfce7,stroke:#22c55e
    style S1 fill:#f3e8ff,stroke:#a855f7
    style D3 fill:#fce7f3,stroke:#ec4899
拓扑 控制流 适用任务 优势 劣势 代表系统
Coordinator 中心化 职责清晰的工程任务 简单、可控、易调试 主 Agent 是瓶颈 Claude Code Agent 工具
Pipeline 线性 固定流程（ETL、审批） 阶段清晰、便于检查 延迟累加、僵化 LangGraph
Swarm 去中心化 探索性、跨领域任务 灵活、可扩展 控制流不可预测 Swarm / handoff 类框架
Debate 投票 事实核查、决策 减少幻觉、提升准确率 成本高、收敛慢 多视角辩论/自审系统（Constitutional AI 是单模型自批判，不属 Debate 拓扑）
Hierarchical 多层中心 大型复杂工程 可以扩展到很大规模 工程复杂度高 AutoGen、CrewAI

下面逐一深入分析。

16.3 Coordinator 模式（树形）

一个"管理者"Agent 负责理解需求、拆分任务、分派给专业子 Agent、汇总结果。这是工业界用得最多的模式。

         ┌──── 研究 Agent (探索代码库)
         │
主 Agent ─┼──── 实现 Agent (编写代码)
         │
         └──── 测试 Agent (运行测试)

Claude Code 的 Agent 工具

Claude Code 的 Agent 工具就是这个模式的典型实现：

// 以下是对 Agent tool 调用的教学化简写（伪代码）。
// 实际 SDK 中，Agent 通常由模型通过 tool_use 触发，结果放在 content[0].text 等字段中。
// 主 Agent 决定启动子 Agent
const result = await Agent({
  description: "探索认证模块结构",
  prompt: `在 /src 目录下搜索所有与 AuthContext 相关的文件，
           分析每个文件中 AuthContext 的用法（读取/修改/创建），
           重点关注：
           1. 是否有直接操作 session token 的代码
           2. 是否有绕过 AuthContext 直接读写 cookie 的代码
           3. 列出所有调用 useAuth() 的组件及其用途
           返回简洁的结构化报告，不要贴原始代码。`,
  subagent_type: "Explore",
})

// 子 Agent 完成后，结果返回给主 Agent
// 主 Agent 基于结果决定下一步
if (result.findings.includes("绕过 AuthContext")) {
  // 启动另一个子 Agent 修复
  await Agent({ subagent_type: "general-purpose", ... })
}

三条关键设计原则

原则一：任务描述的质量决定子 Agent 的输出质量。

子 Agent 的输出 = f(任务描述)。这是单向的函数——糟糕的任务描述不可能产生好的结果。

❌ 坏的任务描述:
"帮我查一下代码"
"看看有没有 bug"
"随便探索一下"

✅ 好的任务描述:
"在 /src/auth 目录下搜索所有使用了 AuthContext 的文件，
 按以下维度列出：
 - 文件路径
 - AuthContext 的用法（读取/修改/创建）
 - 是否有直接操作 session token 的代码
 返回 Markdown 表格格式，不要贴原始代码。"

好的任务描述的特征：

  1. 目标明确——做什么（任务名）+ 为什么（上下文）+ 结果形式（输出规范）
  2. 范围清晰——哪些文件/目录在范围内，哪些不在
  3. 输出格式——Markdown / JSON / 表格 / 纯文本
  4. 禁止事项——"不要贴原始代码"、"不要返回中间过程"

原则二：结果传递要结论化。

子 Agent 的输出会进入主 Agent 的上下文。如果子 Agent 把大量原始文件、日志和中间推理都贴回来，主 Agent 的上下文窗口就会被吞掉一大块。

工程上的原则是：子 Agent 的输出必须结论化。也就是说，子 Agent 不是把读到的内容搬运回来，而是在做信息压缩：读取大量材料，返回少量可执行结论、证据路径和风险点。

// ❌ 糟糕的子 Agent 设计
// 子 Agent 返回：
"以下是我读到的 30 个文件的完整内容：
 file1.ts: [5KB 源码]
 file2.ts: [8KB 源码]
 ..."

// ✅ 好的子 Agent 设计
// 子 Agent 返回：
"认证模块架构分析（共读取 30 个文件）:

核心组件:
1. AuthContext (src/auth/context.ts) - 全局 JWT 状态
2. useAuth hook (src/auth/useAuth.ts) - 组件级访问
3. AuthGuard (src/auth/guard.tsx) - 路由保护

发现的问题:
- src/pages/Profile.tsx:42 绕过 AuthContext 直接读 localStorage
- src/api/user.ts:15 硬编码 token 刷新逻辑

建议修复路径:
- 优先级 P0: Profile.tsx 的绕过问题
- 优先级 P1: user.ts 的硬编码逻辑"

原则三：子 Agent 不应递归无限深度。

主 Agent 启动子 Agent，子 Agent 又启动子子 Agent——这种递归调用很容易失控。实践建议是限制递归深度，优先采用主 → 子的扁平结构；更深的嵌套通常说明任务拆分得不够合理。

本地快照：Claude Code 用 disallowedTools 限制递归

§16.3 的"原则三"在本地 ../claude-code-main（作者本地源码快照，非公开可验证）不只是建议。打开 src/tools/AgentTool/built-in/，Explore、Verification、Plan 三个内置子 Agent 都把 AGENT_TOOL_NAME 放进 disallowedTools：

// exploreAgent.ts:67-73、verificationAgent.ts:139-145、planAgent.ts:77-83
disallowedTools: [
  AGENT_TOOL_NAME,        // ← 子 Agent 不能再启动子 Agent
  EXIT_PLAN_MODE_TOOL_NAME,
  FILE_EDIT_TOOL_NAME,
  FILE_WRITE_TOOL_NAME,
  NOTEBOOK_EDIT_TOOL_NAME,
],

AGENT_TOOL_NAME 出现在 disallowedTools 里，意味着这些子 Agent 不能再调用 Agent 工具。resolveAgentTools() 会把 disallowed tool 从可用工具中过滤掉（src/tools/AgentTool/agentToolUtils.ts:149-160），runAgent.ts 在启动 agent 时使用解析后的工具列表（src/tools/AgentTool/runAgent.ts:500-502）。这把"不要递归启动 Agent"从 prompt 建议变成了能力边界。

三个 subagent 还都禁了文件编辑相关工具（FILE_EDIT / FILE_WRITE / NOTEBOOK_EDIT）。Explore 的定义还设置了 omitClaudeMd: true，并在注释中说明它是 read-only search agent，不需要 commit/PR/lint 规则（src/tools/AgentTool/built-in/exploreAgent.ts:76-82）。prompt 里说明职责，工具列表里去掉能力，这是"指令 + 能力联动"的典型做法。

这是"指令 + 能力联动"的范例：§9 讨论的"指令优先级"靠 prompt 表达，§16 讨论的"递归限制"靠 disallowedTools 强制。Prompt + Capability 双轨比单靠任一边都更可靠。

适用场景

Coordinator 模式适合：

  • 主任务可以清晰分解为独立子任务
  • 子任务之间没有紧耦合
  • 任务结构是"先拆分、再汇总"的扇出扇入模式

不适合：

  • 任务需要多轮对话才能明确（用 Swarm）
  • 子任务之间有强依赖（用 Pipeline）
  • 需要多个 Agent 独立给出观点再综合（用 Debate）

16.4 Pipeline 模式（链式）

Agent 按顺序串联，每个处理一个阶段。适合流程化、阶段性的任务。

需求分析 Agent → 方案设计 Agent → 代码实现 Agent → Review Agent → 测试 Agent

LangGraph 实现

from langgraph.graph import StateGraph, START, END
from typing import TypedDict, List

class PipelineState(TypedDict):
    requirement: str
    design_doc: str
    code: str
    review_feedback: str
    test_results: str
    iteration: int  # 允许 review 打回 developer 修改

def analyst_agent(state: PipelineState) -> PipelineState:
    # 分析需求，产出详细的需求文档
    ...

def architect_agent(state: PipelineState) -> PipelineState:
    # 基于需求设计架构
    ...

def developer_agent(state: PipelineState) -> PipelineState:
    # 基于设计实现代码（考虑 review 反馈）
    ...

def reviewer_agent(state: PipelineState) -> PipelineState:
    # Code Review；每轮 review 都要递增 iteration，否则 route_review
    # 的"超过 3 轮上报人类"永远不会触发
    # return {"review_feedback": ..., "iteration": state["iteration"] + 1}
    ...

def route_review(state: PipelineState) -> str:
    if state["review_feedback"].startswith("APPROVED"):
        return "tester"
    elif state["iteration"] >= 3:
        return "human_escalation"  # 超过 3 轮循环，上报人类
    else:
        return "developer"  # 打回去改

def tester_agent(state: PipelineState) -> PipelineState:
    # 运行测试并记录结果
    ...

def escalation_agent(state: PipelineState) -> PipelineState:
    # 生成需要人工介入的报告
    ...

graph = StateGraph(PipelineState)
graph.add_node("analyst", analyst_agent)
graph.add_node("architect", architect_agent)
graph.add_node("developer", developer_agent)
graph.add_node("reviewer", reviewer_agent)
graph.add_node("tester", tester_agent)
graph.add_node("human_escalation", escalation_agent)

graph.add_edge(START, "analyst")  # 入口边，缺了它 graph 无法启动
graph.add_edge("analyst", "architect")
graph.add_edge("architect", "developer")
graph.add_edge("developer", "reviewer")
graph.add_conditional_edges("reviewer", route_review)  # 循环 + 分支
graph.add_edge("tester", END)
graph.add_edge("human_escalation", END)

app = graph.compile()  # 编译后才能 invoke

Pipeline 的三个陷阱

陷阱一：延迟累加。

单 Agent 延迟 = 1 次 LLM 调用 Pipeline 延迟 = N 次 LLM 调用 × 串行

只要每个阶段都要等待一次模型调用，Pipeline 的延迟就会串行累加。阶段越多，用户越容易感知到等待。

陷阱二：错误传播。

早期阶段的错误会被放大到下游。如果需求分析错了，后续 4 个 Agent 都是在错误的方向上做功。

工程补救：每个阶段后加 sanity check。

def route_after_analyst(state):
    if not state["requirement_analysis"]:
        return "escalate_to_human"
    if contains_ambiguity(state["requirement_analysis"]):
        return "analyst"  # 回炉重造
    return "architect"

陷阱三：反馈循环失控。

Review Agent 可以打回 Developer Agent 重写——这是好事，体现了迭代思维。但如果没有上限，可能会无限循环：

developer → reviewer (不通过) → developer → reviewer (不通过) → ...

必须设置迭代上限，超过就升级到人类或返回部分结果。

16.5 Swarm 模式（去中心化）

没有固定的管理者，Agent 之间根据当前需要动态切换控制权。Swarm / handoff 类框架代表了这种设计方向。

用户 → 路由 Agent → "这是个前端问题" → 前端 Agent
                                       ↓
                            "需要改 API" → 后端 Agent
                                           ↓
                              "需要更新文档" → 文档 Agent

核心概念是 handoff——当前 Agent 发现任务超出自己专长时，主动将控制权交给更合适的 Agent：

def frontend_agent(context):
    # 处理前端问题...
    if needs_api_change:
        return handoff(backend_agent, context="需要修改 /api/users 端点")

def backend_agent(context):
    # 处理后端问题...
    if needs_docs_update:
        return handoff(docs_agent, context="API 变更需要更新文档")

控制流不可预测——这是特性还是 bug？

Swarm 的核心权衡：灵活性 vs 可预测性。

  • 灵活性：Swarm 适合需求不明确的探索性任务。你不知道最终会需要哪些专家介入。
  • 不可预测：你无法保证任务会在有限步骤内完成。可能出现：前端 → 后端 → 数据库 → 后端 → 前端 → 后端 → ...

工程补救：

  1. 强制终止条件：超过 N 次 handoff 就停止
  2. handoff 日志：每次 handoff 都记录，便于事后分析
  3. 显式回到 Router：允许当前 Agent 放弃控制权，让 Router Agent 重新决策

Swarm 的正确打开方式

Swarm 不是"让 Agent 自由协作"——这样会失控。正确的用法：

  • 每个专业 Agent 的领域边界清晰定义
  • handoff 的条件明确——什么情况下该把任务交出去
  • 有全局的终止 Agent——任务完成时由谁宣布结束

16.6 Debate 模式（辩论）

多个 Agent 独立给出观点，然后由一个裁判 Agent 综合。这是一些 reasoning / 多视角评审系统使用的模式；需要注意，Anthropic 的 Constitutional AI 更偏向单模型自批判，不是多 Agent 辩论拓扑。

sequenceDiagram
    participant User
    participant Moderator as 裁判 Agent
    participant Pro as 正方 Agent
    participant Con as 反方 Agent
    participant Critic as 批评家 Agent
    User->>Moderator: 这个架构方案合理吗？
    par 三个独立观点
        Moderator->>Pro: 从"可以"的角度论证
        and
        Moderator->>Con: 从"不可以"的角度论证
        and
        Moderator->>Critic: 指出方案的所有薄弱点
    end
    Pro-->>Moderator: 支持观点 + 理由
    Con-->>Moderator: 反对观点 + 理由
    Critic-->>Moderator: 关键薄弱点清单
    Moderator->>Moderator: 综合三方观点
    Moderator-->>User: 结构化结论 + 推理链

为什么 Debate 有价值？

单 Agent 容易锚定——一旦它给出了一个观点，后续推理都围绕这个观点展开（偏见）。

Debate 的核心机制：强制让不同的 Agent 从不同角度独立思考，再综合。这能帮助暴露：

  • 事实性错误（Factual Errors）
  • 逻辑漏洞（Logical Gaps）
  • 单视角偏见（Perspective Bias）

Debate 的成本

Debate 的代价是更高的成本和延迟：

  • 3 个独立 Agent × N 轮对话 × 长上下文
  • 裁判 Agent 还要读取多方观点并综合

因此 Debate 适合高价值、低容错的任务：

  • 关键架构决策
  • 安全相关代码 Review
  • 医疗/法律场景的建议

不适合：

  • 日常编码任务（成本不划算）
  • 对延迟敏感的任务

16.7 Hierarchical 模式（分层）

Coordinator 模式只有两层。当任务更复杂时，可以扩展到多层：

CEO Agent (战略层)
  ├─ Product Manager Agent (产品层)
  │   ├─ Developer Agent A
  │   └─ Developer Agent B
  ├─ QA Lead Agent (测试层)
  │   ├─ QA Agent 1
  │   └─ QA Agent 2
  └─ Doc Lead Agent (文档层)
      └─ Doc Writer Agent

AutoGen、CrewAI 等框架都支持类似的多层结构。

分层的价值

分层设计带来的核心价值是关注点分离：

  • 战略层：定义目标、分配资源
  • 战术层：拆分任务、协调执行
  • 执行层：具体工作

每一层只需关注自己的职责，复杂度被逐层压缩。

分层的代价

层数每增加一层，整体复杂度都会上升：

  • 通信开销：N 层意味着消息要经过 N 跳
  • 失败放大：任何一层失败都会影响上下游
  • 调试难度：问题排查要跨多层追踪

经验法则：层级越少越好。层级过深通常说明任务拆分得不够合理，或者用错了拓扑。

16.8 上下文隔离：多 Agent 的一级价值

多 Agent 最大的工程价值，不是分工——而是上下文隔离。

单 Agent 的上下文污染问题

单 Agent 处理复杂任务时，上下文窗口会被各种中间结果填满——读的文件、搜索的结果、执行的命令输出、调试的日志。到后期，模型会因为上下文过于拥挤而"迷失"——忘记最初的任务目标、抓不住关键信息、对早期指令的注意力下降。

用户任务
↓
LLM 读了很多文件
↓
LLM 运行多次测试
↓
LLM 多轮调试失败
↓
LLM 最终要输出结果——此时上下文已经很重
     原始任务目标已淹没在噪声里

多 Agent 的上下文隔离

多 Agent 方案中，每个子 Agent 启动时拥有干净的上下文：

// 主 Agent 上下文: 用户需求 + 高层计划
// ↓ 启动子 Agent
// 子 Agent 上下文: 任务描述 + 该任务需要的文件
//   ├─ 子 Agent 读取和搜索大量材料——只在子 Agent 上下文
//   └─ 子 Agent 生成结论——返回给主 Agent
// ↓
// 主 Agent 上下文: 用户需求 + 高层计划 + 子 Agent 结论

关键原则：子 Agent 的中间过程（读了哪些文件、搜了几次）不应污染主 Agent 的上下文。只有最终结论传递回来。

上下文隔离的四条实践原则

  1. 子 Agent 的输出要"结论化"——不是贴原始数据，而是总结和建议
  2. 主 Agent 不读子 Agent 的中间日志——除非调试需要
  3. 每个子 Agent 的系统 prompt 独立设计——让它专注自己的领域
  4. 主 Agent 的上下文要保持"战略层"——不处理战术细节

16.9 通信协议：共享状态 vs 消息传递

Agent 之间如何传递信息？工业界有两种主流方案。

共享状态（LangGraph 风格）

所有 Agent 读写同一个 State 对象：

class SharedState(TypedDict):
    messages: list             # 所有 Agent 共享
    research_results: str      # 研究 Agent 写入
    implementation_plan: str   # 架构 Agent 写入
    code_changes: list         # 实现 Agent 写入
    review_feedback: str       # Review Agent 写入
    test_results: str          # 测试 Agent 写入

def researcher(state: SharedState) -> SharedState:
    # 读：state["messages"]
    # 写：state["research_results"]
    ...

优点：

  • 简单直接，所有 Agent 共享信息
  • 信息传递零成本（内存共享）
  • State 是可检查的——调试、重放、持久化都方便

缺点：

  • Agent 之间耦合度高——修改 State 结构会影响所有 Agent
  • 隔离性弱——Agent 可能读到不相关的其他 Agent 输出
  • 并发安全问题——多 Agent 并行时需要锁

消息传递（Claude Code 风格）

主 Agent 通过 prompt 给子 Agent 传递任务，子 Agent 通过返回值传递结果。没有共享状态——纯消息传递。

// 主 Agent 发送
const result = await Agent({
  prompt: `任务: ${task}\n\n参考信息: ${context}`,
  subagent_type: "Explore",
})

// 子 Agent 返回
return `研究结果: ${findings}\n\n建议: ${recommendations}`

优点：

  • 隔离性强——每个 Agent 只看到自己的上下文
  • 天然支持并行——Agent 之间无共享状态
  • 调试简单——每个 Agent 独立可重放

缺点：

  • 传递大量数据时效率低（需要序列化为文本）
  • Agent 之间无法"实时"同步——只能在调用边界通信

通信模式对比

维度 共享状态（LangGraph） 消息传递（Claude Code）
耦合度 高（所有 Agent 读写同一 State） 低（纯输入→输出）
隔离性 弱（Agent 可能读到无关数据） 强（每个 Agent 只看到自己的上下文）
数据传递效率 高（内存共享） 低（序列化为文本）
并发安全 需要锁/版本机制 天然安全
调试难度 中（State 是可检查的） 低（每个 Agent 独立可重放）
扩展到分布式 难（需要分布式 State） 易（消息天然可网络传输）
适用场景 紧密协作的工作流 松散的任务分发 
sequenceDiagram
    participant Main as 主 Agent
    participant Sub1 as 研究 Agent
    participant Sub2 as 编码 Agent
    Main->>Sub1: "搜索所有认证相关文件\n分析架构..."
    Note over Sub1: 独立上下文<br/>读取 20 个文件<br/>生成分析报告
    Sub1-->>Main: "发现 3 个认证模块:<br/>1. JWT 验证...<br/>2. OAuth..."
    Note over Main: 只收到结论<br/>不收到 20 个文件的原始内容
    Main->>Sub2: "基于以下分析修改认证逻辑:<br/>[研究结论摘要]"
    Note over Sub2: 独立上下文<br/>按需读取相关文件
    Sub2-->>Main: "已修改 auth.ts<br/>测试通过"

选择建议

  • 任务紧密协作（同一个工作流内的不同阶段）→ 共享状态
  • 任务松散独立（可以独立完成）→ 消息传递
  • 需要分布式扩展 → 消息传递
  • 需要 Time Travel 调试 → 共享状态（带 checkpointer）

16.9.1 子 Agent 返回契约

多 Agent 系统最容易失败的地方，不是"子 Agent 做不了事"，而是"子 Agent 返回的东西主 Agent 用不了"。所以每一种子 Agent 都应该有返回契约，而不是自由发挥。

一个实用的返回契约可以包含六个字段：

字段 作用
status success / partial / blocked / failed，让主 Agent 先判断流程
summary 一段可直接放入主上下文的结论
evidence 支撑结论的文件路径、命令、测试名、错误片段
changes 如果有修改，列出文件和行为摘要
risks 不确定点、未验证点、潜在副作用
next_actions 建议主 Agent 下一步做什么
error（可选） 失败时的类型与消息，例如 { type: "network_timeout", message: "..." }
partialResult（可选） status === "partial" 时携带的部分结果，供下游降级使用 
{
  "status": "partial",
  "summary": "认证模块有两处绕过 AuthContext 的访问路径。",
  "evidence": [
    "src/pages/Profile.tsx uses localStorage directly",
    "src/api/user.ts refreshes token outside AuthContext"
  ],
  "changes": [],
  "risks": ["没有检查移动端入口"],
  "next_actions": ["读取 Profile.tsx", "确认 token 刷新逻辑的调用链"]
}

这个契约有三个好处。第一，主 Agent 不需要从长文本里猜子 Agent 是否成功。第二，失败和部分完成能进入流程控制，而不是混在自然语言里。第三，证据路径和风险点能支持后续验证，不需要把子 Agent 的全部上下文搬回来。

如果子 Agent 返回值不满足契约，主 Agent 应该把它当作一次可恢复失败：要求子 Agent 重写摘要、补 evidence，或者降低该结果的可信度并继续执行。返回契约是多 Agent 系统的"接口类型"，没有它，多 Agent 只是多个聊天窗口的拼接。契约还应该被测试：给子 Agent 一个固定任务，检查它是否返回合法状态、证据路径和风险说明，而不是只看自然语言是否顺眼。这样主 Agent 才能稳定编排，而不是临场解释每个子 Agent 的输出，也方便后续记录、回放和评估，降低排障成本。越复杂的拓扑，越需要这种明确接口和稳定边界，否则系统很难长期维护，也更容易做自动化回归。

16.10 成本控制：多 Agent 的组合爆炸

多 Agent 的 API 调用量和上下文量都需要单独建模：

单 Agent 成本 = 主循环 tokens + 工具结果 tokens
多 Agent 成本 = 主 Agent 协调 tokens + Σ(子 Agent tokens) + 结果汇总 tokens

更糟糕的是上下文放大效应：主 Agent 的每一轮都要携带子 Agent 结论列表，这些结论会越积越多。如果子 Agent 返回的是原始数据而不是摘要，主 Agent 的上下文会很快被污染。

预算机制设计

// 预算控制核心类型（教学简化，实际项目可扩展）
interface BudgetConfig {
  maxTokens: number
  maxSubAgents: number
  maxDepth: number
  warnThreshold: number
  spawnStopThreshold: number
}
interface AgentContext {
  depth: number
}
interface CheckResult {
  allowed: boolean
  reason?: string
}
class BudgetExceededError extends Error {}

class MultiAgentBudget {
  private totalTokensUsed = 0
  private subAgentCount = 0
  private readonly config: BudgetConfig

  constructor(config: BudgetConfig) {
    // 例如 { maxTokens, maxSubAgents, maxDepth, warnThreshold, spawnStopThreshold }
    this.config = config
  }

  // 启动子 Agent 前检查
  canSpawn(context: AgentContext): CheckResult {
    if (this.subAgentCount >= this.config.maxSubAgents) {
      return { allowed: false, reason: "max sub-agents reached" }
    }
    if (context.depth >= this.config.maxDepth) {
      return { allowed: false, reason: "max recursion depth reached" }
    }
    if (this.totalTokensUsed > this.config.maxTokens * this.config.spawnStopThreshold) {
      return { allowed: false, reason: "token budget near exhaustion" }
    }
    return { allowed: true }
  }

  // 子 Agent 完成后记录
  recordUsage(tokens: number): void {
    this.totalTokensUsed += tokens
    this.subAgentCount++

    if (this.totalTokensUsed > this.config.maxTokens * this.config.warnThreshold) {
      logger.warn(`Budget usage: ${this.totalTokensUsed}/${this.config.maxTokens}`)
    }
    if (this.totalTokensUsed > this.config.maxTokens) {
      throw new BudgetExceededError(
        `Token budget exhausted: ${this.totalTokensUsed}/${this.config.maxTokens}`
      )
    }
  }

  getTotalTokensUsed(): number {
    return this.totalTokensUsed
  }
}

实践建议

  1. 设置每次任务的 token 上限——作为硬限制
  2. 子 Agent 的数量设上限（如最多 5 个并发）
  3. 对深层递归设限制（通常 ≤ 2 层）
  4. 监控每个子 Agent 的单次消耗——防止单个 Agent 失控
  5. 预留缓冲——不要把预算用到极限，给汇总、重试和用户沟通留空间
  6. 分级预算——重要任务给更多预算，日常任务给紧预算

16.11 错误传播：多 Agent 的失败模式矩阵

子 Agent 失败时，主 Agent 必须优雅处理。多 Agent 的失败模式是组合爆炸——任何一个子 Agent 都可能以多种方式失败。

失败模式分类

graph TD
    Fail[子 Agent 失败]
    Fail --> F1[技术性失败]
    Fail --> F2[语义性失败]
    Fail --> F3[部分完成]
    F1 --> F1a[网络超时]
    F1 --> F1b[Token 超限]
    F1 --> F1c[API 配额耗尽]
    F2 --> F2a[任务理解错误]
    F2 --> F2b[结果质量不佳]
    F2 --> F2c[偏离了主 Agent 预期]
    F3 --> F3a[完成部分子任务]
    F3 --> F3b[发现不可完成]
    F3 --> F3c[需要更多信息]
    style F1 fill:#fee2e2,stroke:#ef4444
    style F2 fill:#fef3c7,stroke:#f59e0b
    style F3 fill:#dbeafe,stroke:#3b82f6

分级处理策略

// 伪代码：spawnAgent、Task、Result、logger 为项目特定实现，返回值遵循 §16.9.1 的契约。
async function coordinateTask(task: Task): Promise<Result> {
  const maxRetries = 2

  for (let attempt = 0; attempt <= maxRetries; attempt++) {
    try {
      const research = await spawnAgent("research", task.description)

      // 技术性失败 - 重试
      if (research.status === "failed" && research.error?.type === "network_timeout") {
        if (attempt < maxRetries) continue
        return { failed: true, reason: "研究阶段网络持续超时" }
      }

      // 语义性失败 - 调整任务描述重试
      if (research.status === "failed" && research.error?.type === "task_misunderstood") {
        task.description = refineDescription(task.description, research.error)
        if (attempt < maxRetries) continue
        return { partial: true, message: "任务理解不一致，需人工介入" }
      }

      // 部分完成 - 继续下一步但标记
      if (research.status === "partial") {
        const next = await spawnAgent("implement", {
          ...task,
          context: research.partialResult,
          warning: "研究阶段只完成了部分目标，请谨慎实现"
        })
        return { ...next, partial: true }
      }

      // 成功路径
      if (research.status === "success") {
        const impl = await spawnAgent("implement", { ...task, research: research.summary })
        return impl
      }

      // 被阻塞或其它未预期状态
      return { failed: true, reason: research.summary || "研究阶段被阻塞" }

    } catch (error) {
      // 不可恢复 - 通知用户
      logger.error("Coordination failure:", error)
      return { failed: true, message: "需要人工介入", error }
    }
  }

  return { failed: true, message: "超过最大重试次数" }
}

三条原则

  1. 子 Agent 的失败不应导致全局崩溃——主 Agent 必须能优雅降级
  2. 区分技术失败和语义失败——处理策略不同
  3. 设置重试上限——不要无限重试

16.12 工程化示例：代码重构任务的完整流程

下面以一个工程任务为例，展示多 Agent 的完整工程化实现。

任务：将某 Node.js 项目的 CommonJS 迁移到 ESM，涉及一批互相依赖的文件。

架构设计

采用 Coordinator + Pipeline 混合拓扑：

graph TD
    User[用户需求] --> Main[主 Agent<br/>Coordinator]
    Main -->|"阶段1"| Explore[探索 Agent<br/>Explore]
    Explore -->|"文件清单 + 依赖图"| Main
    Main -->|"阶段2 (并行)"| Plan[规划 Agent × 3<br/>分组制定迁移计划]
    Plan -->|"迁移方案"| Main
    Main -->|"阶段3 (串行)"| Migrate[迁移 Agent × N<br/>按依赖顺序执行]
    Migrate -->|"修改文件"| Main
    Main -->|"阶段4"| Test[测试 Agent]
    Test -->|"测试结果"| Main
    Main -->|"阶段5 (条件)"| Fix[修复 Agent<br/>仅失败文件]
    Fix --> Test
    style Main fill:#dbeafe,stroke:#3b82f6,stroke-width:2px
    style Explore fill:#dcfce7,stroke:#22c55e
    style Plan fill:#fef3c7,stroke:#f59e0b
    style Migrate fill:#f3e8ff,stroke:#a855f7
    style Test fill:#fce7f3,stroke:#ec4899
    style Fix fill:#fecaca,stroke:#ef4444

实现骨架

// 以下示例为教学化伪代码：Agent(...) 调用和返回结构做了简化。
// 实际 SDK 通常把结果放在 content[0].text 中，usage.totalTokens 记录 token 消耗。
interface AgentOutput {
  content: Array<{ type: "text"; text: string }>
  usage: { totalTokens: number }
}
declare function Agent(options: any): Promise<AgentOutput>

async function runAgent<T>(options: any): Promise<T & { tokens: number }> {
  const out = await Agent(options)
  const data = JSON.parse(out.content[0].text) as T
  return { ...data, tokens: out.usage.totalTokens }
}

interface ExploreResult {
  fileList: string[]
  dependencies: Record<string, string[]>
}
interface MigrationFileResult {
  failed: boolean
  file?: string
}
interface TestResult {
  failedTests: Array<{ name: string; error: string }>
  allPassed: boolean
}
interface MigrationResult {
  success: boolean
  migrations: MigrationFileResult[]
  fixes?: any[]
  budget: number
  failed?: boolean
  reason?: string
  completed?: MigrationFileResult[]
}

// 以下辅助函数/常量为占位声明
declare function partitionByDependency(fileList: string[]): Array<{ name: string; files: string[] }>
declare function topologicalSort(deps: Record<string, string[]>): string[]
declare const TASK_TOKEN_LIMIT: number
declare const TASK_AGENT_LIMIT: number
declare const logger: { warn(msg: string): void }

async function migrateToESM(repoPath: string): Promise<MigrationResult> {
  const budget = new MultiAgentBudget({
    maxTokens: TASK_TOKEN_LIMIT,
    maxSubAgents: TASK_AGENT_LIMIT,
    maxDepth: 2,
    warnThreshold: 0.8,
    spawnStopThreshold: 0.95,
  })

  // ───── 阶段 1: 探索 ─────
  const exploration = await runAgent<ExploreResult>({
    subagent_type: "Explore",
    description: "探索 CommonJS 使用情况",
    prompt: `扫描 ${repoPath}，返回：
             1. 所有 .js 文件的列表
             2. 每个文件的 require() 调用
             3. 每个文件的 module.exports 形式
             4. 文件之间的依赖关系图
             以 JSON 返回。`,
  })
  budget.recordUsage(exploration.tokens)

  // ───── 阶段 2: 规划（并行） ─────
  const fileGroups = partitionByDependency(exploration.fileList)
  const plans = await Promise.all(
    fileGroups.map(group => {
      if (!budget.canSpawn({ depth: 1 }).allowed) return null
      return runAgent<{ plan: string }>({
        subagent_type: "general-purpose",
        description: `规划 ${group.name} 组迁移`,
        prompt: `为以下文件制定 ESM 迁移方案: ${group.files}`,
      })
    })
  )
  plans.forEach(p => p && budget.recordUsage(p.tokens))

  // ───── 阶段 3: 迁移（串行，按依赖顺序） ─────
  const sortedFiles = topologicalSort(exploration.dependencies)
  const results: MigrationFileResult[] = []
  for (const file of sortedFiles) {
    if (!budget.canSpawn({ depth: 1 }).allowed) {
      return { failed: true, reason: "budget exhausted", completed: results }
    }

    const migration = await runAgent<MigrationFileResult>({
      subagent_type: "general-purpose",
      description: `迁移 ${file}`,
      prompt: `按照方案将 ${file} 从 CommonJS 迁移到 ESM...`,
    })
    budget.recordUsage(migration.tokens)
    results.push(migration)

    if (migration.failed) {
      // 单个文件失败不阻塞整体——标记并继续
      logger.warn(`Migration of ${file} failed, will retry`)
    }
  }

  // ───── 阶段 4: 测试 ─────
  const testResult = await runAgent<TestResult>({
    subagent_type: "general-purpose",
    description: "运行所有测试",
    prompt: `在 ${repoPath} 执行 npm test，返回失败的测试清单`,
  })

  // ───── 阶段 5: 条件修复 ─────
  if (testResult.failedTests.length > 0) {
    const fixResults = await Promise.all(
      testResult.failedTests.map(test =>
        runAgent<any>({
          subagent_type: "general-purpose",
          description: `修复测试 ${test.name}`,
          prompt: `测试 ${test.name} 失败: ${test.error}\n修复相关代码`,
        })
      )
    )
    // 递归测试一次（上限 1 轮）
    const retestResult = await runAgent<TestResult>({
      subagent_type: "general-purpose",
      description: "重跑测试",
      prompt: `执行 npm test`,
    })

    return {
      success: retestResult.allPassed,
      migrations: results,
      fixes: fixResults,
      budget: budget.getTotalTokensUsed(),
    }
  }

  return { success: true, migrations: results, budget: budget.getTotalTokensUsed() }
}

工程化的五个关键点

  1. 阶段清晰——5 个阶段各自独立，一眼就能看懂
  2. 并行化——能并行的（阶段 2、5）就并行，能串行的（阶段 3）按依赖顺序
  3. 预算管控——每次启动子 Agent 前检查预算
  4. 失败容忍——单个文件失败不阻塞全局
  5. 降级路径——超预算时返回已完成部分，而非崩溃

16.13 何时使用多 Agent：决策矩阵

多 Agent 不是银弹。使用条件：

决策维度 适合多 Agent 单 Agent 即可
领域跨度 多领域（前端+后端+数据库+文档） 单一领域内
信息量 需处理大量文件 只涉及几个文件
任务结构 可清晰分解为独立子任务 步骤紧密耦合
延迟容忍 对延迟不敏感（批处理） 需要快速响应（对话式）
预算 充足 成本敏感
专业化需求 需要不同角色的视角 单一视角够用
复杂度上限 有多层拆分空间 扁平任务

五个"不要用多 Agent"的信号

  1. 任务在一个普通单 Agent 上下文内就能完成——多 Agent 是过度设计
  2. 需要快速响应——多 Agent 延迟累加会很难看
  3. 成本敏感场景——多 Agent 成本和调试成本都会放大
  4. 子任务之间有紧耦合——分开做会有大量往返开销
  5. 团队没有 multi-agent 经验——复杂度会拖垮项目

如果单 Agent 能搞定，就不要用多 Agent。多 Agent 引入的复杂度（通信、协调、错误处理、成本控制）是有成本的。

16.14 四个反模式与对策

反模式一：Agent 过度增殖

现象：每个小任务都拆成一个子 Agent，拓扑复杂到无法调试。

根因：工程师把多 Agent 当作"万金油"——能拆就拆。

对策：

  • 设置子 Agent 数量上限
  • 问自己三个问题：
    • 这个任务单 Agent 能不能做？
    • 拆分是为了上下文隔离，还是为了"看起来工程化"？
    • 拆分后的调试复杂度能接受吗？

反模式二：子 Agent 返回原始数据

现象：子 Agent 把读到的所有文件内容原样返回给主 Agent，主 Agent 上下文被撑爆。

根因：子 Agent 的 prompt 没有要求"信息压缩"。

对策：

  • 在子 Agent 的 prompt 里明确规定输出格式和长度上限
  • 在返回值处做合法性检查（长度、格式）
  • 不符合要求就要求子 Agent 重新总结

反模式三：无限递归

现象：主 Agent 启动子 Agent，子 Agent 又启动子子 Agent，不断嵌套。

根因：没有深度限制，子 Agent 自行决定是否再启动下一层。

对策：

  • 硬限制递归深度（通常 ≤ 2）
  • 深度参数通过 context 传递
  • 超过限制直接失败，强制扁平化

反模式四：主 Agent 上下文污染

现象：主 Agent 不做"结论抽取"，把子 Agent 的原始输出累积到自己的上下文里。

根因：缺少信息过滤层。

对策：

  • 主 Agent 在保存子 Agent 结论前做一次总结
  • 或者让子 Agent 自己生成摘要，主 Agent 只保留摘要

16.15 本章小结：多 Agent 的七条军规

多 Agent 协调的核心要点：

  1. 上下文隔离 是多 Agent 的一级工程价值——比分工更重要
  2. Coordinator 模式适合有明确主从关系的工程任务（最常用）
  3. Pipeline 模式适合流程化、阶段性的工作（适合 DAG 流）
  4. Swarm 模式适合探索性、跨领域的任务（高灵活性）
  5. Debate 模式适合高价值、低容错的决策任务（贵但准）
  6. 成本控制 必须前置设计——token 预算、并发限制、递归深度限制
  7. 错误处理 要有降级策略——子 Agent 失败不应崩溃全局

七条军规：

  1. 任务描述的质量决定子 Agent 的输出质量——宁可 prompt 长一点，也不要模糊
  2. 子 Agent 的输出要做信息压缩——返回结论、证据路径和风险点，而不是原始上下文
  3. 限制递归深度——更深的嵌套通常意味着任务拆分不合理
  4. 并行能用就用——Coordinator 模式下的独立子 Agent 可以并行
  5. 预算机制前置——不要等到失控才想起来限制
  6. 失败分级处理——技术失败 / 语义失败 / 部分完成 的策略不同
  7. 不要过度设计——单 Agent 够用就不要上多 Agent

下一章将探讨人机协作模式——Agent 不是要取代人类，而是要和人类高效配合。多 Agent 协作好的前提是：每个 Agent 都知道自己该做什么、不该做什么、何时把问题交给人类。