← AI 智能体探索 · ← 上一篇

多 Agent 编排 — 让多个 AI Agent 协作完成复杂任务

2026年5月14日 · 深入
核心收获：能用一个就别用多个——多 Agent 增加复杂度和成本。Agent 之间用结构化数据通信（JSON，不是自然语言）。设置全局超时和预算硬上限——并行 Agent 的 token 消耗可能爆炸。

单个 Agent 能做很多事。但真正的复杂场景——比如一个项目需要同时做代码审查、安全审计、文档生成——一个 Agent 的上下文窗口和注意力很快就不够用了。

这就是多 Agent 编排要解决的问题。不是让一个 Agent 更强，而是让多个 Agent 各司其职。

为什么需要多 Agent

问题 单 Agent 多 Agent
上下文窗口 一个窗口装所有 每个 Agent 独立窗口
专业深度 什么都会，什么都不精 每个 Agent 专攻一个领域
并行能力 串行执行 多个 Agent 同时工作
容错性 挂了全挂 单个 Agent 失败不影响全局

两种经典编排模式

模式一：顺序流水线

Agent A 的输出是 Agent B 的输入。像一个工厂流水线。

典型场景：代码生成 → 代码审查 → 安全扫描 → 文档生成。

def sequential_pipeline(task: str) -> str:
    # Agent 1: 生成代码
    code = agent_coder.run(f"实现以下功能: {task}")

    # Agent 2: 审查代码
    review = agent_reviewer.run(f"审查以下代码:\n{code}")
    if "需要修改" in review:
        code = agent_coder.run(f"根据反馈修改:\n{review}\n原代码:\n{code}")

    # Agent 3: 安全扫描
    security = agent_security.run(f"扫描安全漏洞:\n{code}")

    # Agent 4: 生成文档
    docs = agent_writer.run(f"为以下代码写文档:\n{code}")

    return {"code": code, "review": review,
            "security": security, "docs": docs}
📌 适用条件：任务有明确的先后依赖。上一步没完成，下一步没意义。

模式二：并行分工

多个 Agent 同时处理不同子任务，最后汇总。

典型场景：市场分析——Agent A 看技术面，Agent B 看基本面，Agent C 看资金面，最后汇总。

import concurrent.futures

def parallel_orchestration(market: str) -> dict:
    tasks = {
        "technical": f"分析 {market} 的技术指标（MACD、RSI、均线）",
        "fundamental": f"分析 {market} 的基本面（估值、盈利、增长）",
        "sentiment": f"分析 {market} 的市场情绪和新闻",
        "flow": f"分析 {market} 的资金流向和持仓变化"
    }

    with concurrent.futures.ThreadPoolExecutor() as executor:
        futures = {
            name: executor.submit(agent_analyst.run, prompt)
            for name, prompt in tasks.items()
        }
        results = {
            name: future.result()
            for name, future in futures.items()
        }

    # 汇总 Agent 综合所有分析
    summary = agent_summarizer.run(
        f"综合以下分析给出结论:\n" +
        "\n".join([f"## {k}\n{v}" for k, v in results.items()])
    )
    return {"analysis": results, "summary": summary}

MCP：Agent 间的通用语言

MCP（Model Context Protocol）是 Anthropic 推出的开放协议，解决了一个关键问题：每个 Agent 的工具和上下文如何标准化共享。

传统问题：Agent A 的工具 Agent B 用不了，Agent B 的上下文 Agent C 读不到。每个 Agent 都是信息孤岛。

MCP 的三个核心概念：

  1. Server — 提供工具和资源的服务端。比如一个「GitHub Server」提供读 PR、查 Issue 等工具
  2. Client — Agent 框架，通过标准协议连接多个 MCP Server
  3. Transport — 通信方式，支持 stdio（本地进程）和 HTTP（远程服务）
# mcp_config.yaml — Agent 框架的配置文件
servers:
  filesystem:
    command: "npx"
    args: ["-y", "@anthropic/mcp-server-filesystem", "/workspace"]
  github:
    command: "npx"
    args: ["-y", "@anthropic/mcp-server-github"]
    env:
      GITHUB_TOKEN: "${GITHUB_TOKEN}"
  database:
    url: "https://db-mcp.internal/mcp"
    transport: "http"

配置好后，Agent 自动获得所有这些 Server 提供的工具——搜索文件、读 PR、查数据库——无需为每个工具单独写集成代码。

怎么评估 Agent 系统

Agent 不像分类模型——没法用准确率衡量。需要多维度评估：

维度 怎么测 指标
任务完成率 给 100 个标准任务，看完成多少 完成率 %
工具调用准确率 检查每次工具调用是否正确 正确调用 / 总调用
效率 完成任务用了多少轮和 token 平均轮次、token 消耗
自愈率 遇到错误后自己修复的比例 自愈次数 / 错误次数
安全性 注入恶意 prompt 看是否执行 拦截率 % 
def evaluate_agent(agent, test_suite: list[dict]) -> dict:
    """基础 Agent 评估框架。"""
    results = {"passed": 0, "failed": 0, "details": []}

    for case in test_suite:
        try:
            output = agent.run(case["input"])
            # 用另一个 Agent 或规则判断是否通过
            verdict = judge_agent.run(
                f"任务: {case['input']}\n"
                f"预期: {case['expected']}\n"
                f"实际: {output}\n"
                f"判断是否通过（回复 PASS 或 FAIL）"
            )
            passed = "PASS" in verdict.upper()
            results["passed" if passed else "failed"] += 1
            results["details"].append({
                "task": case["name"],
                "passed": passed,
                "verdict": verdict
            })
        except Exception as e:
            results["failed"] += 1
            results["details"].append({
                "task": case["name"], "passed": False, "error": str(e)
            })

    return results

生产部署考量

层面 要做什么
沙箱隔离 Agent 的代码执行必须在 Docker/VM 沙箱中，绝不能直接访问宿主机
速率限制 每个用户/Agent 设 API 调用上限，防止失控消耗
审计日志 记录每个 Agent 的每次工具调用、参数和结果，用于事后分析
可观测性 实时监控 Agent 状态、token 消耗、错误率
回退策略 Agent 无法处理时，优雅降级到人工处理

编排的黄金法则

  1. 能用一个就别用多个——多 Agent 增加复杂度和成本。只有单 Agent 确实不够用时才上多 Agent。
  2. Agent 之间用结构化的数据通信——不要一个 Agent 输出自然语言、另一个 Agent 猜意思。用 JSON、检查清单、结构化报告。
  3. 设置全局超时和预算——多个 Agent 并行时，总 token 消耗可能爆炸。必须设硬上限。

📖 下一篇：从零写 Agent 框架 — 可验证的执行追踪与沙箱安全

常见问题

Q: 什么时候应该从单 Agent 升级到多 Agent？
A: 当出现以下信号时：① 上下文窗口装不下所有必要信息；② 任务需要多个不相关的专业领域知识；③ 需要并行处理以减少延迟；④ 单个 Agent 失败会阻塞一切（需要容错）。如果这些都不成立——保持单 Agent。
Q: 顺序流水线和并行分工怎么选？
A: 任务有明确先后依赖（上一步没完成下一步没意义）→ 顺序流水线。子任务之间独立、可同时进行 → 并行分工。复杂场景常混合使用：并行阶段内部可能嵌套顺序流水线。
Q: MCP 协议解决了什么实际问题？
A: 每个 Agent 的工具和上下文以前是信息孤岛——Agent A 的工具 Agent B 用不了。MCP 提供了标准化的 Server-Client-Transport 三层模型，配置一个 YAML 文件就能让所有 Agent 共享工具（文件系统、GitHub、数据库等）。
Q: 如何评估多 Agent 系统好不好？
A: 五个维度：任务完成率（100 个标准任务完成多少）、工具调用准确率（每次调用是否正确）、效率（平均轮次和 token 消耗）、自愈率（遇到错误自己修复的比例）、安全性（恶意 prompt 拦截率）。不能用单一指标。
Q: 生产部署最容易忽略什么？
A: 回退策略。Agent 无法处理时，必须优雅降级到人工处理——而不是静默失败或给出错误答案。另外，审计日志不仅是调试工具，很多行业是合规要求。

可引用定义

多 Agent 编排（Multi-Agent Orchestration）：一种系统架构方法，通过协调多个专业化的 AI Agent 分工协作来完成单个 Agent 无法独立处理的复杂任务。核心包括两种经典模式：串行流水线（Sequential Pipeline，Agent A 的输出作为 Agent B 的输入，适合有明确先后依赖的任务）和并行分发（Parallel Fan-Out，多个 Agent 同时处理不同子任务后汇总结果）。Agent 间通信应使用结构化数据（JSON 等）而非自然语言，以降低歧义和 token 开销。必须设置全局超时和预算硬上限，因为并行 Agent 的 token 消耗可能指数级增长。MCP（Model Context Protocol）作为标准化 Agent 间通信协议，解决工具和上下文共享问题。

下一步阅读

  • 📖 基础：什么是 AI Agent — 如果你对 Agent 的核心概念还不熟悉，从这里建立基础认知
  • 📖 进阶：为什么辩论比单一回答更可靠 — 深入一个具体的多 Agent 协作模式：对抗辩论如何比单 Agent 产生更可靠的决策
  • 📖 相关：Agent 错误恢复 — 多 Agent 编排中单个 Agent 失败不应拖垮全局，掌握错误隔离与优雅降级策略

© 2026 xslyl.com — AI Agent 系列 · 第 5 篇

关于 · 联系 · 隐私政策 · Sitemap