第13章：Multi-Agent 系统设计

上一章的结论是：先用单 Agent，遇到上下文瓶颈、专业化需求或并行需求时再拆分。这一章假设你已经确认需要多个 Agent 协作，接下来的问题是——怎么组织它们、怎么让它们通信、怎么防止它们失控。

Multi-Agent 系统的难度不在于"让多个 Agent 跑起来"，而在于"让它们可靠地协作而不互相拖后腿"。一篇 NeurIPS 2025 的研究分析了 1,600 多条执行轨迹后发现，7 个主流 Multi-Agent 框架的任务失败率在 41% 到 86.7% 之间。失败的根源不是模型能力不够，而是系统设计问题——角色规范不清、Agent 之间信息丢失、没有验证机制 [1]。

这一章从架构模式讲起，到通信机制，到状态管理，最后用一个完整的实战案例把这些概念串起来。

13.1 架构模式深度剖析

核心直觉

Multi-Agent 系统的架构模式解决的是一个组织问题：谁来分配任务、谁向谁汇报、谁有权做最终决定。

Supervisor（管理者模式）

Supervisor 模式是最直觉的架构：一个中央 Agent 接收任务，拆解成子任务分配给专用 Worker，收集结果后汇总输出。

为什么这样设计：Supervisor 拥有全局视野，能看到所有 Worker 的输出，做交叉验证和信息整合。这对需要综合多个领域结果的任务特别有效——比如一份研究报告需要同时包含市场数据、技术分析和竞品对比。

在 LangGraph 中，Supervisor 模式通过状态图实现。Supervisor 是一个节点，每个 Worker 也是节点，Supervisor 通过条件边决定下一步调用哪个 Worker：

python

from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import create_react_agent
from langchain_anthropic import ChatAnthropic

model = ChatAnthropic(model="claude-sonnet-4-20250514")

# 定义专用 Worker
research_agent = create_react_agent(
    model, tools=[web_search, arxiv_search],
    prompt="你是研究员，负责搜索和整理指定主题的信息。"
)

coding_agent = create_react_agent(
    model, tools=[code_exec, file_write],
    prompt="你是开发者，负责编写和测试代码。"
)

# Supervisor 决策逻辑
def supervisor(state: MessagesState):
    response = model.invoke([
        {"role": "system", "content": """你是项目管理者。根据当前状态决定下一步：
        - 需要查资料 → 调用 research_worker
        - 需要写代码 → 调用 coding_worker
        - 任务完成 → 结束"""},
        *state["messages"]
    ])
    # 解析 response 决定下一个节点
    return {"next": parse_next_worker(response)}

# 构建状态图
builder = StateGraph(MessagesState)
builder.add_node("supervisor", supervisor)
builder.add_node("research_worker", research_agent)
builder.add_node("coding_worker", coding_agent)

builder.add_edge(START, "supervisor")
builder.add_conditional_edges("supervisor", lambda s: s["next"], {
    "research_worker": "research_worker",
    "coding_worker": "coding_worker",
    "FINISH": END,
})
builder.add_edge("research_worker", "supervisor")
builder.add_edge("coding_worker", "supervisor")

graph = builder.compile()

工程权衡：

优势：全局可控，Supervisor 能做跨 Worker 的结果校验；调试时有清晰的决策链可追踪
代价：Supervisor 本身的上下文会随着 Worker 数量和轮次增长而膨胀；Supervisor 的推理能力成为整个系统的瓶颈——如果它分配任务的判断失误，所有下游 Worker 都白跑
适合：需要汇聚多领域结果、有明确的"综合"步骤的任务

Anthropic 在 2025 年 6 月公开了他们内部的多 Agent 研究系统架构：Lead Agent（Claude Opus）作为 Supervisor，多个 Subagent（Claude Sonnet）作为 Worker。在内部评估中，这个多 Agent 架构比单 Agent 基线的表现提升了 90.2%。但代价也很明确——token 用量是单 Agent 的约 15 倍，且 token 使用量解释了 80% 的性能差异 [2]。

换句话说，Multi-Agent 的性能提升在很大程度上是用更多 token 换来的。

Peer-to-Peer（对等模式）

对等模式没有固定的"老板"。每个 Agent 地位平等，根据任务需要自主决定是否与其他 Agent 交互、是否移交控制权。

为什么这样设计：有些任务没有自然的层级关系。比如一个技术方案评审，分析师、工程师、安全专家各自从不同角度提出意见，谁也不比谁更"权威"。强行指定一个 Supervisor 反而会引入不必要的瓶颈。

OpenAI Agents SDK 的 Handoff 机制天然支持对等模式——任何 Agent 都可以把控制权交给任何其他 Agent：

python

from agents import Agent, handoff

analyst = Agent(
    name="Analyst",
    instructions="从业务角度分析需求。如果需要技术可行性评估，交给工程师。",
)

engineer = Agent(
    name="Engineer",
    instructions="评估技术可行性。如果需要安全审查，交给安全专家。分析完成后交回分析师。",
)

security_expert = Agent(
    name="Security Expert",
    instructions="审查安全风险。完成后交回工程师。",
)

# 先定义 Agent，再补充 handoffs，避免前向引用问题
analyst.handoffs = [handoff(engineer)]
engineer.handoffs = [handoff(security_expert), handoff(analyst)]
security_expert.handoffs = [handoff(engineer)]

工程权衡：

优势：灵活，适合探索性任务；没有单点瓶颈
代价：缺乏全局视野，没有一个 Agent 能看到完整的执行状态；调试困难——控制权在 Agent 之间反复移交，追踪执行路径需要完整的 Trace 支持；最容易出现无限循环（A 交给 B，B 交给 C，C 又交回 A）
适合：多方评审、协商式决策、没有明确层级的协作任务

常见误区：很多人把对等模式理解为"让 Agent 自由讨论"。实际上，没有约束的对等讨论几乎必然退化成无限循环或无效重复。每个 Agent 必须有明确的移交条件和终止条件。

Hierarchical（层级模式）

层级模式是 Supervisor 模式的递归扩展：顶层 Supervisor 管理中层 Supervisor，中层再管理底层 Worker。

为什么这样设计：当任务复杂到单层 Supervisor 无法管理时（Worker 超过 5-7 个，或者子任务之间有复杂的依赖关系），引入中间管理层可以控制每一层的认知负担。

Databricks 在生产中使用了这种架构来处理企业级数据分析任务：顶层 Supervisor 负责理解业务需求，中层 Supervisor 按业务线分管各自的数据 Agent 团队，底层 Worker 执行具体的 SQL 查询和数据转换 [3]。

工程权衡：

优势：可扩展性好，每一层的管理范围可控；符合"分治"思想
代价：延迟叠加严重——每多一层，多一轮 LLM 调用；信息在层级间传递时容易失真（"电话游戏效应"）；系统复杂度高，调试和维护成本大幅上升
适合：大规模、多领域的复杂任务；企业级场景中按部门/业务线组织的分析任务

一个实用判断：如果你的 Worker 数量不超过 5 个，不要用层级模式。直接用 Supervisor 模式就够了。这里的“5 个左右”是工程经验，不是通用规律——层级模式的管理开销只有在 Agent 数量足够多、依赖关系足够复杂时，才能被分治收益抵消。

模式选型决策

维度	Supervisor	Peer-to-Peer	Hierarchical
控制权	中央集中	分布式	分层委托
全局视野	有（Supervisor 持有）	无	部分（每层 Supervisor 有局部视野）
延迟	中等	不可预测	高（层级越深越慢）
调试难度	低	高	中（分层追踪）
适合规模	2-5 个 Worker	2-4 个对等 Agent	5+ 个 Agent
典型场景	研究报告、代码生成	评审、协商	企业数据分析、大规模内容生产

实际系统中，这三种模式经常混合使用。一个层级系统的顶层用 Supervisor 模式，中间某个子团队内部用对等模式做评审，底层 Worker 之间用流水线串联——这种混合架构在生产环境中是常态。

13.2 Agent 间通信

核心直觉

Agent 之间要交换信息才能协作，问题是用什么方式交换、交换多少、谁能看到谁的信息。

通信机制的选择直接决定了系统的可调试性、token 效率和扩展性。当前主流框架大致走了三条路。

消息传递（Message Passing）

每个 Agent 通过显式的消息与其他 Agent 交互。消息可以是对话历史的一部分（OpenAI Agents SDK 的 Handoff），也可以是结构化的任务描述和结果。

OpenAI 的实现：Handoff 本质上是对话历史的转移。当 Agent A Handoff 给 Agent B 时，B 继承了 A 的对话历史（或者经过 input_filter 过滤后的历史）。没有额外的共享状态，所有信息都在消息流里。

python

from agents import Agent, handoff
from agents.handoffs import HandoffInputData

# 控制 Handoff 时传递的历史信息
def filter_for_specialist(handoff_input: HandoffInputData) -> HandoffInputData:
    """只保留最近 5 条输入历史，避免上下文爆炸。"""
    history = handoff_input.input_history
    system_msgs = [item for item in history if item.get("role") == "system"]
    recent = history[-5:]

    return HandoffInputData(
        input_history=system_msgs + recent,
        pre_handoff_items=handoff_input.pre_handoff_items,
        new_items=handoff_input.new_items,
        input_items=handoff_input.input_items,
        run_context=handoff_input.run_context,
    )

specialist = Agent(
    name="Specialist",
    instructions="处理专项任务。",
)

triage = Agent(
    name="Triage",
    handoffs=[
        handoff(specialist, input_filter=filter_for_specialist)
    ],
)

优点：简单、无共享状态需要管理、Agent 之间解耦。缺点：信息只能单向流动（从 A 到 B），如果 B 需要 A 之前产生的某个中间结果，必须在消息里显式携带；消息量大时 token 成本高。

共享状态（Shared State）

所有 Agent 读写同一个状态对象。每个 Agent 执行后更新状态，下一个 Agent 从更新后的状态中读取它需要的信息。

LangGraph 的实现：用 TypedDict 定义共享状态，通过 Annotated 类型的 Reducer 函数控制状态更新方式：

python

from typing import TypedDict, Annotated
from langgraph.graph import add_messages

class TeamState(TypedDict):
    messages: Annotated[list, add_messages]  # 消息自动追加
    research_findings: str                     # 研究结果
    code_artifacts: list[str]                  # 代码产物
    review_comments: list[str]                 # 审查意见
    current_phase: str                         # 当前阶段

def research_worker(state: TeamState) -> dict:
    # 读取 messages，执行研究，写入 research_findings
    findings = do_research(state["messages"])
    return {
        "research_findings": findings,
        "messages": [{"role": "assistant", "content": f"研究完成: {findings[:200]}"}]
    }

def coding_worker(state: TeamState) -> dict:
    # 读取 research_findings，生成代码
    code = generate_code(state["research_findings"])
    return {
        "code_artifacts": [code],
        "messages": [{"role": "assistant", "content": "代码已生成"}]
    }

优点：Agent 之间可以共享结构化数据，不只是对话历史；LangGraph 的 Checkpoint 机制自动持久化状态，支持"时间旅行"（回到任意历史状态重放）。缺点：状态对象容易膨胀；多个 Agent 同时写同一个字段需要冲突解决策略；调试时需要理解完整的状态流转。

黑板系统（Blackboard）

黑板系统是共享状态的一个变体：有一个公共的"黑板"（通常是文件系统、数据库或 KV 存储），所有 Agent 都可以在上面读写。与共享状态的区别在于——黑板是外部持久化的，不是内存中的状态对象。

Anthropic Claude Code 的实现就是一个典型的黑板系统：多个 Sub-agent 通过文件系统协调——一个 Agent 把研究结果写到文件，另一个 Agent 读取文件继续工作。没有直接的 Agent 间消息传递，所有协调通过文件系统这个"黑板"完成。

python

# 伪代码：基于文件系统的黑板通信
import json
from pathlib import Path

BLACKBOARD = Path("/tmp/agent_workspace")

def research_agent():
    findings = search_and_analyze("目标主题")
    # 写入黑板
    (BLACKBOARD / "research.json").write_text(
        json.dumps({"findings": findings, "status": "complete"})
    )

def coding_agent():
    # 从黑板读取
    research = json.loads((BLACKBOARD / "research.json").read_text())
    if research["status"] != "complete":
        return  # 等待研究完成
    code = generate_code(research["findings"])
    (BLACKBOARD / "code_output.py").write_text(code)

def review_agent():
    # 读取代码，写入评审意见
    code = (BLACKBOARD / "code_output.py").read_text()
    review = review_code(code)
    (BLACKBOARD / "review.json").write_text(
        json.dumps({"comments": review, "approved": len(review) == 0})
    )

优点：天然持久化，Agent 崩溃后其他 Agent 仍能继续；与 Agent 的具体实现解耦，甚至可以混合不同框架的 Agent。缺点：需要自己处理并发读写、轮询或通知机制；缺乏类型安全，容易出现格式不一致的问题。

通信模式选型

模式	信息载体	持久化	适合场景
消息传递	对话历史	依赖框架	对话型、Handoff 密集型
共享状态	内存中的状态对象	Checkpoint	需要结构化数据交换的协作
黑板系统	外部存储（文件/DB/KV）	天然持久	长任务、跨框架、异步协作

一个实用的观察：大多数生产系统最终会走向共享状态 + 黑板系统的混合模式——短期的 Agent 间协调用共享状态（快、类型安全），长期的中间产物用外部存储（持久、可审计）。

状态序列化与恢复

不管用哪种通信模式，Multi-Agent 系统都要考虑一个问题：如果系统在执行到一半时崩溃了，能不能恢复？

这要求所有 Agent 间的通信状态都是可序列化的。LangGraph 通过 Checkpoint 自动处理这个问题——每个节点执行完毕后，完整的状态图被序列化到 SQLite 或 Postgres。恢复时从最近的 Checkpoint 加载状态，继续执行。

但如果你用的是消息传递模式（比如 OpenAI Agents SDK），状态恢复需要自己实现。核心思路是：把每一轮 Handoff 的目标 Agent 和消息历史序列化到外部存储，崩溃后从最后一次 Handoff 记录恢复。

python

import json

class HandoffCheckpoint:
    def __init__(self, storage_path: str):
        self.path = storage_path

    def save(self, agent_name: str, messages: list, metadata: dict):
        checkpoint = {
            "agent_name": agent_name,
            "messages": messages,
            "metadata": metadata,
            "timestamp": time.time()
        }
        with open(self.path, "w") as f:
            json.dump(checkpoint, f)

    def restore(self) -> dict | None:
        try:
            with open(self.path) as f:
                return json.load(f)
        except FileNotFoundError:
            return None

13.3 防止常见失败模式

核心直觉

Multi-Agent 系统最常见的死法不是"模型太笨"，而是"系统设计有缺陷"。MAST 研究的数据很清楚：41.77% 的失败来自规范问题，36.94% 来自 Agent 间的对齐问题，只有 21.30% 来自验证缺失 [1]。

无限循环

问题：Agent A 完成后把控制权交给 Agent B，B 觉得结果不够好又交回 A，A 再改再交给 B——反复循环，token 成本指数级增长。

一个真实的案例：2025 年有人公开报告了一个多 Agent 研究工具的事故——系统在没有护栏的情况下运行了 11 天，产生了 47,000 美元的 API 费用 [5]。根本原因是多个 Agent 之间没有共享记忆，也没有全局状态检测，每个 Agent 都认为自己是第一次执行任务。

防御策略：

python

# 策略一：硬性最大轮次限制
MAX_AGENT_TURNS = 15

async def run_multi_agent(task: str) -> str:
    turn_count = 0
    current_agent = triage_agent
    messages = [{"role": "user", "content": task}]

    while turn_count < MAX_AGENT_TURNS:
        result = await current_agent.run(messages)
        turn_count += 1

        if result.is_final:
            return result.output

        if result.handoff_to:
            current_agent = result.handoff_to
            messages = result.messages

    # 超过上限，强制终止并返回当前最佳结果
    return f"任务在 {MAX_AGENT_TURNS} 轮后强制终止。当前结果: {result.output}"

# 策略二：收敛检测——连续 N 轮输出无实质变化则终止
def detect_convergence(outputs: list[str], window: int = 3) -> bool:
    if len(outputs) < window:
        return False
    recent = outputs[-window:]
    # 用简单的字符串相似度检测是否收敛
    from difflib import SequenceMatcher
    for i in range(len(recent) - 1):
        similarity = SequenceMatcher(None, recent[i], recent[i+1]).ratio()
        if similarity < 0.85:
            return False  # 还在变化，未收敛
    return True  # 连续几轮输出高度相似，已收敛

# 策略三：全局预算控制
class TokenBudget:
    def __init__(self, max_tokens: int):
        self.max_tokens = max_tokens
        self.used_tokens = 0

    def consume(self, tokens: int) -> bool:
        self.used_tokens += tokens
        if self.used_tokens > self.max_tokens:
            raise BudgetExhaustedError(
                f"Token 预算已耗尽: {self.used_tokens}/{self.max_tokens}"
            )
        return True

实践建议：三种策略叠加使用。最大轮次是兜底，收敛检测是优化，token 预算是成本安全阀。

目标冲突

问题：两个 Agent 对同一个问题有不同的"目标函数"。比如成本优化 Agent 倾向选最便宜的方案，质量保障 Agent 倾向选最稳的方案。两个 Agent 各执己见，Supervisor 左右为难。

MAST 研究把这类问题归为"Inter-agent Misalignment"——Agent 之间忽略彼此的输出、重复执行相同步骤、或者过早终止协作 [1]。

防御策略：

明确的角色边界和决策权限：每个 Agent 的 System Prompt 里必须写清楚它的决策范围和优先级。"你是成本优化顾问，提出建议但不做最终决策"比"你负责成本优化"要好得多——前者明确了这是建议角色，后者暗示了决策权。
仲裁机制：当多个 Agent 的输出冲突时，Supervisor 不应该"综合一下"，而应该有明确的仲裁规则。比如"安全相关的冲突，安全 Agent 的意见优先；成本相关的冲突，看是否超过预算阈值"。
结构化输出格式：让每个 Agent 以固定格式输出它的结论、置信度和关键假设，方便 Supervisor 做对比决策，而不是靠自然语言"感觉"谁更有道理。

python

from pydantic import BaseModel

class AgentRecommendation(BaseModel):
    recommendation: str          # 具体建议
    confidence: float            # 0-1 置信度
    key_assumptions: list[str]   # 关键假设
    risk_factors: list[str]      # 风险因素
    priority: str                # "must-have" | "nice-to-have" | "optional"

# Supervisor 的仲裁逻辑
def arbitrate(recommendations: list[AgentRecommendation]) -> str:
    # 先过滤掉低置信度的建议
    confident = [r for r in recommendations if r.confidence > 0.7]

    # 按优先级排序
    must_haves = [r for r in confident if r.priority == "must-have"]

    # 检查冲突
    # ...（具体仲裁逻辑根据业务定义）

上下文爆炸

问题：每个 Agent 的每一轮执行都是一次完整的 LLM 调用，带着完整的消息历史。4 个 Agent 讨论 5 轮，至少 20 次 LLM 调用，每次调用的上下文都在累积增长。Anthropic 的数据显示，多 Agent 系统的 token 消耗约为单 Agent 的 15 倍 [2]。

防御策略：

上下文隔离：每个 Agent 维护自己独立的上下文窗口，只传递必要的摘要而非完整历史。这是 Anthropic 推荐的做法——Subagent 执行完后只返回压缩的结果摘要给 Lead Agent，而不是把整个执行过程搬过去。
工具结果清理：Agent 执行过程中积累的中间工具调用结果，在传递给下一个 Agent 之前应该清理掉。第 6 章讲过的 Tool-result Clearing 策略在 Multi-Agent 场景下更加重要。
摘要接力：Agent A 完成后，不是把完整输出传给 Agent B，而是先做一轮摘要，把关键信息压缩到合理的 token 数量。

python

async def run_with_context_isolation(supervisor, workers, task):
    """每个 Worker 在独立上下文中执行，只返回摘要"""
    results = {}

    for worker_name, worker in workers.items():
        # 每个 Worker 独立的上下文，只包含任务描述
        worker_result = await worker.run(
            messages=[{"role": "user", "content": task}]
        )

        # 压缩结果后存入共享状态
        summary = await summarize(worker_result.full_output, max_tokens=500)
        results[worker_name] = summary

    # Supervisor 只看摘要，不看完整输出
    final = await supervisor.run(
        messages=[{
            "role": "user",
            "content": f"任务: {task}\n\n各团队结果:\n" +
                       "\n".join(f"- {k}: {v}" for k, v in results.items())
        }]
    )
    return final

步骤重复与信息丢失

MAST 研究发现的另外两个高频问题 [1]：

步骤重复（Step Repetition）：一个 Agent 重复执行另一个 Agent 已经完成的工作。典型原因是 Agent 之间没有共享"已完成任务"的状态。解决方案是维护一个全局的任务清单，标记每个子任务的完成状态。

信息丢失（Information Withholding）：一个 Agent 产生了重要信息但没有传递给需要这个信息的另一个 Agent。典型原因是通信格式没有定义好，Agent 自行决定"什么值得传"。解决方案是定义结构化的输出 Schema，强制每个 Agent 输出完整的关键字段。

失败模式防御清单

失败模式	发生频率	核心防御	兜底机制
无限循环	极高	最大轮次限制 + 收敛检测	全局 token 预算
目标冲突	高	明确角色边界 + 结构化输出	Supervisor 仲裁规则
上下文爆炸	高	上下文隔离 + 摘要接力	单 Agent token 上限
步骤重复	中	全局任务清单	去重检测
信息丢失	中	结构化输出 Schema	Supervisor 追问
死锁	低	超时机制	强制终止 + 状态恢复

13.4 实战：构建软件开发团队

为什么选这个例子

软件开发是 Multi-Agent 最成熟的应用场景之一。它有自然的角色分工（PM、架构师、开发者、审查者），有明确的阶段性（需求→设计→编码→审查→测试），有客观的验证标准（代码是否通过测试）。

但也要先给一个预期校准：截至写作时，把"AI 软件开发团队"真正推到生产级的案例仍然集中在受控场景（特定代码库、特定任务类型）。Google 的 DORA 2025 报告观察到，AI 采用率上升的同时，Bug 率也上升了约 9%，代码审查时间增加了 91%，PR 大小增加了 154% [6]。所以这个实战的目的是理解 Multi-Agent 协作的工程模式，不是说它可以直接替代人类开发团队。

架构设计

用 Supervisor 模式，一个 Tech Lead Agent 作为管理者，四个专用 Worker：

完整实现

python

import anthropic
import json
from dataclasses import dataclass, field

client = anthropic.Anthropic()
MODEL = "claude-sonnet-4-20250514"

# --- 工具定义 ---

def execute_code(code: str) -> str:
    """在沙箱中执行 Python 代码并返回结果"""
    import subprocess
    result = subprocess.run(
        ["python", "-c", code],
        capture_output=True, text=True, timeout=30
    )
    if result.returncode == 0:
        return f"执行成功:\n{result.stdout}"
    return f"执行失败:\n{result.stderr}"

def run_tests(test_code: str) -> str:
    """运行测试代码"""
    import subprocess
    result = subprocess.run(
        ["python", "-c", test_code],
        capture_output=True, text=True, timeout=60
    )
    passed = result.returncode == 0
    return json.dumps({
        "passed": passed,
        "output": result.stdout if passed else result.stderr
    })

# --- Agent 定义 ---

AGENTS = {
    "pm": {
        "system": """你是产品经理。职责：
1. 把模糊的用户需求拆解成明确的功能点列表
2. 每个功能点必须包含：描述、输入输出、边界条件、验收标准
3. 只输出需求文档，不做技术决策""",
        "tools": []
    },
    "architect": {
        "system": """你是软件架构师。职责：
1. 根据需求文档设计技术方案
2. 输出：模块划分、接口定义、数据结构、关键算法选择
3. 关注可测试性——每个模块必须能独立测试
4. 不写实现代码，只写接口和设计""",
        "tools": []
    },
    "coder": {
        "system": """你是开发者。职责：
1. 严格按照架构设计文档实现代码
2. 每个函数必须有类型注解
3. 同时编写对应的单元测试
4. 代码必须可直接运行，不允许伪代码或省略""",
        "tools": [
            {
                "name": "execute_code",
                "description": "在沙箱中执行 Python 代码，验证代码是否能正常运行",
                "input_schema": {
                    "type": "object",
                    "properties": {"code": {"type": "string"}},
                    "required": ["code"]
                }
            }
        ]
    },
    "reviewer": {
        "system": """你是代码审查者。职责：
1. 审查代码质量：可读性、边界处理、错误处理、性能
2. 运行测试代码验证功能正确性
3. 输出审查结论：APPROVED 或 CHANGES_REQUESTED
4. 如果不通过，给出具体的修改要求（不超过 3 条）""",
        "tools": [
            {
                "name": "run_tests",
                "description": "运行测试代码并返回结果",
                "input_schema": {
                    "type": "object",
                    "properties": {"test_code": {"type": "string"}},
                    "required": ["test_code"]
                }
            }
        ]
    }
}

TOOL_DISPATCH = {
    "execute_code": execute_code,
    "run_tests": run_tests,
}

# --- 单个 Agent 执行循环 ---

def run_single_agent(agent_name: str, task: str, max_turns: int = 10) -> str:
    agent = AGENTS[agent_name]
    messages = [{"role": "user", "content": task}]

    for _ in range(max_turns):
        kwargs = {
            "model": MODEL,
            "max_tokens": 4096,
            "system": agent["system"],
            "messages": messages,
        }
        if agent["tools"]:
            kwargs["tools"] = agent["tools"]

        response = client.messages.create(**kwargs)

        if response.stop_reason == "end_turn":
            return "".join(
                block.text for block in response.content
                if hasattr(block, "text")
            )

        # 处理工具调用
        tool_results = []
        for block in response.content:
            if block.type == "tool_use":
                result = TOOL_DISPATCH[block.name](**block.input)
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": result
                })

        messages.append({"role": "assistant", "content": response.content})
        messages.append({"role": "user", "content": tool_results})

    return "Agent 达到最大轮次限制"

# --- Supervisor 编排 ---

@dataclass
class ProjectState:
    requirement: str = ""
    spec: str = ""
    design: str = ""
    code: str = ""
    review_result: str = ""
    review_rounds: int = 0
    max_review_rounds: int = 3  # 防止无限审查循环

def run_dev_team(user_request: str) -> dict:
    state = ProjectState()

    # Phase 1: 需求分析
    print("[Tech Lead] 分配需求分析任务给 PM...")
    state.spec = run_single_agent("pm", f"用户需求：{user_request}")
    print(f"[PM] 需求文档完成（{len(state.spec)} 字符）")

    # Phase 2: 架构设计
    print("[Tech Lead] 分配架构设计任务给 Architect...")
    state.design = run_single_agent(
        "architect", f"根据以下需求设计技术方案：\n\n{state.spec}"
    )
    print(f"[Architect] 设计文档完成（{len(state.design)} 字符）")

    # Phase 3: 编码 + 审查循环
    while state.review_rounds < state.max_review_rounds:
        # 编码
        code_task = f"根据以下设计实现代码：\n\n{state.design}"
        if state.review_result:
            code_task += f"\n\n上一轮审查意见：\n{state.review_result}"

        print(f"[Tech Lead] 分配编码任务给 Coder（第 {state.review_rounds + 1} 轮）...")
        state.code = run_single_agent("coder", code_task)
        print(f"[Coder] 代码完成（{len(state.code)} 字符）")

        # 审查
        print("[Tech Lead] 分配代码审查任务给 Reviewer...")
        state.review_result = run_single_agent(
            "reviewer",
            f"审查以下代码。需求文档：\n{state.spec}\n\n代码：\n{state.code}"
        )
        state.review_rounds += 1
        print(f"[Reviewer] 审查完成（第 {state.review_rounds} 轮）")

        if "APPROVED" in state.review_result.upper():
            print("[Tech Lead] 代码审查通过，任务完成。")
            break
    else:
        print(f"[Tech Lead] 达到最大审查轮次 ({state.max_review_rounds})，交付当前版本。")

    return {
        "spec": state.spec,
        "design": state.design,
        "code": state.code,
        "review": state.review_result,
        "review_rounds": state.review_rounds,
    }

# 使用
result = run_dev_team("实现一个 LRU Cache，支持 get 和 put 操作，O(1) 时间复杂度")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197

这个实现里的关键设计决策

为什么用流水线而不是自由协作：软件开发有自然的先后顺序（先有需求才能设计，先有设计才能编码），用流水线是最直接的映射。如果改成对等模式让四个 Agent 自由讨论，大概率会陷入无效循环。

为什么编码-审查是循环，其他阶段不是：编码是最容易出错且有客观验证标准（测试能不能通过）的环节。需求和架构阶段的"对不对"很难自动验证，加循环意义不大。

为什么设置 max_review_rounds = 3：这是防止无限循环的兜底。根据经验，如果代码 3 轮审查后仍不通过，继续循环下去改善的概率很低，更可能是需求或架构层面有问题。此时交付当前版本并标记问题，比继续烧 token 更务实。

为什么每个 Agent 的上下文是独立的：注意 run_single_agent 每次调用都是全新的消息列表。Agent 之间的信息传递只通过 Supervisor 的参数（state.spec、state.design），而不是共享对话历史。这就是上下文隔离策略——每个 Agent 只看到与自己相关的信息，避免上下文爆炸。

成本估算

运行一次完整的开发团队流程，假设每个 Agent 平均 2 轮工具调用，审查 2 轮：

PM：~2K input + ~1K output
Architect：~3K input + ~1.5K output
Coder（×2 轮）：~4K input + ~2K output × 2
Reviewer（×2 轮）：~5K input + ~1K output × 2

合计约 30K-40K token。以 Claude Sonnet 的价格（截至 2026.04，$3/M input，$15/M output）估算，单次执行成本约 $0.1-0.2。

这是可控的。但如果把 max_review_rounds 设成 10，或者在每个阶段加上自由讨论环节，成本很容易翻 10 倍以上。

从 Demo 到生产还差什么

上面的代码能跑，但离生产还有明确的差距：

缺少持久化：每个阶段的结果只在内存里。加上 Checkpoint（第 16 章），中途崩溃可以从最后完成的阶段恢复
缺少可观测性：每个 Agent 调用应该有 Trace 和 Span（第 17 章），出了问题能定位到具体是哪个 Agent 的哪一步
缺少 Guardrails：Coder Agent 执行代码时应该在沙箱里（第 16 章 16.3），Reviewer Agent 对代码的安全扫描应该作为 Output Guardrail（第 15 章）
缺少 Human-in-the-Loop：需求确认和架构审批应该有人工检查点（第 12 章 12.4）

有开发者分享过一个务实的经验：成功的生产级 Agent 团队，往往是"单领域专家在持续的人类监督下运行"，而不是"完全自主的 AI 开发团队" [7]。这个观察值得记住。

面试高频题

Q1：设计一个 Multi-Agent 系统，如何防止 Agent 间的无限循环和死锁？

参考答案框架：

防无限循环的三层策略：

第一层：硬性轮次上限。给全局和每个 Agent 分别设置最大执行轮次，到了就强制终止
第二层：收敛检测。监控连续几轮的输出变化，如果相似度超过阈值（表明 Agent 在重复自己），提前终止
第三层：全局 token 预算。设定整个任务的 token 消耗上限，作为成本安全阀

防死锁的策略：

超时机制：每个 Agent 的执行和每次 Handoff 都有超时限制
状态持久化：即使某个 Agent 卡住，其他 Agent 的中间结果不会丢失，可以跳过卡住的环节继续

加分点：提到 MAST 研究的发现——大部分失败来自系统设计而非模型能力，能引用具体的失败模式分类（规范问题 41.77%、对齐问题 36.94%、验证问题 21.30%）会展现出对该领域研究的了解。

Q2：Supervisor 模式和 Peer-to-Peer 模式的核心区别是什么？如何选择？

参考答案框架：

核心区别在于全局视野和控制权：

Supervisor 有全局视野，能看到所有 Worker 的输出并做交叉验证；但 Supervisor 本身成为瓶颈（上下文膨胀、推理能力上限）
Peer-to-Peer 没有全局视野，灵活但不可控；调试困难，最容易出无限循环

选择依据：

需要汇聚多领域结果 → Supervisor
对话型的阶段性任务 → Peer-to-Peer（Handoff）
Agent 数量 > 5 → 考虑 Hierarchical（Supervisor 的递归版本）
不确定 → 先用 Supervisor，它的可控性和可调试性最好

加分点：提到实际系统中三种模式经常混合使用，以及 Anthropic 多 Agent 研究系统的数据——Multi-Agent 比单 Agent 提升 90.2%，但 token 消耗增加约 15 倍。

Q3：如何控制 Multi-Agent 系统的 token 成本？

参考答案框架：

三个层面的控制：

上下文隔离：每个 Agent 维护独立的上下文，只传递压缩摘要而非完整历史。Anthropic 的 Subagent 模式就是这个思路——Subagent 返回摘要，Lead Agent 不需要看完整执行过程
模型路由：Supervisor 用强模型（需要综合判断），简单的执行型 Worker 用弱/小模型。"推理模型规划 + 快速模型执行"的混合架构（第 9 章 9.7 讨论过）在 Multi-Agent 场景下更有价值
执行控制：最大轮次限制、token 预算、收敛检测——防止 Agent 在无效循环中烧钱

加分点：能给出具体的成本估算方法，比如"4 个 Agent 各 2 轮、单轮 3K token、使用 Sonnet 级模型，单次执行约 $0.1-0.2"——展现对实际成本的感知。

参考资料

[1] Why Do Multi-Agent LLM Systems Fail? - Cemri, Pan, Yang et al. (NeurIPS 2025) https://arxiv.org/abs/2503.13657

[2] How we built our multi-agent research system - Anthropic Engineering (2025.06) https://www.anthropic.com/engineering/multi-agent-research-system

[3] Multi-Agent Supervisor Architecture: Orchestrating Enterprise AI at Scale - Databricks (2025) https://www.databricks.com/blog/multi-agent-supervisor-architecture-orchestrating-enterprise-ai-scale

[4] Scaling LLM Test-Time Compute Optimally can be More Effective than Scaling Model Parameters - Google DeepMind (2024) https://arxiv.org/abs/2408.03314

[5] AI Agents Horror Stories: How a $47,000 Failure Exposed the Hype and Hidden Risks of Multi-Agent Systems - TechStartups (2025.11) https://techstartups.com/2025/11/14/ai-agents-horror-stories-how-a-47000-failure-exposed-the-hype-and-hidden-risks-of-multi-agent-systems/

[6] DORA Report 2025 - Google Cloud DevOps Research and Assessment https://dora.dev/research/

[7] LLMOps in Production: What Actually Works - ZenML https://www.zenml.io/blog/llmops-in-production-287-more-case-studies-of-what-actually-works

第13章：Multi-Agent 系统设计 ​

13.1 架构模式深度剖析 ​

核心直觉 ​

Supervisor（管理者模式） ​

Peer-to-Peer（对等模式） ​

Hierarchical（层级模式） ​

模式选型决策 ​

13.2 Agent 间通信 ​

核心直觉 ​

消息传递（Message Passing） ​

共享状态（Shared State） ​

黑板系统（Blackboard） ​

通信模式选型 ​

状态序列化与恢复 ​

13.3 防止常见失败模式 ​

核心直觉 ​

无限循环 ​

目标冲突 ​

上下文爆炸 ​

步骤重复与信息丢失 ​

失败模式防御清单 ​

13.4 实战：构建软件开发团队 ​

为什么选这个例子 ​

架构设计 ​

完整实现 ​

这个实现里的关键设计决策 ​

成本估算 ​

从 Demo 到生产还差什么 ​

面试高频题 ​

参考资料 ​

第13章：Multi-Agent 系统设计

13.1 架构模式深度剖析

核心直觉

Supervisor（管理者模式）

Peer-to-Peer（对等模式）

Hierarchical（层级模式）

模式选型决策

13.2 Agent 间通信

核心直觉

消息传递（Message Passing）

共享状态（Shared State）

黑板系统（Blackboard）

通信模式选型

状态序列化与恢复

13.3 防止常见失败模式

核心直觉

无限循环

目标冲突

上下文爆炸

步骤重复与信息丢失

失败模式防御清单

13.4 实战：构建软件开发团队

为什么选这个例子

架构设计

完整实现

这个实现里的关键设计决策

成本估算

从 Demo 到生产还差什么

面试高频题

参考资料