Plan-and-Execute
React 框架是单步自适应状态机。每一轮循环中，模型都会生成一段思考，然后紧接着生成一个动作，也就是边想边做。 Plan-and-Execute 就是先想后做，是一种多智能体架构。
Planner 将输入目标解析为一个子任务（通常是一个瀑布或者一个 DAG）。 Executor 遍历这个图，它只管输入子任务和必要的 Context，输出 Result。 现在还有一个 Re-planner 在每步结束后，重新输入全局 State，修剪或重构这个 graph。
这种框架在面对长上下文的时候表现往往优于 React 框架，Tokens 消耗、注意力分散以及执行时间都要更优。
它还有一个优势就是单步的失误会被隔离，不会造成整体的错误黑洞。局部错误产生的污染过的上下文并不会影响全局的 Re-planner，Executor 内部也可以增加容错。
模糊不明确、轻量级的任务更适合 React；而明确的研究任务，深度研究更适合 Plan-and-Execute。
一个简单的例子。在 LangGraph 中，对于 Plan-and-Execute，state 状态必须包含“计划蓝图”和“执行足迹”。
from typing import List, Tuple, TypedDict, Optional
import operator
from typing_extensions import Annotated
from pydantic import BaseModel, Field
from langgraph.graph import StateGraph, END
class PlanExecuteState(TypedDict):
input: str
user_id: Optional[str]
plan: List[str] # 任务规划队列
past_steps: Annotated[List[Tuple[str, str]], operator.add] # 追加记录足迹
response: Optional[str] # 最终结果
class StructuredPlan(BaseModel):
steps: List[str] = Field(
description="拆解后的独立执行步骤列表，必须针对具体套餐实体且包含账号核验步骤。"
)
-
Planner Node 输入：用户的复杂 Query（例如：“对比 A 套餐和 B 套餐分别在 2023 年与 2024 年的差异，自己的账号是否还能申请？”）。 Prompt 策略：强制输出 JSON 或 Pydantic 结构。要求它将查询拆解为独立的检索任务。 输出：
["检索 A 套餐 2023、2024 的内容", "检索 B 套餐 2023、2024年的内容", "对比套餐", "检索用户账号权限与状态", "推荐合适的套餐"]。这个列表会被写入 State 的 plan 字段。 -
Executor Node (执行器) 输入：从 plan 中 pop 出第一个任务，结合 past_steps 作为背景。 行为：这通常是一个小型的 ReAct Agent（或者简单的 Tool Calling 模块）。它挂载了向量数据库检索工具、知识图谱查询工具或 API 接口。它只负责把当前这个小任务查清楚。 输出：将执行结果打包成
(当前任务, 执行结果)的元组，追加到 past_steps 中。 -
Re-planner Node (重规划师) 输入：全局 State。 行为：评估 past_steps 里的情报是否已经足够回答 input。 输出逻辑：
- 如果情报足够：生成最终的 response。
- 如果情报不足，且原计划执行完毕：根据新发现的情况，生成全新的 plan。
- 如果情报不足，但原计划还有剩余：保留剩余计划，继续走循环。
def planner_node(state: PlanExecuteState):
"""规划节点：负责把复杂业务 Query 结构化拆解"""
user_query = state["input"]
# 真实业务中：llm.with_structured_output(StructuredPlan).invoke(user_query)
# 此处模拟模型针对特定套餐对比和权限查询拆解出的 5 步静态计划
mock_llm_output = [
"检索 A 套餐 2023、2024 的具体内容与资费标准",
"检索 B 套餐 2023、2024 的具体内容与资费标准",
"对比 A、B 套餐在两年间的核心差异点",
"调用业务 API 检索当前 user_id 的账号权限与在网状态",
"结合套餐对比结果与用户账号状态，生成定制化推荐话术"
]
return {"plan": mock_llm_output}
def executor_node(state: PlanExecuteState):
"""执行节点：消费计划队列，根据任务类型调度异构工具"""
current_plan = state["plan"]
if not current_plan:
return {"past_steps": []}
# 【消费核心】弹出队列中的第一条任务，并把剩余任务传回状态
current_task = current_plan[0]
remaining_plan = current_plan[1:]
user_id = state.get("user_id", "Unknown_User")
# 模拟内部异构工具的调度逻辑
if "API" in current_task or "账号" in current_task:
# 调度内部运营中台或 CRM 系统接口
tool_result = f"【CRM 接口调用结果】用户 {user_id} 当前状态正常，具备老套餐升级特权。"
else:
# 调度知识库检索（RAG 召回）
tool_result = f"【知识库检索结果】已获取到关于 '{current_task}' 的非结构化文档片段。"
return {
"past_steps": [(current_task, tool_result)],
"plan": remaining_plan # 覆盖旧的 plan，强制剔除已做任务，防止死循环
}
def replanner_node(state: PlanExecuteState):
"""重规划节点：评估是否结束，或在环境异常时动态修剪计划"""
# 逻辑 1：如果计划队列已经清空，说明所有步骤执行完毕，生成最终答复
if len(state["plan"]) == 0:
# 真实业务中：基于 state["past_steps"] 的丰富上下文，调用 LLM 生成给用户的话术
final_answer = (
f"根据分析：A 套餐在 24 年降低了流量资费，B 套餐则增加了语音包。\n"
f"核验了您的账号（ID: {state['user_id']}），您完全符合申请条件，建议直接在线升级 A 套餐。"
)
return {"response": final_answer}
# 逻辑 2：如果计划还有剩余，保持原样（不做 response 赋值），继续驱动图走 conditional_edges
return {"plan": state["plan"]}
# ==================== 拓扑图编排与条件路由 ====================
workflow = StateGraph(PlanExecuteState)
# 注册所有业务节点
workflow.add_node("planner", planner_node)
workflow.add_node("executor", executor_node)
workflow.add_node("replan", replanner_node)
# 构筑确定性边
workflow.set_entry_point("planner")
workflow.add_edge("planner", "executor")
workflow.add_edge("executor", "replan")
def route_decision(state: PlanExecuteState):
# 如果 replan 节点认为信息收集完毕并吐出了 response，则终止图
if state.get("response"):
return END
# 计划未完，引导图返回执行器继续消费 plan 队列
return "executor"
# 绑定条件边
workflow.add_conditional_edges(
"replan",
route_decision,
{END: END, "executor": "executor"}
)
app = workflow.compile()
if __name__ == "__main__":
initial_inputs = {
"input": "对比 A 套餐和 B 套餐分别在 2023 年与 2024 年的差异，自己的账号是否还能申请？",
"user_id": "1380XXXXX8000"
}