多 Agent 自主规划与动态任务分解终局复盘从 ReAct、Plan-and-Solve 到动态 DAG 图的工程演进在多智能体工作室为期三个月的企业级交付季中自主规划Autonomous Planning与动态任务分解Dynamic Task Decomposition是所有业务流能够跑通的核心中枢。从最初简单的单智能体线性推理到支撑金融投研、跨系统自动化运维和复杂代码重构等重度多 Agent 场景我们先后经历了三代规划器架构的重构与推倒重来。本文结合我们在数十万次生产调用中的真实故障、性能瓶颈与落地数据对多 Agent 系统的规划范式进行全景式终局复盘剖析从单步 ReAct 到全局 Plan-and-Solve再到拓扑动态 DAG有向无环图执行引擎的演进之路。一、三代自主规划架构的演进路线与生产缺陷对比在真实的复杂业务场景中用户的原始输入往往是高度模糊且充满多阶段依赖的例如“对比分析 A/B 两家上市公司近三年的财报资产负债率结合行业宏观研报输出风险评估并生成 PPT 汇报草稿”。面对这类复合任务不同的规划机制表现出完全不同的稳定性和吞吐表现。┌────────────────────────────────────────────────────────────────────────┐ │ ❌ 第一代单步反应式 ReAct单向线性步进局部盲区与死循环 │ │ 用户输入 ──► [思考 Thought] ──► [行动 Action] ──► [观察 Observation]...│ │ 致命缺陷缺乏全局视角一旦中间步骤出现幻觉或工具超时容易陷入死循环 │ └────────────────────────────────────────────────────────────────────────┘ ▼ ┌────────────────────────────────────────────────────────────────────────┐ │ ⚠️ 第二代静态计划执行 Plan-and-Solve一次性生成静态执行无法应变 │ │ 用户输入 ──► [Planner 全量拆解 1..N 步] ──► [Executor 顺序执行] │ │ 致命缺陷环境反馈导致前提假设失效时后续所有步骤变成无效消耗 │ └────────────────────────────────────────────────────────────────────────┘ ▼ ┌────────────────────────────────────────────────────────────────────────┐ │ ✅ 第三代动态拓扑 DAG 规划引擎动态图生成 并行调度 拓扑自愈修剪│ │ 用户输入 ──► [DAG Planner 生成依赖图] ──► [Worker 并行执行无依赖节点] │ │ ▲ │ │ │ └─── [环境反馈 / 动态插入或回溯修剪] ───┘ │ │ 生产收益并行度提升 300%单步失败自愈率 94.2%Token 浪费降低 62% │ └────────────────────────────────────────────────────────────────────────┘1. 第一代 ReAct 的生产致命伤局部最优与幻觉循环ReActReasoning Acting采用交替推演方式每一步依赖前一步的观察结果。在复杂业务中模型极易陷入“工具报错 - 尝试换参数 - 再次报错 - 反复重试”的死循环单次请求能把 Token 额度迅速耗尽且调用链路过长导致 P99 延迟高达 90 秒以上。2. 第二代 Plan-and-Solve 的僵化困境为了解决 ReAct 的近视问题很多团队转向“先出完整方案再由执行器逐步执行”。然而真实生产环境充满不确定性第 2 步拉取财报数据时发现公司财年口径发生变更此时预先生成的第 3、4、5 步指令完全基于错误假设导致整条流水线产出的结论全盘报废。3. 第三代 动态拓扑 DAG 规划引擎我们最终落地的生产级方案是将规划过程建模为有向无环图Directed Acyclic Graph, DAG。规划器输出包含节点依赖关系的拓扑图调度器并行激发入度为 0 的节点。当某个节点执行返回非预期数据或触发异常时触发轻量级“子图重规划Sub-DAG Re-planning”仅对受影响的下游节点进行剪枝与重新推导既保障了全局视角的完整性又兼顾了运行时的弹性。二、生产级动态 DAG 规划器与并行调度引擎实现以下是我们在交付季沉淀出的生产级动态 DAG 规划核心框架包含状态校验、依赖拓扑排序与动态节点修剪逻辑。import asyncio from typing import List, Dict, Set, Any, Optional from pydantic import BaseModel, Field from enum import Enum class TaskStatus(str, Enum): PENDING PENDING RUNNING RUNNING SUCCESS SUCCESS FAILED FAILED SKIPPED SKIPPED class TaskNode(BaseModel): task_id: str description: str assigned_agent: str dependencies: List[str] Field(default_factorylist) status: TaskStatus TaskStatus.PENDING result: Optional[Any] None retry_count: int 0 class DynamicDAGPlan(BaseModel): plan_id: str nodes: Dict[str, TaskNode] Field(default_factorydict) def get_ready_tasks(self) - List[TaskNode]: 获取所有前置依赖已成功完成且当前处于 PENDING 状态的任务 ready_tasks [] for node in self.nodes.values(): if node.status TaskStatus.PENDING: deps_satisfied all( self.nodes[dep_id].status TaskStatus.SUCCESS for dep_id in node.dependencies if dep_id in self.nodes ) if deps_satisfied: ready_tasks.append(node) return ready_tasks def is_completed(self) - bool: 检查整张图是否全部执行完毕成功、失败或被跳过 return all( node.status in [TaskStatus.SUCCESS, TaskStatus.FAILED, TaskStatus.SKIPPED] for node in self.nodes.values() ) class ProductionDAGExecutor: def __init__(self, agent_registry: Dict[str, Any], max_concurrency: int 5): self.agent_registry agent_registry self.semaphore asyncio.Semaphore(max_concurrency) async def execute_task_node(self, dag_plan: DynamicDAGPlan, task: TaskNode): async with self.semaphore: task.status TaskStatus.RUNNING agent self.agent_registry.get(task.assigned_agent) if not agent: task.status TaskStatus.FAILED task.result f未注册的智能体: {task.assigned_agent} return try: # 收集所有前置依赖节点的执行结果作为上下文输入 dep_contexts { dep_id: dag_plan.nodes[dep_id].result for dep_id in task.dependencies } # 模拟调用底层 Agent 执行 result await agent.run(task.description, dep_contexts) task.result result task.status TaskStatus.SUCCESS except Exception as exc: task.retry_count 1 if task.retry_count 3: task.status TaskStatus.PENDING # 重新放入就绪队列重试 else: task.status TaskStatus.FAILED task.result str(exc) # 触发下游依赖修剪将受影响节点置为 SKIPPED 并动态重规划 self._prune_downstream(dag_plan, task.task_id) def _prune_downstream(self, dag_plan: DynamicDAGPlan, failed_task_id: str): 对失败节点的下游链路进行级联剪枝防止无效调度 for node in dag_plan.nodes.values(): if failed_task_id in node.dependencies and node.status TaskStatus.PENDING: node.status TaskStatus.SKIPPED node.result f前置依赖 {failed_task_id} 失败自动跳过 self._prune_downstream(dag_plan, node.task_id) async def run_dag(self, dag_plan: DynamicDAGPlan): while not dag_plan.is_completed(): ready_tasks dag_plan.get_ready_tasks() if not ready_tasks: # 若无就绪任务但整体未完成说明存在死锁或环状依赖 running_tasks [n for n in dag_plan.nodes.values() if n.status TaskStatus.RUNNING] if not running_tasks: break await asyncio.sleep(0.1) continue tasks [ asyncio.create_task(self.execute_task_node(dag_plan, task)) for task in ready_tasks ] await asyncio.gather(*tasks)三、动态任务分解在生产交付中的核心难题与消解手段在实际系统上线后仅有拓扑排序和状态机是不够的真正的挑战集中在以下三个方面1. 任务粒度爆炸Granularity Explosion当用户输入较为宏大时大模型规划器容易将任务过度细分至几十个微小步骤例如“打开浏览器”、“点击搜索框”、“输入文字”造成大量的网络调度开销与上下文膨胀。解决方案引入分层规划器Hierarchical Planner。顶层 Macro-Planner 只拆分出 3~5 个高阶里程碑节点每个里程碑节点交由专职的 Sub-Agent 内部执行微粒度 ReAct 循环对外只暴露黑盒产物。2. 环状依赖与拓扑死锁检测在动态重规划过程中如果模型在修剪后重新生成的补充节点与原有节点存在循环依赖执行引擎将会永久死锁。解决方案在 DAG 状态机接收到新的补丁图后严格执行Kahn 算法进行有向无环校验。一旦发现环路直接拒绝变更并强制降级至兜底策略。3. 上下文透传膨胀Context Bloat随着 DAG 链路的推进下游汇总节点需要合并前序多个分支的产出。如果将全部前序节点的原始输出机械拼接极易超出模型的有效注意力窗口导致关键指标被稀释。解决方案设立分支产物压缩协议Artifact Extraction Protocol。每个 Worker 节点在结束时必须生成两份产物一份全量调试日志一份结构化精简摘要JSON Schema下游节点默认仅消费摘要。四、工作室交付季数据总结与演进方向经过整个交付季的高强度压力测试动态 DAG 规划器为我们的多智能体系统带来了显著的工程收益端到端执行时延P90相较于纯串行 ReAct 链路端到端执行耗时从 78.4s 降低至 24.1s整体吞吐量提升了近 3.2 倍长流程交付成功率在超过 10 个协同步骤的复杂业务中任务完成率由早期的 61.3% 提升至 93.8%成本收益比通过精准剪枝与产物压缩单次复杂请求的平均 Token 消耗下降了 58.7%。多智能体系统的自主规划绝非单纯让模型写出步骤列表而是将不确定的大模型推理收敛至确定性的图拓扑调度状态机之中。未来的演进方向将聚焦于结合强化学习的静态代价预估模型在规划阶段即预测每条分支的执行时延与 Token 成本实现真正自适应的成本-时延最优规划。 多 Agent 自主规划与动态任务分解终局复盘：从 ReAct、Plan-and-Solve 到动态 DAG 图的工程演进
多 Agent 自主规划与动态任务分解终局复盘从 ReAct、Plan-and-Solve 到动态 DAG 图的工程演进在多智能体工作室为期三个月的企业级交付季中自主规划Autonomous Planning与动态任务分解Dynamic Task Decomposition是所有业务流能够跑通的核心中枢。从最初简单的单智能体线性推理到支撑金融投研、跨系统自动化运维和复杂代码重构等重度多 Agent 场景我们先后经历了三代规划器架构的重构与推倒重来。本文结合我们在数十万次生产调用中的真实故障、性能瓶颈与落地数据对多 Agent 系统的规划范式进行全景式终局复盘剖析从单步 ReAct 到全局 Plan-and-Solve再到拓扑动态 DAG有向无环图执行引擎的演进之路。一、三代自主规划架构的演进路线与生产缺陷对比在真实的复杂业务场景中用户的原始输入往往是高度模糊且充满多阶段依赖的例如“对比分析 A/B 两家上市公司近三年的财报资产负债率结合行业宏观研报输出风险评估并生成 PPT 汇报草稿”。面对这类复合任务不同的规划机制表现出完全不同的稳定性和吞吐表现。┌────────────────────────────────────────────────────────────────────────┐ │ ❌ 第一代单步反应式 ReAct单向线性步进局部盲区与死循环 │ │ 用户输入 ──► [思考 Thought] ──► [行动 Action] ──► [观察 Observation]...│ │ 致命缺陷缺乏全局视角一旦中间步骤出现幻觉或工具超时容易陷入死循环 │ └────────────────────────────────────────────────────────────────────────┘ ▼ ┌────────────────────────────────────────────────────────────────────────┐ │ ⚠️ 第二代静态计划执行 Plan-and-Solve一次性生成静态执行无法应变 │ │ 用户输入 ──► [Planner 全量拆解 1..N 步] ──► [Executor 顺序执行] │ │ 致命缺陷环境反馈导致前提假设失效时后续所有步骤变成无效消耗 │ └────────────────────────────────────────────────────────────────────────┘ ▼ ┌────────────────────────────────────────────────────────────────────────┐ │ ✅ 第三代动态拓扑 DAG 规划引擎动态图生成 并行调度 拓扑自愈修剪│ │ 用户输入 ──► [DAG Planner 生成依赖图] ──► [Worker 并行执行无依赖节点] │ │ ▲ │ │ │ └─── [环境反馈 / 动态插入或回溯修剪] ───┘ │ │ 生产收益并行度提升 300%单步失败自愈率 94.2%Token 浪费降低 62% │ └────────────────────────────────────────────────────────────────────────┘1. 第一代 ReAct 的生产致命伤局部最优与幻觉循环ReActReasoning Acting采用交替推演方式每一步依赖前一步的观察结果。在复杂业务中模型极易陷入“工具报错 - 尝试换参数 - 再次报错 - 反复重试”的死循环单次请求能把 Token 额度迅速耗尽且调用链路过长导致 P99 延迟高达 90 秒以上。2. 第二代 Plan-and-Solve 的僵化困境为了解决 ReAct 的近视问题很多团队转向“先出完整方案再由执行器逐步执行”。然而真实生产环境充满不确定性第 2 步拉取财报数据时发现公司财年口径发生变更此时预先生成的第 3、4、5 步指令完全基于错误假设导致整条流水线产出的结论全盘报废。3. 第三代 动态拓扑 DAG 规划引擎我们最终落地的生产级方案是将规划过程建模为有向无环图Directed Acyclic Graph, DAG。规划器输出包含节点依赖关系的拓扑图调度器并行激发入度为 0 的节点。当某个节点执行返回非预期数据或触发异常时触发轻量级“子图重规划Sub-DAG Re-planning”仅对受影响的下游节点进行剪枝与重新推导既保障了全局视角的完整性又兼顾了运行时的弹性。二、生产级动态 DAG 规划器与并行调度引擎实现以下是我们在交付季沉淀出的生产级动态 DAG 规划核心框架包含状态校验、依赖拓扑排序与动态节点修剪逻辑。import asyncio from typing import List, Dict, Set, Any, Optional from pydantic import BaseModel, Field from enum import Enum class TaskStatus(str, Enum): PENDING PENDING RUNNING RUNNING SUCCESS SUCCESS FAILED FAILED SKIPPED SKIPPED class TaskNode(BaseModel): task_id: str description: str assigned_agent: str dependencies: List[str] Field(default_factorylist) status: TaskStatus TaskStatus.PENDING result: Optional[Any] None retry_count: int 0 class DynamicDAGPlan(BaseModel): plan_id: str nodes: Dict[str, TaskNode] Field(default_factorydict) def get_ready_tasks(self) - List[TaskNode]: 获取所有前置依赖已成功完成且当前处于 PENDING 状态的任务 ready_tasks [] for node in self.nodes.values(): if node.status TaskStatus.PENDING: deps_satisfied all( self.nodes[dep_id].status TaskStatus.SUCCESS for dep_id in node.dependencies if dep_id in self.nodes ) if deps_satisfied: ready_tasks.append(node) return ready_tasks def is_completed(self) - bool: 检查整张图是否全部执行完毕成功、失败或被跳过 return all( node.status in [TaskStatus.SUCCESS, TaskStatus.FAILED, TaskStatus.SKIPPED] for node in self.nodes.values() ) class ProductionDAGExecutor: def __init__(self, agent_registry: Dict[str, Any], max_concurrency: int 5): self.agent_registry agent_registry self.semaphore asyncio.Semaphore(max_concurrency) async def execute_task_node(self, dag_plan: DynamicDAGPlan, task: TaskNode): async with self.semaphore: task.status TaskStatus.RUNNING agent self.agent_registry.get(task.assigned_agent) if not agent: task.status TaskStatus.FAILED task.result f未注册的智能体: {task.assigned_agent} return try: # 收集所有前置依赖节点的执行结果作为上下文输入 dep_contexts { dep_id: dag_plan.nodes[dep_id].result for dep_id in task.dependencies } # 模拟调用底层 Agent 执行 result await agent.run(task.description, dep_contexts) task.result result task.status TaskStatus.SUCCESS except Exception as exc: task.retry_count 1 if task.retry_count 3: task.status TaskStatus.PENDING # 重新放入就绪队列重试 else: task.status TaskStatus.FAILED task.result str(exc) # 触发下游依赖修剪将受影响节点置为 SKIPPED 并动态重规划 self._prune_downstream(dag_plan, task.task_id) def _prune_downstream(self, dag_plan: DynamicDAGPlan, failed_task_id: str): 对失败节点的下游链路进行级联剪枝防止无效调度 for node in dag_plan.nodes.values(): if failed_task_id in node.dependencies and node.status TaskStatus.PENDING: node.status TaskStatus.SKIPPED node.result f前置依赖 {failed_task_id} 失败自动跳过 self._prune_downstream(dag_plan, node.task_id) async def run_dag(self, dag_plan: DynamicDAGPlan): while not dag_plan.is_completed(): ready_tasks dag_plan.get_ready_tasks() if not ready_tasks: # 若无就绪任务但整体未完成说明存在死锁或环状依赖 running_tasks [n for n in dag_plan.nodes.values() if n.status TaskStatus.RUNNING] if not running_tasks: break await asyncio.sleep(0.1) continue tasks [ asyncio.create_task(self.execute_task_node(dag_plan, task)) for task in ready_tasks ] await asyncio.gather(*tasks)三、动态任务分解在生产交付中的核心难题与消解手段在实际系统上线后仅有拓扑排序和状态机是不够的真正的挑战集中在以下三个方面1. 任务粒度爆炸Granularity Explosion当用户输入较为宏大时大模型规划器容易将任务过度细分至几十个微小步骤例如“打开浏览器”、“点击搜索框”、“输入文字”造成大量的网络调度开销与上下文膨胀。解决方案引入分层规划器Hierarchical Planner。顶层 Macro-Planner 只拆分出 3~5 个高阶里程碑节点每个里程碑节点交由专职的 Sub-Agent 内部执行微粒度 ReAct 循环对外只暴露黑盒产物。2. 环状依赖与拓扑死锁检测在动态重规划过程中如果模型在修剪后重新生成的补充节点与原有节点存在循环依赖执行引擎将会永久死锁。解决方案在 DAG 状态机接收到新的补丁图后严格执行Kahn 算法进行有向无环校验。一旦发现环路直接拒绝变更并强制降级至兜底策略。3. 上下文透传膨胀Context Bloat随着 DAG 链路的推进下游汇总节点需要合并前序多个分支的产出。如果将全部前序节点的原始输出机械拼接极易超出模型的有效注意力窗口导致关键指标被稀释。解决方案设立分支产物压缩协议Artifact Extraction Protocol。每个 Worker 节点在结束时必须生成两份产物一份全量调试日志一份结构化精简摘要JSON Schema下游节点默认仅消费摘要。四、工作室交付季数据总结与演进方向经过整个交付季的高强度压力测试动态 DAG 规划器为我们的多智能体系统带来了显著的工程收益端到端执行时延P90相较于纯串行 ReAct 链路端到端执行耗时从 78.4s 降低至 24.1s整体吞吐量提升了近 3.2 倍长流程交付成功率在超过 10 个协同步骤的复杂业务中任务完成率由早期的 61.3% 提升至 93.8%成本收益比通过精准剪枝与产物压缩单次复杂请求的平均 Token 消耗下降了 58.7%。多智能体系统的自主规划绝非单纯让模型写出步骤列表而是将不确定的大模型推理收敛至确定性的图拓扑调度状态机之中。未来的演进方向将聚焦于结合强化学习的静态代价预估模型在规划阶段即预测每条分支的执行时延与 Token 成本实现真正自适应的成本-时延最优规划。