规划执行框架

概述

Plan-and-Execute 是一种将规划（生成步骤列表）和执行（逐步完成）分离的智能体架构。这种分离使得可以用不同的模型处理不同的任务阶段，并支持动态重规划。本文深入分析 Plan-and-Execute 模式、LLMCompiler 并行执行框架、以及层级任务网络（HTN）规划。

1. Plan-and-Execute 模式

1.1 核心架构

graph TD
    TASK[用户任务] --> PLANNER[规划器<br/>Planner LLM]
    PLANNER --> PLAN[步骤列表<br/>Step 1, 2, ..., N]
    PLAN --> EXEC1[执行器执行步骤 1]
    EXEC1 --> RESULT1[结果 1]
    RESULT1 --> EXEC2[执行器执行步骤 2]
    EXEC2 --> RESULT2[结果 2]
    RESULT2 --> DOTS[...]
    DOTS --> EXECN[执行器执行步骤 N]
    EXECN --> RESULTN[结果 N]
    RESULTN --> REPLAN{需要重规划?}
    REPLAN -->|是| PLANNER
    REPLAN -->|否| FINAL[汇总最终结果]

1.2 与 ReAct 的对比

维度	ReAct	Plan-and-Execute
规划方式	每步决定下一步（贪心）	先生成完整计划
全局视野	无（只看当前步）	有（全局计划）
LLM 使用	每步都用大模型	规划用大模型，执行可用小模型
成本	均匀分布	规划成本高，执行成本低
适应性	高（实时调整）	需要显式重规划
可解释性	中等	高（计划步骤可见）

1.3 两阶段设计

阶段一：规划 (Planning)

PLANNER_PROMPT = """
给定用户任务，生成一个分步执行计划。
每个步骤应该是一个清晰、可执行的指令。

任务: {task}

请输出编号列表，每行一个步骤:
1. ...
2. ...
"""

plan = planner_llm.generate(PLANNER_PROMPT.format(task=task))
steps = parse_plan(plan)

阶段二：执行 (Execution)

EXECUTOR_PROMPT = """
你需要执行以下步骤，使用可用的工具完成。

当前步骤: {step}
之前的执行结果: {previous_results}
可用工具: {tools}

请执行当前步骤。
"""

for step in steps:
    result = executor_llm.generate(
        EXECUTOR_PROMPT.format(
            step=step,
            previous_results=results,
            tools=tools
        )
    )
    results.append(result)

1.4 动态重规划

当执行过程中遇到意外情况时，触发重规划：

REPLAN_PROMPT = """
原始计划: {original_plan}
已完成步骤及结果: {completed_steps}
当前问题: {issue}

请基于当前情况修改剩余计划:
"""

def should_replan(step_result, expected):
    """判断是否需要重规划"""
    # 执行失败
    if step_result.error:
        return True
    # 结果与预期严重偏离
    if llm.evaluate(step_result, expected) < threshold:
        return True
    # 发现新信息改变了问题的本质
    if llm.detect_new_info(step_result):
        return True
    return False

2. LLMCompiler：并行执行

2.1 核心思想

Kim et al. (2024) 提出的 LLMCompiler 将任务分解为有向无环图 (DAG)，识别可并行执行的步骤：

graph TD
    TASK[任务: 比较北京和上海的天气和人口] --> PLAN[LLM Planner<br/>生成任务 DAG]
    PLAN --> T1[任务 1: 查询北京天气]
    PLAN --> T2[任务 2: 查询上海天气]
    PLAN --> T3[任务 3: 查询北京人口]
    PLAN --> T4[任务 4: 查询上海人口]
    T1 --> JOIN[Joiner<br/>汇总所有结果]
    T2 --> JOIN
    T3 --> JOIN
    T4 --> JOIN
    JOIN --> ANS[最终比较分析]

2.2 三大组件

Planner（规划器）

生成带依赖关系的任务列表：

任务: 比较北京和上海的天气和人口

1. search("北京 当前天气")          # 无依赖
2. search("上海 当前天气")          # 无依赖
3. search("北京 人口 2024")         # 无依赖
4. search("上海 人口 2024")         # 无依赖
5. join()                           # 依赖 1,2,3,4

关键：Planner 不仅生成任务列表，还标注了依赖关系，使得无依赖的任务可以并行执行。

Task Fetching Unit（任务调度器）

识别 DAG 中的并行机会：

\[ \text{ParallelSet}(t) = \{t_i \mid \text{deps}(t_i) \subseteq \text{completed}\} \]

即所有依赖已完成的任务都可以并行启动。

Joiner（汇总器）

汇总并行执行的结果，决定是否需要重规划：

def joiner(results, original_task):
    # 检查是否所有必要结果都已获取
    if all_results_available(results):
        return llm.synthesize(results, original_task)
    else:
        # 部分失败，决定重规划策略
        return replan(results, original_task)

2.3 性能优势

指标	ReAct	Plan-Execute (串行)	LLMCompiler (并行)
延迟	$N \times L$	$N \times l$	$D \times l$
LLM 调用	$N$	$N + 1$	$1 + 1$
工具调用	$N$ (串行)	$N$ (串行)	$N$ (并行)

其中 $N$ 为步骤数，$L$ 为大模型延迟，$l$ 为小模型/工具延迟，$D$ 为 DAG 的最长路径深度。

3. 层级任务网络 (HTN)

3.1 HTN 规划概述

层级任务网络（Hierarchical Task Network）是经典 AI 中的层级规划方法，将复杂任务自顶向下分解为子任务：

\[ \text{Task} \rightarrow \text{Method} \rightarrow \text{Subtask}_1, \text{Subtask}_2, \ldots \]

graph TD
    T0[准备晚餐] --> M1[方法: 做中餐]
    M1 --> T1[买菜]
    M1 --> T2[烹饪]
    M1 --> T3[摆盘]
    T1 --> T11[列购物清单]
    T1 --> T12[去超市]
    T1 --> T13[选购食材]
    T2 --> T21[洗菜切菜]
    T2 --> T22[炒菜]
    T2 --> T23[煮饭]

3.2 HTN 与 LLM 的结合

LLM 天然适合进行层级任务分解：

HTN_DECOMPOSE_PROMPT = """
将以下任务分解为子任务。每个子任务应该是可以直接执行的
原子操作，或者可以进一步分解的复合任务。

任务: {task}
可用的原子操作: {primitive_actions}

请输出层级分解结果:
任务: {task}
├── 子任务 1: ...
│   ├── 原子操作: ...
│   └── 原子操作: ...
├── 子任务 2: ...
└── 子任务 3: ...
"""

3.3 HTN 在智能体中的优势

优势	描述
可复用性	分解方法可以跨任务复用
抽象层级	在适当层级进行推理和监控
可扩展性	新的分解方法可以增量添加
可解释性	层级结构清晰展示了任务逻辑
失败恢复	可以在子任务级别重试

4. 高级规划策略

4.1 适应性规划

根据执行中获得的信息动态调整计划的粒度和内容：

def adaptive_planning(task, initial_confidence):
    if initial_confidence > 0.9:
        # 高置信度：生成详细计划一次性执行
        plan = detailed_plan(task)
        return execute_all(plan)
    elif initial_confidence > 0.5:
        # 中等置信度：粗略计划 + 逐步细化
        rough_plan = rough_plan(task)
        for step in rough_plan:
            detailed_step = refine_step(step, context)
            result = execute(detailed_step)
            context.update(result)
    else:
        # 低置信度：探索式执行
        return react_loop(task)

4.2 投机性规划

类似于 CPU 的投机执行，预测可能的分支并提前计算：

计划步骤 3: 查询用户的账户状态
  预测结果 A (80% 可能): 账户正常 → 预先准备步骤 4A
  预测结果 B (20% 可能): 账户异常 → 预先准备步骤 4B

4.3 约束规划

在规划中加入显式约束：

\[ \text{Plan}^* = \arg\max_{\pi} R(\pi) \quad \text{s.t.} \quad C(\pi) \leq \text{budget} \]

约束类型：

约束	示例
时间约束	总执行时间 < 60 秒
成本约束	LLM API 调用 < $0.50
安全约束	不执行删除/修改操作
质量约束	每步验证通过率 > 95%

5. 框架实现

5.1 LangGraph 中的 Plan-and-Execute

from langgraph.graph import StateGraph

# 定义状态
class PlanExecuteState(TypedDict):
    task: str
    plan: list[str]
    current_step: int
    results: list[str]
    final_answer: str

# 构建图
workflow = StateGraph(PlanExecuteState)

# 添加节点
workflow.add_node("planner", plan_step)
workflow.add_node("executor", execute_step)
workflow.add_node("replanner", replan_step)

# 添加边
workflow.set_entry_point("planner")
workflow.add_edge("planner", "executor")
workflow.add_conditional_edges(
    "executor",
    should_continue,
    {"replan": "replanner", "next": "executor", "end": END}
)
workflow.add_edge("replanner", "executor")

交叉引用

LangGraph 框架的详细介绍见 LangChain与LangGraph。

5.2 实际部署考量

考量	建议
规划模型	使用最强模型（GPT-4, Claude Opus）确保计划质量
执行模型	可用较小模型（GPT-3.5, Claude Haiku）降低成本
重规划阈值	不宜过于敏感，避免频繁重规划
最大步骤数	设置上限（如 20 步）防止无限循环
步骤粒度	每步应该是一个清晰、可验证的操作
错误处理	区分可重试错误和致命错误

6. 规划质量评估

6.1 计划的质量维度

\[ Q(\text{plan}) = w_1 \cdot \text{完整性} + w_2 \cdot \text{可执行性} + w_3 \cdot \text{效率} + w_4 \cdot \text{鲁棒性} \]

维度	定义	评估方法
完整性	计划是否覆盖了所有必要步骤	检查目标是否可达
可执行性	每个步骤是否可以实际执行	检查工具/API 可用性
效率	步骤数是否最少	比较与最优方案的差距
鲁棒性	对意外情况的容忍度	注入错误后的恢复能力

参考文献

Wang, L. et al. (2023). Plan-and-Solve Prompting: Improving Zero-Shot Chain-of-Thought Reasoning by Large Language Models. ACL 2023.
Kim, S. et al. (2024). An LLM Compiler for Parallel Function Calling. arXiv:2312.04511.
Erol, K. et al. (1994). HTN Planning: Complexity and Expressivity. AAAI 1994.
Huang, W. et al. (2022). Inner Monologue: Embodied Reasoning through Planning with Language Models. CoRL 2022.
Sun, H. et al. (2023). AdaPlanner: Adaptive Planning from Feedback with Language Models. NeurIPS 2023.

指标	ReAct	Plan-Execute (串行)	LLMCompiler (并行)
延迟	\(N \times L\)	\(N \times l\)	\(D \times l\)
LLM 调用	\(N\)	\(N + 1\)	\(1 + 1\)
工具调用	\(N\) (串行)	\(N\) (串行)	\(N\) (并行)