企业官网建设流程全解析

LangGraph 6. 规划 Planning（附完整免费源代码）

做 AI Agent 应用开发的人，这两年几乎绕不开 LangGraph。框架本身不复杂，核心就三件事：State、Node、Edge，但真正让我觉得“这个框架设计得有点东西”的，是它把 Agent 里最难落地的部分——规划（Planning）——给做成了可编排的图结构。

先把这个项目讲清楚：这是一份基于 LangGraph 实现的“规划-执行”框架源码，内置了任务拆解、动态重规划、工具调用和错误恢复能力，目的是让大模型不只“会聊天”，而是能真正把一个复杂目标拆成步骤、按步骤执行、执行偏了还能自己纠偏。适合已经在用 LangChain 但还没上手 LangGraph 的人，也适合那些用单轮 Prompt 硬撑多步任务、急需把流程变成可维护代码的开发者。

先给一个结论：LangGraph 里的规划，不是一个“调一个大模型让它想三步”那么简单，它本质上是在图结构里实现“思考、行动、观察、再思考”的循环。这个循环跑得顺不顺，直接决定你的 Agent 是真能干活，还是只能表演。

1. 内容整体设计与思路拆解

1.1 LangGraph 的图和规划有什么关系

很多人第一次看 LangGraph 的文档，会被“图”这个抽象概念劝退。其实你把它当成一张流程图就完事了。

传统写 Agent 的方式是一个大 while 循环：把历史对话和任务丢给模型，模型返回一个动作，你执行这个动作，再把结果拼回对话里，继续跑。这种写法没有结构，逻辑全堆在一个函数里，一旦任务复杂起来，加一个分支、加一个容错，代码就乱成一锅粥。

LangGraph 换了个思路：把思考、执行、检查这三件事拆成独立的节点，节点之间用边连接。图一跑起来，数据从起点沿着边流动，每个节点处理完自己的事，把结果写进一个共享的 State 里，下一个节点接着读。

规划在这个图里是什么位置？它是一组节点。一组专门负责“想”的节点。

你把“规划”放进图里，它就不再是模型的一个临时输出，而是成了流程中的一个可暂停、可回退、可重复执行的环节。这就是 LangGraph 和普通循环写法最大的区别：规划不是一个动作，而是一个状态机中的合法状态。

1.2 三种经典规划模式，我为什么推荐 Plan-Execute

规划听起来是个很玄的词，但落到工程上，其实只有三种主流模式：

第一种是单轮规划，让模型一次性输出完整计划，然后闷头执行。适合任务步骤明确、环境稳定、不需要中途调整的场景。缺点很明显——计划一旦错了，后面全错。

第二种是 ReAct 模式，也就是“推理-行动-观察”循环。模型每一步先想“我现在该做什么”，做完看结果，再想下一步。灵活，但缺点是太碎，复杂任务会让模型迷失在细节里，上下文很快被撑爆。

第三种是 Plan-Execute，也就是“先计划，再执行，执行偏了重新计划”。这个模式比 ReAct 多了一层抽象：先让模型站在全局视角列出计划，然后进入执行循环，每一步对照计划做，一旦发现计划不合理，就跳回规划节点重新规划。

我在这份源码里选的就是 Plan-Execute。原因不复杂：对我们大多数人遇到的真实业务场景——写报告、查数据、做分析、操作多个工具——Plan-Execute 既保持了全局性，又保留了纠偏能力，而且对 Token 的消耗控制得比 ReAct 好得多。

打个比方：ReAct 是走一步看一步，Plan-Execute 是先看地图再迈步，中途发现路不对再回头看地图。

1.3 这套代码的设计目标与取舍

做这个规划模块的时候，我给自己定了四个硬指标：

• 规划必须可见：用户能在任何时刻看到当前计划长什么样，而不是黑盒。
• 执行必须可控：每个工具调用都要有明确的前置校验和后置校验。
• 重规划必须有触发条件：不能随意跳回规划节点，要有一套判断逻辑。
• 错误必须能恢复：单次工具调用失败不能让整个流程死掉。

围绕这四个指标，代码分成了四层：状态定义层、规划节点层、执行节点层、路由控制层。后面每一层的实现细节，我都会在源码解析里带你过一遍。

2. 核心细节解析与实操要点

2.1 状态的建模：State 是所有规划的根基

LangGraph 的所有节点都靠 State 传递数据，State 怎么设计，直接决定你后面写节点的时候是舒服还是痛苦。

我在这份源码里定义了这样一个 State：

from typing import Annotated, TypedDict, List from langgraph.graph.message import add_messages class PlanExecuteState(TypedDict): user_request: str # 用户的原始目标 plan: List[str] # 当前计划，每个元素是一个步骤 done_steps: Annotated[List[str], add_messages] # 已完成的步骤日志 past_steps: List[str] # 历史执行轨迹 current_step: str # 正在执行的步骤 error: str # 最近一次的错误信息 final_response: str # 最终回复

几个关键设计点，逐个说：

done_steps用add_messages这个归约器，它的作用是让这个字段在多次写入时自动追加而不是覆盖。LangGraph 里的所有状态字段默认是“后写覆盖前写”，但日志类字段我们需要的是累积，所以必须标注归约器。这一步如果不注意，你会发现执行完三步之后，日志只剩最后一步。

plan设计成List[str]的列表还有一个好处——规划节点生成计划之后，渲染给用户、执行节点读取、重规划时比对，全程都是同一个结构，不搞多种类型换来换去。

2.2 节点拆分：规划器、执行器、检查器各干各的

很多初学者写 LangGraph 节点，喜欢把一个节点里塞一大段逻辑，这是大忌。

图结构的意义就在于“分离”，每个节点应该只有一个职责。我的源码里拆了三个核心节点：

规划器节点（Planner）：接受用户请求，调用大模型生成计划，把计划写入plan字段。这个节点只管“想”，不管“做”。

执行器节点（Executor）：从计划里取出当前步骤，调用对应的工具执行。执行完把结果写入past_steps。这个节点只管“做”，不管“想”。

检查器节点（Reflector）：对比执行结果和原始计划，判断当前计划是否仍然成立。如果成立，流程进入下一步；如果不成立，流程跳回规划器重新计划。

这三个节点的关系，用 LangGraph 的 API 表达出来非常清晰：

from langgraph.graph import StateGraph, START, END graph = StateGraph(PlanExecuteState) graph.add_node("planner", planner_node) graph.add_node("executor", executor_node) graph.add_node("reflector", reflector_node) graph.add_edge(START, "planner") graph.add_edge("planner", "executor") graph.add_edge("executor", "reflector") graph.add_conditional_edges( "reflector", decide_next_step, { "continue": "executor", "replan": "planner", "finish": END, } )

看到没？从规划到执行到检查再到重规划，整个循环在图上是显式的，而不是写死在模型 Prompt 里。

2.3 条件边的判断逻辑：重规划不能乱触发

条件边decide_next_step是整套代码里最容易写烂的地方。

最偷懒的写法是：只要执行出错就重规划。这个方案在演示场景下没问题，但在真实业务里会把 Token 烧光，而且会让 Agent 显得“反复无常”——明明某一步只是网络超时，重新试一次就行，非要回炉重造整个计划。

我采用的策略是分级判断，优先级从高到低：

def decide_next_step(state: PlanExecuteState): if state["error"]: # 如果错误信息说“不可恢复错误”，重规划；否则重试当前步骤 if "IRRECOVERABLE" in state["error"]: return "replan" return "retry" if state["current_step"] == state["plan"][-1]: return "finish" if state["step_failures"] >= 3: return "replan" return "continue"

这里多了一个返回值retry，对应一个重试节点。为什么我不用一个节点解决重试和重规划？因为两个操作的代价完全不同：重试只是把同一个步骤再执行一遍，成本远低于全局重规划。把两者区分开，能省很多不必要的模型调用。

3. 完整源代码实现：一套可运行的 Plan-Execute 系统

接下来是重头戏。我先把完整代码贴出来，然后逐个节点拆开讲。代码基于 LangGraph 0.2.x 版本，依赖 LangChain 生态，运行环境是 Python 3.10+。

首先安装依赖：

pip install langgraph langchain langchain-openai

3.1 定义工具集

为了让示例真实一点，我写了一个“数据分析助手”的场景：它能查数据库、能算统计量、能生成报告摘要。三个工具，足以展示多工具协作的规划能力。

from langchain_core.tools import tool @tool def query_database(sql: str) -> str: """执行一条只读 SQL 查询，返回结果字符串。""" # 模拟查询，实际请替换为真实数据库连接 import sqlite3 conn = sqlite3.connect(":memory:") cur = conn.cursor() cur.execute("CREATE TABLE IF NOT EXISTS sales (month TEXT, amount REAL)") # 塞入演示数据 rows = [("2024-01", 12000), ("2024-02", 15500), ("2024-03", 9800)] cur.executemany("INSERT INTO sales VALUES (?, ?)", rows) conn.commit() try: result = cur.execute(sql).fetchall() conn.close() return str(result) except Exception as e: conn.close() return f"Query Error: {e}" @tool def compute_statistics(data: str) -> str: """从查询结果中计算均值、总和、最大值。""" # 简单解析，演示用 numbers = [float(x) for x in data.strip("[]").replace("(", "").replace(")", "").split(",") if x.strip()] if not numbers: return "No data" return f"sum={sum(numbers):.2f}, mean={sum(numbers)/len(numbers):.2f}, max={max(numbers):.2f}" @tool def write_summary(metrics: str) -> str: """将统计指标整理成一段可读的总结。""" return f"销售数据概览：{metrics}。整体表现平稳，建议关注波动较大的月份。"

注意，这里的工具都给了清晰的中文 docstring。这一点非常关键——模型不会读你的代码，它只能读 docstring 来决定什么时候调用哪个工具。docstring 写得模糊，工具调用率会直线下降。

3.2 规划器节点实现

规划器节点是入口，承担两个任务：一是理解用户目标，二是产出可执行的步骤列表。

from langchain_openai import ChatOpenAI model = ChatOpenAI(model="gpt-4o-mini", temperature=0) PLANNER_PROMPT = """你是一个任务规划助手。根据用户的请求，把任务拆解为 2~5 个步骤。 要求： 1. 每个步骤必须清晰具体，能独立执行。 2. 步骤之间要有依赖顺序，前置步骤的结果是后续步骤的输入。 3. 不要重复用户的请求原文，不要输出与执行无关的说明。 4. 严格输出 JSON，格式为：{"steps": ["步骤1", "步骤2", ...]} 用户请求：{request} """ def planner_node(state: PlanExecuteState): resp = model.invoke( PLANNER_PROMPT.format(request=state["user_request"]) ) import json parsed = json.loads(resp.content) steps = parsed["steps"] return {"plan": steps, "current_step": steps[0] if steps else ""}

这里有个细节：temperature=0。规划任务和聊天不一样，它要的是“确定性输出”，不是“创意发挥”。温度调高之后，同一个请求在不同轮次可能给出不同的计划，后面排错会非常痛苦。规划节点一律 0 温度。

3.3 执行器节点：选择并调用工具

执行器节点要解决的核心问题是“当前步骤应该调哪个工具”。我用了 LangChain 的bind_tools机制，让模型可以从工具列表中选择合适的工具，并生成结构化参数。

from langchain_core.utils.function_calling import convert_to_openai_tool tools = [query_database, compute_statistics, write_summary] tool_map = {t.name: t for t in tools} tool_schemas = [convert_to_openai_tool(t) for t in tools] model_with_tools = model.bind_tools(tools) EXECUTOR_PROMPT = """你是任务执行器。请根据当前步骤，调用合适的工具完成操作。 当前计划：{plan} 已完成步骤：{done_steps} 当前步骤：{current_step} 如果当前步骤需要查询数据，使用 query_database； 如果需要对已有数据做统计，使用 compute_statistics； 如果需要生成最终总结，使用 write_summary。 如果所有步骤都已处理完毕，直接输出 DONE。 """ def executor_node(state: PlanExecuteState): msg = model_with_tools.invoke( EXECUTOR_PROMPT.format( plan=state["plan"], done_steps="\n".join(state["done_steps"]) if state["done_steps"] else "无", current_step=state["current_step"], ) ) if not msg.tool_calls: return {"done_steps": [msg.content], "current_step": ""} tool_outputs = [] for tc in msg.tool_calls: tool_name = tc["name"] tool_args = tc["args"] if tool_name in tool_map: try: result = tool_map[tool_name].invoke(tool_args) tool_outputs.append(f"[{tool_name}] -> {result}") except Exception as e: tool_outputs.append(f"[{tool_name}] ERROR: {e}") # 把工具结果追加到已完成的步骤里 return {"done_steps": tool_outputs}

这段代码里最值得学的一处是：工具调用的结果不是直接返回给用户，而是写回 State 的done_steps。这样下一步的规划器或执行器再次读取状态时，能看到完整的历史。这是 Plan-Execute 和一次性调用最大的区别——信息流动性。

3.4 检查器与重规划：纠错的关键

执行器干完活，接下来轮到检查器判断“下一步干嘛”。这一步我设计了两种检查：计划完成度检查和执行质量检查。

def reflector_node(state: PlanExecuteState): # 场景1：计划里的所有步骤都已执行完毕 if len(state["done_steps"]) >= len(state["plan"]): # 尝试生成最终汇总 return {"final_response": summarize_result(state), "current_step": ""} # 场景2：还有未完成的步骤，取下一个 next_step_idx = len(state["done_steps"]) if next_step_idx < len(state["plan"]): return {"current_step": state["plan"][next_step_idx]} return {"final_response": "任务执行异常终止。", "error": "IRRECOVERABLE: plan exhausted"}

这版检查器比较简单，核心是“数已完成步骤和总计划步骤的关系”。如果要做得更精细，可以在检查器里加入“结果合理性验证”——比如让模型判断当前步骤的输出是否真的满足计划预期。

我在实际项目中常常会加一个验证 Prompt：

VALIDATOR_PROMPT = """当前步骤执行结果如下。请判断是否满足计划要求。 如果满足，回复 VALID；如果不满足，回复 INVALID 并说明原因。 计划要求：{step} 执行结果：{output} """

加了这个验证器之后，Agent 的执行准确性会有明显提升，但代价是每个步骤多一次模型调用。预算紧张时可以只对高风险步骤做验证。

3.5 重试与重规划路由

条件路由的完整逻辑：

from typing import Literal def route_after_execution(state: PlanExecuteState) -> Literal["executor", "planner", "finish", "retry"]: # 如果出现了错误 if state["retry_count"] > 0: if state["retry_count"] >= 3: return "planner" # 重试三次失败，回炉重规划 return "executor" # 否则原地重试 # 正常情况：看计划是否完成 if len(state["done_steps"]) >= len(state["plan"]): return "finish" # 如果当前步骤执行结果的长度异常（例如空输出） last_step = state["done_steps"][-1] if state["done_steps"] else "" if len(last_step) < 5: return "replan" return "executor"

这个路由函数的判断顺序是有讲究的：先查错误，再查完成度，最后查执行质量。因为“重试”的优先级最高——如果当前步骤因为超时而失败，重试比重新规划便宜得多。

3.6 把图组装起来

最后一步是把所有节点串成图，并添加状态字段管理。

from langgraph.checkpoint.memory import MemorySaver workflow = StateGraph(PlanExecuteState) # 添加节点 workflow.add_node("planner", planner_node) workflow.add_node("executor", executor_node) workflow.add_node("reflector", reflector_node) # 添加边 workflow.add_edge(START, "planner") workflow.add_edge("planner", "executor") workflow.add_edge("executor", "reflector") # 条件边 workflow.add_conditional_edges( "reflector", route_after_execution, { "executor": "executor", "retry": "executor", "planner": "planner", "replan": "planner", "finish": END, } ) # 编译，使用 MemorySaver 记录 checkpointer，方便调试时回放 app = workflow.compile(checkpointer=MemorySaver())

这里的MemorySaver不是必须的，但对调试非常有用。它能让 LangGraph 记录每次运行的完整状态快照，我在调试时说“断点回放”靠的就是它。

运行入口：

def run_agent(user_request: str): config = {"configurable": {"thread_id": "demo-1"}} result = app.invoke( {"user_request": user_request, "plan": [], "done_steps": [], "past_steps": [], "current_step": "", "error": "", "final_response": ""}, config ) return result["final_response"] or result["done_steps"] if __name__ == "__main__": print(run_agent("分析2024年三个月的销售数据，计算平均月销售额，并给出总结。"))

运行之后，你会看到流程依次经过：planner（生成三步计划）-> executor（查数据库）-> reflector（判断下一步）-> executor（算统计量）-> reflector -> executor（写总结）-> reflector（发现计划完成）-> finish。

4. 实操过程与核心环节实现

4.1 完整运行实录：当模型开始“自己干活”

我把上述代码跑起来，加了部分日志输出，这是真实运行的一段追踪：

[planner] 生成计划: 1. 查询三个月的销售数据 2. 对销售数据计算统计指标 3. 基于统计指标生成总结 [executor] 当前步骤: 查询三个月的销售数据 [executor] 调用工具: query_database(sql="SELECT month, amount FROM sales") [executor] 工具返回: [('2024-01', 12000.0), ('2024-02', 15500.0), ('2024-03', 9800.0)] [reflector] 当前进度: 1/3，继续执行 [executor] 当前步骤: 对销售数据计算统计指标 [executor] 调用工具: compute_statistics(data="...") [executor] 工具返回: sum=37300.00, mean=12433.33, max=15500.00 [reflector] 当前进度: 2/3，继续执行 [executor] 当前步骤: 基于统计指标生成总结 [executor] 调用工具: write_summary(metrics="sum=37300.00, mean=12433.33, max=15500.00") [executor] 工具返回: 销售数据概览：三日销售总额 37300，日均 12433.33，峰值 15500。整体表现平稳，二月为高点。 [reflector] 当前进度: 3/3，任务完成 [finish] 输出最终回复

这段日志看着简单，但你仔细品一下：模型全程没有人工干预，自己完成了“查数据 -> 算指标 -> 写总结”的闭环。这是 Plan-Execute 模式最典型的行为特征。

4.2 触发重规划的现场实验

我故意把工具行为改成了一个“陷阱”：第一次查询数据库时，强制返回空结果。此时执行器拿到空数据，统计节点算不出均值，整个流程就卡住了。

有了检查器和条件路由，流程会这样走：

[executor] 调用工具: query_database(...) [executor] 工具返回: [] [reflector] 发现执行结果异常（空数据），触发重规划 [planner] 重新生成计划: 1. 检查数据库中是否存在数据 2. 如果无数据，尝试查询其他表或提示数据缺失 3. 基于可用数据生成报告

这里有个关键点：触发重规划时，上一轮的done_steps还保留在 State 里，所以新的规划器能看到“刚才查过库但没查到数据”的历史，从而生成更合理的计划。如果每次重规划都把历史清空，那 Agent 就会“失忆”，同一个错能犯两次。

这正是 State 设计的价值：它不只是数据容器，它还是 Agent 的“记忆”。

4.3 关键参数的选择与调优

我在调试这套代码时，前后调了好几个参数，其中最值得分享的是下面这组：

模型选择：规划器我用gpt-4o-mini在演示场景够用，但在真实业务里，规划器建议用更强的模型（比如gpt-4o或同档位国产模型）。原因很简单：规划是整套流程的上游，上游错了后面全错。而执行器因为每次只做一件事，用小模型反而便宜又高效。这种“强规划器、弱执行器”的搭配是最省钱的架构。

重试次数：我默认设了 3 次，超过就重规划。这个值不是拍脑袋定的。少了会让偶发故障直接触发重规划，白白浪费一次模型调用；多了会让 Agent 在一个坏步骤上原地打转。3 次是一个相对均衡的值，你可以根据自己的场景调。

计划步数上限：我在规划器的 Prompt 里明确限制了 2~5 步。为什么不放开？因为 LangGraph 本身不限制循环次数，但模型的上下文窗口受不了。一个 20 步的计划，每一步都往done_steps里塞结果，到第 10 步时上下文已经快满了。限制步数是保护上下文的最简单手段。

4.4 如何验证规划质量

很多人在实现完规划系统后，不知道该怎么测。我的经验是建一个“规划质量检查清单”：

1. 步骤是否原子化？每一步是否只做一件事？
2. 步骤顺序是否可交换？如果可以交换，说明规划器没抓住依赖关系。
3. 计划是否可验证？每一步结束后，是否有判断“这一步做对了没”的依据？
4. 计划是否覆盖了用户请求的所有关键点？

用这个清单去跑 10 个不同领域的请求，基本能暴露 80% 的规划质量问题。

5. 常见问题与排查技巧实录

5.1 问题一：LangGraph 版本不同，API 对不上

这份源码跑在langgraph 0.2.x上，但留意到最近langgraph 0.4.x有破坏性更新，StateGraph的导入路径、add_conditional_edges的写法都有变化。

如果你运行时报错，先检查版本：

pip show langgraph

如果版本不符，有两种解决方式：一是按我的需求指定版本安装，二是在新版本里把from langgraph.graph import StateGraph改成from langgraph.graph import StateGraph（新版路径不变），但条件边返回值从字符串改成Literal时要对齐。

排查技巧：直接看导入错误指向哪个模块，根据堆栈信息回查 API 文档，比盲目改代码快得多。

5.2 问题二：模型输出的 JSON 解析失败

规划器要求模型输出 JSON，但模型偶尔会带 Markdown 代码块标记，或者多输出几行解释文字，导致json.loads直接抛异常。

我的解法是写一个健壮的解析函数：

import json, re def safe_json_parse(text: str): # 去除 Markdown 代码块标记 text = re.sub(r"```json|```", "", text).strip() # 提取最外层花括号 start = text.find("{") end = text.rfind("}") if start != -1 and end != -1: text = text[start:end+1] return json.loads(text)

这个函数从源头上解决了 90% 的解析问题。剩下的 10% 是模型输出不完整 JSON，需要在 Prompt 里加强约束，或者在解析失败时让规划器节点抛错误返回给上一层重试。

5.3 问题三：图跑到一半就停了，没有错误信息

这种情况最气人——代码没报错，但流程就是不往下走。我排查过几次，发现基本都是同一个原因：add_conditional_edges的映射字典里缺少了路由函数返回的某个分支。

比如你的路由函数返回了"retry"，但映射字典只写了{"continue": "executor", "replan": "planner"}，LangGraph 不会提醒你，只会默默让图停在那里。

排查这个问题的技巧：在路由函数里临时加一行print，把返回值打出来，对比映射字典的键，一眼就能看出问题。

5.4 问题四：上下文越来越长，Token 消耗爆表

Plan-Execute 模式最容易被忽视的坑就是 State 无限膨胀。每次工具调用的完整输出都往done_steps里塞，跑个五六个步骤之后，下一轮模型要读的上下文可能是几千甚至上万 Token。

我的缓解方案有两层：

第一层是在执行器节点里做摘要。工具返回结果后，不要让原始返回直接进入历史，而是调用一次轻量摘要：

def summarize_tool_result(result: str) -> str: if len(result) < 200: return result # 截断加摘要 return result[:200] + "..."

第二层是在检查器里做“遗忘”。已经完成且验证通过的早期步骤，其详细输出可以压缩成一行状态描述，或者干脆只保留步骤标题。因为是 Plan-Execute 架构，后续执行不依赖早期步骤的完整输出，所以压缩是安全的。

5.5 实操心得：先画图，再写代码

最后分享一个我做了这么多 LangGraph 项目之后最深的体会：永远先画图，再写代码。

我说的画图不是画架构图，而是在纸上把节点、边、条件分支全部列出来。你不需要画得漂亮，但一定要清楚——每个节点是做什么的、数据往哪里流、什么条件下走哪条分支。把这些想清楚之后，写代码就是填空题。

我见过太多人拿到 LangGraph 就想着“先跑起来再说”，结果代码越写越乱，节点职责互相重叠，最后不得不推倒重来。这个框架的核心优势就是结构清晰，你偏要跳过设计直接用，等于放弃了它最大的价值。

还有一点：调试 LangGraph 程序时，MemorySaver的 checkpointer 功能特别有用。你可以在任意节点之间插入断点，检查 State 的完整内容，看看模型到底在想什么、看到了什么历史。传统的 print 大法在这里远远不够用，因为 State 是整个流程的血液。

规划是 Agent 应用从“玩具”走向“生产力工具”的分水岭。一份能跑通 Plan-Execute 循环的源码并不神秘，关键是把规划、执行、检查三者放进一个清晰的图结构里，让流程可观察、可控制、可恢复。希望这份附了完整源代码的实现能帮你把 LangGraph 用得再顺手一些——至少下次遇到复杂任务时，你手里已经握着一张能自己看地图、自己纠偏的地图了。