1. 为什么企业级 MultiAgent 必须走“先规划、后执行”这条路1.1 单 Agent 处理复杂业务时的三个瓶颈先说结论企业级场景里把一个复杂的业务流程全部塞给单个 Agent让它自己“想一步、做一步”大概率会在上线两周内翻车。这个结论不是拍脑袋得出来的我们团队最早落地 MultiAgent 时就是这么干的然后被线上问题狠狠教育了一轮。当时我们接了一个典型的“工单自动处理”业务用户提交一个售后请求系统需要先理解用户诉求再查订单、查库存、算价格、生成处理方案、调用外部审批接口、最后回执给用户。逻辑链不算特别长但涉及多个内部系统和外部依赖。最初版本就是单 Agent 十几个 function calling 工具结果暴露出三个非常典型的瓶颈。第一上下文摊薄。为了让一个 Agent 能应对所有情况Prompt 里塞满了业务规则、工具说明、输出格式约束、安全红线。表面上很全面实际上模型根本“照顾”不过来。用户消息稍微长一点Agent 就开始抓不住重点该走的流程不走不该调的接口乱调。打个比方让一个新员工同时背十本操作手册再去处理具体工单他必然会漏步骤。第二工具发散。单 Agent 挂了十几个工具模型在每次推理时都有概率选错工具或者在同一个意图上反复横跳。我们的监控日志里出现过大量“工具调用-报错-换一个工具-再报错-又换回来”的循环一次工单处理能烧掉十几轮模型调用。Token 成本飙升的同时用户体验完全不可控。第三也是企业最不能接受的不可复现、不可审计。单 Agent 的自由推理路径每次都不一样同一个用户同类问题今天走 A 流程、明天走 B 流程。业务方来问“这个工单为什么这样处理了中间经历了什么”我们根本给不出一个可靠的过程留痕。这在对外服务场景里几乎等于事故。所以我们的第一个判断是企业级 MultiAgent 不能把“智能”押注在模型随机自由发挥上而要把“智能”收敛到一条可控的执行链路上。这也是我们最终切换到 Plan 模式的最根本动因。1.2 Plan 模式 vs ReAct 模式企业场景的关键差异接触过 Agent 架构的同学应该都知道业界主流的 Agent 工作模式大致分两类一类是 ReAct 模式模型边推理边行动每拿到一个观察结果就决定下一步动作另一类是 Plan 模式模型先基于用户诉求生成一份完整的任务计划再按计划逐步执行。这两种模式在小规模 Demo 里看不出太大差距但在企业生产环境里差别是天壤之别。我整理了一个对比表基本能说明问题的核心。维度ReAct 模式Plan 模式决策时机每步实时决策走一步看一步先全局规划再按步骤执行可控性弱中间步骤不可预判强计划先暴露可人工审核可审计性低执行路径每次不同高任务清单即审计依据出错成本中后段出错可能导致前功尽弃规划错误可在执行前拦截Token 消耗高反复推理和工具试探相对低规划一次后按图索骥适合场景探索型、开放型任务流程型、有明确步骤的任务我们切到 Plan 模式之后最直观的感受就是任务从“黑盒推理”变成了“白盒调度”。用户提一个诉求系统先产出一份类似“任务书”的结构化清单上面写着要做哪几件事、每件事谁来做、依赖关系是什么。工程团队可以在这份任务书出来之后进行校验不合理的地方直接打回重排而不是等 Agent 跑完五六个步骤才发现方向错了。1.3 什么样的业务适合切到 Plan 模式这里要泼一盆冷水Plan 模式不是银弹它不适合所有场景。按照我们内部的经验适合切 Plan 模式的任务有这么几个特征步骤多、有依赖、结果可验证、过程需要留痕。典型的像订单处理、库存调度、供应链审批、自动化测试执行、多系统数据核对。这些任务天然是“流程”形态Plan 模式只是把流程从人肉写死变成了模型动态生成。反之纯聊天、头脑风暴、开放创作这类任务不适合 Plan 模式。你让模型先写一份“聊天计划”再开始聊纯属脱裤子放屁平白增加一次模型调用和延迟。还有一个经验是渐进式切换。不要一上来就把所有流量切到 Plan 模式而是先挑一个流程最固定、出问题影响面最小的业务做试点跑通之后再逐步扩大范围。我们当时选的就是“售后工单自动处理”这个场景因为它链路清晰、边界明确、业务方配合度高非常适合作为 MultiAgent 架构的试验田。2. Plan 模式的核心拆解任务规划器与执行器分离2.1 任务规划器的输出结构化任务清单切换到 Plan 模式之后整个系统的第一个核心组件就是“任务规划器”。它的职责很简单把用户输入转换为一份结构化任务清单。但这份清单的格式设计直接决定了后面调度和执行的质量。我们最终采用了类似 DAG 的任务描述结构。每一份 Plan 都是一个 JSON 数组数组中每个元素是一个原子任务包含任务 ID、名称、类型、依赖关系、输入参数声明、期望输出协议、指定执行的子 Agent 类型等字段。下面是一个简化示例{ plan_id: plan_20250410_001, tasks: [ { task_id: t1, name: 解析用户诉求, type: intent_parse, dependencies: [], input: {raw_message: $user_input}, output_schema: {intent: string, entities: object}, assigned_agent: semantic_agent }, { task_id: t2, name: 查询订单信息, type: order_query, dependencies: [t1], input: {order_id: $t1.entities.order_id}, output_schema: {order_status: string, amount: number, sku_list: array}, assigned_agent: order_agent }, { task_id: t3, name: 生成退款方案, type: refund_plan, dependencies: [t1, t2], input: {intent: $t1.intent, order_info: $t2}, output_schema: {refund_amount: number, reason: string, risk_level: string}, assigned_agent: policy_agent } ] }这里有两个细节值得展开说。一个是依赖声明。每个任务都必须声明它依赖哪些前置任务调度器据此构建执行拓扑。这比简单的“按数组顺序执行”靠谱得多因为模型生成的先后顺序不一定是最优执行顺序有了依赖声明调度器可以并行执行互不依赖的任务缩短整体耗时。另一个是输出协议。每个任务都预先声明了期望的输出结构子 Agent 执行完必须按这个结构回传。这个设计解决了多 Agent 协作里最常见的数据对接问题—子 Agent 自由发挥的输出格式五花八门后端根本没法解析。提前定好协议等于给每个 Agent 套了一个规范边界。2.2 从规划到执行的调度模型Plan 生成之后下一个环节是调度执行。这块企业落地时容易被低估很多人以为把任务清单扔给 Agent 一个个跑就行实际上这里藏着大量细节。我们的调度器是一个独立的 Python 服务核心逻辑可以概括为三步构建依赖图、按拓扑序分批下发、汇总结果并校验。from collections import deque from typing import Dict, List, Set def schedule_tasks(plan: dict): tasks {t[task_id]: t for t in plan[tasks]} # 构建依赖关系 dependency_graph {tid: set(t[dependencies]) for tid, t in tasks.items()} dependents_map {} for tid, deps in dependency_graph.items(): for dep in deps: dependents_map.setdefault(dep, set()).add(tid) # 拓扑排序用入度 0 的节点作为可执行任务 in_degree {tid: len(deps) for tid, deps in dependency_graph.items()} ready_queue deque([tid for tid, deg in in_degree.items() if deg 0]) executed: Set[str] set() results: Dict[str, dict] {} while ready_queue: # 按批次执行所有当前可执行任务并行下发 batch list(ready_queue) ready_queue.clear() batch_results run_agent_batch(batch, tasks, results) for tid, res in batch_results.items(): executed.add(tid) results[tid] res # 解锁下游任务 for dependent in dependents_map.get(tid, []): in_degree[dependent] - 1 if in_degree[dependent] 0: ready_queue.append(dependent) # 如果有任务没被执行说明依赖成环或出现孤立任务 if len(executed) ! len(tasks): raise RuntimeError(fplan execution failed, executed{len(executed)} total{len(tasks)}) return results这段代码不是完整生产实现但已经能说明调度器的几个关键设计点批量并行、依赖解锁、成环检测。生产环境里我们还会在这层做更多事情比如 batch 大小限制、子 Agent 并发数控制、结果协议校验等等后面章节会展开。2.3 规划失败时的降级策略说完调度再说规划环节本身的一个大坑模型生成的 Plan 不一定是合法可执行的。它可能把不该拆的任务拆了可能遗漏必要步骤也可能构造出循环依赖关系。如果不做校验直接进入调度线上事故跑不了。我们当时设计了三层降级策略从轻到重依次是重新规划 - 局部修正 - 人工兜底。第一层规划器每次生成的 Plan 先过一个规则校验器。校验器主要查这几项依赖关系是否成环、任务类型是否在已注册的能力清单里、每个任务的输入参数引用是否指向已存在的前置输出、输出协议是否完整。校验不通过就要求规划器带着错误信息重新生成最多重试两次。第二层如果重试两次之后仍然不通过调度器会尝试局部修正。比如把依赖成环的两三个任务合并成一个任务或者把缺失的参数引用标记为“执行时再解析”。这个修正逻辑看似简单实际需要非常了解业务我们最初是用规则硬写的后来改成由另一个“修正 Agent”来做效果反而更差。原因是修正 Agent 没有足够的上下文判断业务合理性最后还是回归到规则为主。第三层如果前两层都失败系统会把 Plan 丢到一个人工待办队列由业务运营人员在后台查看、手动调整、确认执行。这层兜底虽然看起来“笨”但它是企业系统安全感的来源—模型可以出错但不能让用户承担模型出错的全部后果。3. 主子 Agent 协作用分层结构换可控性3.1 为什么不用“全对等”多智能体规划器解决了“做什么”的问题接下来就是“谁来做”。早期我们构想过一种“全对等”的 MultiAgent 协作模式也就是让多个 Agent 以平级身份聚在一起通过互相发消息的方式完成任务。但真正设计下去之后发现在企业生产环境里这条路基本走不通。对等模式最大的问题是权责边界模糊。几个能力相近的 Agent 同时面对一个任务时没有天然的“负责人”概念谁牵头、谁配合、分歧了怎么裁决全靠模型临场发挥。我们做过一次模拟测试三个对等 Agent 协同处理一个稍复杂的需求最终因为两个 Agent 对某个中间结果的理解不一致互相反复发消息确认来回七八轮才达成勉强共识延迟和 Token 开销都爆炸了。另一个问题是消息风暴。对等模式下每个 Agent 都可以向其他所有 Agent 发起通信通信复杂度接近 O(n²)。Agent 数量从三个增加到五个之后系统整体性能急剧下降日志里充斥着大量“无意义确认”消息。所以最终我们选定了“主子 Agent”的分层协作结构。这个结构可以用一个特别朴素的类比来理解主 Agent 是项目经理子 Agent 是不同工种的工程师。项目经理不亲自干活只负责拆解需求、派发任务、验收结果、协调资源工程师只负责自己专业领域内的执行不跨工种干涉别人。这种组织方式在人类社会里被验证了上百年拿到 Agent 系统里同样适用。3.2 主 Agent 的职责边界主 Agent 在“主子协作”里是一个特殊的存在它的特殊性在于能力不体现在具体执行上而体现在编排和决策上。我们给主 Agent 划定了三个核心职责域任务拆解、任务派发、结果汇总校验。任务拆解就是上一章的 Plan 生成任务派发是根据任务类型找到合适的子 Agent构造调用参数并下发结果汇总校验是收集所有子 Agent 的执行结果按业务规则做最终汇总和决策。同时我们给主 Agent 划了一条红线不允许直接调用业务工具。它只能跟子 Agent 通信不能绕过子 Agent 去查询订单、不能代替子 Agent 去写数据库。这条红线一开始我们觉得有些“死板”后来发现它是整个系统安全性的基石。因为主 Agent 一旦能调用具体工具它的 Prompt 就不得不包含大量工具权限信息被“投毒”或误调用的风险会指数级上升。把工具权限全部下沉到子 Agent主 Agent 保持相对“纯净”攻击面就小了很多。3.3 子 Agent 的注册、发现与调用协议子 Agent 的主业是具体领域执行为了被主 Agent 正确调度每个子 Agent 在上线前都要做一次“能力注册”。注册表里记录了 Agent 的名称、描述、支持的任务类型、输入输出协议、超时阈值、并发上限、模型规格等等。{ agent_name: order_agent, description: 查询订单状态、订单商品明细、订单金额信息不支持改单和售后操作, supported_tasks: [order_query, order_detail], input_schema: {order_id: string, user_id: string}, output_schema: {order_status: string, amount: number, sku_list: array}, timeout_ms: 15000, max_concurrency: 10, model: qwen-plus }主 Agent 在生成 Plan 时会结合“任务类型”和注册表里“supported_tasks”的匹配情况来指派执行者。这一步我们用的是规则匹配 向量召回混合的方式优先做精确匹配匹配不到再用语义检索找最接近的 Agent如果还是没有就认为规划失败走降级策略。子 Agent 之间的调用消息也统一走一套标准信封核心字段包括 task_id、source_agent、target_agent、payload、callback_url。刚开始我们用的是 HTTP 同步调用后来发现同步调用在高并发下很容易互相阻塞就改成了基于消息队列的异步模式。主 Agent 把任务投递到队列子 Agent 消费执行完再回传结果主 Agent 侧通过一个 pending 表跟踪所有未完成的任务。这个改动带来的吞吐量提升非常明显值得每个要上 MultiAgent 的团队提前考虑。4. 状态机驱动的任务生命周期管理4.1 任务状态机与持久化MultiAgent 系统运行起来之后最容易被忽视但最致命的是“任务的中间状态管理”。一个 Plan 被拆成七八个任务每个任务又有自己的生命周期如果中间任何一个环节宕机或超时整个协调就乱了。我们的做法是给每个任务定义一套显式状态机并落库持久化。任务状态我们定义了六种PENDING等待执行、RUNNING执行中、SUCCEEDED执行成功、FAILED执行失败、CANCELLED被取消、SKIPPED依赖失败跳过执行。from enum import Enum, auto class TaskStatus(Enum): PENDING auto() RUNNING auto() SUCCEEDED auto() FAILED auto() CANCELLED auto() SKIPPED auto() # 允许的状态迁移表 ALLOWED_TRANSITIONS { TaskStatus.PENDING: {TaskStatus.RUNNING, TaskStatus.CANCELLED}, TaskStatus.RUNNING: {TaskStatus.SUCCEEDED, TaskStatus.FAILED, TaskStatus.CANCELLED}, TaskStatus.SUCCEEDED: {}, TaskStatus.FAILED: {TaskStatus.RUNNING}, # 允许重试 TaskStatus.CANCELLED: {}, TaskStatus.SKIPPED: {}, } def transition(current: TaskStatus, target: TaskStatus) - TaskStatus: if target not in ALLOWED_TRANSITIONS[current]: raise IllegalStateTransitionError(current, target) return target这套状态机的好处是开发、测试、运维所有角色对“任务现在处于什么阶段”有一个统一的认知。出现问题时只要查某个 task_id 的状态历史就能复现整个生命周期。它把 MultiAgent 这种看起来“黑盒”的智能系统拉回到了普通分布式系统可以管理的维度。状态持久化我们直接用了关系型数据库每个任务一行记录字段包括 plan_id、task_id、status、current_retry_count、max_retry_count、input_snapshot、output_snapshot、error_message、timestamps。没有引入额外组件运维成本低排查问题也方便。4.2 上下文裁剪子 Agent 只回传结构化结果不做全量对话回传这块是我们节省 Token 成本最有效的一个设计也是我认为最有价值的一条经验。最早版本里子 Agent 执行完会把完整的对话历史回传给主 Agent导致主 Agent 的上下文迅速膨胀越到后面推理质量越差Token 费用也水涨船高。后来我们做了个“冷酷”的改造子 Agent 只允许回传 output_schema 里声明的结构化结果对话过程全部丢弃只在日志里留一份压缩摘要。主 Agent 拿到的是干净的字段值而不是一大坨对话记录。这个改动上线后单任务的 Token 消耗下降约四成主 Agent 的上下文长度稳定在一个恒定水平推理延迟也明显降下来了。当然这需要配合子 Agent 的“能力设计”来做。因为子 Agent 只回传结构化结果就要求它在执行时的 Prompt 设计里把自己的推理过程和结果抽取解耦。简单说每个子 Agent 内部是“自由推理 结构化输出”对外则是纯结构化结果。外部看不到它的思维过程但能看到准确的结果。4.3 超时、重试与熔断企业级系统必须有最基本的容错能力MultiAgent 也不例外。我们给每个子 Agent 配了三个核心参数超时时间、最大重试次数、熔断阈值。超时时间根据子 Agent 的具体任务类型单独配置。比如订单查询超时设置为 15 秒因为底层接口一般比较快而政策推导类任务超时设置为 30 秒因为模型推理本身耗时更长。这里的一个教训是不要给所有 Agent 设置同一套超时时间。以前我们统一设 20 秒结果查询类任务等到超时还没返回白白浪费等待时间推理类任务又经常因为 20 秒太短而误判失败。重试方面我们的策略是“最多重试两次且重试逻辑只在 FAILED 状态上触发”。但要注意重试并不适合所有任务。如果一个子 Agent 是因为底层业务数据本身有问题导致失败重试多少次结果都一样只有那些因为网络抖动、模型超时等瞬时原因导致失败的任务重试才有意义。所以我们在任务定义里增加了一个字段retryable规划器生成任务清单时综合判断这个任务是否可重试。熔断机制同样重要。我们给每个子 Agent 统计了单位时间内的失败率如果失败率连续三分钟超过 50%就自动摘除该 Agent 的调度资格并将依赖它的任务全部标记为 SKIPPED同时触发告警。这个机制防止了“一个下游接口抖动导致所有 Agent 被拖垮”的雪崩效应。5. 可观测性企业落地 MultiAgent 的隐形刚需5.1 一次完整 MultiAgent 调用的链路追踪很多团队在做 MultiAgent 时关注点全在模型效果上对可观测性基本没概念。直到线上出了问题翻遍日志都拼不出一次完整的任务链路才意识到这是个巨大的坑。MultiAgent 系统天然是分布式调用一次用户请求会触发规划器、多个子 Agent、外部系统之间的多次交互没有链路追踪根本没法排查。我们的方案启用了全局唯一 request_id 和 trace_id 机制。一次用户请求进入系统时生成 request_id从规划器开始到每个子任务执行结束所有日志都携带这个 trace_id。每个子 Agent 的内部日志也通过环境变量透传 trace_id 给底层模型调用和外部系统调用这样一行日志就能串起全链路。具体落地时我们直接复用了内部现有的 OpenTelemetry 基础设施没有额外造轮子。在任务下发入口埋一个 Span子 Agent 执行时包裹一个子 Span带上 task_id、agent_name、model_name、输入输出大小、耗时等属性。配合 APM 系统的拓扑视图一眼就能看到一次 MultiAgent 调用经过了哪些 Agent、在哪一层耗时最高、哪个环节报错。这套东西上线之后排查问题的平均耗时从小时级降到了分钟级。5.2 成本与延迟的精细化度量可观测性的第二个价值是“算得清账”。MultiAgent 系统的成本构成比普通接口复杂得多规划器调一次模型、每个子 Agent 调至少一次模型、失败重试会叠加中间步骤、上下文长短直接影响 Token 消耗。这些成本如果不做精细化度量财务月报出来时你根本解释不了为什么费用涨了三四倍。我们在日志里为每个 Span 记录了token_usage输入/输出 Token 数、model_price_per_1k_tokens、estimated_cost、latency_ms等字段。最后汇总到一张任务成本明细表里可以按业务线、按 Agent、按用户维度做多维分析。这张表帮我们发现了不少优化机会。比如某个子 Agent 的实际调用 Token 数远高于预估拆开日志一看是它的 Prompt 里塞了一段特别长的历史消息而这段历史消息对该任务的推理根本不提供有效信息。删掉之后这个子 Agent 的单次调用成本直接下降了 35%。这类优化如果只靠拍脑袋是完全发现不了的。6. 落地的效果复盘与下一步演进方向切到“Plan 模式 主子 Agent 协作”这套架构之后线上表现跟之前相比有了质的改善。我挑几个最直观的维度说说。首先是可控性。现在每次用户请求都会先产出一份 Plan运营团队可以提前看到系统“打算怎么做”而不是等它做完才知道结果。即便规划器偶尔生成不合理的计划也有校验器和人工兜底两道防线拦截基本杜绝了“Agent 自由发挥造成不可逆影响”的情况。其次是可观测性。每笔工单处理都有一套完整的 trace 日志能回溯到每一步的输入输出。业务方来问异常工单时审计成本非常低。这个变化对技术团队的价值远超预期因为它把 MultiAgent 系统纳入了公司已有的监控、告警、审计体系不再是一个游离在外的“黑箱玩具”。从量化数据看我们试点场景的规划一次通过率稳定在 85% 以上加上二次修正后接近 95%。单笔工单的平均处理延迟从最初 ReAct 模式的三四十秒降到了十秒级别不包含人工兜底的场景。Token 成本在引入“子 Agent 只回传结构化结果”的机制后下降了四成左右。当然这些数据只代表我们自己的业务场景如果你的业务复杂度不同数字会有出入但趋势应该是类似的。关于下一步演进我目前比较关注两个方向。一个是动态规划修正能力现在 Plan 一旦生成并开始执行中途基本不能大改。但有些业务在执行过程中会出现新信息比如查询订单时发现该用户同时还有一个未关闭的投诉单这时候理想状态是动态插入一个新任务而不是傻傻地按原计划跑完再回头补救。另一个是多 Plan 并发协同当用户诉求本身不是一个线性流程而是多个相对独立的目标需要并行推进时当前的单 Plan 结构就不够用了需要升级成多 Plan 并行、主 Agent 统一汇总的模式。这两个方向我们都在做原型验证等有阶段性成果再出来分享。最后聊一点个人的体会。做企业级 MultiAgent 落地真正的难点其实不在“让模型更聪明”而在“让模型的聪明可控、可预期、可治理”。Plan 模式是把模型推理收敛到一份可校验的任务清单上主子 Agent 协作是把执行收敛到一套清晰的权限与调度边界内。这两件事都没有用到什么高深技术但它们的价值远远超过花大力气调一个“更智能的单体 Agent”。如果你的团队也正在做 MultiAgent 落地我建议先用这两个思路把骨架搭稳再考虑让它更“聪明”的事。 企业级MultiAgent落地：Plan模式与主子Agent协作架构实践
1. 为什么企业级 MultiAgent 必须走“先规划、后执行”这条路1.1 单 Agent 处理复杂业务时的三个瓶颈先说结论企业级场景里把一个复杂的业务流程全部塞给单个 Agent让它自己“想一步、做一步”大概率会在上线两周内翻车。这个结论不是拍脑袋得出来的我们团队最早落地 MultiAgent 时就是这么干的然后被线上问题狠狠教育了一轮。当时我们接了一个典型的“工单自动处理”业务用户提交一个售后请求系统需要先理解用户诉求再查订单、查库存、算价格、生成处理方案、调用外部审批接口、最后回执给用户。逻辑链不算特别长但涉及多个内部系统和外部依赖。最初版本就是单 Agent 十几个 function calling 工具结果暴露出三个非常典型的瓶颈。第一上下文摊薄。为了让一个 Agent 能应对所有情况Prompt 里塞满了业务规则、工具说明、输出格式约束、安全红线。表面上很全面实际上模型根本“照顾”不过来。用户消息稍微长一点Agent 就开始抓不住重点该走的流程不走不该调的接口乱调。打个比方让一个新员工同时背十本操作手册再去处理具体工单他必然会漏步骤。第二工具发散。单 Agent 挂了十几个工具模型在每次推理时都有概率选错工具或者在同一个意图上反复横跳。我们的监控日志里出现过大量“工具调用-报错-换一个工具-再报错-又换回来”的循环一次工单处理能烧掉十几轮模型调用。Token 成本飙升的同时用户体验完全不可控。第三也是企业最不能接受的不可复现、不可审计。单 Agent 的自由推理路径每次都不一样同一个用户同类问题今天走 A 流程、明天走 B 流程。业务方来问“这个工单为什么这样处理了中间经历了什么”我们根本给不出一个可靠的过程留痕。这在对外服务场景里几乎等于事故。所以我们的第一个判断是企业级 MultiAgent 不能把“智能”押注在模型随机自由发挥上而要把“智能”收敛到一条可控的执行链路上。这也是我们最终切换到 Plan 模式的最根本动因。1.2 Plan 模式 vs ReAct 模式企业场景的关键差异接触过 Agent 架构的同学应该都知道业界主流的 Agent 工作模式大致分两类一类是 ReAct 模式模型边推理边行动每拿到一个观察结果就决定下一步动作另一类是 Plan 模式模型先基于用户诉求生成一份完整的任务计划再按计划逐步执行。这两种模式在小规模 Demo 里看不出太大差距但在企业生产环境里差别是天壤之别。我整理了一个对比表基本能说明问题的核心。维度ReAct 模式Plan 模式决策时机每步实时决策走一步看一步先全局规划再按步骤执行可控性弱中间步骤不可预判强计划先暴露可人工审核可审计性低执行路径每次不同高任务清单即审计依据出错成本中后段出错可能导致前功尽弃规划错误可在执行前拦截Token 消耗高反复推理和工具试探相对低规划一次后按图索骥适合场景探索型、开放型任务流程型、有明确步骤的任务我们切到 Plan 模式之后最直观的感受就是任务从“黑盒推理”变成了“白盒调度”。用户提一个诉求系统先产出一份类似“任务书”的结构化清单上面写着要做哪几件事、每件事谁来做、依赖关系是什么。工程团队可以在这份任务书出来之后进行校验不合理的地方直接打回重排而不是等 Agent 跑完五六个步骤才发现方向错了。1.3 什么样的业务适合切到 Plan 模式这里要泼一盆冷水Plan 模式不是银弹它不适合所有场景。按照我们内部的经验适合切 Plan 模式的任务有这么几个特征步骤多、有依赖、结果可验证、过程需要留痕。典型的像订单处理、库存调度、供应链审批、自动化测试执行、多系统数据核对。这些任务天然是“流程”形态Plan 模式只是把流程从人肉写死变成了模型动态生成。反之纯聊天、头脑风暴、开放创作这类任务不适合 Plan 模式。你让模型先写一份“聊天计划”再开始聊纯属脱裤子放屁平白增加一次模型调用和延迟。还有一个经验是渐进式切换。不要一上来就把所有流量切到 Plan 模式而是先挑一个流程最固定、出问题影响面最小的业务做试点跑通之后再逐步扩大范围。我们当时选的就是“售后工单自动处理”这个场景因为它链路清晰、边界明确、业务方配合度高非常适合作为 MultiAgent 架构的试验田。2. Plan 模式的核心拆解任务规划器与执行器分离2.1 任务规划器的输出结构化任务清单切换到 Plan 模式之后整个系统的第一个核心组件就是“任务规划器”。它的职责很简单把用户输入转换为一份结构化任务清单。但这份清单的格式设计直接决定了后面调度和执行的质量。我们最终采用了类似 DAG 的任务描述结构。每一份 Plan 都是一个 JSON 数组数组中每个元素是一个原子任务包含任务 ID、名称、类型、依赖关系、输入参数声明、期望输出协议、指定执行的子 Agent 类型等字段。下面是一个简化示例{ plan_id: plan_20250410_001, tasks: [ { task_id: t1, name: 解析用户诉求, type: intent_parse, dependencies: [], input: {raw_message: $user_input}, output_schema: {intent: string, entities: object}, assigned_agent: semantic_agent }, { task_id: t2, name: 查询订单信息, type: order_query, dependencies: [t1], input: {order_id: $t1.entities.order_id}, output_schema: {order_status: string, amount: number, sku_list: array}, assigned_agent: order_agent }, { task_id: t3, name: 生成退款方案, type: refund_plan, dependencies: [t1, t2], input: {intent: $t1.intent, order_info: $t2}, output_schema: {refund_amount: number, reason: string, risk_level: string}, assigned_agent: policy_agent } ] }这里有两个细节值得展开说。一个是依赖声明。每个任务都必须声明它依赖哪些前置任务调度器据此构建执行拓扑。这比简单的“按数组顺序执行”靠谱得多因为模型生成的先后顺序不一定是最优执行顺序有了依赖声明调度器可以并行执行互不依赖的任务缩短整体耗时。另一个是输出协议。每个任务都预先声明了期望的输出结构子 Agent 执行完必须按这个结构回传。这个设计解决了多 Agent 协作里最常见的数据对接问题—子 Agent 自由发挥的输出格式五花八门后端根本没法解析。提前定好协议等于给每个 Agent 套了一个规范边界。2.2 从规划到执行的调度模型Plan 生成之后下一个环节是调度执行。这块企业落地时容易被低估很多人以为把任务清单扔给 Agent 一个个跑就行实际上这里藏着大量细节。我们的调度器是一个独立的 Python 服务核心逻辑可以概括为三步构建依赖图、按拓扑序分批下发、汇总结果并校验。from collections import deque from typing import Dict, List, Set def schedule_tasks(plan: dict): tasks {t[task_id]: t for t in plan[tasks]} # 构建依赖关系 dependency_graph {tid: set(t[dependencies]) for tid, t in tasks.items()} dependents_map {} for tid, deps in dependency_graph.items(): for dep in deps: dependents_map.setdefault(dep, set()).add(tid) # 拓扑排序用入度 0 的节点作为可执行任务 in_degree {tid: len(deps) for tid, deps in dependency_graph.items()} ready_queue deque([tid for tid, deg in in_degree.items() if deg 0]) executed: Set[str] set() results: Dict[str, dict] {} while ready_queue: # 按批次执行所有当前可执行任务并行下发 batch list(ready_queue) ready_queue.clear() batch_results run_agent_batch(batch, tasks, results) for tid, res in batch_results.items(): executed.add(tid) results[tid] res # 解锁下游任务 for dependent in dependents_map.get(tid, []): in_degree[dependent] - 1 if in_degree[dependent] 0: ready_queue.append(dependent) # 如果有任务没被执行说明依赖成环或出现孤立任务 if len(executed) ! len(tasks): raise RuntimeError(fplan execution failed, executed{len(executed)} total{len(tasks)}) return results这段代码不是完整生产实现但已经能说明调度器的几个关键设计点批量并行、依赖解锁、成环检测。生产环境里我们还会在这层做更多事情比如 batch 大小限制、子 Agent 并发数控制、结果协议校验等等后面章节会展开。2.3 规划失败时的降级策略说完调度再说规划环节本身的一个大坑模型生成的 Plan 不一定是合法可执行的。它可能把不该拆的任务拆了可能遗漏必要步骤也可能构造出循环依赖关系。如果不做校验直接进入调度线上事故跑不了。我们当时设计了三层降级策略从轻到重依次是重新规划 - 局部修正 - 人工兜底。第一层规划器每次生成的 Plan 先过一个规则校验器。校验器主要查这几项依赖关系是否成环、任务类型是否在已注册的能力清单里、每个任务的输入参数引用是否指向已存在的前置输出、输出协议是否完整。校验不通过就要求规划器带着错误信息重新生成最多重试两次。第二层如果重试两次之后仍然不通过调度器会尝试局部修正。比如把依赖成环的两三个任务合并成一个任务或者把缺失的参数引用标记为“执行时再解析”。这个修正逻辑看似简单实际需要非常了解业务我们最初是用规则硬写的后来改成由另一个“修正 Agent”来做效果反而更差。原因是修正 Agent 没有足够的上下文判断业务合理性最后还是回归到规则为主。第三层如果前两层都失败系统会把 Plan 丢到一个人工待办队列由业务运营人员在后台查看、手动调整、确认执行。这层兜底虽然看起来“笨”但它是企业系统安全感的来源—模型可以出错但不能让用户承担模型出错的全部后果。3. 主子 Agent 协作用分层结构换可控性3.1 为什么不用“全对等”多智能体规划器解决了“做什么”的问题接下来就是“谁来做”。早期我们构想过一种“全对等”的 MultiAgent 协作模式也就是让多个 Agent 以平级身份聚在一起通过互相发消息的方式完成任务。但真正设计下去之后发现在企业生产环境里这条路基本走不通。对等模式最大的问题是权责边界模糊。几个能力相近的 Agent 同时面对一个任务时没有天然的“负责人”概念谁牵头、谁配合、分歧了怎么裁决全靠模型临场发挥。我们做过一次模拟测试三个对等 Agent 协同处理一个稍复杂的需求最终因为两个 Agent 对某个中间结果的理解不一致互相反复发消息确认来回七八轮才达成勉强共识延迟和 Token 开销都爆炸了。另一个问题是消息风暴。对等模式下每个 Agent 都可以向其他所有 Agent 发起通信通信复杂度接近 O(n²)。Agent 数量从三个增加到五个之后系统整体性能急剧下降日志里充斥着大量“无意义确认”消息。所以最终我们选定了“主子 Agent”的分层协作结构。这个结构可以用一个特别朴素的类比来理解主 Agent 是项目经理子 Agent 是不同工种的工程师。项目经理不亲自干活只负责拆解需求、派发任务、验收结果、协调资源工程师只负责自己专业领域内的执行不跨工种干涉别人。这种组织方式在人类社会里被验证了上百年拿到 Agent 系统里同样适用。3.2 主 Agent 的职责边界主 Agent 在“主子协作”里是一个特殊的存在它的特殊性在于能力不体现在具体执行上而体现在编排和决策上。我们给主 Agent 划定了三个核心职责域任务拆解、任务派发、结果汇总校验。任务拆解就是上一章的 Plan 生成任务派发是根据任务类型找到合适的子 Agent构造调用参数并下发结果汇总校验是收集所有子 Agent 的执行结果按业务规则做最终汇总和决策。同时我们给主 Agent 划了一条红线不允许直接调用业务工具。它只能跟子 Agent 通信不能绕过子 Agent 去查询订单、不能代替子 Agent 去写数据库。这条红线一开始我们觉得有些“死板”后来发现它是整个系统安全性的基石。因为主 Agent 一旦能调用具体工具它的 Prompt 就不得不包含大量工具权限信息被“投毒”或误调用的风险会指数级上升。把工具权限全部下沉到子 Agent主 Agent 保持相对“纯净”攻击面就小了很多。3.3 子 Agent 的注册、发现与调用协议子 Agent 的主业是具体领域执行为了被主 Agent 正确调度每个子 Agent 在上线前都要做一次“能力注册”。注册表里记录了 Agent 的名称、描述、支持的任务类型、输入输出协议、超时阈值、并发上限、模型规格等等。{ agent_name: order_agent, description: 查询订单状态、订单商品明细、订单金额信息不支持改单和售后操作, supported_tasks: [order_query, order_detail], input_schema: {order_id: string, user_id: string}, output_schema: {order_status: string, amount: number, sku_list: array}, timeout_ms: 15000, max_concurrency: 10, model: qwen-plus }主 Agent 在生成 Plan 时会结合“任务类型”和注册表里“supported_tasks”的匹配情况来指派执行者。这一步我们用的是规则匹配 向量召回混合的方式优先做精确匹配匹配不到再用语义检索找最接近的 Agent如果还是没有就认为规划失败走降级策略。子 Agent 之间的调用消息也统一走一套标准信封核心字段包括 task_id、source_agent、target_agent、payload、callback_url。刚开始我们用的是 HTTP 同步调用后来发现同步调用在高并发下很容易互相阻塞就改成了基于消息队列的异步模式。主 Agent 把任务投递到队列子 Agent 消费执行完再回传结果主 Agent 侧通过一个 pending 表跟踪所有未完成的任务。这个改动带来的吞吐量提升非常明显值得每个要上 MultiAgent 的团队提前考虑。4. 状态机驱动的任务生命周期管理4.1 任务状态机与持久化MultiAgent 系统运行起来之后最容易被忽视但最致命的是“任务的中间状态管理”。一个 Plan 被拆成七八个任务每个任务又有自己的生命周期如果中间任何一个环节宕机或超时整个协调就乱了。我们的做法是给每个任务定义一套显式状态机并落库持久化。任务状态我们定义了六种PENDING等待执行、RUNNING执行中、SUCCEEDED执行成功、FAILED执行失败、CANCELLED被取消、SKIPPED依赖失败跳过执行。from enum import Enum, auto class TaskStatus(Enum): PENDING auto() RUNNING auto() SUCCEEDED auto() FAILED auto() CANCELLED auto() SKIPPED auto() # 允许的状态迁移表 ALLOWED_TRANSITIONS { TaskStatus.PENDING: {TaskStatus.RUNNING, TaskStatus.CANCELLED}, TaskStatus.RUNNING: {TaskStatus.SUCCEEDED, TaskStatus.FAILED, TaskStatus.CANCELLED}, TaskStatus.SUCCEEDED: {}, TaskStatus.FAILED: {TaskStatus.RUNNING}, # 允许重试 TaskStatus.CANCELLED: {}, TaskStatus.SKIPPED: {}, } def transition(current: TaskStatus, target: TaskStatus) - TaskStatus: if target not in ALLOWED_TRANSITIONS[current]: raise IllegalStateTransitionError(current, target) return target这套状态机的好处是开发、测试、运维所有角色对“任务现在处于什么阶段”有一个统一的认知。出现问题时只要查某个 task_id 的状态历史就能复现整个生命周期。它把 MultiAgent 这种看起来“黑盒”的智能系统拉回到了普通分布式系统可以管理的维度。状态持久化我们直接用了关系型数据库每个任务一行记录字段包括 plan_id、task_id、status、current_retry_count、max_retry_count、input_snapshot、output_snapshot、error_message、timestamps。没有引入额外组件运维成本低排查问题也方便。4.2 上下文裁剪子 Agent 只回传结构化结果不做全量对话回传这块是我们节省 Token 成本最有效的一个设计也是我认为最有价值的一条经验。最早版本里子 Agent 执行完会把完整的对话历史回传给主 Agent导致主 Agent 的上下文迅速膨胀越到后面推理质量越差Token 费用也水涨船高。后来我们做了个“冷酷”的改造子 Agent 只允许回传 output_schema 里声明的结构化结果对话过程全部丢弃只在日志里留一份压缩摘要。主 Agent 拿到的是干净的字段值而不是一大坨对话记录。这个改动上线后单任务的 Token 消耗下降约四成主 Agent 的上下文长度稳定在一个恒定水平推理延迟也明显降下来了。当然这需要配合子 Agent 的“能力设计”来做。因为子 Agent 只回传结构化结果就要求它在执行时的 Prompt 设计里把自己的推理过程和结果抽取解耦。简单说每个子 Agent 内部是“自由推理 结构化输出”对外则是纯结构化结果。外部看不到它的思维过程但能看到准确的结果。4.3 超时、重试与熔断企业级系统必须有最基本的容错能力MultiAgent 也不例外。我们给每个子 Agent 配了三个核心参数超时时间、最大重试次数、熔断阈值。超时时间根据子 Agent 的具体任务类型单独配置。比如订单查询超时设置为 15 秒因为底层接口一般比较快而政策推导类任务超时设置为 30 秒因为模型推理本身耗时更长。这里的一个教训是不要给所有 Agent 设置同一套超时时间。以前我们统一设 20 秒结果查询类任务等到超时还没返回白白浪费等待时间推理类任务又经常因为 20 秒太短而误判失败。重试方面我们的策略是“最多重试两次且重试逻辑只在 FAILED 状态上触发”。但要注意重试并不适合所有任务。如果一个子 Agent 是因为底层业务数据本身有问题导致失败重试多少次结果都一样只有那些因为网络抖动、模型超时等瞬时原因导致失败的任务重试才有意义。所以我们在任务定义里增加了一个字段retryable规划器生成任务清单时综合判断这个任务是否可重试。熔断机制同样重要。我们给每个子 Agent 统计了单位时间内的失败率如果失败率连续三分钟超过 50%就自动摘除该 Agent 的调度资格并将依赖它的任务全部标记为 SKIPPED同时触发告警。这个机制防止了“一个下游接口抖动导致所有 Agent 被拖垮”的雪崩效应。5. 可观测性企业落地 MultiAgent 的隐形刚需5.1 一次完整 MultiAgent 调用的链路追踪很多团队在做 MultiAgent 时关注点全在模型效果上对可观测性基本没概念。直到线上出了问题翻遍日志都拼不出一次完整的任务链路才意识到这是个巨大的坑。MultiAgent 系统天然是分布式调用一次用户请求会触发规划器、多个子 Agent、外部系统之间的多次交互没有链路追踪根本没法排查。我们的方案启用了全局唯一 request_id 和 trace_id 机制。一次用户请求进入系统时生成 request_id从规划器开始到每个子任务执行结束所有日志都携带这个 trace_id。每个子 Agent 的内部日志也通过环境变量透传 trace_id 给底层模型调用和外部系统调用这样一行日志就能串起全链路。具体落地时我们直接复用了内部现有的 OpenTelemetry 基础设施没有额外造轮子。在任务下发入口埋一个 Span子 Agent 执行时包裹一个子 Span带上 task_id、agent_name、model_name、输入输出大小、耗时等属性。配合 APM 系统的拓扑视图一眼就能看到一次 MultiAgent 调用经过了哪些 Agent、在哪一层耗时最高、哪个环节报错。这套东西上线之后排查问题的平均耗时从小时级降到了分钟级。5.2 成本与延迟的精细化度量可观测性的第二个价值是“算得清账”。MultiAgent 系统的成本构成比普通接口复杂得多规划器调一次模型、每个子 Agent 调至少一次模型、失败重试会叠加中间步骤、上下文长短直接影响 Token 消耗。这些成本如果不做精细化度量财务月报出来时你根本解释不了为什么费用涨了三四倍。我们在日志里为每个 Span 记录了token_usage输入/输出 Token 数、model_price_per_1k_tokens、estimated_cost、latency_ms等字段。最后汇总到一张任务成本明细表里可以按业务线、按 Agent、按用户维度做多维分析。这张表帮我们发现了不少优化机会。比如某个子 Agent 的实际调用 Token 数远高于预估拆开日志一看是它的 Prompt 里塞了一段特别长的历史消息而这段历史消息对该任务的推理根本不提供有效信息。删掉之后这个子 Agent 的单次调用成本直接下降了 35%。这类优化如果只靠拍脑袋是完全发现不了的。6. 落地的效果复盘与下一步演进方向切到“Plan 模式 主子 Agent 协作”这套架构之后线上表现跟之前相比有了质的改善。我挑几个最直观的维度说说。首先是可控性。现在每次用户请求都会先产出一份 Plan运营团队可以提前看到系统“打算怎么做”而不是等它做完才知道结果。即便规划器偶尔生成不合理的计划也有校验器和人工兜底两道防线拦截基本杜绝了“Agent 自由发挥造成不可逆影响”的情况。其次是可观测性。每笔工单处理都有一套完整的 trace 日志能回溯到每一步的输入输出。业务方来问异常工单时审计成本非常低。这个变化对技术团队的价值远超预期因为它把 MultiAgent 系统纳入了公司已有的监控、告警、审计体系不再是一个游离在外的“黑箱玩具”。从量化数据看我们试点场景的规划一次通过率稳定在 85% 以上加上二次修正后接近 95%。单笔工单的平均处理延迟从最初 ReAct 模式的三四十秒降到了十秒级别不包含人工兜底的场景。Token 成本在引入“子 Agent 只回传结构化结果”的机制后下降了四成左右。当然这些数据只代表我们自己的业务场景如果你的业务复杂度不同数字会有出入但趋势应该是类似的。关于下一步演进我目前比较关注两个方向。一个是动态规划修正能力现在 Plan 一旦生成并开始执行中途基本不能大改。但有些业务在执行过程中会出现新信息比如查询订单时发现该用户同时还有一个未关闭的投诉单这时候理想状态是动态插入一个新任务而不是傻傻地按原计划跑完再回头补救。另一个是多 Plan 并发协同当用户诉求本身不是一个线性流程而是多个相对独立的目标需要并行推进时当前的单 Plan 结构就不够用了需要升级成多 Plan 并行、主 Agent 统一汇总的模式。这两个方向我们都在做原型验证等有阶段性成果再出来分享。最后聊一点个人的体会。做企业级 MultiAgent 落地真正的难点其实不在“让模型更聪明”而在“让模型的聪明可控、可预期、可治理”。Plan 模式是把模型推理收敛到一份可校验的任务清单上主子 Agent 协作是把执行收敛到一套清晰的权限与调度边界内。这两件事都没有用到什么高深技术但它们的价值远远超过花大力气调一个“更智能的单体 Agent”。如果你的团队也正在做 MultiAgent 落地我建议先用这两个思路把骨架搭稳再考虑让它更“聪明”的事。