Plan-and-Execute

规划-执行

Plan-and-Execute Agents

Plan-and-Execute 是一种智能体工作模式：先让模型把复杂目标拆成可执行步骤，再由模型或工具逐步完成这些步骤。

它的核心是把两件事分开：

• Planner：负责想清楚路线，输出步骤、依赖关系和结束条件；
• Executor：负责完成当前步骤，调用工具、读取资料、计算或生成结果；
• Replanning：当执行结果和预期不一致时，根据已有状态更新剩余计划。

基本流程可以写成：

$$\text{Goal}\to \text{Plan}\to \text{Execute}\to \text{Observe}\to \text{Replan}$$

相比让模型一边想一边做，Plan-and-Execute 更适合长任务、多工具任务和需要中间状态管理的任务。它能减少漏步骤、顺序混乱、重复劳动和工具调用不合适的问题。

核心流程

典型流程如下：

1. 用户给出目标。
2. Planner 生成可执行计划。
3. Executor 执行当前步骤。
4. 系统记录执行结果。
5. 判断任务是否完成。
6. 如果计划失效，则 Replanning；否则继续下一步。

关键点不是计划越长越好，而是每一步都应该能被执行器真正完成。

Planner

Planner 不负责解决细节，而是把目标拆成合理步骤。

好的计划通常满足三点：

• 步骤不能太粗，否则执行器不知道怎么下手；
• 步骤不能太碎，否则系统会变慢，上下文也会膨胀；
• 每一步最好有明确输入和输出，方便检查执行结果。

例如：

步骤 1：从文档中提取主要实验设置。
输出：实验设置摘要。

步骤 2：从文档中提取主要结果。
输出：结果摘要。

步骤 3：合并实验设置和结果，写出论文贡献。
输出：贡献总结。

Executor

Executor 只负责当前步骤，不应该重新规划整个任务。

它关心的是：

给定当前步骤和已有上下文，我应该怎么完成这一步？

如果当前步骤是计算实验结果，就调用计算工具或直接计算；如果当前步骤是读取网页，就调用浏览器或检索工具；如果当前步骤是修改代码，就进入代码编辑和验证流程。

执行器的职责边界要窄，否则系统容易变成“每一步都重新开始想一遍”，导致行为混乱。

Replanning

计划不是一次写完就永远正确。执行过程中常见的重规划触发条件包括：

• 工具调用失败；
• 页面、文件或接口不存在；
• 中间结果和预期不一致；
• 用户目标被进一步澄清；
• 已完成步骤说明原计划不再合适。

重规划不是从头乱来，而是基于当前状态更新剩余计划：

$$\text{New Plan}=\mathrm{Planner}(\text{Goal},\text{Old Plan},\text{Execution History})$$

Execution History 应该包含已经完成的步骤、工具返回结果、失败信息和当前剩余任务。

适用场景

Plan-and-Execute 适合这些任务：

• 目标比较大，需要拆成多个子任务；
• 步骤之间有明显依赖关系；
• 需要调用搜索、浏览器、数据库、代码执行器等多个工具；
• 需要产出结构化结果，例如表格、报告、代码修改、实验分析；
• 中间结果可能影响后续步骤，需要反馈和调整。

典型例子：

调研某个技术方向，阅读资料，提炼观点，生成对比表。

把一个数据分析任务拆成数据清洗、统计分析、图表生成和结论总结。

不适用场景

如果任务一步就能回答，Plan-and-Execute 会显得笨重。

例如：

解释一下 sigmoid 是什么。

这类问题直接回答即可。

另外，如果计划器本身不可靠，生成了错误路线，执行器机械照做会放大错误。因此复杂系统里通常还需要结果检查、失败重试和重规划机制。

和相关范式的区别

范式	重点	和 Plan-and-Execute 的区别
Chain-of-Thought	让模型逐步推理	更偏模型内部思考，不一定形成外部可执行计划
ReAct	Thought-Action-Observation 循环	更适合频繁环境交互，边观察边决定下一步
Plan-and-Solve	先生成计划，再按计划解题	更像提示策略，主要服务多步推理题
Plan-and-Execute	计划与执行分离	更像智能体架构，可接入工具和外部系统

实际系统里这些范式常混合使用。例如 Planner 给出大方向，Executor 内部用 ReAct 风格调用工具并处理观察结果。

常见问题

问题	表现	处理方式
计划太粗	例如“完成整个项目”	拆成有输入输出的阶段
计划太碎	例如“读第一句、读第二句”	合并成有意义的工作单元
执行器越权	当前步骤没做完，又开始重新规划	提示执行器只完成当前步骤
缺少检查	错误结果一路传递	每步后判断是否完成、是否可用
不会重规划	工具失败后继续原计划	根据失败信息更新剩余步骤

官方实现

Python

from typing import Annotated, TypedDict, Union
import operator

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_community.tools.tavily_search import TavilySearchResults
from langgraph.graph import END, StateGraph
from langgraph.prebuilt import create_react_agent
from pydantic import BaseModel, Field


class Plan(BaseModel):
    steps: list[str] = Field(
        description="Steps to follow, sorted in execution order."
    )


class Response(BaseModel):
    response: str


class Act(BaseModel):
    action: Union[Plan, Response]


class PlanExecute(TypedDict):
    input: str
    plan: list[str]
    past_steps: Annotated[list[tuple[str, str]], operator.add]
    response: str


llm = ChatOpenAI(model="gpt-4o")
tools = [TavilySearchResults(max_results=3)]
agent_executor = create_react_agent(llm, tools)

planner_prompt = ChatPromptTemplate.from_template(
    """For the given objective, create a simple step-by-step plan.
Do not add unnecessary steps.
The final step should produce the final answer.

Objective: {objective}"""
)

replanner_prompt = ChatPromptTemplate.from_template(
    """Update the plan using the execution history.
If the task is complete, return a final response.
Otherwise, return only the remaining steps.

Objective: {input}
Original plan: {plan}
Completed steps: {past_steps}"""
)

planner = planner_prompt | llm.with_structured_output(Plan)
replanner = replanner_prompt | llm.with_structured_output(Act)


async def plan_step(state: PlanExecute):
    plan = await planner.ainvoke({"objective": state["input"]})
    return {"plan": plan.steps}


async def execute_step(state: PlanExecute):
    task = state["plan"][0]
    result = await agent_executor.ainvoke(
        {"messages": [("user", task)]}
    )
    output = result["messages"][-1].content
    return {"past_steps": [(task, output)]}


async def replan_step(state: PlanExecute):
    result = await replanner.ainvoke(state)
    if isinstance(result.action, Response):
        return {"response": result.action.response}
    return {"plan": result.action.steps}


def should_end(state: PlanExecute):
    return END if state.get("response") else "agent"


workflow = StateGraph(PlanExecute)
workflow.add_node("planner", plan_step)
workflow.add_node("agent", execute_step)
workflow.add_node("replan", replan_step)

workflow.set_entry_point("planner")
workflow.add_edge("planner", "agent")
workflow.add_edge("agent", "replan")
workflow.add_conditional_edges("replan", should_end, ["agent", END])

app = workflow.compile()

result = await app.ainvoke({
    "input": "Research the current weather in San Francisco and write a short summary."
})

TypeScript

import { z } from "zod";
import { ChatOpenAI } from "@langchain/openai";
import { HumanMessage } from "@langchain/core/messages";
import { ChatPromptTemplate } from "@langchain/core/prompts";
import { TavilySearch } from "@langchain/tavily";
import {
  Annotation,
  END,
  START,
  StateGraph,
} from "@langchain/langgraph";
import { createReactAgent } from "@langchain/langgraph/prebuilt";

const PlanExecuteState = Annotation.Root({
  input: Annotation<string>({
    reducer: (x, y) => y ?? x ?? "",
  }),
  plan: Annotation<string[]>({
    reducer: (x, y) => y ?? x ?? [],
  }),
  pastSteps: Annotation<[string, string][]>({
    reducer: (x, y) => x.concat(y),
    default: () => [],
  }),
  response: Annotation<string>({
    reducer: (x, y) => y ?? x ?? "",
  }),
});

const llm = new ChatOpenAI({ model: "gpt-4o" });
const tools = [new TavilySearch({ maxResults: 3 })];
const agentExecutor = createReactAgent({ llm, tools });

const planSchema = z.object({
  steps: z.array(z.string()).describe("Steps in execution order."),
});

const actSchema = z.object({
  action: z.union([
    z.object({ steps: z.array(z.string()) }),
    z.object({ response: z.string() }),
  ]),
});

const plannerPrompt = ChatPromptTemplate.fromTemplate(
  `For the given objective, create a simple step-by-step plan.
Do not add unnecessary steps.
The final step should produce the final answer.

Objective: {objective}`,
);

const replannerPrompt = ChatPromptTemplate.fromTemplate(
  `Update the plan using the execution history.
If the task is complete, return a final response.
Otherwise, return only the remaining steps.

Objective: {input}
Original plan: {plan}
Completed steps: {pastSteps}`,
);

const planner = plannerPrompt.pipe(llm.withStructuredOutput(planSchema));
const replanner = replannerPrompt.pipe(llm.withStructuredOutput(actSchema));

async function planStep(state: typeof PlanExecuteState.State) {
  const plan = await planner.invoke({ objective: state.input });
  return { plan: plan.steps };
}

async function executeStep(state: typeof PlanExecuteState.State) {
  const task = state.plan[0];
  const result = await agentExecutor.invoke({
    messages: [new HumanMessage(task)],
  });
  const output = result.messages.at(-1)?.content?.toString() ?? "";
  return { pastSteps: [[task, output] as [string, string]] };
}

async function replanStep(state: typeof PlanExecuteState.State) {
  const result = await replanner.invoke(state);
  if ("response" in result.action) {
    return { response: result.action.response };
  }
  return { plan: result.action.steps };
}

function shouldEnd(state: typeof PlanExecuteState.State) {
  return state.response ? END : "agent";
}

const workflow = new StateGraph(PlanExecuteState)
  .addNode("planner", planStep)
  .addNode("agent", executeStep)
  .addNode("replan", replanStep)
  .addEdge(START, "planner")
  .addEdge("planner", "agent")
  .addEdge("agent", "replan")
  .addConditionalEdges("replan", shouldEnd, ["agent", END]);

const app = workflow.compile();

const result = await app.invoke({
  input:
    "Research the current weather in San Francisco and write a short summary.",
});