｜Agentic AI Group
智能体工程课程
2026 版
进阶8 学时更新：2026-07-26

03｜多智能体协同、路由与编排

理解多智能体不是简单增加角色数量，而是通过任务边界、共享状态、路由和质量门禁提升复杂任务可靠性。

前置知识
  • 理解单智能体状态与工具调用
  • 掌握 Python asyncio 基础
学习成果
  • 能比较层级式、对等式与市场式协同
  • 能实现任务分发、并行执行、重试和降级
  • 能根据任务选择 LangGraph、AutoGen 或 CrewAI 等框架
层级式多智能体协同
图 3-1：协调者负责目标和质量，专业智能体负责可验证的局部交付。

1. 多智能体为何有效，又为何会失败

单一智能体在长任务中常见问题：上下文过长、角色目标混杂、工具太多导致选择困难、错误在后续步骤放大。多智能体通过角色分工和上下文隔离缓解这些问题。

但多智能体并不天然更好。增加一个 Agent 会增加模型调用、消息传递、状态同步和错误面。只有满足以下条件时才值得拆分：

  • 子任务可清晰定义输入输出；
  • 子任务需要不同工具、知识或权限；
  • 子任务可以并行或独立评估；
  • 总协调成本低于拆分带来的收益。

1.1 三种架构模式

模式调度方式优点风险适用
层级式协调者分派与汇总目标统一、易审计协调者成为瓶颈报告、研发、审批
对等式Agent 直接协商灵活、无单点决策易产生重复和争论探索性讨论、协作模拟
市场式按能力、成本或竞价选择资源利用灵活机制复杂、难调试大规模技能平台

初学项目优先使用层级式。它最容易定义责任、预算和停止条件。

2. 主流框架与选型

框架主要抽象强项适合教学内容
LangGraphState、Node、Edge、Checkpoint显式状态图、持久化、HITL路由、循环、恢复、工程化
AutoGen AgentChatAgent、Team、Message多 Agent 对话与团队模式Selector、Swarm、人机交互
CrewAIAgent、Task、Crew、Flow角色任务与事件流顺序/层级流程、快速业务原型
OpenAI Agents SDKAgent、Handoff、Guardrail、Trace轻量 Agent、交接与追踪工具调用、交接、安全门禁
LlamaIndex WorkflowsEvent、Step、Context数据与 RAG 工作流结合知识型多智能体
选型原则先确定状态、流程和质量要求，再选框架。不要根据“角色描述写起来最像人”来选生产框架。

3. 编排与路由策略

3.1 编排的四个职责

  1. 任务拆解：把目标转成可验收子任务；
  2. 执行调度：决定串行、并行、分支或循环；
  3. 状态合并：处理多个 Agent 的结果、冲突和版本；
  4. 异常恢复：重试、替代、降级、人工处理。

3.2 路由信号

路由器可以使用：任务类型、复杂度、数据敏感度、所需工具、时限、预算、历史成功率和当前负载。生产中建议“规则优先、模型补充”：

if task.requires_write and not user.approved:
    route = "human_approval"
elif task.type == "retrieval":
    route = "research_agent"
elif task.complexity <= 2:
    route = "small_model_agent"
else:
    route = classifier_agent(task)

3.3 大小模型分级路由

  • 规则或小模型负责分类、抽取、格式检查；
  • 强模型负责规划、综合和困难推理；
  • 低置信度或关键决策升级到更强模型或人工；
  • 所有升级记录原因，避免系统默默增加成本。

3.4 负载均衡与优先级

将任务放入队列，使用 priority、deadline、estimated_cost 和 required_capability 调度。高优先级不意味着无限抢占，应设置每类任务配额，防止普通任务长期饥饿。

4. 容错、检查点与人机协同

4.1 错误分类

错误处理策略
临时网络失败指数退避重试，限制次数
模型限流切换备用模型或排队
工具参数错误返回结构化错误，允许 Agent 修正一次
证据不足补充检索或转人工
结果冲突交给审校 Agent，并保留原始结果
高风险动作暂停并请求人工审批

4.2 幂等与检查点

每个任务和工具调用都应有唯一 ID。重复执行时先检查是否已完成，避免重复发邮件、重复扣款或重复写入。检查点保存当前节点、输入、已完成任务、预算和错误，支持进程重启后恢复。

4.3 可运行的异步协调器

完整文件：examples/03_multi_agent_orchestrator.py。

import asyncio
from dataclasses import dataclass
from typing import Awaitable, Callable

@dataclass
class Task:
    name: str
    payload: str
    retries: int = 1

Agent = Callable[[Task], Awaitable[str]]

async def run_with_retry(agent: Agent, task: Task) -> str:
    last_error: Exception | None = None
    for attempt in range(task.retries + 1):
        try:
            return await asyncio.wait_for(agent(task), timeout=3)
        except Exception as exc:
            last_error = exc
            await asyncio.sleep(0.2 * (2 ** attempt))
    raise RuntimeError(f"{task.name} 失败") from last_error

async def orchestrate(topic: str, agents: dict[str, Agent]) -> dict[str, str]:
    tasks = {
        "research": Task("research", topic, retries=2),
        "implementation": Task("implementation", topic, retries=1),
    }
    research, implementation = await asyncio.gather(
        run_with_retry(agents["research"], tasks["research"]),
        run_with_retry(agents["implementation"], tasks["implementation"]),
    )
    draft = f"研究证据：{research}\n实现方案：{implementation}"
    review = await run_with_retry(
        agents["review"], Task("review", draft, retries=1)
    )
    return {"draft": draft, "review": review}

5. 应用案例：研究报告多智能体团队

5.1 角色与契约

  • 研究 Agent：输出来源、摘要、发布日期和可信度；
  • 分析 Agent：只使用研究 Agent 提供的证据；
  • 实现 Agent：输出可运行代码和环境说明；
  • 审校 Agent：检查事实、引用、结构、风险和遗漏；
  • 协调者：控制预算、合并结果、决定是否返工。

每个 Agent 的输出必须是结构化 JSON，而不是自由聊天：

{
  "status": "completed",
  "claims": [{"text": "...", "source_id": "S1"}],
  "open_questions": [],
  "confidence": 0.86
}

5.2 质量门禁

  • 无来源的事实不能进入最终报告；
  • 代码必须通过语法检查；
  • 审校 Agent 不能直接覆盖原文，只提出问题和修改建议；
  • 最终发布由协调者或人工负责人确认。

5.3 工程复现与故障演练

cd repository/enterprise-agent-lab
python -m app.cli --multi-agent \
  --question "设备 P-100 最近有哪些类似故障？" \
  --user manager-001
python -m unittest discover -s tests -v

app/orchestrator.py 将研究、分析、执行结果放入结构化状态；不要用 Agent 之间的自由聊天替代接口。实践时给一个角色制造异常，要求协调器标记失败而非伪造缺失结果；再把两个无依赖子任务并行化，记录串行/并行延迟和合并规则。

部署仍复用 app.server 与项目 DEPLOY.md，但上线前要增加角色级超时、总轮次/Token 预算、检查点和人工接管。验收证据包括角色输入输出、一次 Trace、重试次数、最终状态和失败恢复记录。

6. 自测

何时不应拆成多智能体？

子任务无法独立验收、状态共享成本很高、单个确定性工作流已经足够时。

为什么审校 Agent 不应直接修改最终事实？

需要保留来源与责任链，审校应提出可追踪的修改请求，由原负责 Agent 或协调者确认。

并行执行最重要的前提是什么？

子任务之间不存在必须先后满足的数据依赖，且结果可以在之后安全合并。

7. 可靠参考资料