授权与合规声明
本文为技术实操分享,所有示例均在自建隔离环境中验证或为可复现的演示代码;文中版本、命令与数据均标注来源,无法核验的已标「待验证」,不含任何攻击步骤、利用载荷与绕过手法。转载请注明出处。
一、为什么单个 Agent 会撞到天花板
很多人第一次把大模型做成 Agent,是一个「万能 prompt + 一堆工具」的单体:一个系统提示词描述全部目标,一个while循环反复调用模型直到任务结束。Demo 阶段很爽,一到复杂任务就露馅。
1.1 单体 Agent 的三道墙
第一道是上下文窗口墙。所有中间思考、工具返回、历史轨迹都堆在同一段上下文里,任务一长就被挤爆,要么截断丢信息,要么摘要丢细节。
第二道是职责混杂墙。一个 prompt 既要做规划,又要做执行,还要做校验,模型在三种角色间反复横跳,输出稳定性明显下降。
第三道是错误放大墙。中间某一步工具调用出错,单体没有隔离边界,错误顺着轨迹一路传染,最后整轮任务失败,且你很难定位是哪一步。
1.2 协作的本质是「分权」
多 Agent 协作不是「多叫几个模型」,而是把「规划」和「执行」拆开:一个负责拆解与调度,多个负责把单一子任务做扎实。这正好对应工业界常用的 Orchestrator-Worker(编排者-工作者)模式。
二、Orchestrator-Worker 是什么
2.1 角色分工
- Orchestrator(编排者):只做三件事——把用户目标拆成可并行的子任务、把子任务派发给合适的 Worker、汇总 Worker 的结果。它不直接干脏活。
- Worker(工作者):每个 Worker 只认一个明确的子任务和一个窄职责(例如「查数据库」「写代码」「做校验」),上下文短、目标单一,反而更稳。
2.2 与单体 Agent 的判据差异
判据很简单:当任务能被拆成「彼此相对独立、可并行、结果可汇总」的子任务时,协作模式胜出;当任务强依赖、必须一步步串行推理时,单体反而更可控。不要为了用协作而用协作。
三、一个最小可跑的编排骨架
下面是一段示意性的 Python 骨架,核心是「拆解 → 派发 → 汇总」,不依赖任何具体框架。
⚠️ 代码待验证
fromdataclassesimportdataclass,fieldimportasyncio@dataclassclassSubTask:id:strinstruction:strresult:str=""status:str="pending"# pending | done | failedasyncdefworker_execute(task:SubTask,llm_call)->SubTask:"""单个 Worker:只完成自己的子任务。"""try:task.result=awaitllm_call(task.instruction)task.status="done"exceptExceptionase:task.status="failed"task.result=f"ERROR:{e}"returntaskasyncdeforchestrate(goal:str,llm_call)->str:"""Orchestrator:拆任务、并发派发、汇总。"""plan=awaitllm_call(f"把下面目标拆成 3 个独立子任务,每句一行:{goal}")subtasks=[SubTask(id=f"t{i}",instruction=line.strip())fori,lineinenumerate(plan.splitlines())ifline.strip()]# 并发派发,单个失败不影响其他results=awaitasyncio.gather(*(worker_execute(t,llm_call)fortinsubtasks))ok=[rforrinresultsifr.status=="done"]failed=[rforrinresultsifr.status=="failed"]summary_prompt="汇总以下子任务结果,给出最终答复:\n"+"\n".join(r.resultforrinok)final=awaitllm_call(summary_prompt)iffailed:final+=f"\n(注:{len(failed)}个子任务失败,已隔离,不影响其余结果)"returnfinal3.1 任务拆解的写法
拆解提示词要输出「结构化、可独立验证」的子任务,而不是一段话。建议让 Orchestrator 输出 JSON 列表,每个元素带id与instruction,方便后续追踪与重试。
3.2 Worker 执行与回传
Worker 拿到的是「单一指令 + 所需入参」,返回「结果文本 + 状态」。注意它不持有全局目标,这样上下文最短,也最容易做单测。
四、结果汇聚与冲突消解
4.1 汇总策略
汇总不是简单拼接。推荐两种:
- 分治汇总:先并行得到子结果,再交给一个汇总 Worker 做一致性对齐,适合子任务答案可能冲突的场景。
- Reduce 式汇总:把子结果分批两两合并,最后得到单一结论,适合结果量大的场景。
4.2 失败隔离
单个 Worker 失败必须「就地标记、不抛出」,否则asyncio.gather任一异常会拖垮整体。上面的status字段就是为了这个——失败可观测、可重试、可降级。
五、生产环境必须补的三件事
5.1 超时与重试
每个 Worker 调用都要有独立超时和有限重试,且重试要带退避。Orchestrator 本身也要有总体超时,避免一个卡死的子任务拖垮整轮。
5.2 成本控制
多 Agent 最大的隐性成本是「模型调用次数乘子」。给每个 Worker 指定合适的模型档位(简单子任务用小模型),并在 Orchestrator 层限制最大子任务数与递归深度,防止任务爆炸。
5.3 可观测
每个子任务的入参、出参、耗时、token 消耗都要落日志。没有这些,线上出问题你连「是哪个 Worker 飘了」都看不出来。
配套资料:这一章讲的超时、重试与成本控制,仓库里整理了一份可直接套用的多 Agent 韧性设计清单,扫码即可获取:
六、什么时候不该用多 Agent
6.1 判据清单
出现以下任一条,先别上协作:
- 任务本身强串行、一步错步步错;
- 子任务之间频繁互相依赖,拆开反而要不断回传上下文;
- 体量小、调用频次低,单体 Agent 已经够稳。
6.2 常见误用
把「多 Agent」当银弹,给每个环节都挂一个模型,结果调用次数翻了好几倍、稳定性却没提升。记住:协作解决的是「复杂度的拆分」,不是「能力的叠加」。
七、从 Demo 到上线的检查单
7.1 上线前自查
- 每个 Worker 是否有独立的超时与重试?
- 失败是否被隔离、可观测、可重试?
- 最大子任务数、递归深度是否设了上限?
- 汇总步骤能否处理「部分失败」?
7.2 复盘指标
盯三个数:整体任务成功率、平均子任务耗时、单轮平均模型调用次数。前两个要涨,最后一个要压住,三者一起看才有意义。
配套资料:文末这份资料包里有一份「多 Agent 上线检查单」的对照表,和上面七章一一对应,扫码即可获取:
附表 A:本文引用事实与出处对照表
| 事实/结论 | 出处 | 本文位置 |
|---|---|---|
| Orchestrator-Worker 是多 Agent 系统的常见编排模式 | 通用工程实践(待验证具体文档出处) | 第二章 |
| 单体 Agent 在长任务下面临上下文、职责、错误放大三类问题 | 本文基于工程经验归纳 | 第一章 |
子任务失败需隔离,避免gather异常拖垮整体 | Python asyncio 通用实践 | 第三、四章 |
| 多 Agent 会放大模型调用次数(成本乘子) | 本文基于工程经验归纳 | 第五章 |
附表 B:术语速查表
| 术语 | 含义 |
|---|---|
| Orchestrator | 编排者,负责拆解目标、派发与汇总 |
| Worker | 工作者,只完成单一子任务的窄职责单元 |
| 失败隔离 | 单个子任务失败不影响整体的设计 |
| 分治汇总 | 先得子结果再做一致性对齐的汇总方式 |
写在最后:这篇用到的资料
写这篇文章时,我把多 Agent 协作的骨架、韧性设计和上线检查单整理成了可对照的清单,顺手也整理了几份配套的东西:
- 多 Agent 编排骨架示例:上面第三章代码的完整可运行版
- 上线检查单对照表:七章要点逐条可勾选
- 成本控制配置模板:子任务上限与模型档位设置示例
资料是我自己整理的,放在下面这个码上,扫码即可获取:
资料较多,建议先看「全套 AGI 大模型学习路线」,再挑一个实战项目跟练。