引言：被 demo 欺骗的三年
从 AutoGen、CrewAI 到 LangGraph，多智能体（Multi-Agent）框架在过去两年几乎成了"AI 应用成熟度"的代名词。
官方示例里，一个 Planner 拆任务、三个 Researcher 并行检索、一个 Writer 汇总、一个 Critic 挑刺，最后输出一份漂亮的报告——看起来比单次 LLM 调用聪明得多。
但把它搬到生产环境，多数团队会撞上同一堵墙：
- 成本与延迟线性放大。5 个 Agent 串行跑，Token 消耗和端到端延迟往往是单 Agent 的 3～10 倍。
- 错误级联传播。一个 Agent 输出偏了，下游全跟着偏，且越往后越难定位是谁的锅。
- 可靠性反而下降。假设单个 Agent 单步正确率 90%，三个环节串联就是0.93≈72.9%0.9^3 \approx 72.9\%0.93≈72.9%——多智能体天然是"多个单点故障的串联"。
这里有个被 demo 掩盖的真相：demo 里的多智能体之所以"看起来更强"，往往是因为它的任务本身很简单。演示用的任务通常有唯一、清晰、易于验证的答案（比如"总结这篇文章"），单 Agent 也能做对，多 Agent 只是把同一个正确答案用更贵的方式又算了一遍。而一旦任务变成长链条、多依赖、需要外部事实支撑的复杂工作，demo 的光环就会瞬间褪色——这正是生产环境和 notebook 的分水岭。
所以问题不该是"多智能体是不是更强"，而应该是：在什么条件下，多智能体的收益能覆盖它的开销？
我的结论很明确：多智能体的价值不来自"Agent 数量"，而来自三个可工程化的机制——任务分工、结果校验、失败重试。缺了这三样，多智能体只是一次昂贵的、分布式的幻觉。
本文会按"原理 → 分工 → 校验 → 实战 → 踩坑 → FAQ"的顺序，把这三个机制拆成可以抄进项目的代码。所有代码只依赖pydantic和标准库，不绑定任何框架，因为框架会变，机制不会。
一、核心原理：增益与损耗从哪里来
1.1 多智能体的真实增益
（1）上下文隔离。这是最被低估的收益。单 Agent 处理长任务时，检索结果、草稿、修改意见全堆在一个上下文里，很容易触发"lost in the middle"和 context rot（上下文腐烂）。所谓 “lost in the middle”，是指模型对上下文首尾的信息记得最牢，中间部分容易被忽略；当一份 8000 token 的草稿里夹着一句关键的约束时，模型很可能视而不见。多智能体让每个角色只看到自己需要的那一小块上下文，注意力不被稀释，输出分布更尖锐。
举个具体例子：单 Agent 场景下，你让它"先写初稿再自我批评再修改"，三件事挤在一个上下文里，模型在修改时往往会"护短"自己的初稿，或者干脆把批评和修改混在一起、越改越乱。而把 Writer 和 Critic 拆成两个独立上下文，Critic 面对的是一份"别人的稿子"，没有自我防御的心理负担，挑刺会更狠、更准。
（2）角色专业化。Prompt 越聚焦，模型行为的方差越小。"你是一个只做事实核查的证据审计员"比"你是全能助手"可控得多。从信息论的角度看，一个角色 Prompt 实际上给模型的输出分布加了一个强先验，把大量无关的生成路径直接砍掉了。角色越窄，模型的输出空间越小，你需要处理的长尾就越少。
（3）并行分治。相互独立的子任务可以 fan-out 并行执行，把延迟从 O(n) 压回 O(1)。这一点在 API 限流宽松、子任务彼此独立时收益巨大——原本 5 个串行子任务 50 秒，并行后可能只要 12 秒。
1.2 不可忽视的损耗
（1）错误传播的乘性衰减。如上文公式，串联结构下每多一环，整体成功率就乘一次。更糟糕的是，这个衰减不是线性的：当环节数从 3 增加到 6，0.96≈53%0.9^6 \approx 53\%0.96≈53%，已经跌破一半。也就是说，你越是"多"加 Agent，系统就越脆。这是很多人直觉相反的地方——大家觉得"多一层检查多一份保险"，但如果没有校验和重试，多出来的那一层本身就是新的故障源。
（2）协调开销。任务描述、上下文传递、结果合并，全是额外的 Token。一个 5 Agent 的流水线，可能 40% 的 Token 花在了"沟通"而不是"干活"上。你可以在日志里分别统计"业务 Token"和"编排 Token"，很多团队第一次看到这个比例都会心疼。
（3）责任真空。当输出错误时，是 Planner 拆错了？Researcher 检索错了？还是 Writer 归纳错了？没有结构化日志和校验点，这就是一笔糊涂账。责任真空带来的次生伤害是：你无法迭代。因为不知道哪一环是瓶颈，你只能靠猜去调 Prompt，最后变成玄学调参。
因此，多智能体的可靠性不能靠"让 Agent 更聪明"，只能靠外部机制把乘性衰减拉回来：把不可靠的串联，改造成"校验 + 重试"的闭环。
一个有用的心智模型是：把每个 Agent 当作一个"概率性的、会偶尔犯错的远程服务"。你不会指望一个会 10% 概率返回 500 的微服务靠"更努力"来解决问题，你会给它加超时、加重试、加熔断、加监控。对 Agent 也应该一样——不要幻想换更强的模型就能消灭错误，而要假设错误必然发生，然后设计容纳错误的系统。
二、任务分工：别让 LLM 干它干不好的事
任务分工有两种模式：
| 模式 | 说明 | 适用场景 |
|---|---|---|
| 静态分工 | 拓扑结构在代码里写死（DAG） | 流程稳定、可测试、要上生产 |
| 动态分工 | Planner Agent 运行时拆解 | 开放式探索、需求不确定 |
我的建议是：默认静态，动态只作补充。由 LLM 决定"用几个 Agent、谁依赖谁"，本质上是把编排逻辑交给了一个不可靠的组件。更好的做法是：拓扑由代码决定，LLM 只负责产出一个符合 Schema 的任务列表。
这里有一个容易被忽略的取舍：动态分工的"灵活性"是一把双刃剑。它让系统能处理没见过的任务形态，但也让测试变得几乎不可能——你无法为"任意拓扑"写单元测试，只能做端到端抽查，这在生产环境是灾难。静态分工牺牲了一部分灵活性，换来的是确定性、可测试性和可观测性，这笔交易在绝大多数业务场景里都是划算的。
而这一切的前提是接口契约先行——每个 Agent 的输入输出都必须是结构化的，而不是自由文本。
from__future__importannotationsfromtypingimportAny,LiteralfrompydanticimportBaseModel,Field,ValidationErrorclassSubTask(BaseModel):id:str=Field(pattern=r"^t\d+$")kind:Literal["research","write","review"]instruction:str=Field(min_length=10,max_length=500)depends_on:list[str]=Field(default_factory=list)classPlan(BaseModel):"""Planner 的唯一合法输出。schema 本身就是最好的 prompt。"""tasks:list[SubTask]=Field(min_length=1,max_length=8)deftopo_order(self)->list[SubTask]:ids={t.idfortinself.tasks}fortinself.tasks:ifnotset(t.depends_on)<=ids:raiseValueError(f"依赖不存在:{t.depends_on}not in{ids}")done,order=set(),[]remaining=list(self.tasks)whileremaining:ready=[tfortinremainingifset(t.depends_on)<=done]ifnotready:# 存在环raiseValueError("依赖图中存在环")fortinready:order.append(t);done.add(t.id);remaining.remove(t)returnorderclassAgentResult(BaseModel):task_id:strstatus:Literal["ok","failed"]payload:dict[str,Any]notes:str=""关键点：Plan.topo_order()抛出的异常是确定性校验——依赖成环、引用了不存在的 task_id，这些错误不需要任何 LLM 来判断，一段 20 行的代码就能兜住。凡是能用代码判定的，永远不要交给模型。
再补充几个分工阶段的实战建议：
- 限制任务数量上限。
max_length=8不是随口写的。任务越多，协调开销越大、错误面越广。经验值是 3～6 个，超过 8 个基本说明该重新审视任务设计本身了。 instruction要可执行、可验证。"研究一下 AI 市场"是坏指令；"列出 2023 年全球 AI 芯片市场的三家主要厂商及其市占率，并给出信息来源 URL"是好指令。好指令的一个重要特征是：它自带验收标准，后续的校验器能直接照着它写断言。- 让 id 稳定且可预测。用
t1、t2、t3而不是让模型自由发挥 UUID，这样日志、结果表、依赖关系都能对齐，排查问题时一眼就能定位。 - Schema 即 Prompt。把
Literal["research", "write", "review"]这类约束写进 Pydantic 模型，并通过 JSON Schema 注入到给模型的指令里，比在自然语言里反复叮嘱"只能选这三个之一"有效得多。模型对结构化约束的遵守率远高于对自然语言约定的遵守率。
三、结果校验：多智能体的安全气囊
校验必须分层，且层与层之间职责清晰：
- 语法层：能不能解析成 JSON？符不符合 Pydantic Schema？（确定性）
- 语义层：数值是否在范围内？引用的 URL 是否真的存在？代码能不能跑通测试？（确定性为主）
- 交叉层：由独立的 Verifier Agent 判断"论证是否自洽"“证据是否支撑结论”。（LLM 判断）
核心原则：能用代码校验的，绝不用 LLM 校验。LLM-as-a-judge 只应该用在真正主观、开放的部分。把客观事实交给模型判断，等于主动引入噪声。举个典型反例：判断"返回的 JSON 里 count 字段是否等于数组长度"，模型可能给你一个看似合理的"是"，但它其实没数——这段逻辑用一行len(payload["items"]) == payload["count"]就能确定性地完成。
另一个容易被忽略的点是校验器独立性：如果 Verifier 和被校验的 Agent 用同一个模型、同一套 Prompt 模板，它们会共享同样的偏差，产生大量"假阳性通过"。至少要做到：不同角色 Prompt、不同的温度参数，条件允许时换一个模型家族。这背后的道理很简单——同源误差无法自我纠错。一个模型很难发现自己的盲区，就像一个人很难通过"再想一遍"发现自己的认知偏见。换一个模型，哪怕它整体更弱，只要它的偏差方向不同，就能起到互补作用。
下面是一个带"错误回注"的重试执行器——这是让重试真正有效的关键：
importtimefromdataclassesimportdataclass,fieldfromtypingimportCallable,Generic,TypeVar T=TypeVar("T",bound=BaseModel)@dataclassclassRetryPolicy:max_attempts:int=3base_delay:float=0.5deadline_s:float=90.0_start:float=field(default_factory=time.monotonic)defdelay(self,attempt:int)->float:returnself.base_delay*(2**(attempt-1))# 指数退避defexhausted(self,attempt:int)->bool:returnattempt>=self.max_attemptsor\(time.monotonic()-self._start)>self.deadline_sdefrun_with_validation(task_prompt:str,schema:type[T],llm_call:Callable[[str],str],validator:Callable[[T],list[str]]|None=None,policy:RetryPolicy|None=None,)->T|None:"""调用 LLM → 解析 → 校验 → 失败则把错误回注重试。"""policy=policyorRetryPolicy()feedback:list[str]=[]attempt=0whilenotpolicy.exhausted(attempt):attempt+=1prompt=task_promptiffeedback:# 关键：只回注结构化错误摘要，而不是全量历史，避免上下文膨胀prompt+="\n\n上一轮输出被拒绝，原因如下，请针对性修正：\n- "+\"\n- ".join(feedback[-5:])raw=llm_call(prompt)try:obj=schema.model_validate_json(raw)exceptValidationErrorase:feedback.append(f"Schema 校验失败:{e.errors()[:3]}")time.sleep(policy.delay(attempt))continueerrors=validator(obj)ifvalidatorelse[]iferrors:feedback.append("业务校验失败: "+"; ".join(errors))time.sleep(policy.delay(attempt))continuereturnobjreturnNone# 明确失败，交给上层降级/熔断，而不是抛半成品三个设计细节值得强调：
- 失败必须回注具体原因。不加反馈的重试等于重掷骰子，同样的错误会以同样的概率再次出现。带上 Pydantic 的
errors()或业务断言信息，才算一次"有信息的重试"（思路类似 Reflexion）。 - 只回注摘要，不回注全量历史。把所有失败尝试的原始输出塞回 Prompt，会迅速膨胀上下文，反而降低成功率。保留最近 3～5 条结构化错误即可。
- 超时要封顶。
max_attempts和deadline_s双保险，防止在 CI 或线上无限烧钱。
还有一个常被忽略的细节：指数退避里应该加抖动（jitter）。上面的delay()是纯指数退避，若多个 Agent 同时失败，它们会在同一时刻一起重试，形成"重试风暴"，把下游 API 打出 429。生产环境里建议改成base_delay * (2 ** (attempt - 1)) * (0.5 + random.random())，用随机抖动把重试错开。
最后一句return None的语义很重要：它代表**“这一环明确、干净地失败了”**，而不是"勉强返回一个半成品"。返回半成品是最危险的做法，因为下游无法区分"这是好结果"还是"这是坏结果"，错误会静默地流向最终输出。宁可显式失败，也不要静默污染。
四、实战：一条可上线的"研究报告"流水线
把上面两块拼起来。场景：给定主题，生成一份带引用的事实性报告。
importloggingfromconcurrent.futuresimportThreadPoolExecutor log=logging.getLogger("orchestrator")classOrchestrator:def__init__(self,llm,tool_registry,max_workers:int=4):self.llm=llm self.tools=tool_registry self.pool=ThreadPoolExecutor(max_workers=max_workers)self.results:dict[str,AgentResult]={}# ---- 工具：把上游依赖的产物拼成本任务的上下文 ----def_context_for(self,task:SubTask)->str:parts=[]fordepintask.depends_on:r=self.results.get(dep)ifrandr.status=="ok":parts.append(f"### 来自{dep}的产物\n{r.payload}")return"\n\n".join(parts)ifpartselse"（无上游依赖）"# ---- 单个任务的执行单元 ----def_run_one(self,task:SubTask)->AgentResult:log.info("start task %s kind=%s",task.id,task.kind)prompt=(f"任务类型:{task.kind}\n"f"指令:{task.instruction}\n\n"f"可用上下文:\n{self._context_for(task)}")# 统一走"校验 + 重试"通道；schema 按任务类型选择schema=SCHEMA_BY_KIND[task.kind]validator=VALIDATOR_BY_KIND.get(task.kind)obj=run_with_validation(prompt,schema,llm_call=lambdap:self.llm(p),validator=validator,)ifobjisNone:log.warning("task %s failed after retries",task.id)returnAgentResult(task_id=task.id,status="failed",payload={},notes="retries exhausted")returnAgentResult(task_id=task.id,status="ok",payload=obj.model_dump())# ---- 编排主循环：按拓扑序推进，就绪的任务并发执行 ----defrun(self,plan:Plan)->dict[str,AgentResult]:order=plan.topo_order()# 确定性校验：成环/悬空依赖在此暴露done:set[str]=set()whilelen(done)<len(order):ready=[tfortinorderift.idnotindoneandset(t.depends_on)<=done]ifnotready:raiseRuntimeError("调度死锁：就绪集为空")futures={self.pool.submit(self._run_one,t):tfortinready}forfut,taskinfutures.items():res=fut.result()self.results[task.id]=res done.add(task.id)# 关键：上游失败时，不再让下游硬着头皮跑ifres.status=="failed":self._mark_downstream_failed(order,task.id)done.update(t.idfortinorderift.idinself.resultsandself.results[t.id].status=="failed")returnself.resultsdef_mark_downstream_failed(self,order,failed_id:str):"""把失败任务的所有下游标记为 skipped，避免污染扩散。"""fortinorder:iffailed_idint.depends_onandt.idnotinself.results:self.results[t.id]=AgentResult(task_id=t.id,status="failed",payload={},notes=f"上游{failed_id}失败，已短路")上面这段代码里有三个生产级的关键设计，值得逐一说明：
- 就绪集驱动，而不是固定串行。每一轮只挑"依赖全部完成"的任务并发执行，这样既保持了拓扑约束，又最大化了并行度。比"按顺序 for 循环"高效得多。
- 失败短路（fail-fast）。上游一失败，立刻把它所有下游标成
failed并注明原因，而不是让下游拿着空上下文继续跑。这直接砍掉了错误级联——原来会传播到全链路，现在只影响失败节点及其直接下游。 - 调度死锁检测。
就绪集为空这个判断是兜底保险。正常情况下topo_order()已经保证了有解，但如果运行时动态修改了依赖，这个断言能第一时间暴露问题。
调用它：
if__name__=="__main__":topic="2024 年开源大模型推理框架的现状"plan_prompt=("你是任务规划器。针对下面的主题，产出 3~5 个子任务，""只输出符合 Plan schema 的 JSON。\n主题: "+topic)plan_obj=run_with_validation(plan_prompt,Plan,llm_call=LLM)assertplan_objisnotNone,"规划阶段失败，直接降级到单 Agent"orch=Orchestrator(llm=LLM,tool_registry=TOOLS)results=orch.run(plan_obj)ok=sum(r.status=="ok"forrinresults.values())log.info("pipeline finished: %d/%d ok",ok,len(results))注意最外层还有一层降级：如果规划阶段本身就失败了，直接退回到"单 Agent + 单轮"的老路，而不是让整个请求挂掉。多智能体系统的第一设计原则应该是——它的失败模式必须比单 Agent 更优雅，而不是更惨烈。
五、踩坑与优化建议
这一节是"文档里不会写、但线上一定会遇到"的部分。
坑 1：上下文膨胀导致重试越重越差。
有人在重试时把每次的原始输出全塞回 Prompt，结果第 3 次尝试时上下文已经上万 token，模型反而更糊涂。务必只回注结构化的错误摘要。
坑 2：重试风暴。
所有 Agent 用同一条固定退避曲线，失败时会在同一时刻集体重试。加随机抖动（jitter），并给下游 API 设置并发上限。
坑 3：缺乏幂等。
如果 Agent 带副作用（写数据库、发请求），重试会导致重复写入。要么让操作幂等，要么用一个task_id + attempt作为幂等键。
坑 4：把温度调成 0 以为能得到确定性。
温度 0 只是让采样更集中，不是真正的确定性，尤其在并发和不同硬件下。永远不要假设 Agent 输出可复现，要靠校验而不是靠"它应该稳定"。
坑 5：只记最终结果，不记中间产物。
多智能体排障的核心是可观测性。每个 Agent 的输入 Prompt、原始输出、校验错误、耗时、Token 消耗都应落结构化日志。没有这些，责任真空会让调优变成玄学。
优化建议：把"业务 Token"和"编排 Token"分开计费；给每个任务设独立的超时；用缓存（相同 Prompt 命中直接复用）降低重复调用成本；在 CI 里用固定的假 LLM（返回预设 JSON）跑全链路，保证编排逻辑本身可测——编排正确性和模型能力要解耦测试。
六、FAQ
Q1：多智能体和"单 Agent + 多轮工具调用"到底差在哪？
本质区别在于上下文是否隔离。单 Agent + 多轮工具调用，所有中间结果都在同一个上下文里，长任务必然遭遇上下文腐烂。多智能体的核心收益就是让每个角色只看一小块上下文。但如果你发现各个子任务其实高度耦合、共享大量上下文，那多智能体就是过度设计。
Q2：到底用几个 Agent 合适？
先问"能不能用 2 个解决"。经验上，3～6 个是甜点区。超过 6 个，协调开销和错误面会迅速吃掉收益。记住那个0.9n0.9^n0.9n的公式——每加一环，都是在乘一个小于 1 的数。
Q3：Verifier 用同一个模型行不行？
能跑，但会浪费大量"假阳性通过"。至少换 Prompt、换温度；条件允许换模型家族。同源误差无法自我纠错。
Q4：重试为什么有时候反而更差？
三种可能：（a）回注的错误信息太长，撑爆上下文；（b）任务本身无解，重试只是徒劳烧钱；（c）退避不足，触发了限流。对策分别是：只回注摘要、设置明确的失败出路、加抖动。
Q5：怎么控制成本？
三招：静态拓扑（避免 LLM 决定编排）、结果缓存（相同输入复用）、分层校验（能用代码的绝不用 LLM）。再补一条：给每个任务设 Token 预算，超了就熔断。
Q6：什么时候不该用多智能体？
任务短、答案唯一、无需外部事实核查时。这类任务单 Agent 又快又稳，多智能体只会把正确答案算得更贵。判断标准很简单：先上单 Agent，把它的失败模式记下来；如果失败集中在"上下文太满"和"角色混杂"，那才轮到多智能体上场。
七、结论：把"更强"变成"更可靠"
回到标题的问题：多智能体协作真的更强吗？
答案是：它不会自动更强，但它可以被工程化为更可靠——前提是你补齐三个机制。
- 任务分工：拓扑由代码决定，LLM 只产出符合 Schema 的任务列表。凡是能用代码判定的，绝不交给模型。
- 结果校验：分层校验，能用代码的绝不用 LLM；Verifier 必须独立，避免同源误差。
- 失败重试：带回注、带抖动、带超时封顶，失败要显式（
return None）而不是静默污染。
多智能体的价值不在"Agent 数量"，而在于它能不能把不可靠的串联，改造成校验 + 重试的闭环。做对了，它是安全气囊；做错了，它只是一次昂贵的、分布式的幻觉。
最后给一条务实的落地路径：**先写单 Agent 版本，记录它的失败模式；
更多硬核网安与AI工具包，请扫码获取完整源码！
只有当失败集中在"上下文超载"和"角色混淆"这两类问题上，才引入多智能体，并且先只加一个 Verifier**。用最小改动验证收益，再考虑扩拓扑。技术选型里，克制往往比激进更值钱。
