04. Plan-and-Execute:先规划,再执行,失败时重规划

ReAct 是一个反应式的模式——每一步只看当下,决定下一步做什么。这在短任务里没问题,但任务一长就出问题。

一个典型场景:你让 Agent 做"帮我分析 data/ 下所有 CSV 文件,找出销售额 top 5 的产品,生成一份报告并存到 output/report.md"。这是一个 6~10 步的任务。ReAct 跑到第 4 步时,上下文里已经是一大堆工具调用和观察结果。模型在决定"下一步做什么"时,最初的目标已经被淹没在观察结果堆里。它可能跑去生成报告,忘了先过滤 top 5;也可能在某个 CSV 解析出错后,再也想不起还有其他 CSV 没读。

这种现象叫 goal drift——目标漂移。ReAct 天然不抗这个,因为它没有"全局视图"。

Plan-and-Execute 的核心洞察

解法很简单但很有效:把规划和执行分开。

在任务开始时,让一个 Planner(通常是更强的推理模型)一次性生成完整的计划——N 个步骤,每步做什么、用什么工具、预期产出是什么。然后一个 Executor 按这个计划逐步执行,每做完一步更新计划状态。

用户任务: 分析 data/ 下 CSV,找 top 5 产品,生成报告

Planner 输出:
  Step 1: list_files("data/", ext="csv") -> 文件列表
  Step 2: for each file: read_csv(f) -> dataframe
  Step 3: merge all dfs, group by product, sum sales
  Step 4: sort desc, take top 5
  Step 5: generate markdown report
  Step 6: write_file("output/report.md", content)

Executor 按顺序执行,每步拿到 Observation 后:
  - 如果结果符合预期: 推进到下一步
  - 如果偏离: 触发 Replan

这个架构的意义在于:全局目标不再依赖模型每轮从观察堆里"记起来",而是作为一个数据结构固定下来。Executor 每一步只需要回答"按计划,这一步做什么",不需要做长程推理。

最小可跑版本

用 Python 手写一个最小的 Plan-and-Execute Agent,不依赖任何框架:

import json
from openai import OpenAI
from pydantic import BaseModel

client = OpenAI()

class PlanStep(BaseModel):
    id: int
    action: str  # 工具名或 "think"
    args: dict
    goal: str    # 这一步要达成的子目标
    status: str = "pending"  # pending | done | failed
    result: str | None = None

class Plan(BaseModel):
    task: str
    steps: list[PlanStep]

def make_plan(task: str, available_tools: list[str]) -> Plan:
    sys = f"""你是一个规划专家。给定用户任务,拆成有序步骤。
可用工具: {', '.join(available_tools)}
返回严格的 JSON,格式为: {{"task": "...", "steps": [{{"id":1,"action":"tool_name","args":{{...}},"goal":"..."}}, ...]}}"""
    resp = client.chat.completions.create(
        model="o4-mini",
        messages=[{"role": "system", "content": sys},
                  {"role": "user", "content": task}],
        response_format={"type": "json_object"},
    )
    return Plan.model_validate_json(resp.choices[0].message.content)

def execute_step(step: PlanStep, tools: dict) -> PlanStep:
    try:
        fn = tools[step.action]
        step.result = str(fn(**step.args))
        step.status = "done"
    except Exception as e:
        step.result = f"错误: {e}"
        step.status = "failed"
    return step

def should_replan(plan: Plan, new_obs: str) -> bool:
    sys = """你在审查一个执行计划是否还有效。看最新一步的结果,判断:
- 如果结果符合预期,后续步骤可以正常执行,回答 "continue"
- 如果结果和预期严重偏离,需要重新规划,回答 "replan"
只输出一个词。"""
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": sys},
            {"role": "user", "content": f"计划: {plan.model_dump_json()}\n最新结果: {new_obs}"},
        ],
    )
    return "replan" in resp.choices[0].message.content.lower()

def run(task: str, tools: dict, max_replan: int = 2):
    plan = make_plan(task, list(tools.keys()))
    replan_count = 0

    while True:
        pending = [s for s in plan.steps if s.status == "pending"]
        if not pending:
            return plan  # 所有步骤完成

        step = execute_step(pending[0], tools)

        if step.status == "failed" or should_replan(plan, step.result):
            if replan_count >= max_replan:
                return plan  # 达到 replan 上限
            replan_count += 1
            # 重规划:把当前进度喂回去
            new_task = f"""原任务: {task}
已完成: {[s for s in plan.steps if s.status == 'done']}
刚失败: {step}
请生成剩余步骤的新计划"""
            plan = make_plan(new_task, list(tools.keys()))

这段代码的灵魂在三个地方:

make_plan——用推理模型一次性出计划。推理模型之所以重要,是因为规划需要全局考虑所有步骤之间的依赖,这是 CoT 能发挥最大作用的地方。

should_replan——这是 Plan-Execute 和纯"线性脚本"的分水岭。每一步执行完都让模型判断"计划还对不对"。如果偏了,触发重规划。

状态化的 Plan 对象——所有步骤的状态和结果都存在 Plan 里。无论重规划多少次,已完成的工作不丢,不需要重做。

重规划的策略:全重还是局部改

重规划分两种模式:

全量重规划——把当前进度和失败原因整个喂给 Planner,让它从"当前状态"出发生成剩余计划。上面的代码是这种。优点是简单、能应对大幅偏离。缺点是频繁重规划时上下文会膨胀。

局部修补——只让 Planner 修改"从失败步骤往后"的那几步,前面已完成的保持不变。更省 token,但实现复杂一点,需要让 Planner 明确返回"要替换哪些步骤"。生产 Agent(LangGraph、CrewAI)倾向用这种。

实务上经验是:失败的原因如果是"工具返回和预期不一致",用局部修补;如果是"我对任务理解错了",用全量重规划。前者只需要换一两步,后者需要重新思考整个路径。

Plan-Execute 的几个衍生变体

这一套"分规划和执行"的思想出来后,衍生了几个变体,每个都值得了解:

ReWOO (Reasoning WithOut Observation)——2023 年的论文。让 Planner 出完整计划,所有步骤并行执行(如果没有依赖),最后 Solver 综合所有观察给答案。比传统 Plan-Execute 更省 token,因为不需要每步都回到模型做判断。适合步骤相互独立的任务,比如"查 10 个人的信息"。

LLMCompiler——Plan 阶段直接输出一个 DAG,Executor 按依赖关系并行执行。本质上是把 Agent 编译成一个可并行的计算图。对"大量独立工具调用"的场景延迟减少 30~50%。

LATS (Language Agent Tree Search)——把 Plan-Execute 和 Tree Search 结合。每步有多个可能的 Action,用 Monte Carlo 搜索评估不同路径。计算成本高,但对难题成功率显著提升。下一篇 Reflection 和第 06 篇 Tree of Thoughts 会涉及类似思路。

LangGraph 里的 Plan-Execute

如果你用 LangGraph,上面的逻辑基本就是它的 StateGraph 模式。官方甚至提供了 create_planner_worker 这样的 helper。LangGraph 把"状态"和"节点之间的流转"做成了显式的图结构:

from langgraph.graph import StateGraph, END

def planner_node(state): ...
def executor_node(state): ...
def should_continue(state) -> str:
    if all_done(state): return END
    if should_replan(state): return "planner"
    return "executor"

graph = StateGraph(State)
graph.add_node("planner", planner_node)
graph.add_node("executor", executor_node)
graph.add_edge("planner", "executor")
graph.add_conditional_edges("executor", should_continue, {
    "planner": "planner", "executor": "executor", END: END,
})

对比我们上面手写的版本,LangGraph 的价值在于:状态流转变成声明式的图,调试时能可视化地看到每次从哪个节点跳到哪个节点。对长任务是决定性的可观测性优势。第 18 篇会系统对比 LangGraph 和其他框架。

什么时候用 Plan-Execute,什么时候不用

不是所有任务都适合。经验:

用 Plan-Execute:任务超过 5 步;步骤之间有明确依赖;有明显的"整体策略"需要固定;需要展示进度给用户。

用纯 ReAct:任务 1~3 步就完;每一步的结果决定下一步做什么(高分支);对话式、探索式任务,不适合预先规划。

一个实用的混合模式:用 Plan-Execute 做主干,用 ReAct 做步内。例如 Plan 里的某一步是"调研竞品"——这个步内部需要多轮搜索和筛选,那就让 Executor 在执行这一步时跑一个小 ReAct 循环,做完返回结果。这是 Claude Code、Cursor 这类工具实际采用的架构。

下一步

Plan-Execute 解决了"全局目标钉住"。但它有个弱点——Planner 出的计划不一定对,Executor 按计划执行不代表结果合格。

下一篇 Reflection 就是补这个短板:让 Agent 执行完一个步骤或整个任务后,自己审视结果、挑毛病、重做。这是把 Agent 从"能跑完"推到"跑得好"的关键一步。