多 Agent 协作：任务拆解与编排工程实践

2026-06-26

多 Agent 协作不是”堆更多 Agent 就更聪明”，而是把一个大任务拆成可并行、可监控的子任务，让每个 Agent 专注于自己最擅长的能力边界。理解编排拓扑和通信协议，是让多 Agent 系统从 Demo 走向生产的关键。

你大概率是这样入坑的：先写一个大 prompt 塞满所有指令，让单个 Agent 又查资料又写代码又自己审核，结果它一会儿忘了前面的要求，一会儿把审核步骤跳过直接给结果，你只能靠加更长的 prompt 去”提醒”它。这条路走到头就是 prompt 越堆越长、上下文越吃越满、输出却越来越不可控。真正的解法不是让一个 Agent 更聪明，而是把职责拆开：谁查资料、谁写代码、谁挑错，各司其职，互相之间用结构化协议对话，而不是靠一坨自然语言指令硬撑。下面这些拓扑和踩过的坑，都是从”单 Agent 顶不住”这个痛点倒推出来的。

三种编排拓扑对比

拓扑	结构	适用场景	缺点
单链（Pipeline）	A → B → C 顺序传递	流程固定、步骤依赖强	任一环节失败全链阻塞
DAG（有向无环图）	并行子任务 + 汇总节点	子任务相互独立可并行	编排复杂，调试成本高
Supervisor 模式	主 Agent 路由，子 Agent 执行	任务种类多、动态分配	主 Agent 成为性能瓶颈

选型建议：需求不明确时从 Supervisor 开始，瓶颈出现后再拆成 DAG 局部并行。

选型时还有一个容易被忽略的维度：延迟预算。单链最慢，因为每一步都要等上一步的模型返回，三步顺序调用哪怕每步 2 秒，用户也要等 6 秒起步；DAG 把互不依赖的子任务并行发出去，总耗时接近”最慢的那一个子任务”而不是”所有子任务之和”，这也是为什么信息聚合类任务（同时查天气、查新闻、查股价再汇总）几乎都该走 DAG；Supervisor 模式的延迟取决于路由决策本身要不要调一次模型——如果路由逻辑简单到能用规则判断（关键词匹配、正则），就别浪费一次模型调用去做路由，直接用代码分支，把 Supervisor 降级成一个纯函数，成本和延迟都能砍掉一截。

Supervisor 模式实现

主 Agent 负责理解任务、分配子 Agent、汇总结果：

SUPERVISOR_PROMPT = """
你是任务调度器。根据用户需求，决定调用哪个子 Agent：
- research_agent：负责信息检索和资料整理
- code_agent：负责编写和调试代码
- review_agent：负责质量检查和风险评估
以 JSON 格式返回: {"agent": "<name>", "task": "<具体指令>"}
"""

def supervisor_loop(user_input: str) -> str:
    history = []
    while True:
        decision = call_llm(SUPERVISOR_PROMPT, user_input, history)
        agent_name = decision["agent"]
        task = decision["task"]

        if agent_name == "DONE":
            return decision["result"]

        result = AGENTS[agent_name].run(task)
        history.append({"agent": agent_name, "task": task, "result": result})

这段代码看着简单，真正上生产前你至少要补三个东西，不然线上迟早出事：

第一，decision["agent"] 这行会炸。模型偶尔不按 JSON 格式返回，或者返回的 JSON 里少了 agent 字段，你会直接吃一个 KeyError 或者 json.JSONDecodeError。这不是小概率事件，长 history 场景下模型”看漏一句指令”的概率会明显上升。稳妥的做法是用 try/except 包住解析，解析失败就把原始文本回灌给模型让它自纠，或者干脆用支持 response_format={"type": "json_object"} 的接口强制约束输出结构，比事后修补靠谱得多。

第二，while True 没有出口保护。如果 Supervisor 的判断逻辑有 bug，一直选不中 "DONE"，这个循环会一直转下去，每转一圈就是一次模型调用、一次真金白银的 token 花费。生产环境必须加一个硬顶：比如最多循环 8 次，超过就强制返回当前已有结果并标记”未完全收敛”，宁可给一个不完美的答案，也不能让成本失控。

第三，history 是无限增长的。跑到第 10 轮的时候，你传给 Supervisor 的 history 已经把前 9 轮的子 Agent 输出全堆进去了，很容易顶到模型的上下文上限，报出类似 context_length_exceeded 的错误。实操中通常只保留最近 2-3 轮的完整记录，更早的轮次用一句话摘要替代（“第 1-5 轮已完成资料检索，结论是……”），既省 token 又不丢关键信息。

并行 DAG 执行

子任务无依赖时并行执行，用 asyncio 或线程池加速：

import asyncio

async def run_parallel_agents(subtasks: list[dict]) -> list[str]:
    """并行执行多个子 Agent 任务"""
    coros = [AGENTS[t["agent"]].arun(t["task"]) for t in subtasks]
    results = await asyncio.gather(*coros, return_exceptions=True)

    # 处理局部失败：不因一个子任务失败而全停
    return [
        r if not isinstance(r, Exception) else f"子任务失败: {r}"
        for r in results
    ]

关键设计：return_exceptions=True 让并行任务局部失败可恢复，而非全部中断。

这里有个新手常踩的认知误区：以为 asyncio.gather 是”多线程并行”，其实 Python 的 asyncio 是单线程事件循环，真正的并发发生在网络 I/O 等待期间——发出请求后 CPU 立刻切去处理下一个协程，等某个请求的响应回来了再切回去处理，本质是”等待时间的复用”而不是”计算能力的叠加”。这意味着如果你的子 Agent 里混了 CPU 密集型任务（比如本地跑一个复杂的正则匹配或者大 JSON 解析），asyncio 帮不了你，该多久还是多久，甚至会因为事件循环被阻塞而拖慢其他协程，这时候才需要考虑线程池或多进程。

第二个坑是超时。asyncio.gather 本身不带超时机制，如果某个子 Agent 调用的模型接口卡住不返回，其他任务哪怕早就跑完了，整个 gather 也会一直挂着不返回。生产代码里一定要套一层 asyncio.wait_for，给每个子任务设一个明确的超时上限（比如 30 秒），超时就当成一种 Exception 走 return_exceptions=True 的降级分支处理，而不是让用户对着转圈的界面干等：

async def run_with_timeout(agent, task, timeout=30):
    try:
        return await asyncio.wait_for(agent.arun(task), timeout=timeout)
    except asyncio.TimeoutError:
        return f"子任务超时（>{timeout}s），已跳过"

状态共享与通信

多 Agent 之间的状态传递是最容易踩坑的环节：

方案	实现	适用
直接传参	上游结果作为下游输入参数	链式流程，结构简单
共享状态对象	线程安全的字典或 Pydantic 模型	多 Agent 读写同一状态
消息队列	Redis / RabbitMQ	跨进程/跨服务异步协作
黑板模式（Blackboard）	所有 Agent 读写同一个上下文文档	协作写作、报告生成

避坑：共享状态要做版本控制或加锁，防止并发写入导致状态覆盖。

黑板模式最容易出的问题是”隐性覆盖”：两个 Agent 同时读到了同一版本的文档，各自基于这个版本改完再写回去，后写的那个会把先写的改动整个覆盖掉，而且不会报任何错——你只会发现”明明让 Agent A 加的那段内容怎么不见了”，排查起来非常费劲，因为日志里两次写入都”成功”了。简单有效的做法是给共享文档加一个版本号字段，写入前先比对版本号，不一致就拒绝写入并触发重试（这就是乐观锁的思路），比直接加全局互斥锁更适合”读多写少、偶尔冲突”的协作场景，也不会把本该并行的 Agent 硬生生排成串行。

生产落地清单

问题	应对方案
子 Agent 循环调用	记录调用链路，检测环路，超深度强制中断
结果质量不一致	review_agent 做最终输出校验，不符合则重试
调试困难	每个 Agent 记录 input/output/cost，用 trace ID 串联
总 token 成本爆炸	主 Agent 传摘要而非全文；子任务用轻量模型
局部失败处理	设计降级策略：子任务失败时用默认值或人工兜底

常见问题

多 Agent 和单 Agent 加多工具有什么区别？ 单 Agent 多工具是在一个上下文中顺序推理；多 Agent 可以真正并行、隔离上下文，避免工具过多导致的”工具幻觉”问题。当工具超过 15 个或任务可并行时，考虑拆成多 Agent。

Supervisor Agent 用什么模型？ Supervisor 负责路由决策，推理要准确，建议用能力强的模型（如 gpt-4o、claude-3.5-sonnet）；子 Agent 执行具体任务，可视任务复杂度选用轻量模型降低成本。

如何避免子 Agent 之间的输出格式不一致？ 在每个子 Agent 的 system prompt 中强制规定输出 JSON Schema，并在汇总节点做 schema 校验。失败时让子 Agent 重试，最多 2 次，仍失败则走降级。

要不要上 LangGraph / CrewAI 这类现成框架，还是自己手写编排？ 这个问题没有标准答案，看你的拓扑复杂度和团队对”黑盒”的容忍度。如果你的编排逻辑就是本文里的 Supervisor 或简单 DAG，几十行代码能写清楚，自己手写反而更可控——出问题的时候你能一行行调试，不用先去啃框架源码猜它内部到底怎么传的状态。但如果你的图变得很复杂，比如带条件分支、循环回退、需要持久化中断点（用户中途暂停、第二天继续跑），这些框架把状态机、检查点这类脏活都封装好了，自己重新造一遍性价比不高。判断标准很简单：先手写一版能跑通的最小实现，等真的遇到”状态管理写不动了”或者”分支逻辑绕不清楚了”这种具体的痛点，再评估要不要换框架，而不是一上来就为了用框架而用框架。

← 返回应用模式总览：从 Prompt 到 Agent | 应用模式专题

多 Agent 系统依赖多家模型供应商？力达云聚合 API 提供统一接口，自动在 GPT-4o / Claude / DeepSeek 间路由，子 Agent 按需切换模型无需改代码。