Multi-agent Systems

反模式：80% 的「多 agent」是单 agent 加上花架子

背景与原理

Cognition 在 Don't Build Multi-Agents（2025 年 6 月）里把这个反模式讲透了。核心 claim：multi-agent 系统的失败不是 agent 不够聪明，是它们看不到彼此。一个 subagent 决定「先 refactor 这个文件再加 feature」，另一个并发 subagent 决定「直接加 feature 不动结构」——两个都在自己 context 里逻辑自洽，merge 时冲突，orchestrator 既无法判优劣也无法回滚。这种隐式决策冲突（implicit decision conflict）是多 agent 最难治的失败模式，因为没有报错、没有 stack trace，只有一个看起来「奇怪但能跑」的结果。

Cognition 的两条工程原则：(1) Share full context, not summaries——任何 subagent 都要能看到整个 trace（包括 orchestrator 的目标、其他 agent 已经做了什么），summarization 是有损压缩，几乎一定会丢关键信息；(2) Actions carry implicit decisions——每个 tool call 都隐含一个判断（「我选择这条路而不是另一条」），并发 agent 各自做隐式决策时没有协商机制，所以编码、规划这类每一步依赖前一步的任务，单 agent 严格优于多 agent。

这跟 Anthropic 的 Research 系统不矛盾——它们的任务结构正好相反：并行检索 100 个网页、然后 orchestrator 聚合答案。每个 subagent 之间本来就独立，没有 implicit decision conflict 问题。所以判断要不要多 agent 的第一个问题不是「这个任务复杂吗」，是「子任务之间有没有隐式共享状态」。

另一个常被忽视的反模式：把单 agent 的 prompt 拆成多个 agent 来「降低复杂度」——typical 例子是「PM agent + Engineer agent + Reviewer agent」串行 pipeline。这种拆法几乎从不带来质量提升，只是把 prompt 工程的复杂度从 1 个 system prompt 转移到 3 个 agent 间的 message format 设计。MetaGPT 这类「SOP 编码进 agent 角色」的论文展示了上限，但需要把每个角色的 prompt 都调到生产质量——这本身工作量比单 agent 高 3×，且必然引入 agent 间 message drift 问题。除非你的产品 SOP 真的稳定到可以编码成 agent 角色（极少），否则单 agent 永远是 baseline。

实战示例

用之前的 ROI 决策树问自己 6 个问题——如果有 ≥ 3 个 No，多 agent 就是过度工程：

# multi_agent_check.py —— 开建前的 sanity check
def should_use_multi_agent(task) -> str:
    checks = {
        "parallelizable":
            task.can_split_into_independent_subtasks,
        "low_implicit_state":
            not task.requires_step_n_depends_on_step_n_minus_1,
        "context_truly_overflows":
            task.estimated_tokens > 200_000,  # long-ctx 真不够
        "value_per_query_high":
            task.dollar_value_per_run >= 1.0,  # 15× token 才划算
        "subagent_output_verifiable":
            task.each_subagent_result_can_be_validated_independently,
        "single_agent_tried_and_failed":
            task.has_single_agent_baseline_with_eval_score,
    }
    failed = [k for k,v in checks.items() if not v]
    if len(failed) >= 3:
        return f"❌ Stay on single agent. Missing: {failed}"
    if "parallelizable" in failed or "low_implicit_state" in failed:
        return "❌ Task structure fights multi-agent. Use single agent + long ctx."
    return "✅ Good fit. Start with orchestrator-worker; measure 15× token budget."

Orchestrator-Worker：Anthropic 的赌注与真实成本

背景与原理

Anthropic 在 2025 年 6 月的 How we built our multi-agent research system 博客里把这个模式讲得最具体。架构：用户 query → Lead Agent (Opus 4) 做规划、拆分子任务、spawn N 个并行 Subagents (Sonnet 4) 各自查不同方向 → Lead 聚合所有 subagent 的发现 → 生成最终答案。关键数字：内部 research eval 上比 single-agent Opus 4 高 90.2%，但 token 消耗约为普通 chat 的 15×（普通 agent 大约 4×）。

四个让这个模式真正 work 的工程约束：

• (1) Spawn 准则：Lead Agent 不能随便 spawn——它要被明确教会「什么样的子查询值得开一个 subagent」。Anthropic 的 prompt 给了 lead 「衡量任务广度、复杂度、并行收益」的判断准则。反例：让 lead 「能 spawn 就 spawn」→ 它会为每个搜索词都开一个 subagent，token 爆炸而无质量提升。
• (2) Subagent prompt 隔离：每个 subagent 拿到的 prompt 必须包含任务目标但不包含其他 subagent 的工作——这是和 Cognition 立场相反的地方。Anthropic 这样设计的前提：subagent 之间本来就独立（查不同子方向），互相看反而引入 confirmation bias。这个设计只在「真正并行」任务上 work，coding/refactoring 这种串行任务上必然崩。
• (3) 聚合策略：Lead Agent 不是简单 concat subagent 结果——它做二次推理：交叉验证、去重、识别矛盾、补漏。Anthropic 强调这一步用更强的模型（Opus）至关重要，弱模型的聚合等于把 5 个噪音平均成 1 个噪音。
• (4) 并发上限：subagent 数量不是越多越好——超过 5-10 个之后收益曲线骤平（多了的 agent 只是重复已经覆盖的角度），同时 lead 聚合复杂度随 N 上升而下降。Anthropic 没给具体数字但暗示常用区间是 3-7 个 subagent。

15× token 成本怎么理解？普通 chat 1 turn = 1 个 LLM 调用；普通 agent ≈ 4× 因为 ReAct loop 多轮；multi-agent research ≈ 15× 因为：lead 自己跑多轮（~4×）+ 每个 subagent 各跑多轮（~4×）× N 个并发 subagent（~3-5）+ 聚合。这意味着每 query 成本要 > $1 才有 ROI。低价值通用任务用 multi-agent 是直接烧钱；研究、尽职调查、跨工具调研、深度 SWE-bench 这类结果一份值钱的场景才匹配。

实战示例

最小可工作的 orchestrator-worker（Anthropic SDK，伪代码——抓住模式而非细节）：

import anthropic, asyncio, json

client = anthropic.AsyncAnthropic()

# ============ Subagent: 独立查一个子方向 ============
async def subagent(task: dict) -> dict:
    # 注意：prompt 只给本子任务，不给其他 subagent 的工作
    msg = await client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=2000,
        system="You are a research subagent. Focus ONLY on the assigned "
               "sub-question. Use search tools. Return: findings + sources.",
        tools=[search_tool, fetch_tool],
        messages=[{"role":"user",
                   "content": f"Sub-question: {task['question']}\n"
                              f"Scope: {task['scope']}"}],
    )
    # tool loop 省略，实际要循环到 stop_reason='end_turn'
    return {"task": task, "findings": msg.content[0].text}

# ============ Lead Agent: 规划 + 聚合 ============
async def lead_agent(user_query: str) -> str:
    # Step 1: 规划——用强模型决定要不要拆、拆成什么
    plan = await client.messages.create(
        model="claude-opus-4-7",
        max_tokens=1500,
        system="You are a research lead. Decide if this query benefits from "
               "parallel subagents. If YES, output JSON list of 3-7 INDEPENDENT "
               "sub-questions. If NO (task is sequential or simple), output "
               "{\"single\": true} and you'll handle it alone.",
        messages=[{"role":"user","content":user_query}],
    )
    spec = json.loads(plan.content[0].text)

    # Step 2a: 任务不适合拆 → fallback 单 agent (绝大多数 query)
    if spec.get("single"):
        return await single_agent_run(user_query)

    # Step 2b: 并行 spawn subagents
    results = await asyncio.gather(*[subagent(t) for t in spec["tasks"]])

    # Step 3: 聚合——再次用强模型，必须做 cross-check 而非 concat
    aggregation = await client.messages.create(
        model="claude-opus-4-7",
        max_tokens=4000,
        system="You are the research lead synthesizing N subagent reports. "
               "Cross-validate, identify contradictions, fill gaps. "
               "Output: comprehensive answer + confidence notes.",
        messages=[{"role":"user",
                   "content":f"Original query: {user_query}\n\n"
                             f"Subagent findings:\n{json.dumps(results, indent=2)}"}],
    )
    return aggregation.content[0].text

# ============ Cost guard：硬上限防失控 ============
# 在 production 必须包：单次 budget 上限 + N_subagent <= 7 + timeout

Debate / Cross-check：什么时候多视角真的提升质量

背景与原理

Du et al. 2023（Improving Factuality and Reasoning in Language Models through Multiagent Debate，arXiv 2305.14325）是这个范式的起点。核心机制：让 N 个 LLM 各自独立回答 → 互相看到对方答案 + 论据 → 多轮迭代后输出共识答案。论文在 MMLU、数学题、事实 QA 上证明显著优于单次回答和 self-consistency。这个结果到 2026 年仍稳健——当任务有明确正确答案。

但 debate 在 3 个场景上几乎不 work：

• (1) 开放主观任务——「写一段产品文案」「评估这个商业计划」——没有 ground truth，agent 之间会 converge 到一个「大家都接受的平庸答案」，远不如让一个 agent 直接生成 + 人评筛选。
• (2) 同模型同 prompt 的 N 次重复——「同源 diversity」很弱，debate 退化为 self-consistency，正向 bias 同步放大（所有 agent 都倾向 yes-man，debate 后反而更 yes-man）。要让 debate work，必须构造真 diversity：用 Claude vs GPT vs Gemini、或同模型但角色 prompt 强对立（「你是支持者 / 你是反对者」）。
• (3) 长 form generation——debate 适合「短答案有对错」的任务；让 agent 对 5000 字文章逐段 debate，几乎一定退化为相互奉承或卡死。

更高 ROI 的「多视角」工程模式是 asymmetric cross-check（不对称交叉验证）：用一个不同的模型 / 不同的 prompt / 单独的 verifier agent 来检查主 agent 的输出。例如：主 agent 用 Claude Opus 写代码 → verifier agent 用 GPT-5 单独跑 testing 思路 + 找 edge case。这种「主+审计」结构比 N-agent debate 在工程上更稳：责任清晰、成本可控（2× 而非 N×）、失败可独立 attribute。Constitutional AI 自己审、self-refine、Reflexion 都是同源的「主+审」结构在不同时间维度的展开。

还有一个反直觉发现：让模型「debate 自己之前的答案」（self-debate / self-reflection）效果常不如「让另一个 agent 来 critique」。Cognition 的 backward rationalization 问题：模型一旦给了答案 A，它会倾向于为 A 辩护、找证据支持，而非真正质疑——self-debate 经常变成 self-confirmation。所以cross-check 比 self-reflection 更可靠，这是 multi-agent 在工程上真正的甜点。

实战示例

Asymmetric cross-check 模式（主 agent + 不同模型 verifier），2× 成本换可观质量提升：

import anthropic, openai

claude = anthropic.Anthropic()
gpt = openai.OpenAI()

def primary_solve(question: str) -> str:
    # 主 agent：Claude Opus 出答案
    msg = claude.messages.create(
        model="claude-opus-4-7", max_tokens=1500,
        messages=[{"role":"user","content":question}],
    )
    return msg.content[0].text

def cross_check(question: str, primary_answer: str) -> dict:
    # Verifier：用不同厂商模型 + adversarial prompt
    # 关键 1：换模型（diversity）  关键 2：角色对立（反方）
    prompt = f"""You are a SKEPTICAL reviewer. Your job is to find what is
WRONG with the candidate answer below. Do NOT defend it.

QUESTION: {question}

CANDIDATE ANSWER:
{primary_answer}

Output JSON:
{{"verdict": "correct"|"incorrect"|"uncertain",
  "issues": [],
  "missing": []}}"""
    rsp = gpt.chat.completions.create(
        model="gpt-5", temperature=0.2,
        messages=[{"role":"user","content":prompt}],
        response_format={"type":"json_object"},
    )
    return json.loads(rsp.choices[0].message.content)

def arbitrate(question, primary, review) -> str:
    # 仅当 verifier 找出问题时才重新跑——成本可控
    if review["verdict"] == "correct" and not review["issues"]:
        return primary
    # 让主 agent 看到 critique 后修订（不是简单覆盖）
    revision = claude.messages.create(
        model="claude-opus-4-7", max_tokens=2000,
        messages=[
            {"role":"user","content":question},
            {"role":"assistant","content":primary},
            {"role":"user","content":
              f"A skeptical reviewer raised these issues:\n"
              f"{json.dumps(review, indent=2)}\n\n"
              f"Revise ONLY if the issues are valid. If you disagree, explain why."}],
    )
    return revision.content[0].text

# 用法
answer = primary_solve(question)
review = cross_check(question, answer)
final  = arbitrate(question, answer, review)
# 成本：~2× single agent。质量：对 factual/code 任务通常 +10-20pp

协调税与通信协议：A2A、MCP、和 N² 的代价

背景与原理

多 agent 系统的「协调税」是分布式系统老问题在 LLM 时代的翻版。三个关键开销：

• (1) 通信复杂度——agent 之间任意通信是 O(N²) 条边，每条边都要消息编码、token 消耗、上下文同步；超过 5-7 个 agent 就显著掉性能（每个 agent 上下文里塞太多别的 agent 状态）。
• (2) 决策延迟叠加——串行 agent pipeline 是 latency 累加（5 个 agent × 8s = 40s）；并行 agent 受限于最慢 subagent + 聚合延迟（约 1.5× 单 agent 时间）；用户感知延迟通常是多 agent 反对方向最大的工程压力。
• (3) 失败 attribution——单 agent 失败容易定位（看 trace 找出错那一步）；multi-agent 失败要回答「哪个 agent 在哪一步做了什么导致最终错」——subagent 各自 trace + lead 聚合 trace + 跨 agent 状态，debug 复杂度暴涨。Anthropic 的工程博客承认这是 multi-agent 最难解的工程问题，他们用每个 agent 全 trace 落库 + 跨 trace 关联 id 才勉强能 debug。

2025 年 4 月 Google 推出 A2A（Agent-to-Agent）协议，标准化「不同框架不同厂商的 agent 互相发现 + 委派任务 + 协调」。技术栈：HTTP + SSE + JSON-RPC 2.0，每个 agent 用 Agent Card（JSON 描述自身能力）发布，client agent 通过 Card 找合适的目标 agent，Task 抽象封装委派。6 月 Google 把 A2A 捐给 Linux Foundation，已有 150+ 组织支持。

A2A vs MCP 的分工经常被混：

• MCP（Anthropic 2024）：标准化 LLM agent ↔ tools/data sources。一个 agent 通过 MCP server 调用 GitHub、Slack、本地文件——是纵向连接（agent 拿到能力）。
• A2A（Google 2025）：标准化 agent ↔ agent。一个 agent 通过 A2A 把任务委派给另一个 agent——是横向连接（agent 之间协作）。
• 互补不互斥：实际系统通常agent 通过 MCP 拿工具、通过 A2A 拿别的 agent 帮忙。Google 自己的官方说法就是 "A2A and MCP are complementary"。

但对独立开发者：不要先上协议。2-3 个 agent 的系统用裸 Python function call + asyncio.gather 几小时能搞定；上了 A2A/MCP 这种通用协议反而增加序列化、网络、版本兼容的复杂度。协议的价值在跨组织 / 跨厂商 / 长期演进的 agent 生态——你一个人的产品里 3 个 agent，直接函数调用比 JSON-RPC over HTTP 简单 10×、快 10×、易 debug 10×。等系统真的需要把内部 agent 暴露给外部、或者整合多个第三方 agent 时，再迁移到 A2A 也不晚。

实战示例

独立开发者最务实的「无协议」多 agent 通信模板（asyncio + shared trace）：

import asyncio, uuid, time, json
from dataclasses import dataclass, field

# —— 共享 trace：debug 多 agent 的命脉 ——
@dataclass
class SharedTrace:
    run_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    events: list = field(default_factory=list)

    def log(self, agent: str, kind: str, payload: dict):
        self.events.append({
            "t": time.time(), "agent": agent,
            "kind": kind, "payload": payload,
        })

# —— Subagent：直接 async 函数，不上 RPC ——
async def subagent(name: str, task: dict, trace: SharedTrace,
                   budget_tokens: int = 8000) -> dict:
    trace.log(name, "start", {"task": task})
    used = 0
    try:
        # LLM call + tool loop (省略)
        result = await run_llm_loop(task, budget_tokens=budget_tokens)
        used = result["tokens_used"]
        trace.log(name, "done", {"tokens": used, "preview": result["text"][:200]})
        return result
    except Exception as e:
        trace.log(name, "error", {"err": str(e), "tokens": used})
        return {"failed": True, "reason": str(e)}

# —— Orchestrator: 并发 + budget + timeout 三层防护 ——
async def orchestrate(user_query: str) -> dict:
    trace = SharedTrace()
    trace.log("lead", "query", {"query": user_query})

    plan = await lead_plan(user_query, trace)
    if len(plan["tasks"]) > 7:
        plan["tasks"] = plan["tasks"][:7]   # 硬上限

    # 三层保护：每 agent budget / 总超时 / 失败不阻塞
    try:
        results = await asyncio.wait_for(
            asyncio.gather(*[
                subagent(f"sub-{i}", t, trace, budget_tokens=8000)
                for i, t in enumerate(plan["tasks"])
            ], return_exceptions=True),
            timeout=120,                 # 总 budget 2 分钟
        )
    except asyncio.TimeoutError:
        trace.log("lead", "timeout", {})
        results = [{"failed": True, "reason": "global timeout"}]

    # 部分成功也能聚合 (failure isolation)
    good = [r for r in results if not (isinstance(r,dict) and r.get("failed"))]
    final = await lead_aggregate(user_query, good, trace)

    # 落库：trace 是后续 debug + eval 的命脉
    json.dump({"run_id":trace.run_id, "events":trace.events, "final":final},
              open(f"runs/{trace.run_id}.json","w"), ensure_ascii=False)
    return {"answer": final, "trace_id": trace.run_id}