本篇约 1.2 万字阅读时长约 30 分钟。从 ReAct 的固有缺陷切入拆解 LangGraph 上 Plan-and-Execute Agent 的完整实现涵盖状态设计、Schema 定义、三大节点深度剖析、Provider 路由与 REPL 流式输出。在 AI Agent 落地的过程中ReAct推理-行动模式是大家最熟悉的套路模型先想一想再调个工具拿到结果再想循环往复。但问题来了——当任务复杂、链路变长时ReAct的 LLM 每一步都要重新审视全局上下文再做决策token 消耗大中间步骤容易跑偏。举个例子你让 Agent 去查2024年澳网男单冠军的家乡在哪。ReAct的做法是先搜索冠军是谁再搜索这个人的家乡。听起来没毛病但每一步推理都带着完整历史消息模型在第二步可能因为上下文太长而忘了第一步搜到的名字或者干脆自作主张用参数知识直接回答。那么有没有一种更靠谱的思路有就是Plan-and-Execute先规划再执行。本篇就基于LangGraph从零拆解一套完整的 Plan-and-Execute Agent 源码涵盖状态设计、Schema 定义、节点逻辑、图编排以及实际运行中的踩坑经验。项目源码:https://github.com/stefanxfy/LangGraph-Learning/tree/master/react-agent-plan-and-execute一、ReAct 的瓶颈与 Plan-and-Execute 的破局要理解 Plan-and-Execute先得搞清楚它在解决ReAct的什么毛病。1、ReAct 模式的固有缺陷ReAct的核心循环是 Think - Act - Observe模型每走一步都要把完整的历史消息重新过一遍。这带来三个问题token 爆炸每一步都把之前的对话历史塞进 prompt步数越多上下文越长成本线性增长决策漂移模型在长链路中容易丢失初始目标越走越偏无法并行因为是边想边做前一步的结果决定了下一步的方向串行依赖无法打破用一句定性结论来说ReAct是个优秀的执行者但不是个好规划者。2、Plan-and-Execute 的核心思路Plan-and-Execute 的设计哲学很朴素先想清楚再动手。它把 Agent 的职责拆成三个角色Planner军师拿到用户目标一次性生成完整的步骤列表运筹帷幄Executor先锋严格按步骤执行每做完一步汇报结果冲锋陷阵Replanner裁判审视执行结果决定是继续执行剩余步骤还是直接给出最终答案决定继续还是收兵三位角色各司其职形成闭环。3、两种模式对比对比维度ReActPlan-and-Execute决策方式边想边做逐步推理先规划全局再逐步执行token 消耗高每步重载全量上下文低执行时只看当前步骤错误纠正只能通过观察被动纠正Replanner 可主动调整计划并行能力差串行依赖强好步骤间可并行设计适用场景简单任务、短链路复杂任务、多步骤组合鱼和熊掌不可兼得ReAct胜在灵活Plan-and-Execute 胜在可控。二、整体架构与流程图那么这套 Plan-and-Execute 的图结构长什么样先用 mermaid 看一下节点连线关系的骨架核心链路就一条线START - planner - agent - replan然后replan通过should_end条件边决定是回到agent还是走向END。但光看连线骨架还不够。三个角色各自读写 State 的哪些字段should_end怎么路由用一张全景图把全貌画出来这张图把三件事讲清楚了节点的执行顺序、should_end的路由分支、以及PlanExecuteState 在各节点间的读写关系。这里有一个关键设计每次只执行plan[0]而不是一次性跑完所有步骤。为什么因为replan节点需要在每一步执行后审视全局可能调整后续计划。这种执行一步 - 回顾一次的节奏保证了灵活性。个人认为这种设计在工程上是很务实的——既不像ReAct那样每步都要从头推理也不像批处理那样无法动态调整。三、状态设计与 Schema 定义图有了节点之间的数据怎么流转靠的是State。1、PlanExecute 状态结构LangGraph的核心概念之一就是State状态它是图中所有节点共享的数据容器。来看PlanExecute的定义class PlanExecute(TypedDict): The plan-and-execute agents state. input: str plan: list[str] past_steps: Annotated[list[tuple[str, str]], operator.add] response: str四个字段职责分明input用户的原始输入plan当前待执行的步骤列表past_steps已完成的步骤及结果用operator.add做 reducerresponse最终答案这里最值得关注的是past_steps的注解past_steps: Annotated[list[tuple[str, str]], operator.add]Annotatedoperator.add是LangGraph的reducer 机制。普通字段在节点更新时会被直接覆盖而加了operator.add的字段会做累加追加永远不会被覆盖。用 ASCII 图来对比一下两种更新方式的区别普通字段覆盖模式┌─────────────────────────────────────────┐│ 节点A: past [step1] ││ 节点B: past [step2] ← 直接覆盖 ││ 最终: past [step2] │└─────────────────────────────────────────┘reducer operator.add追加模式┌─────────────────────────────────────────┐│ 节点A: past [step1] ││ 节点B: past [step2] ← operator.add ││ 最终: past [step1, step2] │└─────────────────────────────────────────┘这意味着每次agent节点执行完一步结果会 append 到past_steps里而不会把之前的记录冲掉。这是有必要的——replanner需要看到所有已完成的步骤才能做出判断。2、Plan Schemaclass Plan(BaseModel): Plan to follow in future. steps: list[str] Field( descriptiondifferent steps to follow, should be in sorted order )Plan很简单就是一个字符串列表。planner节点的输出会被结构化为这个 Schema确保每一步都是一个清晰的任务描述。3、Act Schema——一个踩坑点Act是replanner的输出 Schema它需要表达两种可能要么返回最终答案要么返回更新后的计划。先看 notebook 原版的设计# notebook 原版用 Unionaction: Union[Response, Plan]看起来很优雅但实际跑起来会出问题。当对接 GLM/MiniMax 等 OpenAI 兼容的 Provider 时langchain的解析器会把 Union 字段展平和模型返回的嵌套结构{action: {...}}对不上导致解析失败。这算是踩了个大坑。所以本项目把它展平成了Literal判别符 两个可选字段class Act(BaseModel): Action to perform (replanner output). action_type: Literal[response, plan] Field( descriptionUse response to answer the user, plan to keep working. ) response: str | None Field( defaultNone, descriptionFinal answer to the user; set when action_typeresponse., ) steps: list[str] | None Field( defaultNone, descriptionRemaining steps to do; set when action_typeplan., )用action_type做判别符response和steps根据类型填充。没有嵌套结构兼容性直接拉满。这是一个务实的选择——牺牲了一点类型层面的优雅换来了跨 Provider 的稳定性。四、三大节点深度剖析有了 State 和 Schema接下来看图中的三个核心节点。每个节点都是一个工厂函数返回一个接收PlanExecute状态、返回状态更新的闭包。1、planner 节点军师出招def make_plan_step(planner) - Callable[[PlanExecute], dict]: def plan_step(state: PlanExecute) - dict: plan planner.invoke({messages: [(user, state[input])]}) steps plan.steps if (plan and plan.steps) else [state[input]] return {plan: steps} return plan_step逻辑很直白把用户输入丢给planner拿到Plan结构提取steps返回。但有一行防御代码值得注意steps plan.steps if (plan and plan.steps) else [state[input]]如果模型没有返回有效的steps比如直接用自然语言回答了就把整个输入当作一个步骤。这样即使 planner 罢工了executor 也还能跑起来不至于直接崩。这是工程上的兜底策略——别让一个节点的异常拖垮整条链路。2、execute_step 节点先锋冲锋——本篇最核心的设计取舍这个节点是整个项目最有深度的部分也是踩坑最多的地方。当局者迷旁观者清——每一步都看不全大局的 executor先看完整代码def make_execute_step(agent_executor) - Callable[[PlanExecute], dict]: def execute_step(state: PlanExecute) - dict: plan state[plan] plan_str \n.join(f{i 1}. {step} for i, step in enumerate(plan)) task plan[0] past state.get(past_steps) or [] past_str ( \n.join(f- {step}: {result} for step, result in past) or (none yet) ) task_formatted ( fFor the following plan:\n{plan_str}\n\n fYou are tasked with executing step 1, {task}. ) agent_response agent_executor.invoke({messages: [(user, task_formatted)]}) return {past_steps: [(task, agent_response[messages][-1].content)]} return execute_step先说正常流程取plan[0]格式化成任务描述丢给agent_executor一个预构建的ReActAgent执行把结果 append 到past_steps。但注意看task_formatted的构造——它只带了plan_str和当前task没有带past_str。代码里其实算出了past_str但最终没用上。为什么execute_step 的上下文传递设计——全文最关键的取舍这要分两种场景来分析。简单任务场景比如查询2024年澳网男单冠军的家乡。planner生成的计划是查 2024 年澳网男单冠军是谁查这个人的家乡第一步执行后replanner拿到结果Jannik Sinner可以直接用 LLM 自身知识回答他的家乡不需要执行第二步。这种情况下不带历史上下文反而更省 token。复杂任务场景如果任务是多步骤深度关联的比如查某公司最近的财报数据根据财报数据计算关键指标根据指标生成投资建议第二步依赖第一步的数值结果第三步依赖第二步的计算结果。如果execute_step不带历史上下文第二步就不知道第一步算出了什么。那么会怎样下一步的执行就永远得不到正确答案。整个流程会陷入 step-replan 的死循环对系统稳定性是毁灭性打击。解决方案在代码注释里写得很清楚——把past_str加进task_formatted# 携带历史 step 上下文的版本task_formatted ( fFor the following plan:\n{plan_str}\n\n fSteps already completed and their results:\n{past_str}\n\n fYou are tasked with executing step 1, {task}.\n fReuse the results above when they are relevant — do not redo fwork that is already done.)但本项目最终选择不带上下文注释说明了原因# task_formatted 不携带历史 step 上下文并不是省 token 的好办法# 而是对简单任务的取舍实际复杂任务每一个 step 是需要更多历史 step 的上下文的个人认为这个取舍的核心在于简单任务的演示价值在于展示 Plan-and-Execute 的完整闭环流程而不是追求复杂场景的生产级可用。复杂任务需要的不仅仅是past_str而是更完善的上下文共享机制比如把中间结构化结果存入 State而不只是文本摘要。用 ASCII 图把 execute_step 的上下文流转画出来两种模式的对比更直观┌──────────── 不带 past_str默认────────────┐│ ││ execute_step 调用 agent_executor 时 ││ ││ task_formatted plan_str task ││ ┌──────┐ ││ │ agent│ ← 看不到 step1 结果 ││ └──────┘ ││ 结果step2 不知道 step1 产出了什么 ││ → 简单任务 OK复杂任务死循环 ││ │├──────────── 带 past_str改进版─────────────┤│ ││ execute_step 调用 agent_executor 时 ││ ││ task_formatted plan_str past_str task ││ ┌──────┐ ││ │ agent│ ← step1 结果在手 ││ └──────┘ ││ 结果step2 能复用 step1 的输出 ││ → 复杂任务也能跑通 ││ │└────────────────────────────────────────────────┘3、replan 节点裁判定夺def make_replan_step(replanner) - Callable[[PlanExecute], dict]: def replan_step(state: PlanExecute) - dict: output: Act replanner.invoke(state) if output.action_type response: return {response: output.response} return {plan: output.steps or []} return replan_stepreplanner拿到当前完整 State包含input、plan、past_steps输出Act结构。如果action_type response把最终答案写入response字段否则用output.steps更新plan剩余步骤注意output.steps or []这个写法——如果模型返回了plan类型但steps为空就用空列表兜底避免plan[0]越界。4、should_end路由条件def should_end(state: PlanExecute) - str: Route: stop when a response exists, otherwise keep executing. if state.get(response): return END return agent一句话有response就结束没有就回agent继续干。简单粗暴但有效。五、图编排与编译——build_app 的依赖注入设计三个节点都有了接下来就是把它们组装成图。核心是build_app函数def build_app( *, model: BaseChatModel | None None, tools: list | None None, planner: Runnable | None None, replanner: Runnable | None None, agent_executor: Any | None None, checkpointer: Any | None None,) - CompiledStateGraph:每一个参数都是可选注入的。留None的角色会从model和tools自动推导。为什么这么设计因为测试需要。测试时可以注入 fake 的planner、replanner、agent_executor完全不需要真实的 LLM Provider就能跑通整个图的逻辑。这个依赖注入整得挺靠谱。来看核心装配逻辑# 只有角色还没构建时才需要 model所以注入了全部角色的测试不需要 Provider Key if planner is None or replanner is None or agent_executor is None: if model is None: model make_default_model() if planner is None: planner planner_prompt | model.with_structured_output( Plan, methodfunction_calling ) if replanner is None: replanner replanner_prompt | model.with_structured_output( Act, methodfunction_calling ) if agent_executor is None: agent_executor create_react_agent( model, tools, promptYou are a helpful assistant. )这里有一个关键细节methodfunction_calling。为什么不用json_schema因为 GLM 和 MiniMax 的json_schemaresponse_format 不靠谱返回的是纯文本而非 JSON。而function_calling方式通过工具调用的结构化返回兼容性更好。接下来是图的组装workflow StateGraph(PlanExecute) workflow.add_node(planner, make_plan_step(planner)) workflow.add_node(agent, make_execute_step(agent_executor)) workflow.add_node(replan, make_replan_step(replanner)) workflow.add_edge(START, planner) workflow.add_edge(planner, agent) workflow.add_edge(agent, replan) workflow.add_conditional_edges(replan, should_end, [agent, END]) return workflow.compile(checkpointercheckpointer)四行add_edge就把整张图画完了START - planner - agent - replanreplan通过条件边should_end路由到agent或ENDcheckpointer默认用InMemorySaver()支持LangGraph的断点续传和状态回溯。六、Provider 配置与工具体系1、Provider 路由一个字典搞定多模型切换本项目支持 GLM 和 MiniMax 两种 Provider切换方式是通过环境变量LLM_PROVIDER控制。核心设计如下dataclass(frozenTrue)class Provider: key_env: str # API Key 的环境变量名 model_env: str # 模型名覆盖的环境变量名 model_default: str # 默认模型名 base_url_env: str # Base URL 覆盖的环境变量名 base_url_default: str # 默认 Base URL两个 Provider 的配置对照Providerkey_envmodel_defaultbase_url_defaultglmZHIPUAI_API_KEYGLM-5.1https://open.bigmodel.cn/api/coding/paas/v4minimaxMINIMAX_API_KEYMinMax-M3https://api.minimaxi.com/v1这个设计的精妙之处在于开闭原则要加一个新 Provider只需要在PROVIDERS字典里加一条记录路由逻辑一行都不用改。实际使用时.env配置示例如下# .env 文件LLM_PROVIDERglmZHIPUAI_API_KEYyour_api_key_here# 可选覆盖默认模型# GLM_MODELGLM-5.1# 可选覆盖默认 Base URL# GLM_BASE_URLhttps://open.bigmodel.cn/api/coding/paas/v4# 如果想切到 MiniMax# LLM_PROVIDERminimax# MINIMAX_API_KEYyour_minimax_keymake_default_model的路由逻辑也很干脆def make_default_model() - ChatOpenAI: provider os.getenv(LLM_PROVIDER, DEFAULT_PROVIDER).lower() if provider not in PROVIDERS: sys.stderr.write( ferror: unknown LLM_PROVIDER{provider}. fChoose one of: {, .join(sorted(PROVIDERS))}.\n ) sys.exit(2) cfg PROVIDERS[provider] api_key require_env(cfg.key_env) model os.getenv(cfg.model_env, cfg.model_default) base_url os.getenv(cfg.base_url_env, cfg.base_url_default) return ChatOpenAI(modelmodel, api_keyapi_key, base_urlbase_url)找不到 Provider 或者 Key 没配直接sys.exit(2)退出绝不静默降级。这是必须的——配置错误就要第一时间暴露别等到运行到一半才报错。2、工具体系离线模式与在线模式tools.py提供了一个内置的离线搜索引擎内置了一组关键词-答案映射_KB: list[tuple[tuple[str, ...], str]] [ ((australian open, 2024, men, winner), Jannik Sinner won the 2024 Australian Open mens singles title.), ((sinner, hometown), Jannik Sinner is from Sexten (Sesto), in South Tyrol, Italy.), # ...]这个离线知识库的目的是让多步规划端到端跑通不需要任何外部 API Key。如果配置了TAVILY_API_KEY会自动切换到真实的Tavily搜索def get_tools() - list: if os.getenv(TAVILY_API_KEY): from langchain_community.tools.tavily_search import TavilySearchResults return [TavilySearchResults(max_results3)] return [search]两种模式无缝切换对上层图结构完全透明。七、REPL 运行与流式输出最后来看入口文件__main__.py它提供了一个交互式 REPL用stream_modeupdates实时打印每一步的执行过程。1、流式追踪设计def _print_trace(node: str, update: dict) - None: if node planner: plan update.get(plan) or [] print(\n--- plan ---) for i, step in enumerate(plan, 1): print(f{i}. {step}) elif node agent: for task, result in update.get(past_steps) or []: print(f\n▶ executing: {task}) print(f - {result}) elif node replan: if update.get(response) is not None: print(\n--- replan: done (final answer) ---) print(update[response]) else: remaining update.get(plan) or [] print(\n--- replan: continue, remaining plan ---) for i, step in enumerate(remaining, 1): print(f{i}. {step})stream_modeupdates的机制是每跑完一个节点立刻推送该节点的状态更新。这样计划、执行步骤、重规划决策都是实时打印的而不是等全部跑完一次性 dump。2、REPL 主循环def repl() - None: load_env() app build_app() while True: try: user_input input(\n ).strip() except (EOFError, KeyboardInterrupt): print(\nbye) return if user_input :quit: return thread_id str(uuid.uuid4()) config {configurable: {thread_id: thread_id}, recursion_limit: 50} for chunk in app.stream({input: user_input}, config, stream_modeupdates): for node, update in chunk.items(): _print_trace(node, update)每次查询都是独立的thread_id保证 plan、past_steps、response 不会跨查询泄漏。recursion_limit设为 50因为 plan-execute-replan 的循环可能跑好几轮默认的 25 可能不够用。REPL 还支持Ctrl-DEOF和Ctrl-C键盘中断优雅退出输入:quit也能直接退出。代码里用try/except (EOFError, KeyboardInterrupt)捕获这两种中断信号保证不会因为用户的误操作而抛出一堆 traceback。3、实际运行效果以查询who won the 2024 Australian Open men’s title and where is that person from为例运行时输出大致如下--- plan ---1. Identify the winner of the 2024 Australian Open mens singles title.2. Research the hometown/birthplace of that winner.▶ executing: Identify the winner of the 2024 Australian Open mens singles title. - Jannik Sinner won the 2024 Australian Open mens singles title.--- replan: continue, remaining plan ---1. Research the hometown/birthplace of that winner.▶ executing: Research the hometown/birthplace of that winner. - Jannik Sinner is from Sexten (Sesto), in South Tyrol, Italy.--- replan: done (final answer) ---The winner is Jannik Sinner, from Sexten (Sesto), South Tyrol, Italy.计划 - 执行 - 重规划 - 执行 - 最终答案完整闭环这流程 yyds。八、Prompt 模板planner 与 replanner 的提示词设计1、planner_promptplanner_prompt ChatPromptTemplate.from_messages([ (system, For the given objective, come up with a simple step by step plan. This plan should involve individual tasks, that if executed correctly will yield the correct answer. Do not add any superfluous steps. The result of the final step should be the final answer. Make sure that each step has all the information needed - do not skip steps.), (placeholder, {messages}),])核心要求四点生成逐步计划每步是独立可执行的任务不要多余步骤每步要包含足够信息2、replanner_prompt宽松版 vs 严格版两个版本共享相同的模板变量槽位{input}用户的原始目标{plan}当前计划{past_steps}已完成的步骤及结果区别在于关键约束段不同。先看当前生效的宽松版完整模板replanner_prompt ChatPromptTemplate.from_template( For the given objective, come up with a simple step by step plan. # ...前段与 planner_prompt 一致 Your objective was this:\n{input}\n\n Your original plan was this:\n{plan}\n\n You have currently done the follow steps:\n{past_steps}\n\n Update your plan accordingly. If no more steps are needed and you can return to the user, then respond with that. Otherwise, fill out the plan. Only add steps to the plan that still NEED to be done. Do not return previously done steps as part of the plan.\n\n Return action_typeresponse (with the final answer in response) when you can answer the user, otherwise action_typeplan (with the remaining steps in steps).)再看注释掉的严格版关键差异段如下# 严格版注释中的差异段Prefer executing the remaining steps over answering early: return a final response only when the objective is fully covered by tool results already in past_steps. Do NOT short-circuit an unexecuted step by answering from your own knowledge...the steps field must contain ONLY the original-plan steps that have not yet been executed, kept VERBATIM and in their original order — do not rephrase, merge, split, or add new steps.两个版本的核心差异一目了然策略维度宽松版默认严格版提前回答允许如果认为可以回答就直接出最终结果禁止除非目标已被工具结果完全覆盖步骤改写允许调整剩余步骤的措辞禁止只允许删除已完成步骤剩余步骤必须原样保留代词消解replanner 可改写步骤让代词指向明确步骤原样保留代词无法被消解适用场景简单任务、演示闭环模拟复杂任务的上下文缺失问题为什么默认不用严格版因为严格版禁止改写步骤措辞这意味着如果 step2 依赖 step1 的结果比如查那个人的家乡step2 中的代词那个人就永远无法被解析。所以两种版本的选择本质上是宽松版适合简单任务的全流程演示严格版适合验证复杂任务在缺少上下文共享时的问题。这是一个典型的工程权衡——没有银弹。九、要点总结Plan-and-Execute 的核心拓扑是START - planner - agent - replan - (agent | END)每次只执行plan[0]由replanner决定继续还是收兵past_steps使用operator.addreducer 做累加追加保证已完成的步骤记录不会被覆盖ActSchema 从Union[Response, Plan]展平为Literal判别符 可选字段解决 OpenAI 兼容 Provider 的结构化输出解析问题execute_step是否携带历史上下文是核心设计取舍简单任务不带可省 token复杂任务不带会陷入死循环build_app的全参数依赖注入设计让测试可以完全脱离真实 LLM Provider 离线运行methodfunction_calling比json_schema更适合对接 GLM/MiniMax 等国产模型Provider 路由用字典注册实现开闭原则加模型只需加一条配置stream_modeupdates实现节点级实时追踪计划和执行过程逐行打印学AI大模型的正确顺序千万不要搞错了2026年AI风口已来各行各业的AI渗透肉眼可见超多公司要么转型做AI相关产品要么高薪挖AI技术人才机遇直接摆在眼前有往AI方向发展或者本身有后端编程基础的朋友直接冲AI大模型应用开发转岗超合适就算暂时不打算转岗了解大模型、RAG、Prompt、Agent这些热门概念能上手做简单项目也绝对是求职加分王给大家整理了超全最新的AI大模型应用开发学习清单和资料手把手帮你快速入门学习路线:✅大模型基础认知—大模型核心原理、发展历程、主流模型GPT、文心一言等特点解析✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑✅开发基础能力—Python进阶、API接口调用、大模型开发框架LangChain等实操✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经以上6大模块看似清晰好上手实则每个部分都有扎实的核心内容需要吃透我把大模型的学习全流程已经整理好了抓住AI时代风口轻松解锁职业新可能希望大家都能把握机遇实现薪资/职业跃迁这份完整版的大模型 AI 学习资料已经上传CSDN朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】