多智能体协作真的更强吗？用 Python 实现任务分工、结果校验与失败重试机制

引言被 demo 欺骗的三年从 AutoGen、CrewAI 到 LangGraph多智能体Multi-Agent框架在过去两年几乎成了AI 应用成熟度的代名词。官方示例里一个 Planner 拆任务、三个 Researcher 并行检索、一个 Writer 汇总、一个 Critic 挑刺最后输出一份漂亮的报告——看起来比单次 LLM 调用聪明得多。但把它搬到生产环境多数团队会撞上同一堵墙成本与延迟线性放大。5 个 Agent 串行跑Token 消耗和端到端延迟往往是单 Agent 的 310 倍。错误级联传播。一个 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 增加到 60.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:strField(patternr^t\d$)kind:Literal[research,write,review]instruction:strField(min_length10,max_length500)depends_on:list[str]Field(default_factorylist)classPlan(BaseModel):Planner 的唯一合法输出。schema 本身就是最好的 prompt。tasks:list[SubTask]Field(min_length1,max_length8)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,orderset(),[]remaininglist(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_length8不是随口写的。任务越多协调开销越大、错误面越广。经验值是 36 个超过 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 TTypeVar(T,boundBaseModel)dataclassclassRetryPolicy:max_attempts:int3base_delay:float0.5deadline_s:float90.0_start:floatfield(default_factorytime.monotonic)defdelay(self,attempt:int)-float:returnself.base_delay*(2**(attempt-1))# 指数退避defexhausted(self,attempt:int)-bool:returnattemptself.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]]|NoneNone,policy:RetryPolicy|NoneNone,)-T|None:调用 LLM → 解析 → 校验 → 失败则把错误回注重试。policypolicyorRetryPolicy()feedback:list[str][]attempt0whilenotpolicy.exhausted(attempt):attempt1prompttask_promptiffeedback:# 关键只回注结构化错误摘要而不是全量历史避免上下文膨胀prompt\n\n上一轮输出被拒绝原因如下请针对性修正\n- \\n- .join(feedback[-5:])rawllm_call(prompt)try:objschema.model_validate_json(raw)exceptValidationErrorase:feedback.append(fSchema 校验失败:{e.errors()[:3]})time.sleep(policy.delay(attempt))continueerrorsvalidator(obj)ifvalidatorelse[]iferrors:feedback.append(业务校验失败: ; .join(errors))time.sleep(policy.delay(attempt))continuereturnobjreturnNone# 明确失败交给上层降级/熔断而不是抛半成品三个设计细节值得强调失败必须回注具体原因。不加反馈的重试等于重掷骰子同样的错误会以同样的概率再次出现。带上 Pydantic 的errors()或业务断言信息才算一次有信息的重试思路类似 Reflexion。只回注摘要不回注全量历史。把所有失败尝试的原始输出塞回 Prompt会迅速膨胀上下文反而降低成功率。保留最近 35 条结构化错误即可。超时要封顶。max_attempts和deadline_s双保险防止在 CI 或线上无限烧钱。还有一个常被忽略的细节指数退避里应该加抖动jitter。上面的delay()是纯指数退避若多个 Agent 同时失败它们会在同一时刻一起重试形成重试风暴把下游 API 打出 429。生产环境里建议改成base_delay * (2 ** (attempt - 1)) * (0.5 random.random())用随机抖动把重试错开。最后一句return None的语义很重要它代表**“这一环明确、干净地失败了”**而不是勉强返回一个半成品。返回半成品是最危险的做法因为下游无法区分这是好结果还是这是坏结果错误会静默地流向最终输出。宁可显式失败也不要静默污染。四、实战一条可上线的研究报告流水线把上面两块拼起来。场景给定主题生成一份带引用的事实性报告。importloggingfromconcurrent.futuresimportThreadPoolExecutor loglogging.getLogger(orchestrator)classOrchestrator:def__init__(self,llm,tool_registry,max_workers:int4):self.llmllm self.toolstool_registry self.poolThreadPoolExecutor(max_workersmax_workers)self.results:dict[str,AgentResult]{}# ---- 工具把上游依赖的产物拼成本任务的上下文 ----def_context_for(self,task:SubTask)-str:parts[]fordepintask.depends_on:rself.results.get(dep)ifrandr.statusok: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}\nf指令:{task.instruction}\n\nf可用上下文:\n{self._context_for(task)})# 统一走校验 重试通道schema 按任务类型选择schemaSCHEMA_BY_KIND[task.kind]validatorVALIDATOR_BY_KIND.get(task.kind)objrun_with_validation(prompt,schema,llm_calllambdap:self.llm(p),validatorvalidator,)ifobjisNone:log.warning(task %s failed after retries,task.id)returnAgentResult(task_idtask.id,statusfailed,payload{},notesretries exhausted)returnAgentResult(task_idtask.id,statusok,payloadobj.model_dump())# ---- 编排主循环按拓扑序推进就绪的任务并发执行 ----defrun(self,plan:Plan)-dict[str,AgentResult]:orderplan.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():resfut.result()self.results[task.id]res done.add(task.id)# 关键上游失败时不再让下游硬着头皮跑ifres.statusfailed:self._mark_downstream_failed(order,task.id)done.update(t.idfortinorderift.idinself.resultsandself.results[t.id].statusfailed)returnself.resultsdef_mark_downstream_failed(self,order,failed_id:str):把失败任务的所有下游标记为 skipped避免污染扩散。fortinorder:iffailed_idint.depends_onandt.idnotinself.results:self.results[t.id]AgentResult(task_idt.id,statusfailed,payload{},notesf上游{failed_id}失败已短路)上面这段代码里有三个生产级的关键设计值得逐一说明就绪集驱动而不是固定串行。每一轮只挑依赖全部完成的任务并发执行这样既保持了拓扑约束又最大化了并行度。比按顺序 for 循环高效得多。失败短路fail-fast。上游一失败立刻把它所有下游标成failed并注明原因而不是让下游拿着空上下文继续跑。这直接砍掉了错误级联——原来会传播到全链路现在只影响失败节点及其直接下游。调度死锁检测。就绪集为空这个判断是兜底保险。正常情况下topo_order()已经保证了有解但如果运行时动态修改了依赖这个断言能第一时间暴露问题。调用它if__name____main__:topic2024 年开源大模型推理框架的现状plan_prompt(你是任务规划器。针对下面的主题产出 3~5 个子任务只输出符合 Plan schema 的 JSON。\n主题: topic)plan_objrun_with_validation(plan_prompt,Plan,llm_callLLM)assertplan_objisnotNone,规划阶段失败直接降级到单 AgentorchOrchestrator(llmLLM,tool_registryTOOLS)resultsorch.run(plan_obj)oksum(r.statusokforrinresults.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跑全链路保证编排逻辑本身可测——编排正确性和模型能力要解耦测试。六、FAQQ1多智能体和单 Agent 多轮工具调用到底差在哪本质区别在于上下文是否隔离。单 Agent 多轮工具调用所有中间结果都在同一个上下文里长任务必然遭遇上下文腐烂。多智能体的核心收益就是让每个角色只看一小块上下文。但如果你发现各个子任务其实高度耦合、共享大量上下文那多智能体就是过度设计。Q2到底用几个 Agent 合适先问能不能用 2 个解决。经验上36 个是甜点区。超过 6 个协调开销和错误面会迅速吃掉收益。记住那个0.9n0.9^n0.9n的公式——每加一环都是在乘一个小于 1 的数。Q3Verifier 用同一个模型行不行能跑但会浪费大量假阳性通过。至少换 Prompt、换温度条件允许换模型家族。同源误差无法自我纠错。Q4重试为什么有时候反而更差三种可能a回注的错误信息太长撑爆上下文b任务本身无解重试只是徒劳烧钱c退避不足触发了限流。对策分别是只回注摘要、设置明确的失败出路、加抖动。Q5怎么控制成本三招静态拓扑避免 LLM 决定编排、结果缓存相同输入复用、分层校验能用代码的绝不用 LLM。再补一条给每个任务设 Token 预算超了就熔断。Q6什么时候不该用多智能体任务短、答案唯一、无需外部事实核查时。这类任务单 Agent 又快又稳多智能体只会把正确答案算得更贵。判断标准很简单先上单 Agent把它的失败模式记下来如果失败集中在上下文太满和角色混杂那才轮到多智能体上场。七、结论把更强变成更可靠回到标题的问题多智能体协作真的更强吗答案是它不会自动更强但它可以被工程化为更可靠——前提是你补齐三个机制。任务分工拓扑由代码决定LLM 只产出符合 Schema 的任务列表。凡是能用代码判定的绝不交给模型。结果校验分层校验能用代码的绝不用 LLMVerifier 必须独立避免同源误差。失败重试带回注、带抖动、带超时封顶失败要显式return None而不是静默污染。多智能体的价值不在Agent 数量而在于它能不能把不可靠的串联改造成校验 重试的闭环。做对了它是安全气囊做错了它只是一次昂贵的、分布式的幻觉。最后给一条务实的落地路径**先写单 Agent 版本记录它的失败模式更多硬核网安与AI工具包请扫码获取完整源码只有当失败集中在上下文超载和角色混淆这两类问题上才引入多智能体并且先只加一个 Verifier**。用最小改动验证收益再考虑扩拓扑。技术选型里克制往往比激进更值钱。