40% 的 Multi-Agent 项目在生产中挂掉：任务分发、结果聚合与级联失败防护的工程实践

为什么你的 Multi-Agent 系统会在生产里挂掉

2026 年 5 月，arxiv 上出现了一篇让人不舒服的论文（2605.03310）。研究者对 1600+ 条执行 trace 分析后得出结论：Multi-Agent LLM 系统的生产失败率在 41%～87% 之间。

更让人警觉的是：这些失败大部分不是因为模型回答质量差，而是因为协调层设计缺陷——任务说明模糊、子 agent 之间对齐失败、验证缺口。换句话说，你可以用最好的模型，但如果协调层没设计好，系统照样挂。

这和我自己踩过的坑完全吻合。

我在生产中维护过一个 5 个子 agent 并行处理文档分析的系统。早期版本的平均成功率是 67%——这意味着每 3 次任务就有 1 次需要人工介入或重跑。问题不在模型，在设计。

本文把我们踩过的 5 个核心陷阱和对应解法整理出来，附可运行的代码。

Multi-Agent 协调的三个核心问题

在讨论陷阱之前，先把问题域拆清楚。Multi-Agent 系统的协调层要解决三件事：

 复制代码1. 任务分发（Task Dispatch）：把父任务拆成什么粒度，分给谁，怎么传参
2. 结果聚合（Result Aggregation）：子 agent 都跑完（或部分失败）后，结果怎么合并
3. 失败隔离（Failure Isolation）：一个子 agent 的失败不应传染给其他子 agent 或父 agent

大多数教程只讲第一个，生产里挂的大多是第二和第三个。

陷阱一：任务粒度不对，超时和调度开销都吃光利润

问题

"把文档拆成小段给多个 agent 并行处理"——这听起来合理，但粒度不对会出两种极端：

• 粒度太大：每个子 agent 要处理 100 页文档，单个子 agent 超时 → 整个任务失败
• 粒度太小：1000 个子 agent 各处理 1 句话 → 调度开销（LLM 调用的 prompt overhead + context）远大于计算价值

我们测试过一个文档摘要任务：

结论：最优粒度通常是"一个子 agent 可以在 1/3 父任务超时内完成的工作量"。太细的拆分反而因为 context 开销让 token 成本暴增，还降低了成功率。

解法：超时预算传播

父 agent 应该把自己剩余的 deadline budget 按比例传给子 agent，而不是给每个子 agent 固定超时：

 复制代码import asyncio
import time
from dataclasses import dataclass
from typing import Optional@dataclass
class TaskBudget:
    """超时预算：从父任务传播给子任务"""
    deadline: float  # Unix timestamp
    
    @property
    def remaining_seconds(self) -> float:
        return max(0, self.deadline - time.time())
    
    def child_budget(self, fraction: float = 0.8) -> "TaskBudget":
        """
        给子任务的 budget = 剩余时间 × fraction
        fraction < 1 保留父任务的聚合/清理时间
        """
        remaining = self.remaining_seconds
        return TaskBudget(deadline=time.time() + remaining * fraction)
    
    def is_expired(self) -> bool:
        return self.remaining_seconds <= 0
async def dispatch_with_budget(
    tasks: list[dict],
    worker_fn,
    parent_budget: TaskBudget,
    max_concurrency: int = 5,
) -> list[dict]:
    """
    带超时预算传播的 fan-out 分发器
    """
    if parent_budget.is_expired():
        raise TimeoutError("父任务预算已耗尽，拒绝分发")
    
    # 每个子任务拿到 80% 的剩余 budget
    child_budget = parent_budget.child_budget(fraction=0.8)
    
    semaphore = asyncio.Semaphore(max_concurrency)
    
    async def run_one(task: dict) -> dict:
        async with semaphore:
            timeout = child_budget.remaining_seconds
            if timeout <= 0:
                return {"task": task, "status": "skipped", "reason": "budget_expired"}
            try:
                result = await asyncio.wait_for(
                    worker_fn(task, budget=child_budget),
                    timeout=timeout,
                )
                return {"task": task, "status": "success", "result": result}
            except asyncio.TimeoutError:
                return {"task": task, "status": "timeout", "reason": "child_budget_exceeded"}
            except Exception as e:
                return {"task": task, "status": "error", "reason": str(e)}
    
    return await asyncio.gather(*[run_one(t) for t in tasks])

关键点：

1. child_budget.remaining_seconds 每次调用都是动态计算的，不是固定值
2. fraction=0.8 给父 agent 保留 20% 时间做聚合
3. semaphore 控制并发，避免 LLM API 限流

陷阱二：Partial Failure 聚合策略没想好，要么过严要么过松

问题

5 个子 agent 并行跑，3 个成功、2 个超时。你应该怎么处理？

大多数实现只有两种策略：

1. All-or-nothing：任何一个失败 → 整体失败。太严，生产可用性差
2. Best-effort：收集所有成功结果，忽略失败。太松，可能返回残缺数据给下游

两种都是错的。正确答案是按任务类型定义聚合策略：

解法：声明式聚合策略

 复制代码from enum import Enum
from typing import Callable, Anyclass AggregationPolicy(Enum):
    ALL_SUCCESS = "all_success"          # 全部成功才算成功
    MAJORITY = "majority"                # 超过 threshold 比例成功
    AT_LEAST_ONE = "at_least_one"        # 至少 1 个成功
    BEST_EFFORT = "best_effort"          # 收集所有成功，不因失败报错@dataclass
class AggregationConfig:
    policy: AggregationPolicy
    success_threshold: float = 0.6       # MAJORITY 模式下的成功比例
    required_keys: list[str] = None      # ALL_SUCCESS 模式下的必填字段
def aggregate_results(
    raw_results: list[dict],
    config: AggregationConfig,
    merge_fn: Callable[[list[dict]], Any] = None,
) -> dict:
    """
    按策略聚合 fan-out 结果
    """
    successes = [r for r in raw_results if r.get("status") == "success"]
    failures = [r for r in raw_results if r.get("status") != "success"]
    total = len(raw_results)
    success_rate = len(successes) / total if total > 0 else 0    # 判断是否达到聚合条件
    if config.policy == AggregationPolicy.ALL_SUCCESS:
        if len(failures) > 0:
            return {
                "status": "partial_failure",
                "success_count": len(successes),
                "failure_count": len(failures),
                "failures": failures,
                "data": None,
            }
    elif config.policy == AggregationPolicy.MAJORITY:
        if success_rate < config.success_threshold:
            return {
                "status": "below_threshold",
                "success_rate": success_rate,
                "required": config.success_threshold,
                "data": None,
            }
    elif config.policy == AggregationPolicy.AT_LEAST_ONE:
        if len(successes) == 0:
            return {"status": "all_failed", "data": None}    # 合并成功结果
    merged = merge_fn([r["result"] for r in successes]) if merge_fn else [r["result"] for r in successes]
    
    return {
        "status": "success",
        "success_rate": success_rate,
        "failure_count": len(failures),
        "data": merged,
        "partial_failures": failures if config.policy == AggregationPolicy.BEST_EFFORT else [],
    }
# 使用示例：文档分析任务
config = AggregationConfig(
    policy=AggregationPolicy.MAJORITY,
    success_threshold=0.7,  # 70% 成功才通过
)result = aggregate_results(raw_results, config, merge_fn=lambda results: "nn".join(results))
if result["status"] != "success":
    # 记录哪些 chunk 失败了，标记数据缺口
    log_partial_failure(result["failures"])

陷阱三：级联失败——一个 Agent 幻觉，下游全部污染

问题

这是最危险的陷阱。考虑这个 Pipeline：

 复制代码搜索 Agent → 提取 Agent → 分析 Agent → 报告 Agent

搜索 Agent 因为上下文窗口不足，偷偷幻觉了一段"搜索结果"。提取 Agent 不知道输入是假的，认真提取了。分析 Agent 基于错误提取结果做了分析。报告 Agent 生成了一份看起来专业实则全错的报告。

没有一个 Agent 报错，但输出完全是垃圾。

MAST 数据集分析的 14 种失败模式中，"specification ambiguity"（任务说明不明确）和 "verification gaps"（验证缺口）是最高频的两种，二者都会导致级联污染。

解法一：每个 Agent 输出结构化验证

 复制代码from pydantic import BaseModel, field_validator
from typing import Literalclass SearchAgentOutput(BaseModel):
    """搜索 Agent 的输出契约"""
    status: Literal["success", "no_results", "error"]
    results: list[dict]
    source_count: int
    confidence: float  # 0.0-1.0，低于 0.5 触发降级逻辑
    
    @field_validator("confidence")
    @classmethod
    def validate_confidence(cls, v):
        if v < 0.0 or v > 1.0:
            raise ValueError(f"confidence 必须在 0-1 之间，得到 {v}")
        return v
    
    @field_validator("results")
    @classmethod
    def results_not_empty_when_success(cls, v, info):
        if info.data.get("status") == "success" and len(v) == 0:
            raise ValueError("status=success 但 results 为空，可能是幻觉")
        return v
async def run_search_agent_with_validation(query: str, budget: TaskBudget) -> SearchAgentOutput:
    """带输出验证的搜索 Agent"""
    raw_output = await call_llm_agent(
        system_prompt=SEARCH_AGENT_PROMPT,
        user_message=query,
        response_format=SearchAgentOutput,  # 使用 structured output
        timeout=budget.remaining_seconds,
    )
    
    # Pydantic 验证会在 parse 时自动触发
    output = SearchAgentOutput.model_validate(raw_output)
    
    # 额外语义检查
    if output.confidence < 0.5:
        raise LowConfidenceError(
            f"搜索 Agent 置信度 {output.confidence:.2f} 低于阈值 0.5，"
            f"拒绝传递给下游 Agent"
        )
    
    return output

解法二：Bulkhead（舱壁隔离）

Circuit Breaker 解决的是"重复调用失败 API"问题，但 Multi-Agent 的级联失败更适合用 Bulkhead 模式——每个 agent 的资源（token budget、并发槽位、超时）都是隔离的，一个 agent 资源耗尽不影响其他 agent。

 复制代码import asyncio
from collections import defaultdictclass AgentBulkhead:
    """
    舱壁隔离：每个 agent 类型有独立的并发和 token budget
    """
    def __init__(self, limits: dict[str, dict]):
        """
        limits = {
            "search": {"max_concurrent": 3, "token_budget": 50000},
            "extract": {"max_concurrent": 5, "token_budget": 30000},
            "analyze": {"max_concurrent": 2, "token_budget": 80000},
        }
        """
        self._semaphores = {
            name: asyncio.Semaphore(cfg["max_concurrent"])
            for name, cfg in limits.items()
        }
        self._token_budgets = {
            name: cfg["token_budget"]
            for name, cfg in limits.items()
        }
        self._token_used = defaultdict(int)
    
    async def run(self, agent_type: str, fn, estimated_tokens: int = 1000):
        """
        在舱壁内运行一个 agent 调用
        """
        sem = self._semaphores.get(agent_type)
        if sem is None:
            raise ValueError(f"未知 agent 类型: {agent_type}")
        
        # 检查 token budget
        budget = self._token_budgets[agent_type]
        used = self._token_used[agent_type]
        if used + estimated_tokens > budget:
            raise BudgetExhaustedError(
                f"{agent_type} 的 token budget 已耗尽 "
                f"({used}/{budget} tokens used)"
            )
        
        async with sem:
            result = await fn()
            # 更新实际 token 使用（从 LLM 响应中拿）
            if hasattr(result, "usage"):
                self._token_used[agent_type] += result.usage.total_tokens
            return result
# 使用
bulkhead = AgentBulkhead({
    "search": {"max_concurrent": 3, "token_budget": 50_000},
    "extract": {"max_concurrent": 5, "token_budget": 30_000},
    "analyze": {"max_concurrent": 2, "token_budget": 80_000},
})# search agent 并发超过 3 时自动排队，token 超出时报错而不是继续烧
result = await bulkhead.run(
    "search",
    lambda: run_search_agent(query, budget),
    estimated_tokens=2000,
)

陷阱四：子 Agent 幂等性缺失，重试导致重复执行

问题

LLM API 调用因网络抖动失败，你的 orchestrator 自动重试。但如果子 agent 已经执行了一半（比如写了数据库、发了邮件），重试会导致重复副作用。

这在 tool-calling agent 中尤为危险：

 复制代码Agent 调用 write_to_db(data) → 网络超时 → Orchestrator 重试整个子任务
→ Agent 再次调用 write_to_db(data) → 数据库写入两次

解法：幂等令牌（Idempotency Key）

 复制代码import hashlib
import json
from datetime import datetimedef make_idempotency_key(
    task_id: str,
    agent_type: str,
    attempt: int = 0,
) -> str:
    """
    生成幂等令牌。同一任务的同一 attempt，令牌相同。
    重试时 attempt 递增，避免缓存旧结果。
    """
    payload = json.dumps({
        "task_id": task_id,
        "agent_type": agent_type,
        "attempt": attempt,
    }, sort_keys=True)
    return hashlib.sha256(payload.encode()).hexdigest()[:16]
class IdempotentAgentRunner:
    """
    带幂等缓存的 Agent 运行器
    """
    def __init__(self, cache_backend):
        self.cache = cache_backend  # Redis / 内存 dict
    
    async def run(
        self,
        task_id: str,
        agent_type: str,
        agent_fn,
        attempt: int = 0,
        ttl_seconds: int = 3600,
    ):
        key = make_idempotency_key(task_id, agent_type, attempt)
        
        # 检查是否已有结果
        cached = await self.cache.get(key)
        if cached:
            return json.loads(cached)
        
        # 执行 agent
        result = await agent_fn()
        
        # 缓存结果
        await self.cache.set(key, json.dumps(result), ex=ttl_seconds)
        return result
# 使用：重试时 attempt 递增，确保不复用失败 attempt 的缓存
for attempt in range(3):
    try:
        result = await runner.run(
            task_id=parent_task_id,
            agent_type="extract",
            agent_fn=lambda: extract_agent(document_chunk),
            attempt=attempt,
        )
        break
    except Exception as e:
        if attempt == 2:
            raise
        await asyncio.sleep(2 ** attempt)  # 指数退避

关键设计：

• attempt 参数确保每次重试有独立的幂等 key，不会复用失败的缓存
• TTL 设为 1 小时，避免因网络问题导致长期缓存旧结果
• tool-calling agent 的工具调用结果也要在 agent 侧做幂等（数据库层用 INSERT ... ON CONFLICT DO NOTHING）

陷阱五：Orchestrator 不知道子 Agent 的"内部状态"，调试地狱

问题

生产出问题了，你知道整个任务失败了，但不知道哪个子 agent 在哪一步、用了多少 token、中间输出是什么。日志里只有 AgentError: task failed。

解法：结构化 Span 追踪

 复制代码import uuid
from contextlib import asynccontextmanager
from dataclasses import dataclass, field
import time@dataclass
class AgentSpan:
    """单个 Agent 调用的追踪 Span"""
    span_id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
    agent_type: str = ""
    task_id: str = ""
    parent_span_id: str = None
    started_at: float = field(default_factory=time.time)
    ended_at: float = None
    status: str = "running"  # running / success / error / timeout
    input_summary: str = ""  # 不存全量，只存摘要
    output_summary: str = ""
    token_usage: dict = field(default_factory=dict)
    error: str = None
    
    def finish(self, status: str, output_summary: str = "", token_usage: dict = None, error: str = None):
        self.ended_at = time.time()
        self.status = status
        self.output_summary = output_summary
        self.token_usage = token_usage or {}
        self.error = error
        return self
    
    @property
    def duration_seconds(self) -> float:
        end = self.ended_at or time.time()
        return end - self.started_at
class AgentTracer:
    """
    轻量级 Agent 调用追踪器
    生产中可替换为 OpenTelemetry / LangSmith / 自研 trace backend
    """
    def __init__(self, trace_backend=None):
        self._spans: list[AgentSpan] = []
        self._backend = trace_backend
    
    @asynccontextmanager
    async def span(
        self,
        agent_type: str,
        task_id: str,
        parent_span_id: str = None,
        input_summary: str = "",
    ):
        s = AgentSpan(
            agent_type=agent_type,
            task_id=task_id,
            parent_span_id=parent_span_id,
            input_summary=input_summary,
        )
        self._spans.append(s)
        try:
            yield s
            if s.status == "running":
                s.finish("success")
        except asyncio.TimeoutError:
            s.finish("timeout", error="TimeoutError")
            raise
        except Exception as e:
            s.finish("error", error=str(e))
            raise
        finally:
            if self._backend:
                await self._backend.record(s)
    
    def get_trace_summary(self) -> dict:
        """返回整个执行链路的摘要，用于错误报告"""
        return {
            "total_spans": len(self._spans),
            "failed_spans": [
                {"span_id": s.span_id, "agent": s.agent_type, "error": s.error}
                for s in self._spans if s.status in ("error", "timeout")
            ],
            "total_tokens": sum(
                s.token_usage.get("total_tokens", 0) for s in self._spans
            ),
            "total_duration": max((s.duration_seconds for s in self._spans), default=0),
        }
# 使用示例
tracer = AgentTracer(trace_backend=your_trace_backend)# 父 Agent
async with tracer.span("orchestrator", task_id="doc-analysis-001") as parent_span:
    tasks = split_document(document)
    
    # 子 Agent（携带 parent_span_id）
    async def run_child(chunk, idx):
        async with tracer.span(
            "extract",
            task_id=f"doc-analysis-001-chunk-{idx}",
            parent_span_id=parent_span.span_id,
            input_summary=f"chunk {idx}: {len(chunk)} chars",
        ) as child_span:
            result = await extract_agent(chunk, budget)
            child_span.finish(
                "success",
                output_summary=f"{len(result.entities)} entities extracted",
                token_usage=result.usage.__dict__,
            )
            return result
    
    results = await asyncio.gather(*[run_child(c, i) for i, c in enumerate(tasks)])# 失败时打印整链路摘要
print(tracer.get_trace_summary())

生产部署检查表

按上面五个陷阱，整理成一张过关检查表：

最后一条常被忽视。学术研究（2024-2026 多篇）多次指出：在很多任务上，设计良好的单 Agent 系统和简单多 Agent 系统性能相当，但成本低得多。Multi-Agent 是手段不是目的——先证明你的任务需要它，再付出协调复杂度的代价。

小结

Multi-Agent 系统的生产可靠性不是模型问题，是工程问题。五个核心陷阱：

1. 任务粒度不对 → 超时预算传播
2. 聚合策略不清 → 声明式聚合 + partial failure 处理
3. 级联失败 → 输出验证 + Bulkhead 隔离
4. 幂等性缺失 → idempotency key + 重试策略
5. 可观测性缺失 → 结构化 Span 追踪

代码都可以直接用。如果你的 Multi-Agent 系统目前的生产成功率低于 90%，大概率是这五个里的某几个没做好。

参考资料

• Shuvalov, P. (2026). "Coordination as an Architectural Layer for LLM-Based Multi-Agent Systems". arXiv:2605.03310
• Cemri et al. (2025). MAST: Multi-Agent System Taxonomy. 1600+ execution traces failure analysis
• 相关学术研究（2024-2026）: Multi-Agent Baselines Comparisons