进阶大模型应用：构建基于 Plan-and-Execute 架构的自适应 AI Agent

一、 引言为什么复杂的任务需要“计划与执行”分离在前面探讨的 ReAct 或简单多智能体协作中智能体通常是“走一步看一步”。这种动态决策在处理短流程任务时非常灵活但在面对需要几十步甚至上百步流转的超长周期任务时会暴露两个致命缺陷目标迷失Goal Drifting随着执行步骤的增加上下文被大量的中间观察结果Observations填满大模型容易忘记最初的终极目标。效率低下每次行动都需要大模型重新推理全局Token 消耗呈指数级上升且容易在某个局部错误中陷入无意义的死循环。为了攻克这一难题业界提出了Plan-and-Execute计划-执行架构。其核心思想是将“高层宏观规划”与“底层微观执行”彻底解耦Planner计划者负责将用户的复杂需求拆解为一张有序的“步骤清单Todo List”。Executor执行者聚焦于当前分配到的具体单一子任务调用工具将其攻克。Replanner重规划者根据执行者的反馈结果动态更新、删除或重组剩余的步骤清单。这种架构不仅大幅提升了任务成功率还让整个执行过程具备了极高的可观测性和人类可干预性。二、 计划-执行架构的状态机模型在 LangGraph 中我们可以通过一个包含循环的有向图来完美复刻这个架构。整个系统的状态流转如下用户输入需求 -Planner生成步骤清单。检查清单如果仍有未完成的步骤 - 提取第一步交给Executor。Executor调用工具完成任务输出阶段性成果。Replanner审视成果结合原计划动态更新步骤清单增加新步骤或标记已完成然后跳回步骤 2。清单全部完成 - 输出最终结果END。三、 基于 LangGraph 的核心代码实现1. 定义全局有状态的 PlanState我们需要一个结构来记录原始输入、当前的计划清单、已经完成的步骤以及最终的输出。Pythonfrom typing import List, Tuple, TypedDict, Annotated import operator # 定义计划的结构 class PlanState(TypedDict): input: str # 用户的原始需求 plan: List[str] # 动态的步骤清单 past_steps: List[Tuple[str, str]] # 已完成的步骤记录格式为 (步骤描述, 执行结果) response: str # 最终输出结果2. 编写 Planner 节点Planner 只负责宏观拆解不负责具体执行。Pythonfrom langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from pydantic import BaseModel, Field llm ChatOpenAI(modelgpt-4o-mini, temperature0) # 使用 Pydantic 强制让大模型输出结构化的计划列表 class ActPlan(BaseModel): steps: List[str] Field(description为了完成最终目标需要按顺序执行的独立步骤列表) planner_llm llm.with_structured_output(ActPlan) def planner_node(state: PlanState): prompt ChatPromptTemplate.from_template( 你是一个高层项目规划专家。请针对用户的最终需求拆解出一个详细的、按顺序执行的步骤计划。\n 最终需求: {input}\n 请只返回步骤列表。 ) planner_chain prompt | planner_llm result planner_chain.invoke({input: state[input]}) return {plan: result.steps}3. 编写 Executor 节点Executor 专注于处理state[plan]中的第一个任务并结合past_steps提供的上下文信息。Pythonfrom langchain.agents import load_tools # 假设这里引入了标准的搜索或计算工具 def executor_node(state: PlanState): current_task state[plan][0] # 获取当前需要执行的第一步 past_steps state[past_steps] context \n.join([f步骤: {task}\n结果: {res} for task, res in past_steps]) prompt ChatPromptTemplate.from_template( 你是一个高效的执行者。请根据已有的上下文信息完成当前的特定任务。\n 已有历史信息:\n{context}\n\n 当前需要完成的任务: {current_task}\n 请直接输出该步骤的执行结果或答案。 ) executor_chain prompt | llm response executor_chain.invoke({context: context, current_task: current_task}) # 将当前任务和结果作为元组返回用于追加到 past_steps 中 return {past_steps: [(current_task, response.content)]}4. 编写 Replanner 节点动态调整核心Replanner 是系统的灵魂它决定了系统是继续执行、修改计划还是直接收尾。Pythonclass ActReposition(BaseModel): action: str Field(description下一步动作如果是继续执行后面的计划返回 continue如果任务已圆满完成返回 final。) new_steps: List[str] Field(default[], description如果发现原计划有遗漏或需要调整返回更新后剩余的步骤列表。如果无需调整留空。) final_response: str Field(default, description如果 action 为 final在此填入给用户的最终完美答复。) replanner_llm llm.with_structured_output(ActReposition) def replanner_node(state: PlanState): prompt ChatPromptTemplate.from_template( 你是一个项目监控与重规划专家。\n 原始需求: {input}\n 原定剩余计划: {plan}\n 已执行的步骤及结果:\n{past_steps}\n\n 请评估当前进展。如果目标已达成请设置 action 为 final 并给出最终答复。\n 如果目标未达成请设置 action 为 continue。你可以根据最新情况调整或重写剩余计划new_steps。 ) context \n.join([f步骤: {task}\n结果: {res} for task, res in state[past_steps]]) remaining_plan state[plan][1:] # 移除刚刚已经执行过的那一步 replanner_chain prompt | replanner_llm result replanner_chain.invoke({ input: state[input], plan: remaining_plan, past_steps: context }) if result.action final: return {response: result.final_response, plan: []} else: # 如果大模型返回了更新后的步骤则使用新步骤否则使用移除第一步后的剩余原计划 updated_plan result.new_steps if result.new_steps else remaining_plan return {plan: updated_plan}5. 构建并编译 LangGraph 工作流有了这三个节点我们用条件路由将它们串联起来。Pythonfrom langgraph.graph import StateGraph, END workflow StateGraph(PlanState) # 注册节点 workflow.add_node(Planner, planner_node) workflow.add_node(Executor, executor_node) workflow.add_node(Replanner, replanner_node) # 设置入口 workflow.set_entry_point(Planner) # 确定基础连线 workflow.add_edge(Planner, Executor) workflow.add_edge(Executor, Replanner) # 核心条件路由根据 plan 清单是否为空决定是进入下一次循环还是终止 def should_continue(state: PlanState): if not state[plan]: return END return Executor workflow.add_conditional_edges( Replanner, should_continue, { Executor: Executor, END: END } ) app workflow.compile()四、 执行效果演示向该系统输入一个复杂的、依赖递进关系的模糊任务输入“分析 2026 年生成式 AI 在医疗行业的最新三个应用场景并挑出其中落地最快的一个场景评估其潜在法律风险。”在app.stream()的运行过程中系统的执行轨迹如下[Planner] 拆出初始清单步骤1: 检索 2026 医疗 AI 的最新三个应用场景。步骤2: 对比三个场景的落地成熟度选出最快的一个。步骤3: 针对该成熟场景分析其数据隐私与法律风险。[Executor]执行步骤1返回三个场景AI影像辅助诊断、电子病历自动生成、AI靶点预测。[Replanner]检查发现步骤1完成计划正常命令继续。[Executor]执行步骤2对比后得出“电子病历自动生成”由于不直接触及核心临床诊断落地速度最快。[Replanner]在这一步大模型敏锐地发现原计划的步骤3范围太广它自主在剩余计划中临时增加了“查阅医疗数据跨境传输新规”这一子步骤。[Executor] [Replanner]循环交替直到所有动态调整后的清单清空。[END]输出结构高度严密、逻辑天衣无缝的深度报告。五、 生产环境的最佳实践总结执行器的容错性Fallback实际生产中Executor内部通常包裹着一个标准的 ReAct Agent允许它在执行某一个单一子步骤时可以多次尝试调用各种不同的工具。长文本记忆压缩随着循环迭代past_steps积累的文本量会极大。在 Replanner 输入前可以引入一个“总结节点”专门将过去几步的 Observation 提炼压缩只保留核心 Data 指标。人机协同Human-in-the-loop在 LangGraph 的Planner到Executor的连线中间可以配置interrupt_before[Executor]。这样系统生成计划后会暂停等待人类架构师在后台点击“确认”或手动微调 Todo List 后再开始执行这是目前企业落地复杂 AI 工作流时最稳妥的做法。