多 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 的生产致命伤：局部最优与幻觉循环

ReAct（Reasoning + 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_factory=list) status: TaskStatus = TaskStatus.PENDING result: Optional[Any] = None retry_count: int = 0 class DynamicDAGPlan(BaseModel): plan_id: str nodes: Dict[str, TaskNode] = Field(default_factory=dict) 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 规划器为我们的多智能体系统带来了显著的工程收益：

1. 端到端执行时延（P90）：相较于纯串行 ReAct 链路，端到端执行耗时从 78.4s 降低至 24.1s，整体吞吐量提升了近 3.2 倍；
2. 长流程交付成功率：在超过 10 个协同步骤的复杂业务中，任务完成率由早期的 61.3% 提升至 93.8%；
3. 成本收益比：通过精准剪枝与产物压缩，单次复杂请求的平均 Token 消耗下降了 58.7%。

多智能体系统的自主规划绝非单纯让模型写出步骤列表，而是将不确定的大模型推理收敛至确定性的图拓扑调度状态机之中。未来的演进方向将聚焦于结合强化学习的静态代价预估模型，在规划阶段即预测每条分支的执行时延与 Token 成本，实现真正自适应的成本-时延最优规划。