LangGraph Plan-and-Execute：先规划再执行，别让 Agent 边想边做

个人主页： > for_ever_love__ <（欢迎各位大佬莅临😊）

其他栏目: > 大模型开发从0到1 <

其他栏目: > iOS项目总结大全 <

其他栏目: > 我想学python了 <

其他栏目: > iOS UI <

文章目录

• LangGraph Plan-and-Execute：先规划再执行，别让 Agent 边想边做
• • 一、边想边做 vs 先规划后执行
  • 二、什么时候该用 Plan-and-Execute
  • 三、最小实现
  • • 状态定义
    • 规划节点
    • 执行节点
    • 重规划节点
    • 路由函数
    • 组装图
  • 四、三个必须加的约束
  • 五、和 ReAct 怎么选
  • 六、调试时的观察点
  • 小结

LangGraph Plan-and-Execute：先规划再执行，别让 Agent 边想边做

ReAct 式 Agent 的最大问题不是能力不够，而是它走一步看一步。
遇到"查三个数据源，对比完再出结论"这种任务，
它往往查完第一个就开始编答案了。

Plan-and-Execute 的思路很朴素：先把计划定下来，再按计划执行。

一、边想边做 vs 先规划后执行

用同一个任务对比一下。任务：“对比 A、B 两个方案的成本和性能，给出选型建议。”

ReAct（边想边做）的典型轨迹：

思考：我需要先查 A 方案的成本 调用：search("A 方案 成本") 观察：... 思考：A 方案成本是 X，接下来应该看性能 调用：search("A 方案 性能") 观察：... 思考：我已经了解了 A，可以给出建议了 ← 问题出现：B 还没查 输出：选 A 吧……

问题在于每一步决策只看"上一步的结果"，没有全局视角。
任务越长，跑偏概率越高。

Plan-and-Execute 的轨迹：

规划：1. 查 A 方案成本 2. 查 A 方案性能 3. 查 B 方案成本 4. 查 B 方案性能 5. 对比并给出建议 执行：步骤 1 → 步骤 2 → 步骤 3 → 步骤 4 → 步骤 5 （如果中途发现计划不对，回到规划节点重排）

两者的本质区别：规划是一次性的全局决策，执行是局部的机械动作。

二、什么时候该用 Plan-and-Execute

不是所有场景都适合，别盲目上：

适合	不适合
任务能拆成多个可枚举的子步骤	任务本质是一问一答
步骤之间有依赖（先查再算再写）	步骤之间毫无关系
任务较长（超过 5 步）	一两句话就能完成
需要可审计、可回放	追求最低延迟

代价要说清楚：多了一次规划调用 = 多一次 LLM 往返 = 更慢更贵。
简单问题用它属于浪费。

三、最小实现

LangGraph 官方教程里有个经典实现，我们写一个精简但完整的版本。

状态定义

fromtypingimportAnnotated,TypedDict,Literal,OptionalimportoperatorfrompydanticimportBaseModel,FieldclassPlan(BaseModel):"""规划节点的结构化输出"""steps:list[str]=Field(description="执行步骤，按顺序排列，每步一句话")classPlanExecuteState(TypedDict):input:strplan:list[str]past_steps:Annotated[list[tuple[str,str]],operator.add]response:str

past_steps用operator.add做 reducer，这样并行/多次执行都会追加而不是覆盖。

规划节点

fromlangchain.chat_modelsimportinit_chat_modelfromlangchain_core.messagesimportSystemMessage,HumanMessage planner_llm=init_chat_model("gpt-4o").with_structured_output(Plan)PLANNER_PROMPT="""你是一个任务规划助手。 针对用户的目标，产出一个简单、逐步执行的计划。 计划应当是独立的、可逐步执行的具体动作。 如果不需要额外工具就能回答，直接给最终答案即可。 不要添加多余步骤，结果不要超过 {max_steps} 步。 """asyncdefplan_step(state:PlanExecuteState):plan=awaitplanner_llm.ainvoke([SystemMessage(PLANNER_PROMPT.format(max_steps=8)),HumanMessage(state["input"]),])return{"plan":plan.steps}

执行节点

fromlanggraph.prebuiltimportcreate_react_agent agent=create_react_agent(llm,tools)asyncdefexecute_step(state:PlanExecuteState):task=state["plan"][0]# 把历史步骤也喂给执行器，避免重复劳动history="\n".join(f"{t}=>{r}"fort,rinstate.get("past_steps",[]))prompt=f"""你现在负责执行计划中的一个步骤。 总体目标：{state['input']}已经完成的步骤：{historyor'无'}请你只完成这一个步骤：{task}不要做多余的事，输出这一步的结果。"""result=awaitagent.ainvoke({"messages":[HumanMessage(prompt)]})return{"past_steps":[(task,result["messages"][-1].content)]}

注意past_steps里我们同时存了任务描述和结果，
这样后续步骤能看到前面做了什么，不会重复查同一件事。

重规划节点

执行完一步后，判断是继续还是收尾：

classResponse(BaseModel):response:str=Field(description="给用户的最终回答")classAct(BaseModel):action:Literal["continue","finish"]=Field(description="continue = 还要执行更多步骤；finish = 可以直接回答")classReplanDecision(BaseModel):action:Union[Act,Response]replanner_llm=init_chat_model("gpt-4o").with_structured_output(ReplanDecision)REPLANNER_PROMPT="""回顾用户的总体目标，以及目前已完成的步骤。 判断是继续执行剩余步骤，还是已经可以给出最终答案。 如果需要继续，给出**更新后的完整计划**（保留已完成步骤记录）； 如果可以结束，直接给出最终回答。不要输出多余解释。"""asyncdefreplan_step(state:PlanExecuteState):output=awaitreplanner_llm.ainvoke([SystemMessage(REPLANNER_PROMPT),HumanMessage(f"目标：{state['input']}\n原计划：{state['plan']}\n"f"已完成：{state.get('past_steps',[])}"),])ifisinstance(output.action,Response):return{"response":output.action.response}return{"plan":output.action.steps_to_take}# 用新计划替换旧计划

这里有个容易写错的地方：重规划返回的是"更新后的完整计划"，
它替换掉state["plan"]，而执行节点永远取plan[0]。
所以已完成的那一步必须从新计划里剔除，否则会死循环。

这也是为什么要把past_steps传给重规划器——它得知道哪些做过了。

路由函数

defshould_end(state:PlanExecuteState)->str:return"end"ifstate.get("response")else"continue"deftake_next(state:PlanExecuteState)->str:return"agent"ifstate["plan"]else"end"

组装图

fromlanggraph.graphimportStateGraph,START,END builder=StateGraph(PlanExecuteState)builder.add_node("planner",plan_step)builder.add_node("agent",execute_step)builder.add_node("replan",replan_step)builder.add_edge(START,"planner")builder.add_edge("planner","agent")builder.add_edge("agent","replan")builder.add_conditional_edges("replan",should_end,{"end":END,"continue":"agent"},)graph=builder.compile()

跑起来：

asyncforeventingraph.astream({"input":"查一下北京和上海今天的天气，哪个更适合户外跑步？"},stream_mode="updates",):print(event)

四、三个必须加的约束

裸跑上面的图，大概率会遇到无限循环。加这三条：

1. 最大步数截断

classPlanExecuteState(TypedDict):...step_count:intasyncdefexecute_step(state):...return{"past_steps":[...],"step_count":state.get("step_count",0)+1}defshould_end(state):ifstate.get("response"):return"end"ifstate.get("step_count",0)>=10:# ← 硬上限return"end"ifnotstate.get("plan"):# 计划空了也要停return"end"return"continue"

2. 递归深度上限

graph=builder.compile()result=awaitgraph.ainvoke({"input":"..."},config={"recursion_limit":50},)

LangGraph 默认recursion_limit是 25，超了会抛GraphRecursionError。
这不是 bug，是保护机制，别一味调大。

3. 执行步骤超时

llm=init_chat_model("gpt-4o",timeout=60,max_retries=2)

某一步卡住 5 分钟，整个流程就废了。

五、和 ReAct 怎么选

我的判断标准：

• 任务明确、步骤可枚举、需要多步依赖→ Plan-and-Execute；
• 需要动态探索（不知道该查什么，查了才知道下一步）→ ReAct；
• 两者都不是→ 别上 Agent，直接写个 Chain。

还有个折中方案值得提：只在开头规划一次，之后不重规划。
去掉replan节点，规划完直接顺序执行。
这样省掉每次的规划调用，代价是遇到意外情况没法调整。
对流程稳定的批处理任务，这个折中版本性价比更高。

六、调试时的观察点

流程图跑不起来时，按这个顺序查：

1. 规划输出是不是空列表—— 提示词没约束好，模型觉得不需要规划；
2. 重规划一直返回 continue—— 检查past_steps有没有真的喂进去，
   模型看不到进展就会一直要求继续；
3. 新计划里包含已完成的步骤—— 重规划提示词要明确写"剔除已完成步骤"；
4. 执行节点做了计划外的事—— 执行提示词强调了"只做这一步"吗？

第 2、3 条占了实际问题的八成，重点检查提示词里有没有把past_steps传全。

小结

• Plan-and-Execute = 规划节点 + 执行节点 + 重规划节点 + 条件路由；
• 执行节点每次只做plan[0]，重规划负责更新计划或收尾；
• 必须加最大步数、recursion_limit、超时三重保险；
• 适合长流程、步骤可枚举的任务；简单问答别用；
• 流程稳定的批处理可以去掉重规划，省一次调用。