复杂任务规划器：Plan-and-Solve 与 DAG 依赖图并发调度

本节目标

纯 ReAct 的短板与 Plan-and-Solve 的破局

ReAct 是一种典型的贪心局部决策（Greedy Local Search）：模型每一步只看眼前这一步该调什么工具。当用户提出一个复合型长任务（例如：「对比项目中 A、B、C 三个模块的依赖体积，找出循环引用，生成重构方案并补齐单元测试」）时，纯 ReAct 往往在第 4 步修某个细节报错时彻底忘了后面还要做另外两个模块，或者把本可以并行执行的 3 个模块的扫描任务串行跑了 3 遍，耗时翻了 3 倍。

DAG（有向无环图）规划调度器将任务拆分为两个阶段：

  1. Planner（全局规划器）：先把复杂目标拆解成一张带有显式前驱依赖（dependsOn）的子任务图。
  2. DAG Executor（拓扑并发执行器）：自动找出当前所有前驱依赖已完成的就绪节点（Ready Nodes），用 Promise.all 并发执行，并将上游节点的产出精准注入下游节点的上下文。
          ┌──► [Task B: 扫描模块 A] ──┐
[Task A] ─┼──► [Task C: 扫描模块 B] ──┼──► [Task E: 汇总对比并生成重构补丁]
(解析配置) └──► [Task D: 扫描模块 C] ──┘
               (B, C, D 同层并发执行!)

手写带环路检测与分层并发执行的 DAG 任务调度器

export class DagTaskScheduler {
  constructor({ workerRunner, onReplan }) {
    this.workerRunner = workerRunner // async (task, upstreamOutputs) => result
    this.onReplan = onReplan         // async ({ failedTask, error, completed }) => newTasks
  }

  // 校验 DAG 合法性（检测未知依赖与循环依赖 Cycle）
  validateGraph(tasks) {
    const map = new Map(tasks.map((t) => [t.id, t]))
    const visiting = new Set()
    const visited = new Set()

    const dfs = (id) => {
      if (visiting.has(id)) throw new Error(`检测到任务循环依赖 (Cycle): ${id}`)
      if (visited.has(id)) return
      visiting.add(id)
      const node = map.get(id)
      if (!node) throw new Error(`存在未定义的依赖任务节点: ${id}`)
      for (const depId of node.dependsOn || []) {
        dfs(depId)
      }
      visiting.delete(id)
      visited.add(id)
    }

    for (const t of tasks) dfs(t.id)
    return true
  }

  async executePlan(initialTasks) {
    let tasks = [...initialTasks]
    this.validateGraph(tasks)

    const completed = new Map() // taskId -> output
    const failed = new Map()

    while (completed.size < tasks.length) {
      // 1. 筛选出所有尚未完成、且所有前置依赖均已成功完成的就绪节点
      const readyBatch = tasks.filter(
        (t) =>
          !completed.has(t.id) &&
          !failed.has(t.id) &&
          (t.dependsOn || []).every((depId) => completed.has(depId))
      )

      if (readyBatch.length === 0) {
        throw new Error('DAG 调度死锁：无就绪任务可执行，但计划尚未全部完成')
      }

      // 2. 同层无依赖冲突的子任务全部并行启动
      const batchResults = await Promise.allSettled(
        readyBatch.map(async (task) => {
          const upstreamContext = Object.fromEntries(
            (task.dependsOn || []).map((depId) => [depId, completed.get(depId)])
          )
          const output = await this.workerRunner(task, upstreamContext)
          return { task, output }
        })
      )

      // 3. 收集本批次结果，若遇节点失败则触发动态重规划（Re-planning）
      for (let i = 0; i < batchResults.length; i++) {
        const res = batchResults[i]
        const currentTask = readyBatch[i]

        if (res.status === 'fulfilled') {
          completed.set(currentTask.id, res.value.output)
        } else {
          failed.set(currentTask.id, res.reason)
          if (!this.onReplan) throw res.reason

          // 调用 Planner 根据已完成进度与报错动态调整剩余子图
          tasks = await this.onReplan({
            failedTask: currentTask,
            error: res.reason?.message || String(res.reason),
            completed: Object.fromEntries(completed)
          })
          this.validateGraph(tasks)
          break
        }
      }
    }

    return Object.fromEntries(completed)
  }
}

课后练习

练习 1：在拿到大模型生成的 DAG 任务列表后，为什么在启动执行前必须先跑一遍基于深度优先搜索（DFS 三色标记法：visiting vs visited）的 validateGraph？

答案：大模型在拆分复杂任务时偶尔会写出循环依赖（例如 Task A dependsOn Task B 同时 Task B dependsOn Task A）或引用了拼写错误的 taskId。若不提前用 DFS 检测环路与悬空引用，while 调度循环会因为找不到入度为 0 的就绪节点而直接死锁。

练习 2：为什么静态一次性规划（Static Planning）在真实世界往往不够用，必须在节点失败时引入 onReplan（动态重规划）？

答案：Planner 在任务开始前制定计划时，尚未看到工具返回的真实环境数据（例如计划第 2 步去读取某个配置文件，执行时才发现该项目改用了环境变量配置）。当假设与真实环境不符导致某步失败时，动态重规划能够保留已完成的成果（completed），仅重写后续受影响的子图。