多 Agent 架构实战：主控 Agent 如何拆解任务并调度子 Agent 完成协作

为什么单个 Agent 撑不住复杂任务

把一个 10 步流程全塞进一个 Agent 的 system prompt，它能跑，但会出问题：context window 被消耗殆尽、中间错误会污染后续推理、单点失败无法局部恢复。实测中，当任务步骤超过 6 步、涉及 2 个以上外部工具时，单 Agent 的成功率会从 85% 左右跌到 50% 以下。

多 Agent 架构的核心价值不是"更多 AI"，而是职责隔离：每个 Agent 持有最小必要上下文，专注单一职责，主控层（orchestrator）只负责规划与调度，执行层（worker）只负责具体操作。这种分层让整个系统的可观测性和可恢复性都大幅提升。

---

Orchestrator-Worker 模式的基本结构

这是目前最成熟的多 Agent 架构拓扑，结构如下：

User Request
     │
     ▼
┌─────────────┐
│ Orchestrator │  ← 规划、分解、调度、聚合
└──────┬──────┘
       │  dispatches subtasks
  ┌────┼────┐
  ▼    ▼    ▼
[W1] [W2] [W3]   ← 各自持有独立 context，并行或串行执行

Orchestrator 的职责边界

• 接收用户意图，输出结构化任务图（task graph）
• 决定哪些子任务可以并行，哪些必须串行（依赖关系）
• 监听每个 worker 的返回状态，决定是否重试或降级
• 汇总所有 worker 结果，生成最终响应

Worker 的职责边界

• 只接受来自 orchestrator 的标准化输入（通常是 JSON schema）
• 拥有独立的 tool call 权限（如：搜索 worker 只能调用 websearch，代码 worker 只能调用 codeinterpreter）
• 返回结构化结果，不做跨任务推理

权限隔离是关键。给每个 worker 最小权限集，不仅降低安全风险，也让 orchestrator 的调度逻辑更可预测。

---

任务分解策略：从意图到可执行子任务

任务分解是整个架构最难的环节，直接影响后续所有步骤的质量。有三种常用策略：

策略 1：静态分解（适合流程固定的场景）

在 orchestrator 的 system prompt 里预定义任务模板，用户输入触发对应模板。适合客服、审批流等场景。优点是延迟低（无需 LLM 推理分解步骤），缺点是灵活性差。

策略 2：LLM 动态分解（适合开放式任务）

让 orchestrator 本身是一个强推理模型（如 Claude Opus 4.8 或 GPT-5.5），给它一个 JSON schema，要求它输出 task graph：

{
  "tasks": [
    {
      "id": "t1",
      "worker": "search_agent",
      "input": {"query": "2025 EV market share"},
      "depends_on": []
    },
    {
      "id": "t2",
      "worker": "analyst_agent",
      "input": {"topic": "EV market trends"},
      "depends_on": ["t1"]
    }
  ]
}

实践中建议在 prompt 里明确限制 task 数量上限（通常 5~8 个），避免 orchestrator 过度拆解导致调度开销反超收益。

策略 3：混合模式

对已知任务类型走静态路由，对未知类型降级到 LLM 动态分解。这是生产环境最常见的选择，兼顾速度与灵活性。

策略	首次延迟	灵活性	维护成本	适用场景
静态分解	~0ms	低	中	固定流程
LLM 动态分解	800~2000ms	高	低	开放任务
混合模式	~200ms	中高	中	生产系统

---

并行调度与结果聚合

拿到 task graph 之后，orchestrator 需要根据 depends_on 字段构建 DAG（有向无环图），然后按拓扑序调度执行。

并行执行的实现方式

用 Python asyncio 是最直接的方案：

import asyncio

async def run_parallel_tasks(tasks: list[Task]) -> list[Result]:
    # 筛选出当前层无依赖的任务
    ready = [t for t in tasks if all_deps_done(t)]
    results = await asyncio.gather(
        *[dispatch_to_worker(t) for t in ready],
        return_exceptions=True  # 关键：不让单个失败中断整批
    )
    return results

return_exceptions=True 是生产代码里容易漏的细节。没有它，任何一个 worker 抛异常都会让整个 gather 失败，而有了它，你可以在聚合层单独处理每个任务的成败。

结果聚合策略

• 全量聚合：等所有 worker 完成再汇总，适合报告生成类任务
• 流式聚合：worker 每完成一个就实时更新结果，适合有进度反馈需求的场景
• 阈值聚合：收到 N 个结果后不等剩余 worker，用于对延迟敏感的场景（类似 speculative execution）

聚合层通常也需要一个 LLM 调用来做最终语义整合，这里推荐用较轻量的模型（如 Claude Haiku 4.5 或 GPT-5.4-mini），降低成本，因为聚合逻辑通常比分解简单得多。

---

失败重试与降级机制

分布式系统的经典问题在多 Agent 架构里同样存在，而且因为 LLM 调用本身具有非确定性，失败模式更复杂。

三类失败及对应处理

失败类型	表现	推荐处理
工具调用超时	worker 等待外部 API 无响应	指数退避重试，最多 3 次
输出格式不合规	worker 返回的 JSON 解析失败	注入 validation error 重新调用，最多 2 次
语义质量不达标	orchestrator 评估 worker 输出不满足要求	换更强的模型重试一次，或标记为人工审核

指数退避重试示例

import asyncio

async def call_with_retry(worker_fn, input_data, max_retries=3):
    for attempt in range(max_retries):
        try:
            result = await worker_fn(input_data)
            if validate_output(result):
                return result
        except Exception as e:
            if attempt == max_retries - 1:
                return {"status": "failed", "error": str(e)}
            await asyncio.sleep(2 ** attempt)  # 1s, 2s, 4s

降级策略：当某个 worker 持续失败时，不应该让整个请求失败。常见做法是：标记该子任务为 skipped，在最终聚合时注明"部分结果不可用"，让用户获得一个不完整但有价值的答案，而不是硬错误。这在用户体验上差异巨大。

Orchestrator 本身的失败：这是最危险的单点。实践中建议对 orchestrator 的输出做 schema 严格校验，一旦 task graph 格式非法，直接回退到静态分解或返回错误，不要让格式错误的任务图流入调度层。

---

构建多 Agent 系统最耗时的环节往往不是写代码，而是调用大量模型时的 API 管理成本。我自己在测试不同 orchestrator 和 worker 模型组合时，会用 XycAi 词元平台——200+ 模型通过一个 OpenAI 兼容 API 接入，Claude Opus 4.8、GPT-5.5 等旗舰模型低至官方价 1.4 折起，Claude Code / Codex / Gemini CLI 也支持一键接入。多模型混用的架构里，这种统一接口能省掉大量切换和密钥管理的麻烦。