1. 背景
Plan-and-Execute 模式是一种 “先计划、后执行、按计划循环迭代” 先输出一份结构化的执行计划，将复杂任务拆解为若干明确步骤，然后严格按照步骤逐步执行；当某一步需要外部信息或能力（如搜索、计算等）时，则触发工具调用，完成后继续推进计划，直到满足终止条件。
实际应用场景 以 Cursor 为例：当 Cursor 要编写复杂的代码时，通常会先执行规划，生成一个 to-do list，之后按照这个 to-do list 的顺序依次执行任务。这种”先规划、后执行”的方式能够有效处理复杂任务，避免遗漏关键步骤。
Plan Mode 的三层结构 Plan Mode 通常可以抽象出三层结构：
计划层 ：用自然语言清晰列出步骤、编号、依赖、输入输出。执行层 ：逐步消费计划中的步骤，产出结果或中间态。工具层 ：在执行层遇到”需要能力/数据”的节点时，面向具体工具进行调用（如价格查询、表达式计算等）。
优势 这种”先规划、后执行”的范式具备以下优势：
可解释性强
可回溯
易扩展（新增工具即可拓展能力）
在设计 Agent 时已经被广泛采用。
2. 构建 Plan-and-Execute Agent 使用 LangGraph 来构建一个简单的 Plan-and-Execute Agent，节点结构包括：
Plan Node（计划节点）
Execute Node（执行节点）
Tool Node（工具节点）
Conditional Edge（条件边）
2.1 基础准备 定义工具 首先开发需要用到的工具（tools）。这里定义了两个函数：一个是查询汇率信息，另一个是计算器，然后利用 LangChain 的 @tool 装饰器，将它们包装成可被 LLM 调用的工具。
代码说明：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 from langchain.tools import tool@tool def get_exchange_rate (currency: str ) -> str : """ 获取货币汇率信息 这个工具用于查询指定货币对人民币的汇率。 在实际应用中，这里可以对接真实的汇率 API（如 ExchangeRate-API）。 Args: currency: 货币代码，例如"USD"（美元）、"EUR"（欧元） Returns: 返回格式化的汇率信息字符串 """ if currency == "USD" or currency == "美元" : return "美元对人民币汇率：1 USD = 7.25 CNY" if currency == "EUR" or currency == "欧元" : return "欧元对人民币汇率：1 EUR = 7.85 CNY" return f"没有找到{currency} 的汇率信息" @tool def calculate (expression: str ) -> str : """ 计算表达式 这个工具用于执行数学表达式计算。 注意：实际生产环境中应该使用更安全的表达式解析库（如 numexpr）， 而不是直接使用 eval()，这里仅用于演示。 Args: expression: 数学表达式字符串，例如"100 * 7.25" Returns: 计算结果字符串 """ return str (eval (expression)) TOOLS = [get_exchange_rate, calculate] TOOL_DICT = {tool.name: tool for tool in TOOLS}
关键点说明：
@tool 装饰器会将普通函数转换为 LangChain 的工具对象，LLM 可以识别并调用工具函数的文档字符串（docstring）会被 LLM 读取，用于理解工具的用途和参数
TOOL_DICT 用于在执行阶段快速查找和调用对应的工具
定义模型 定义大语言模型（LLM），这里使用 DeepSeek Chat 模型。然后将模型与工具进行绑定，使 LLM 能够识别和调用我们定义的工具。
代码说明：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 from langchain_openai import ChatOpenAIimport osLLM = ChatOpenAI( model="deepseek-chat" , api_key=os.getenv("DEEPSEEK_API_KEY" ), base_url=os.getenv("DEEPSEEK_BASE_URL" ), ) LLM_WITH_TOOLS = LLM.bind_tools(TOOLS)
关键点说明：
bind_tools() 方法会将工具的函数签名和描述信息传递给 LLM绑定后，LLM 在生成回复时可能会返回 tool_calls 字段，表示需要调用工具
工具调用信息包含工具名称、参数和调用 ID，用于后续执行
定义状态 定义 Agent 运行过程中需要维护的状态。自定义了 PlanAndExecuteState，它继承 LangGraph 的 MessagesState，可以自动保存消息列表（包括用户消息、AI 回复、工具调用结果等），并且额外扩展了两个属性，分别表示生成的计划和当前执行的步骤。
代码说明：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 from langgraph.graph.message import MessagesStateclass PlanAndExecuteState (MessagesState ): """ 计划和执行状态定义 这个状态类用于在整个 Agent 执行过程中保存和传递信息。 继承自 MessagesState 的属性： messages: List[BaseMessage] - 消息列表，包含对话历史、工具调用等 自定义属性： plan: str - 生成的执行计划文本 step: int - 当前执行的步骤编号（从 0 开始） """ plan: str step: int
关键点说明：
MessagesState 提供了消息管理功能，自动处理消息的追加和更新plan 字段存储计划节点生成的执行计划，供执行节点使用step 字段用于跟踪执行进度，便于调试和日志记录状态对象在节点之间传递，每个节点可以读取和修改状态
2.2 Plan Node（计划节点） Plan Node 是整个 Agent 的第一个节点，负责根据用户提出的问题，生成详细的、结构化的执行计划。这个计划将指导后续的执行步骤。
代码说明：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 from state import PlanAndExecuteStatefrom llm import LLM_WITH_TOOLSfrom langchain_core.messages import SystemMessageSYSTEM_PROMPT = """ 你是一位智能助手，擅长解决用户提出的各种问题。请为用户提出的问题创建分析方案步骤。 如果有需要，可以调用工具。 你可以调用工具的列表如下： get_exchange_rate: 获取指定货币对人民币的汇率信息 Parameters: ----------- currency : str 货币代码，例如"USD"（美元）、"EUR"（欧元） Returns: -------- str 指定货币的汇率信息 calculate: 计算表达式 Parameters: ----------- expression : str 表达式内容 Returns: -------- str 表达式的计算结果 要求： 1.用中文列出清晰步骤 2.每个步骤标记序号 3.明确说明需要分析和执行的内容 4.只需输出计划内容，不要做任何额外的解释和说明 5.设计的方案步骤要紧紧贴合我的工具所能返回的内容，不要超出工具返回的内容 """ def plan_node (state: PlanAndExecuteState ) -> PlanAndExecuteState: """ 生成计划节点 这个节点的职责： 1. 接收用户的问题（在 state["messages"] 中） 2. 调用 LLM 生成执行计划 3. 将计划保存到状态中，供后续节点使用 Args: state: 当前状态，包含用户消息 Returns: 更新后的状态，包含生成的计划 """ messages = [SystemMessage(content=SYSTEM_PROMPT)] + state["messages" ] response = LLM_WITH_TOOLS.invoke(input =messages) plan = response.content print (f"生成的计划: \n{plan} " ) state["plan" ] = plan return state
工作流程说明：
节点接收包含用户问题的状态
构造包含 System Prompt 和用户消息的完整消息列表
调用 LLM 生成执行计划（纯文本格式，如”1. 查询美元汇率 2. 查询欧元汇率 3. 计算兑换金额”）
将计划保存到状态中
返回更新后的状态，流程进入下一个节点
注意 ：为了方便演示，直接将工具描述信息写到 System Prompt 中了。更理想的方案是接入 MCP（Model Context Protocol）或使用 LangChain 的工具描述自动生成功能，感兴趣的同学可以自己实现。
2.3 Execute Node（执行节点） Execute Node 是 Agent 的核心执行节点，负责根据生成的计划逐步执行任务。LLM 会根据当前计划、对话历史和工具调用结果，决定下一步操作：可能是直接回复，也可能是调用工具。
代码说明：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 from state import PlanAndExecuteStatefrom llm import LLM_WITH_TOOLSfrom langchain_core.messages import SystemMessageSYSTEM_PROMPT = """ 你是一位思路清晰、有条理的智能助手，你必须严格按照以下计划执行任务： 当前计划： {plan} 如果你认为计划已经执行到最后一步了，请在内容的末尾加上 Final Answer 字样 示例： 这个问题的最终答案是99.99。 Final Answer """ def execute_node (state: PlanAndExecuteState ) -> PlanAndExecuteState: """ 执行计划节点 这个节点的职责： 1. 读取计划节点生成的计划 2. 根据计划、对话历史和工具调用结果，执行当前步骤 3. LLM 可能直接回复，也可能生成工具调用请求 4. 更新执行步骤计数器 Args: state: 当前状态，包含计划、消息历史和步骤编号 Returns: 更新后的状态，包含执行结果 """ step = state["step" ] system_message = SystemMessage(content=SYSTEM_PROMPT.format (plan=state["plan" ])) messages = [system_message] + state["messages" ] resp = LLM_WITH_TOOLS.invoke(input =messages) print (f"\n【第 {step + 1 } 步】{resp.content} " ) state["messages" ].append(resp) state["step" ] = step + 1 return state
工作流程说明：
节点读取状态中的计划（由 plan_node 生成）
构造包含计划和完整对话历史的消息列表
调用 LLM，LLM 会：
分析当前应该执行计划的哪一步
如果需要数据，生成工具调用请求（tool_calls 字段）
如果可以直接回答，生成文本回复（content 字段）
将 LLM 的回复保存到消息历史中
更新步骤计数器
返回状态，由条件边决定下一步流向
关键点：
LLM 的回复可能包含 tool_calls（需要调用工具）或 content（直接回复）
如果回复末尾包含 “Final Answer”，表示任务完成
步骤计数器用于跟踪执行进度，便于调试
Tool Node 负责执行工具调用。当 Execute Node 中的 LLM 生成工具调用请求时，流程会进入 Tool Node，执行实际的工具调用，并将结果保存到消息历史中，供后续节点使用。
代码说明：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 from state import PlanAndExecuteStatefrom tools import TOOL_DICTfrom langchain_core.messages import ToolMessagedef tool_node (state: PlanAndExecuteState ) -> PlanAndExecuteState: """ 工具调用节点 这个节点的职责： 1. 从最后一条消息中提取工具调用信息 2. 根据工具名称查找对应的工具对象 3. 使用工具参数执行工具调用 4. 将工具调用结果封装为 ToolMessage，添加到消息历史中 Args: state: 当前状态，最后一条消息应包含 tool_calls Returns: 更新后的状态，包含工具调用结果 """ last_message = state["messages" ][-1 ] tool_calls = last_message.tool_calls if tool_calls is None : return state for tool_call in tool_calls: tool_name = tool_call["name" ] tool_args = tool_call["args" ] tool = TOOL_DICT.get(tool_name) if tool is None : continue tool_call_result = tool.invoke(input =tool_args) print (f"\n调用工具: {tool_name} \n参数: {tool_args} \n执行结果: {tool_call_result} " ) state["messages" ].append( ToolMessage( content=f"工具调用结果: {tool_call_result} " , tool_call_id=tool_call["id" ], ) ) return state
工作流程说明：
节点从最后一条消息中提取 tool_calls 信息
遍历每个工具调用请求：
提取工具名称和参数
从 TOOL_DICT 中查找对应的工具对象
使用参数调用工具函数
将结果封装为 ToolMessage 并添加到消息历史
返回更新后的状态，流程回到 execute_node 继续执行
关键点：
ToolMessage 必须包含 tool_call_id，用于关联对应的工具调用请求LLM 在后续回复时，会根据 tool_call_id 匹配工具调用结果
一个 LLM 回复可能包含多个工具调用，需要遍历处理
工具调用结果会被添加到消息历史中，供后续节点使用
2.5 条件边（Conditional Edge） 定义好节点之后，还需要一个 Conditional Edge（条件边）来控制节点之间的流转规则。条件边会根据当前状态决定下一步应该执行哪个节点。
代码说明：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 from state import PlanAndExecuteStatefrom langgraph.graph import ENDdef conditional_edge (state: PlanAndExecuteState ) -> str : """ 条件边函数 这个函数决定 execute_node 执行后的下一步流向： - 如果 LLM 回复中包含 "Final Answer"，返回 END，表示任务完成 - 否则返回 "tool_node"，表示需要调用工具 Args: state: 当前状态，包含执行结果 Returns: 下一个节点的名称，或 END 表示结束 """ last_message = state["messages" ][-1 ] if "Final Answer" in last_message.content: return END else : return "tool_node"
工作流程说明：
条件边函数接收当前状态作为参数
检查最后一条消息（LLM 的回复）中是否包含 “Final Answer”
如果包含，返回 END，流程结束
如果不包含，返回 "tool_node"，流程进入工具节点
关键点：
条件边是 LangGraph 中实现分支逻辑的关键机制
返回值必须是已定义的节点名称或 END
这里的判断逻辑比较简单，实际应用中可能需要更复杂的条件判断
“Final Answer” 标识是在 System Prompt 中约定的终止条件
2.6 构造 Workflow（工作流） State、Node 和 Edge 都定义好了之后，就可以构造完整的 Workflow（工作流）了。Workflow 定义了整个 Agent 的执行流程和节点之间的连接关系。
代码说明：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 from langgraph.graph import StateGraph, START, ENDfrom langgraph.graph.state import CompiledStateGraphfrom langchain_core.messages import HumanMessagedef build_agent () -> CompiledStateGraph: """ 构建 Plan-and-Execute Agent 这个函数负责： 1. 创建状态图（StateGraph） 2. 添加所有节点 3. 添加节点之间的边（普通边和条件边） 4. 编译成可执行的 Agent Returns: 编译后的 Agent，可以用于执行任务 """ graph = StateGraph(PlanAndExecuteState) graph.add_node("plan_node" , plan_node) graph.add_node("execute_node" , execute_node) graph.add_node("tool_node" , tool_node) graph.add_edge(START, "plan_node" ) graph.add_edge("plan_node" , "execute_node" ) graph.add_conditional_edges( source="execute_node" , path=conditional_edge, path_map={ "tool_node" : "tool_node" , END: END, }, ) graph.add_edge("tool_node" , "execute_node" ) agent = graph.compile () agent.get_graph().draw_mermaid_png(output_file_path="./agent.png" ) return agent def run_agent (agent: CompiledStateGraph, user_query: str ) -> str : """ 运行 Agent 这个函数负责： 1. 初始化 Agent 状态 2. 执行 Agent 工作流 3. 返回最终结果 Args: agent: 编译后的 Agent 实例 user_query: 用户的问题 Returns: 最终的回答文本 """ messages = [HumanMessage(content=user_query)] init_state = PlanAndExecuteState( plan="" , step=0 , messages=messages, ) result = agent.invoke(init_state) return result["messages" ][-1 ].content
工作流结构说明：
整个 Agent 的执行流程如下：
1 2 3 4 5 START → plan_node → execute_node → [条件判断] ↓ ↓ tool_node END ↓ execute_node (循环)
START → plan_node ：工作流开始，进入计划节点生成计划plan_node → execute_node ：计划生成后，进入执行节点开始执行execute_node → [条件判断] ：
如果包含 “Final Answer”，流程结束（END）
否则进入 tool_node 执行工具调用
tool_node → execute_node ：工具调用完成后，回到执行节点继续执行循环执行 ：步骤 3-4 会循环执行，直到 LLM 输出 “Final Answer”
关键点：
StateGraph 是 LangGraph 的核心，用于构建有状态的工作流compile() 方法将图编译成可执行的状态机invoke() 方法执行整个工作流，从 START 到 END状态在节点之间传递，每个节点可以读取和修改状态
条件边实现了分支逻辑，使工作流能够根据条件选择不同的执行路径
这样，一个完整的 Plan-and-Execute Agent 就构建完成了！
3. 结果演示 测试一下：执行主函数，向 agent 提问 “我想用100美元和50欧元兑换人民币，总共能兑换多少人民币？”。
代码说明：
1 2 3 4 5 6 7 8 9 10 11 12 13 if __name__ == "__main__" : agent: CompiledStateGraph = build_agent() result = run_agent(agent, "我想用100美元和50欧元兑换人民币，总共能兑换多少人民币？" ) print (f"\n最终答案: {result} " )
预期执行流程：
Plan Node ：生成计划
1 2 3 1. 查询美元对人民币的汇率 2. 查询欧元对人民币的汇率 3. 计算100美元和50欧元总共能兑换多少人民币
Execute Node（第1步） ：执行计划第1步
LLM 决定调用 get_exchange_rate 工具查询美元汇率
生成工具调用请求
Tool Node ：执行工具调用
调用 get_exchange_rate({"currency": "USD"})
返回：”美元对人民币汇率：1 USD = 7.25 CNY”
Execute Node（第2步） ：执行计划第2步
LLM 看到美元汇率结果，决定调用 get_exchange_rate 查询欧元汇率
生成工具调用请求
Tool Node ：执行工具调用
调用 get_exchange_rate({"currency": "EUR"})
返回：”欧元对人民币汇率：1 EUR = 7.85 CNY”
Execute Node（第3步） ：执行计划第3步
LLM 看到两个汇率结果，决定调用 calculate 计算总兑换金额
调用 calculate("100 * 7.25 + 50 * 7.85")
返回计算结果：1107.5
Execute Node（最终） ：生成最终答案
LLM 整合所有信息，生成最终答案
输出：”100美元可以兑换725元人民币，50欧元可以兑换392.5元人民币，总共可以兑换1117.5元人民币。Final Answer”
条件边检测到 “Final Answer”，流程结束
4. 总结 4.1 Plan-and-Execute 模式的核心流程 Plan-and-Execute 模式的核心流程可以概括为以下五个阶段：
计划阶段（Plan） ：根据用户问题生成结构化的执行计划
将复杂任务拆解为若干明确步骤
每个步骤都有清晰的输入输出说明
执行阶段（Execute） ：按照计划逐步执行
LLM 根据当前步骤和上下文决定下一步操作
可能需要调用工具获取外部数据或能力
工具调用（Tool Calling） ：当需要外部能力时，调用相应工具
提取工具调用请求中的参数
执行工具函数并获取结果
将结果保存到消息历史中
循环迭代（Iteration） ：执行完成后判断是否结束
如果未完成，继续执行下一步
工具调用结果会作为上下文传递给下一次执行
终止条件（Termination） ：当检测到终止标识时，任务完成
4.2 模式优势 Plan-and-Execute 模式具有以下优势：
可解释性强 ：每个步骤都有明确的计划，用户可以清楚地看到 Agent 的执行过程可回溯 ：完整的消息历史记录了整个执行过程，便于调试和问题排查易扩展 ：新增工具只需定义函数并用 @tool 装饰，无需修改核心逻辑结构化 ：计划提供了清晰的任务分解，有助于 LLM 更好地理解和执行任务
4.3 适用场景 这种模式特别适合处理：
复杂的多步骤任务 ：需要多个步骤才能完成的任务需要外部数据的任务 ：需要查询数据库、调用 API 等需要计算的任务 ：需要进行数学计算、数据分析等需要可解释性的任务 ：用户需要了解 Agent 的执行过程
4.4 改进方向 在实际应用中，可以考虑以下改进：
更智能的计划生成 ：使用更先进的规划算法，考虑步骤之间的依赖关系动态计划调整 ：根据执行结果动态调整计划错误处理 ：添加错误处理和重试机制工具描述自动化 ：使用 MCP 或自动生成工具描述，而不是手动编写更复杂的终止条件 ：不仅依赖 “Final Answer”，还可以考虑步骤完成度、时间限制等
通过 LangGraph 构建 Plan-and-Execute Agent，我们可以轻松实现一个功能强大、可解释、易扩展的智能助手。
