企业级 AI Agent“计划与执行”（Plan-and-Execute）架构实战：破解长程任务拓扑解耦、动态重规划（Re-planning）与断点容错

在企业级智能体落地过程中，当业务需求从单轮检索或单步工具调用升级到跨 ERP/MES 供应链排产调整、企业财务年度账目核验、数据中心批量离线巡检等长程任务（Long-horizon Tasks）时，传统的单步循环思考（ReAct / Reflexion）模式暴露出显著瓶颈：由于每一步决策都高度依赖完整对话历史，随着执行步数增加，上下文急剧膨胀，不仅导致推理成本呈二次方激增，更极易引发“长视野迷失（Horizon Loss）”、工具调用陷入死循环以及无法并发执行互不依赖的子步骤。为了攻克长程复杂业务编排难题，“计划与执行”（Plan-and-Execute）架构成为了现代企业级智能体系统的核心范式。本文结合百信企业智能体平台的私有化生产实践，深度拆解由 Planner 进行任务拓扑解耦与 DAG 有向无环图生成、由执行器（Executor）进行上下文隔离与高并发调度、通过增量动态重规划（Re-planning）应对真实系统环境变化，以及结合 Checkpointer 机制实现秒级断点恢复与幂等容错的完整架构。

企业长程复杂任务痛点：ReAct 架构局限与 Plan-and-Execute 范式演进

在大模型驱动的智能体系统演进中，经典的 ReAct（Reasoning + Acting）架构通过“思考-行动-观察（Thought-Action-Observation）”的紧密交替循环，在解决 3~5 步内的短链任务中表现卓越。然而，一旦面对跨系统、多前置条件、耗时数十分钟甚至需要人工审批的长程企业任务，其底层机理便暴露出三大深层次困境：

1. 单步贪婪试错的“视野狭隘”与死循环陷阱（Horizon Loss & Tool Oscillation）

ReAct 的核心假设是“每一步行动都基于前一步的即时反馈进行贪婪推断”。这种缺乏全局蓝图的局部试错机制，在复杂业务中极易导致：

• 局部最优陷阱：智能体在处理供应链突发缺料预警时，为了满足局部某个紧急订单，直接调用采购接口下单，却忽略了后续步骤中由于仓储库容不足、关税申报排期冲突带来的全局系统性瘫痪；
• 工具调用震荡与死锁：当查询接口返回“参数不完整”或“权限受限”时，ReAct 容易在相邻两个接口之间反复切换重试，直到耗尽最大轮次或预算熔断，无法从更高维度意识到需要重新规划执行路径。

2. 线性上下文滚雪球与注意力稀释（Context Dilution）

在 ReAct 循环中，每次观察到的外部接口返回数据（如一段数百行的 JSON 结构体）都会不加筛选地追加到总 Prompt 中：

• 注意力分散与指令漂移：当执行到第 10 步时，上下文已累积数万 Token，底层模型对最初系统设定的核心业务规则（如“严禁在未确认预算前生成订单”）的遵从率显著下滑；
• 无法并行（Sequential Bottleneck）：即使任务中包含 4 个彼此毫无依赖关系的子系统数据拉取任务（例如同时拉取 ERP 财务凭证、CRM 客户评级、WMS 库位状态和税务发票池），ReAct 也只能按序串行执行，导致端到端耗时被无限拉长。

核心架构解构：Planner 拓扑解耦、DAG 有向无环图与 Executor 分布式执行

在 Plan-and-Execute 架构中，系统将“思考要做什么（宏观规划）”与“具体怎么做（微观执行）”进行物理级解耦，构建了由规划器（Planner）、任务拓扑（DAG Graph）和执行器池（Executor Engine）组成的三层系统结构：

1. 规划阶段（Planning Phase）：全局意图向有向无环图（DAG）映射

Planner 通常由高推理能力的基座大模型（如通过系统提示词专门强化的任务分解器）承担。它并不直接调用业务系统接口，而是专注将用户的非结构化宏观目标，分解为一组具有严格依赖关系的结构化子任务节点：

• 显式依赖声明（Explicit Dependency Declaration）：每个任务节点明确声明前置依赖项（如 dependencies: ["task_fetch_erp", "task_verify_wms"]），形成合法的 DAG 拓扑，杜绝隐式时序假设；
• 局部上下文最小化（Context Minimization）：Planner 为每个任务节点生成高度自包含的执行目标与参数期望，Executor 在执行具体节点时无需加载其他无关分支的细节日志。

2. 结构化任务节点规范（Task Node Schema）

在企业级私有化引擎中，任务节点必须遵循强类型的 JSON Schema 契约：

// 生产级任务节点数据结构定义
export interface PlanTaskNode {
  id: string;                         // 任务唯一标识，如 task_01_query_erp
  title: string;                      // 任务人类可读名称
  description: string;                // 执行指令描述，用于引导 Worker 专注于目标
  assignedWorker: string;             // 指定执行该任务的专业 Worker 或工具集
  dependencies: string[];             // 前置任务 ID 列表，全量完成方可激活本节点
  status: 'pending' | 'ready' | 'running' | 'completed' | 'failed' | 'skipped';
  inputSchema?: Record<string, any>;  // 输入参数映射（支持引用前置节点输出变量）
  outputResult?: any;                 // 节点真实执行输出，作为后续节点的上下文输入
  retryCount: number;                 // 当前重试次数
  maxRetries: number;                 // 最大允许局部重试次数
}

3. 执行阶段（Execution Phase）：拓扑排序与高并发并发派发（Fan-out / Fan-in）

执行引擎维护一个动态任务队列，通过对 DAG 持续进行拓扑排序（Topological Sort）推进流程：

• 就绪检测与并行扇出（Fan-out）：当检测到多个节点的 dependencies 均已进入 completed 状态，调度器将这些节点同时分发给独立的 Executor Worker 进行高并发处理，大幅缩减总链路耗时；
• 状态汇总与屏障同步（Fan-in Barrier）：对于汇聚节点（如“根据 ERP 采购单与 MES 质检结果综合生成入库单”），执行引擎设置屏障等待，直到所有上游依赖节点全部写入中间结果黑板，才激活下游节点。

动态增量重规划（Re-planning）：执行反馈闭环、拓扑变更条件与幻觉防御

传统的静态工作流一旦遭遇外部接口返回非预期数据（如 ERP 提示“物料已被占用”或“价格超过审批红线”）就会立即报错退出。Plan-and-Execute 架构的核心威力在于其具备**运行时的闭环动态重规划（Re-planning）**能力：

1. 动态重规划触发守卫条件（Re-planning Guardrails）

重规划是一个消耗模型推理算力的高开销操作，不能无节制触发。百信平台设计了三道严格的触发守卫（Guard Conditions）：

• 重大环境事实分歧（Environmental Divergence）：当执行节点返回的外部业务事实与 Planner 最初的前置假设冲突（例如原本规划调用 A 供应商采购，接口返回其停产整改），局部重试无法修复时触发；
• 前置节点执行彻底失败（Terminal Step Failure）：某任务在达到 maxRetries 后依然失败，且该任务被下游多个核心任务强依赖，无法通过简单降级忽略时触发；
• 新增动态前置条件发现（Dynamic Pre-condition Discovered）：在执行某查询步骤后，返回结果指明需要先通过一道全新的法务审批流程，原 DAG 拓扑缺失该关键节点。

2. 增量拓扑重写算法（Incremental Re-planning Algorithm）

在触发重规划时，Replanner 绝不是推倒重来，而是实施**局部增量手术**：

1. 冻结已完成节点（Freeze Completed Nodes）：已执行成功节点（状态为 completed）及其产生的业务副作用（如已生成的采购预占单）被完全锁定为不可变上下文（Immutable Context），严禁被重新规划，防止重复写入系统；
2. 裁剪失效未执行子图（Prune Orphan Branches）：移除受阻断节点影响且已无意义的后续 pending 节点；
3. 增量注入补救节点（Inject Remedial Sub-graph）：Replanner 根据当前最新黑板上下文，为剩余未满足的目标生成新的替代路径（如由“调用主供应商”切换为“拆解为两家备选供应商采购并补发比价审批”），拼接到现有 DAG 中继续推进。

3. 重规划次数防雪崩熔断与防死锁（Budget & Loop Prevention）

为了防止大模型在复杂边界条件下陷入“规划-失败-重规划-再次失败”的无限震荡死循环：

• 设置重规划预算上限（Max Re-planning Budget）：全链路硬性限制重规划次数（通常建议设为 3 次）。一旦达到上限，直接将上下文与当前 DAG 状态打包，升级为人工审批或安全优雅降级；
• 相似计划检测与语义去重（Semantic Plan Deduplication）：通过计算新规划子图与历史已失败子图的编辑距离或语义相似度，若检测到 Replanner 试图复用已证明失败的执行策略，强行中断并提示模型更换求解逻辑。

状态治理与断点容错：中间变量持久化、Checkpointer 机制与幂等回退

在企业生产环境中，一个长程规划任务可能持续数小时（例如涉及跨部门审批与批量报表合成），中间难免遭遇网络波动、服务重启或异步等待。因此，状态治理与容错恢复机制是决定架构能否商用的分水岭：

1. 全局状态机与中间变量黑板（State Machine & Blackboard）

系统维护一个结构化的全局状态黑板，所有节点必须且只能通过黑板进行数据交互：

• 变量引用契约（Variable Scoping & Referencing）：下游任务可以通过统一语法（如 {{task_query_stock.output.available_qty}}）引用上游产物，调度器在派发任务前自动完成参数插值与合法性校验；
• 敏感信息掩码（Data Masking）：对于涉及客户隐私或财务敏感字段，黑板支持字段级脱敏隔离，避免无关任务节点的 Worker 接收到超权限数据。

2. 检查点机制（Checkpointer）与秒级断点续跑

系统在每一个任务节点状态流转（如 running -> completed）或重规划完成的瞬间，触发 Checkpointer 执行不可变快照持久化：

{
  "checkpointId": "chk_20261105_04_008",
  "planId": "plan_erp_inventory_reconcile",
  "version": 4,
  "timestamp": "2026-11-05T09:35:12.000Z",
  "dagNodes": [
    { "id": "t1", "status": "completed", "output": { "discrepancyCount": 12 } },
    { "id": "t2", "status": "completed", "output": { "approvalStatus": "APPROVED" } },
    { "id": "t3", "status": "running", "retryCount": 1 }
  ],
  "blackboardState": { "auditId": "AUD-99812", "operator": "auto_agent_worker" }
}

当宿主进程崩溃或因维护重启后，系统能够根据 checkpointId 从外部数据库（如 PostgreSQL / Redis）直接恢复 DAG 内存状态，跳过所有已完成节点，实现真正无缝的断点续跑。

3. 分布式补偿与 Saga 幂等回退（Saga Compensation & Idempotency）

在涉及银行扣款、ERP 记账、工单关闭等具备真实副作用的外部系统调用时：

• 节点绑定反向补偿逻辑（Compensation Actions）：在声明写操作任务时，必须同时登记其反向回滚接口（如 execute: "createPurchaseOrder" 对应 compensate: "cancelPurchaseOrder"）；
• 全局中止时的有序逆序回滚：当整条长程任务因不可抗力彻底失败且需退出时，系统沿着已完成的写操作节点路径进行逆序（Reverse-order）补偿调用，确保外部异构系统的数据强一致性。

生产级 TypeScript 引擎实战：DAG 调度器、动态重规划器与容错执行流

以下为百信在生产环境中运行的核心 Plan-and-Execute 调度引擎精简实现，完整展示了 DAG 拓扑解析、并行分发、反馈检测与增量重规划逻辑：

/**
 * 企业级 Plan-and-Execute 核心调度引擎
 * 职责：实现 DAG 拓扑排序、多 Worker 并行派发、环境变化感知与动态重规划闭环
 * 遵循严格生产规范与全中文注释说明
 */

// 任务节点定义接口
export interface PlanTask {
  id: string;                         // 任务全局唯一标识
  title: string;                      // 任务中文名称
  description: string;                // 执行指令细则
  dependencies: string[];             // 前置依赖任务 ID 数组
  status: 'pending' | 'ready' | 'running' | 'completed' | 'failed';
  result?: any;                       // 真实业务执行结果
  retryCount: number;                 // 重试计数
}

// 执行上下文与黑板共享状态
export interface ExecutionContext {
  planId: string;                     // 规划流程唯一 ID
  globalGoal: string;                 // 用户原始宏观业务目标
  tasks: Map<string, PlanTask>;       // 动态 DAG 任务哈希表
  blackboard: Record<string, any>;    // 共享上下文变量池
  replanningCount: number;            // 当前已触发的重规划次数
  maxReplanningBudget: number;        // 最大允许重规划次数
}

export class PlanAndExecuteEngine {
  private ctx: ExecutionContext;

  constructor(planId: string, globalGoal: string, initialTasks: PlanTask[]) {
    this.ctx = {
      planId,
      globalGoal,
      tasks: new Map(initialTasks.map(t => [t.id, { ...t }])),
      blackboard: {},
      replanningCount: 0,
      maxReplanningBudget: 3,
    };
  }

  /**
   * 核心主执行循环
   * 采用事件与拓扑驱动，持续派发无前置依赖的就绪任务，检测异常并触发重规划
   */
  public async executePipeline(): Promise<{ success: boolean; data: any }> {
    console.log(`[Plan-and-Execute] 启动流程: ${this.ctx.planId}, 目标: ${this.ctx.globalGoal}`);

    while (true) {
      // 1. 检查全量任务是否已全部顺利完成
      const allTasks = Array.from(this.ctx.tasks.values());
      const allCompleted = allTasks.every(t => t.status === 'completed');
      if (allCompleted) {
        console.log(`[Plan-and-Execute] 全流程执行完毕，已达成业务目标。`);
        return { success: true, data: this.ctx.blackboard };
      }

      // 2. 筛选当前满足依赖条件的“就绪”任务（Fan-out 并发源）
      const readyTasks = allTasks.filter(t => {
        if (t.status !== 'pending') return false;
        // 检查所有前置依赖项是否均已完成
        return t.dependencies.every(depId => {
          const depTask = this.ctx.tasks.get(depId);
          return depTask && depTask.status === 'completed';
        });
      });

      // 若没有就绪任务但整体未完成，检测是否存在失败节点
      if (readyTasks.length === 0) {
        const failedTask = allTasks.find(t => t.status === 'failed');
        if (failedTask) {
          console.warn(`[Plan-and-Execute] 捕获任务执行受阻节点: ${failedTask.title} (${failedTask.id})`);

          // 尝试触发增量动态重规划
          const replanned = await this.triggerReplanning(failedTask);
          if (!replanned) {
            // 重规划预算耗尽或无法生成替代路径，安全熔断
            return { success: false, data: { error: 'REPLANNING_EXHAUSTED', failedTask } };
          }
          // 重规划成功，重新进入主循环
          continue;
        }

        // 若无就绪任务且无失败任务（可能存在循环依赖死锁）
        const runningTasks = allTasks.filter(t => t.status === 'running');
        if (runningTasks.length === 0) {
          throw new Error('检测到 DAG 拓扑死锁或无法解析的前置依赖！');
        }
      }

      // 3. 并发派发所有就绪任务（Fan-out 并行执行）
      console.log(`[Plan-and-Execute] 本轮激活 ${readyTasks.length} 个并发就绪任务节点`);
      await Promise.all(readyTasks.map(task => this.runSingleTask(task)));

      // 4. 持久化检查点（Checkpointer Snapshot）
      this.saveCheckpoint();
    }
  }

  /**
   * 单任务节点沙箱化执行
   */
  private async runSingleTask(task: PlanTask): Promise<void> {
    task.status = 'running';
    console.log(`-> 正在执行子任务: [${task.id}] ${task.title}`);

    try {
      // 模拟调用底层真实工具/业务系统接口
      // 实际生产中此处派发给对应的 Tool / Worker Agent
      const mockResult = await this.mockExternalSystemCall(task);

      // 记录产出并更新黑板变量
      task.result = mockResult;
      task.status = 'completed';
      this.ctx.blackboard[task.id] = mockResult;
      console.log(`✓ 子任务顺利完成: [${task.id}]`);
    } catch (err: any) {
      task.retryCount++;
      console.error(`✗ 子任务异常: [${task.id}] 错误信息: ${err.message}`);

      // 简单的局部重试控制（最多重试 1 次，超出后标记为 failed 交给 Replanner）
      if (task.retryCount <= 1) {
        console.log(`[局部重试] 任务 ${task.id} 即将执行第 ${task.retryCount} 次重试...`);
        task.status = 'pending';
      } else {
        task.status = 'failed';
      }
    }
  }

  /**
   * 闭环动态增量重规划逻辑（Replanner）
   */
  private async triggerReplanning(failedTask: PlanTask): Promise<boolean> {
    this.ctx.replanningCount++;
    console.log(`[Replanner] 正在触发第 ${this.ctx.replanningCount} / ${this.ctx.maxReplanningBudget} 次动态重规划...`);

    if (this.ctx.replanningCount > this.ctx.maxReplanningBudget) {
      console.error(`[Replanner 熔断] 已达到重规划预算上限，终止自动重规划。`);
      return false;
    }

    // 1. 冻结不可变的 completed 任务，收集当前有效上下文
    const completedTasks = Array.from(this.ctx.tasks.values()).filter(t => t.status === 'completed');
    console.log(`[Replanner] 冻结 ${completedTasks.length} 个已完成历史节点，保证外部系统幂等。`);

    // 2. 模拟 LLM 动态评估：根据受阻原因，生成针对性的替代子路径
    // 假设失败原因是某个供应商缺货，动态重规划为拆单采购并补充审批
    console.log(`[Replanner] 正在重写依赖于 ${failedTask.id} 的后续子图...`);

    // 移除原失败节点及受阻的旧待定节点
    this.ctx.tasks.delete(failedTask.id);

    // 增量注入替代任务节点
    const altTask1: PlanTask = {
      id: `${failedTask.id}_alt_fallback_query`,
      title: '调用备选物料库并进行库存再分配',
      description: '主路径受阻，转向备选合格供应商物料目录进行可用性复核',
      dependencies: completedTasks.map(t => t.id).slice(-1), // 依赖最近一个完成项
      status: 'pending',
      retryCount: 0,
    };

    const altTask2: PlanTask = {
      id: `${failedTask.id}_alt_human_confirm`,
      title: '生成变动审计差异单并提交审批',
      description: '对价格与交期差异进行合规留痕',
      dependencies: [altTask1.id],
      status: 'pending',
      retryCount: 0,
    };

    // 将新节点注入 DAG
    this.ctx.tasks.set(altTask1.id, altTask1);
    this.ctx.tasks.set(altTask2.id, altTask2);

    console.log(`[Replanner] 增量注入 2 个补救节点成功，DAG 拓扑已重构。`);
    return true;
  }

  /**
   * 模拟外部业务系统调用
   */
  private async mockExternalSystemCall(task: PlanTask): Promise<any> {
    // 模拟微小网络时延
    await new Promise(r => setTimeout(r, 100));

    // 演示特定任务抛出受阻异常以展示重规划机制
    if (task.id === 'task_verify_primary_supplier' && task.retryCount === 0) {
      throw new Error('主供货商核心物料库存不足（系统返回 STOCK_EXHAUSTED）');
    }

    return { code: 200, message: '执行成功', timestamp: Date.now() };
  }

  /**
   * 检查点持久化保存（模拟存储到分布式存储）
   */
  private saveCheckpoint(): void {
    // 实际生产中将 this.ctx 序列化存储至 Redis / PostgreSQL
    // console.log(`[Checkpointer] 快照已持久化。当前任务数: ${this.ctx.tasks.size}`);
  }
}

落地工程边界与选型决策：ReAct、Plan-and-Execute 与分层 Supervisor 的适用场景对比

在企业级私有化 AI 平台落地中，并不存在适用于所有场景的“万能架构”。架构师必须根据任务的复杂度、耗时跨度、容错要求以及算力消耗进行科学选型：

1. 三大编排范式的工程选型决策树

• 选用 ReAct 的场景：
  • 步骤少于 5 步，逻辑高度线性，且每一步必须紧跟最新外部反馈才能判断下一步；
  • 典型场景：简单客服问答中的知识库检索扩展、单接口参数自纠错、日常简单文档润色。
• 选用 Plan-and-Execute 的场景：
  • 任务具备明确的全局最终交付目标，中间步骤多达 8~30 步，存在大量互不依赖可并发的子环节；
  • 执行耗时较长，对网络波动具备容忍度，强依赖断点续跑与不可变中间结果审计；
  • 典型场景：跨 ERP/MES 生产调度复核、多数据源财务并表稽核、自动化代码库安全大范围巡检。
• 选用分层主管（Hierarchical Supervisor）的场景：
  • 任务需要高度垂直的专业领域知识，且不同角色具有截然不同的权限级别与工具集隔离需求；
  • 典型场景：大型企业多部门联合应急处置（法务、财务、研发与运维智能体跨部门联合攻关）。

2. 私有化部署下的 Token 吞吐与显存优化权衡

在企业私有云算力受限（如仅配备 2~4 张本地算力卡）的环境中，Plan-and-Execute 展现出压倒性的工程经济性：

• 总 Token 消耗降低 40%~60%：相比 ReAct 将全量历史反复喂给大模型，Plan-and-Execute 的 Executor 仅接收当前节点的专用微观提示词，极大缩短了单次 Prompt 长度，避开了注意力机制的二次方显存爆炸；
• 首字延迟与吞吐优化：Planner 生成全量骨架后，通过异步并发派发，使得算力卡的批处理（Batching）利用率提升至 85% 以上，大幅改善了企业级并发响应能力。

通过构建**全局 DAG 拓扑解耦、增量闭环动态重规划与 Checkpointer 状态持久化**的完整体系，企业智能体平台得以跨越玩具级 Demo 的死循环与上下文脆弱性泥潭，真正为工业级复杂长程业务提供稳健、确定且高可用的生产力支撑。