5.3　在多 Agent 系统中实现错误传播策略

这一条考什么

这一条考你在"协调者 + 子 Agent"架构里如何设计错误的上报方式，让协调者能根据错误信息做出合理的恢复决策。你要会返回结构化的错误上下文，能区分"访问失败"和"查询成功但结果为空"，知道为什么泛化的错误信息、静默吞掉错误、一处失败就终止整个流程都是反模式。最终整合的输出还要带上覆盖说明（coverage notes）。

核心概念

结构化错误上下文

子 Agent 失败时，如果只回一句"搜索失败"，协调者就只能猜：要不要重试？换个数据源？还是放弃这个子课题？结构化的错误上下文应该包括：

• 失败类型：timeout、rate_limited、auth_denied、not_found、invalid_query、source_unavailable……
• 尝试过的内容：具体查询、参数、尝试次数；
• 部分结果：失败之前已经拿到的数据；
• 可能的替代方案：换关键词、换数据源、缩小时间范围。

{
  "status": "partial_failure",
  "error_type": "timeout",
  "retryable": true,
  "attempted": [
    {"query": "solid-state battery patents 2024", "source": "patent_db", "attempts": 3, "result": "timeout after 30s"}
  ],
  "partial_results": [{"title": "EP4012345 ...", "url": "https://example.org/ep4012345"}],
  "alternatives": ["改用 WIPO 检索接口", "把时间范围缩小到 2024-H2"]
}

有了这些信息，协调者就能判断：这是暂时性故障，可以稍后重试；部分结果已经够用；或者应该派另一个子 Agent 走替代路线。

访问失败 ≠ 有效的空结果

这是本条最核心的区分：

• 访问失败：超时、401/403、服务不可用。查询根本没有执行成功，不知道有没有数据。需要考虑重试或换数据源。
• 有效的空结果：查询成功执行，确实没有匹配项。这本身就是一个有意义的发现（"2024 年该领域没有相关专利"），不应该重试。

如果两者都表现为"返回空列表"，协调者就可能把"没查到"当作"不存在"，把一次超时写成结论，或者对一个确实没有结果的查询无休止地重试。所以错误报告里必须显式区分，比如用 status: "success", results: [] 和 status: "error", error_type: "timeout" 两种表示。

泛化错误状态为什么有害

"搜索不可用"这种消息把所有原因混在一起：是限流（等一下就好）、认证过期（要换凭证）、查询语法错误（要改查询），还是服务宕机（要换数据源）？协调者拿不到区分依据，只能一刀切，要么全部重试，要么全部放弃，两种都不对。

两个极端的反模式

• 静默吞掉错误：子 Agent 遇到异常后返回空结果，并标记成功。下游以为"没有数据"，最终报告里自信地写"未发现相关研究"，这个错误会一路传到用户那里，而且没人察觉。
• 一处失败就终止整个流程：5 个子课题里有 1 个的数据源超时，就让整个研究任务报错退出，另外 4 个已经完成的结果也全部作废。正确做法是继续完成能完成的部分，在输出里标明哪里有缺口。

本地恢复 + 只上报解决不了的错误

子 Agent 应该先在本地处理暂时性故障：带退避的有限次重试、换一个等价接口。只有本地解决不了的错误才上报给协调者，并附上尝试过的内容和部分结果。这样协调者不会被每一次网络抖动打扰，同时又能拿到真正需要它决策的信息。

def search_with_recovery(query, max_attempts=3):
    tried = []
    for i in range(max_attempts):
        try:
            hits = search_api(query)
            return {"status": "success", "results": hits}  # hits 可以是 []，也是有效结果
        except TimeoutError:
            tried.append({"query": query, "attempt": i + 1, "error": "timeout"})
            time.sleep(2 ** i)
        except PermissionError:
            return {"status": "error", "error_type": "auth_denied", "retryable": False, "attempted": tried}
    return {"status": "error", "error_type": "timeout", "retryable": True,
            "attempted": tried, "partial_results": [], "alternatives": ["换用备用检索源"]}

在 MCP 工具里，同样的思路体现为：返回 isError: true，并在内容里给出结构化的错误类型和是否可重试（参见 2.2）。

带覆盖说明的整合输出

协调者整合时，要在最终输出里明确写出覆盖情况：哪些结论有充分来源支持，哪些子课题因为数据源不可用而存在缺口。

## 覆盖说明
- 市场规模：3 个独立来源，结论充分
- 专利布局：专利数据库超时（已重试 3 次），仅基于 2 条部分结果，结论待补充
- 政策环境：检索成功，未找到 2024 年后的新法规（有效空结果）

读者因此能分清"确实没有"和"没查到"。

正确做法与反模式

• 正确：返回失败类型、尝试过的内容、部分结果和替代方案。反模式：只回"操作失败"、"搜索不可用"这类泛化消息。
• 正确：显式区分访问失败和有效的空结果。反模式：超时的时候也返回空列表并标记成功。
• 正确：子 Agent 在本地对暂时性故障做有限次重试，解决不了再上报。反模式：每次抖动都上报协调者，或者在本地无限重试。
• 正确：单个子任务失败时继续完成其他部分，并在输出中标明缺口。反模式：一处失败就终止整个工作流。
• 正确：整合输出带上覆盖说明，区分证据充分的结论和有缺口的结论。反模式：缺口被悄悄略过，报告读起来像是全面覆盖了。

常见陷阱

• "出错时返回空结果，保证流程继续"：表面上很健壮，其实是静默吞掉错误，会把"没查到"伪装成"不存在"。
• "任何子 Agent 失败都让整个任务失败，以保证结果完整"：浪费已有的成果。正确做法是降级完成，并标注缺口。
• "统一返回一个错误码，方便协调者处理"：泛化信息恰恰让协调者没法区分该重试、该换源还是该放弃。
• "由协调者负责所有重试"：暂时性故障应该在子 Agent 本地恢复，协调者只处理本地解决不了的问题。
• 把有效的空结果当作错误去重试：查询成功但没有匹配项，就是结论，重试只会浪费调用。
• 把错误信息写成长段自然语言：协调者（和代码）难以可靠解析。应该使用结构化字段。

实操练习

目标：模拟 3 个子 Agent（一个成功、一个返回有效空结果、一个超时），让协调者根据结构化错误生成带覆盖说明的报告。

1. 用 Python 函数模拟 3 个子 Agent 的返回值。
2. 把它们的 JSON 放进协调者的提示词。
3. 要求协调者输出报告和"覆盖说明"一节。

import json
import anthropic

client = anthropic.Anthropic()

sub_results = {
    "market_size": {"status": "success", "results": [
        {"claim": "2025 年全球市场规模约 120 亿美元", "source": "https://example.org/report-a"}]},
    "regulation": {"status": "success", "results": []},
    "patents": {"status": "error", "error_type": "timeout", "retryable": True,
                "attempted": [{"query": "solid-state battery patent 2024", "attempts": 3}],
                "partial_results": [{"title": "EP4012345", "url": "https://example.org/ep4012345"}],
                "alternatives": ["改用 WIPO 检索", "缩小到 2024-H2"]},
}

prompt = f"""你是研究协调者。以下是各子 Agent 的结构化返回：
{json.dumps(sub_results, ensure_ascii=False, indent=1)}

请：
1. 对每个子课题决定下一步（采用 / 视为有效空结果 / 重试或换源），说明理由；
2. 输出简短报告，并以"## 覆盖说明"结尾，区分证据充分、有效空结果、因访问失败而存在缺口的部分。"""

r = client.messages.create(model="claude-sonnet-4-5", max_tokens=1200,
                           messages=[{"role": "user", "content": prompt}])
print(r.content[0].text)

要检查的结果：协调者应该把 regulation 视为"已检索、无新法规"，并且不要求重试；把 patents 判断为可重试的访问失败，并给出替代方案（比如 WIPO）；覆盖说明里明确写出专利部分有缺口。然后把 patents 改成 {"status": "success", "results": []} 再跑一次，观察报告是不是误写成了"没有相关专利"，体会静默吞错的危害。

自测清单

• 结构化错误上下文应该包含哪四类信息？各自帮助协调者做什么决策？
• "访问失败"和"有效的空结果"有什么区别？混在一起会导致什么后果？
• 为什么"搜索不可用"这类泛化错误信息有害？
• 静默吞掉错误和一处失败就终止全流程，分别错在哪里？
• 哪些错误应该由子 Agent 在本地恢复，哪些应该上报给协调者？
• 整合输出里的覆盖说明应该写哪些内容？