1. 从“直接问”到“先规划”为什么Agent需要Plan-and-Solve如果你用过早期的AI助手或者直接向大语言模型LLM提过一个稍微复杂点的问题比如“帮我策划一个为期三天的北京旅游行程要包含文化古迹、特色美食和适合亲子活动的场所预算控制在人均5000元以内”你大概率会收到一份看起来“还行”的清单。景点、餐厅、活动似乎都提到了但仔细一看行程安排可能极不合理第一天上午在故宫下午就安排去了八达岭长城完全忽略了交通时间或者推荐的餐厅人均消费远超预算导致总花费超标。这种回答就是典型的“直接动手边做边想”模式——模型根据你的指令直接开始生成内容缺乏一个全局的、结构化的思考过程。这就是“Plan-and-Solve”范式要解决的核心痛点。在AI Agent智能体的开发领域我们越来越意识到让LLM直接生成最终答案对于简单任务尚可但对于多步骤、有约束条件、需要逻辑推理的复杂任务效果往往不尽人意。Agent作为能够感知环境、进行决策并执行动作的智能程序其核心能力之一就是“规划”。“Plan-and-Solve”直译过来就是“先规划再解决”它强调在执行具体动作之前先让Agent或者说驱动Agent的LLM制定一个清晰的、分步骤的计划。这听起来像是常识但在AI领域实现它却需要一套精巧的框架设计。它不仅仅是让模型在回答前加一句“让我想想”而是要求模型能明确地分解任务、识别子目标、评估资源与约束并规划出可行的执行路径。这与人类处理复杂问题的方式高度一致工程师在写代码前会画架构图项目经理在启动项目前会做WBS工作分解结构我们旅行前也会做攻略。“Plan-and-Solve”的本质是将LLM从一个“即时反应”的文本生成器升级为一个具备“前瞻性思考”和“结构化问题解决”能力的推理引擎。近年来随着LangChain、LangGraph等框架的流行以及ReActReasoning and Acting等范式的提出为Agent实现“Plan-and-Solve”提供了强大的工具链。ReAct范式鼓励模型将“思考”Reasoning和“行动”Acting交织进行在每一步行动前都输出一段推理链。而“Plan-and-Solve”可以看作是ReAct的一种前置强化和结构化延伸在开始任何“行动”之前先进行一轮集中的、宏观的“规划”Plan产出整个任务的工作流蓝图然后再按照这个蓝图去逐步“解决”Solve。这能显著提升复杂任务处理的可靠性、一致性和可解释性。2. Plan-and-Solve的核心组件与工作流拆解一个典型的、实现了Plan-and-Solve范式的Agent工作流通常包含几个关键组件它们协同工作将“规划”的思想落地。理解这些组件是设计和开发此类Agent的基础。2.1 规划器Planner任务的“总设计师”规划器是Plan-and-Solve架构的大脑通常由一个或多个LLM调用构成。它的核心职责是接收用户的初始请求或称为“目标”并输出一个结构化的计划。这个计划不是一段笼统的描述而应该是一个可操作的任务分解清单。规划器需要完成以下工作任务理解与澄清首先它需要精准理解用户意图。对于模糊的请求它可能需要发起追问例如“您说的‘高端’餐厅具体是指人均消费在什么范围”。这一步确保了规划的起点是准确的。约束与条件识别自动提取或确认任务中的限制条件如时间、预算、格式要求、可用工具API等。例如从“生成一份季度报告”中识别出需要财务数据API从“订机票”中识别出日期和预算。任务分解Decomposition将宏观目标拆解为一系列有序的、原子级的子任务。例如“策划旅游行程”可分解为1) 确定每日主题2) 为每天上午、下午、晚上筛选景点3) 为每个时间段附近的景点匹配餐厅4) 估算交通时间和方式5) 核算总预算。依赖关系分析确定子任务之间的前后顺序和依赖关系。有些任务可以并行有些必须串行。例如“查询航班信息”必须在“比较航班价格”之前而“预订酒店”和“预订景点门票”在时间确定后可以并行进行。资源分配与工具选择为每个子任务分配合适的“工具”Tool。Agent的能力来源于其可调用的工具如搜索API、计算器、代码执行器、数据库查询等。规划器需要决定“查询天气”用天气API“计算汇率”用计算器工具。一个规划良好的输出应该接近于一个项目计划表可以是JSON、YAML格式或者是一段结构极其清晰的文本。例如{ “goal”: “策划三天北京亲子游行程预算人均5000元” “constraints”: [“总时长: 3天”, “预算: ≤5000元/人”, “主题: 文化美食亲子”], “plan”: [ {“step”: 1, “task”: “确定每日核心主题和住宿区域”, “tool”: “reasoning”, “depends_on”: []}, {“step”: 2, “task”: “基于主题搜索并筛选第1天上午的文化古迹景点”, “tool”: “search_api”, “depends_on”: [1]}, {“step”: 3, “task”: “为第1天上午的景点匹配附近适合家庭的午餐餐厅并估算费用”, “tool”: “search_api calculator”, “depends_on”: [2]}, // ... 更多步骤 {“step”: N, “task”: “汇总所有信息生成格式优美的最终行程文档”, “tool”: “report_generator”, “depends_on”: [所有前置步骤]} ] }2.2 执行器Executor/ 调度器Orchestrator计划的“施工队长”规划器产出蓝图后需要有一个组件来按图施工这就是执行器在LangGraph等框架中常称为“调度器”。它的职责是工作流驱动按照规划器输出的步骤顺序依次启动每个子任务。它需要管理任务状态知道当前执行到哪一步下一步该执行什么。工具调用对于每个需要工具的子任务执行器负责准备输入参数调用对应的工具函数并获取返回结果。例如调用搜索API时需要将“北京适合亲子的博物馆”作为查询关键词传入。上下文管理这是关键所在。执行器需要维护一个“工作上下文”它包含了初始目标、已完成的步骤及其结果、当前步骤的输入输出等所有信息。这个上下文会随着执行流程逐步丰富并作为后续步骤的输入。例如步骤2搜索到的景点列表会被传递给步骤3用于寻找附近的餐厅。异常处理与重规划计划赶不上变化。当工具调用失败如API返回错误、返回结果不符合预期如搜索不到合适景点或遇到未预见的约束冲突如预算超标时执行器不能直接崩溃。它需要有能力捕获异常并将当前状态包括错误信息反馈给规划器或一个专门的“重规划”模块触发局部或全局的重新规划。LangGraph框架在这方面提供了强大的支持。它允许你用“图”来定义Agent的工作流节点Node代表状态检查或工具调用边Edge代表状态流转的条件。通过这种显式的图结构你可以非常直观地构建出包含规划、执行、条件判断、循环用于重试或重规划的复杂Agent逻辑。2.3 反思器Reflector与重规划机制系统的“质检员”一个健壮的Plan-and-Solve系统不能是“一锤子买卖”。最初的计划可能基于不完整的信息或错误的假设。因此引入“反思”环节至关重要。反思器通常在以下时机被触发每个主要步骤完成后检查结果是否满足该子任务的目标质量如何。遇到失败或意外结果时分析失败原因。所有步骤执行完毕后最终检查评估最终结果是否真正满足了用户的初始目标。反思器本身也可以是一个LLM调用它接收当前上下文和最新结果并回答诸如“当前的结果是否解决了子任务X”“预算是否仍然可控”“到目前为止的行程安排在交通时间上是否合理”等问题。如果反思器判断当前路径有问题它可以建议退回某一步进行重试或者直接触发“重规划”Replanning。重规划不一定需要推倒重来。根据问题的严重性可以是局部调整只重新规划当前出错步骤及后续受影响的步骤。全局重规划完全重新分析任务生成新计划。这通常发生在发现核心假设错误时例如用户后来补充了“其中一天要去天津”。将反思和重规划机制嵌入工作流使得Agent具备了动态适应和从错误中学习的能力大大提升了其在复杂、不确定环境中的鲁棒性。3. 实战用LangGraph构建一个Plan-and-Solve旅行规划Agent理论说得再多不如动手实现一个。下面我们以构建一个“智能旅行规划Agent”为例展示如何用LangGraph框架实现一个基本的Plan-and-Solve工作流。我们将使用Python和OpenAI的GPT-4作为规划与反思的LLM。注意以下代码为示例逻辑需要你具备基本的Python和LangChain/LangGraph知识并配置好OpenAI API密钥。3.1 定义状态与工具首先我们需要定义Agent工作流中流转的“状态”。这是一个Pydantic模型包含了所有必要的信息。from typing import TypedDict, List, Annotated, Union from langgraph.graph import StateGraph, END from langchain_core.messages import HumanMessage, SystemMessage from langchain_openai import ChatOpenAI import operator from datetime import datetime import sys # 定义工作流状态 class AgentState(TypedDict): # 用户输入 goal: str constraints: List[str] # 规划阶段产出 plan: List[dict] # 每个元素如 {step_id: int, “task”: str, “tool”: str, “status”: “pending”/“done”/“failed”} # 执行上下文 current_step: int context: dict # 存储每一步的输入输出例如 {“step_1”: {“input”: “…”, “output”: “…”}} # 最终结果 final_answer: Union[str, None] # 反思与错误信息 reflection: str needs_replan: bool接下来定义几个简单的“工具”。在实际应用中这些工具应该连接真实的API。# 模拟工具函数 def search_attractions(query: str, filters: dict None) - str: 模拟搜索景点的工具 # 这里应该是调用真实API如Google Places, 百度地图等 print(f“[工具调用] search_attractions: {query}”) # 返回模拟数据 mock_data [ {“name”: “故宫博物院”, “type”: “文化古迹”, “time_needed”: “4小时”, “ticket_price”: 60, “location”: “东城区”}, {“name”: “中国科学技术馆”, “type”: “亲子科普”, “time_needed”: “3小时”, “ticket_price”: 30, “location”: “朝阳区”}, {“name”: “颐和园”, “type”: “文化古迹”, “time_needed”: “3小时”, “ticket_price”: 30, “location”: “海淀区”}, ] return f“根据查询‘{query}’找到以下景点{mock_data}” def search_restaurants(location: str, cuisine: str None, budget_per_person: int None) - str: 模拟搜索餐厅的工具 print(f“[工具调用] search_restaurants: 位置{location}, 菜系{cuisine}, 预算{budget_per_person}”) mock_data [ {“name”: “全聚德王府井店”, “cuisine”: “北京烤鸭”, “avg_cost”: 150, “location”: “东城区”}, {“name”: “小吊梨汤”, “cuisine”: “京菜”, “avg_cost”: 80, “location”: “多家分店”}, ] return f“在{location}附近找到餐厅{mock_data}” def calculate_budget(items: List[dict]) - str: 模拟计算预算的工具 print(f“[工具调用] calculate_budget: 计算{len(items)}项花费”) total sum(item.get(‘price‘, 0) for item in items) return f“总计预算为{total}元” # 将工具封装为LangChain Tool对象此处简化实际需用tool装饰器或StructuredTool tools [search_attractions, search_restaurants, calculate_budget] tool_map {func.__name__: func for func in tools}3.2 构建规划节点规划节点接收初始状态包含用户目标调用LLM生成计划。llm ChatOpenAI(model“gpt-4-turbo”, temperature0) def planner_node(state: AgentState) - AgentState: 规划节点根据目标和约束生成计划 goal state[“goal”] constraints state.get(“constraints”, []) # 构建给LLM的提示词 system_prompt “””你是一个专业的旅行规划专家。请将用户的旅行规划请求分解为具体、可执行、有逻辑顺序的子任务步骤。 每个步骤需要说明1. 步骤ID 2. 任务描述 3. 建议使用的工具从以下工具中选择search_attractions, search_restaurants, calculate_budget, reasoning。 其中‘reasoning’工具表示仅需推理不调用外部API。 输出格式请严格遵循以下JSON示例 { “plan”: [ {“step_id”: 1, “task”: “任务描述1”, “tool”: “tool_name1”}, {“step_id”: 2, “task”: “任务描述2”, “tool”: “tool_name2”} ] } “”” user_prompt f“”” 用户目标{goal} 约束条件{‘ ‘.join(constraints) if constraints else ‘无’} 请生成规划。 “”” messages [SystemMessage(contentsystem_prompt), HumanMessage(contentuser_prompt)] response llm.invoke(messages) # 解析LLM的回复这里假设它返回了合法的JSON import json try: plan_data json.loads(response.content) plan_list plan_data.get(“plan”, []) # 为每个计划项初始化状态 for item in plan_list: item[“status”] “pending” state[“plan”] plan_list state[“current_step”] 0 # 准备从第一个步骤开始 print(f“[规划完成] 生成了{len(plan_list)}个步骤的计划。”) except json.JSONDecodeError: state[“reflection”] “规划器返回了无法解析的格式。” state[“needs_replan”] True print(“[规划错误] LLM返回非JSON格式。”) return state3.3 构建执行路由与工具调用节点我们需要一个路由函数决定当前是执行工具还是进入反思或结束。def should_continue(state: AgentState) - str: 判断下一步该去哪执行工具、反思、还是结束 if state.get(“needs_replan”): return “replan” if state[“current_step”] len(state[“plan”]): return “finalize” return “execute” def execution_node(state: AgentState) - AgentState: 执行节点运行当前步骤对应的工具 current_step_index state[“current_step”] plan_item state[“plan”][current_step_index] task_desc plan_item[“task”] tool_name plan_item[“tool”] print(f“[执行步骤 {current_step_index 1}] {task_desc} (使用工具: {tool_name})”) # 准备工具输入这里简化处理实际应根据任务描述动态提取参数 # 更复杂的实现可以用一个LLM来解析任务描述提取工具调用参数。 tool_input task_desc # 简化直接把任务描述当输入 try: if tool_name “reasoning”: # 对于推理步骤直接让LLM思考 reasoning_prompt f“请根据已有上下文进行推理分析。任务{task_desc}。当前上下文{state.get(‘context‘, {})}” result llm.invoke([HumanMessage(contentreasoning_prompt)]).content elif tool_name in tool_map: # 调用对应的工具函数 result tool_map[tool_name](tool_input) else: result f“错误未知的工具 ‘{tool_name}’” # 记录结果到上下文 step_key f“step_{current_step_index}” state.setdefault(“context”, {})[step_key] {“input”: task_desc, “output”: result, “tool”: tool_name} # 标记该步骤完成 plan_item[“status”] “done” print(f“[执行成功] 结果摘要{result[:100]}…”) # 移动到下一步 state[“current_step”] 1 except Exception as e: # 工具调用失败 plan_item[“status”] “failed” state[“reflection”] f“步骤 {current_step_index 1} 执行失败错误{str(e)}” state[“needs_replan”] True print(f“[执行失败] 错误{e}”) return state3.4 构建反思与重规划节点反思节点评估当前状态决定是否需要以及如何重规划。def reflection_node(state: AgentState) - AgentState: 反思节点分析问题决定是否及如何重规划 reflection state.get(“reflection”, “”) print(f“[进入反思] 反思信息{reflection}”) # 这里可以调用LLM进行更复杂的分析判断是局部调整还是全局重规划 # 例如询问LLM“基于当前错误‘{reflection}’我们应该修改计划中的哪些步骤还是重新生成整个计划” # 本例中我们简化处理如果失败就回到规划节点让规划器根据当前上下文重新规划。 # 首先我们需要将已完成的步骤信息提供给规划器作为上下文。 # 清空“needs_replan”标志以便重新进入规划流程。 state[“needs_replan”] False # 注意这里我们选择回到规划节点。更精细的设计可以有一个独立的‘replanner’节点。 # 为了简化我们在规划器节点中增加根据已有上下文进行重规划的逻辑。 # 这需要修改之前的planner_node使其能接收‘partial_plan’和‘context’。 # 由于篇幅我们在此仅示意触发重规划意味着回到规划节点但状态中携带了错误信息。 print(“[反思决策] 决定触发重规划。”) # 重规划时可以保留已成功步骤的结果只重新规划失败及后续步骤。 # 这里我们简单地将当前失败步骤的状态重置为pending并再次尝试执行实际可能需调整计划。 # 更优解是调用一个‘replan_node’专门处理。 return state def finalize_node(state: AgentState) - AgentState: 最终节点汇总所有结果生成最终答案 print(“[进入最终汇总]”) context state.get(“context”, {}) # 调用LLM根据所有步骤的结果生成一份友好的最终报告 summary_prompt f“”” 你是一个旅行助理。请根据以下所有步骤的执行结果整合成一份完整的、用户友好的旅行规划方案。 步骤上下文信息 {context} 用户最初的目标是{state[‘goal’]} 请直接输出最终方案。 “”” final_report llm.invoke([HumanMessage(contentsummary_prompt)]).content state[“final_answer”] final_report print(“[流程结束] 最终报告已生成。”) return state3.5 组装工作流图最后使用LangGraph将这些节点连接起来形成一个完整的工作流。# 初始化状态图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(“planner”, planner_node) workflow.add_node(“executor”, execution_node) workflow.add_node(“reflector”, reflection_node) workflow.add_node(“finalizer”, finalize_node) # 设置入口点 workflow.set_entry_point(“planner”) # 添加边定义节点间的流转逻辑 workflow.add_conditional_edges( “planner”, # 规划完成后根据状态决定下一步 lambda state: “executor” if state.get(“plan”) and not state.get(“needs_replan”) else “reflector”, {“executor”: “executor”, “reflector”: “reflector”} ) workflow.add_conditional_edges( “executor”, should_continue, # 这个函数判断下一步去哪 {“execute”: “executor”, “replan”: “reflector”, “finalize”: “finalizer”} ) workflow.add_edge(“reflector”, “planner”) # 反思后回到规划器重新规划 workflow.add_edge(“finalizer”, END) # 汇总后结束 # 编译图 app workflow.compile() # 运行Agent initial_state { “goal”: “为我规划一个为期两天的北京家庭游第一天侧重历史文化第二天侧重现代科技与亲子总预算一家三口不超过5000元。”, “constraints”: [“时间: 2天”, “人员: 3人家庭”, “总预算: ≤5000元”, “主题: 历史科技亲子”], “plan”: [], “current_step”: 0, “context”: {}, “final_answer”: None, “reflection”: “”, “needs_replan”: False } print(“开始执行旅行规划Agent...\n”) final_state app.invoke(initial_state) print(“\n” “”*50) print(“最终规划方案”) print(“”*50) print(final_state[“final_answer”])这个示例虽然简化但清晰地展示了Plan-and-Solve在LangGraph中的实现骨架规划Planner - 执行Executor - 条件判断 - 可选反思/重规划Reflector - 最终汇总Finalizer。在实际开发中你需要强化每个环节特别是规划器的提示工程、执行器的参数解析以及反思器的决策逻辑。4. 避坑指南Plan-and-Solve实战中的常见挑战与对策将Plan-and-Solve范式应用到生产环境时你会遇到许多在Demo中不曾显现的挑战。以下是我在多个项目中总结出的关键问题和应对策略。4.1 规划器的“幻觉”与不稳定性问题LLM作为规划器其生成的计划可能包含不存在的工具、逻辑上不可行的步骤顺序或者完全误解了任务约束。每次调用可能产生差异较大的计划导致系统行为不稳定。对策结构化输出与强验证必须强制LLM以严格的、可解析的格式如JSON Schema、Pydantic模型输出计划。在接收到计划后立即进行程序化验证检查工具名是否在注册列表中检查步骤依赖关系是否形成循环检查必要参数是否齐全。少样本提示Few-Shot Prompting在系统提示词中提供2-3个高质量、不同场景的规划示例。这能极大地对齐LLM的输出格式和逻辑质量。约束显式化不要依赖LLM从自然语言描述中自行提取所有约束。在提示词中明确列出所有已知约束如可用工具[A, B, C]必须遵守的格式XXX并要求规划器确认。多规划器投票Ensemble Planning对于关键任务可以并行调用多个规划器例如使用不同提示词或不同模型然后通过一个简单的投票或选择机制如选择步骤数最合理、工具使用最规范的来确定最终计划。这能提高计划的可靠性。4.2 上下文管理与信息传递的损耗问题随着执行步骤增多上下文所有历史步骤的输入输出会变得非常庞大。直接将完整的上下文扔给下一步的LLM或工具会很快耗尽令牌限制且包含大量无关信息干扰模型判断。对策摘要与提炼在每个主要阶段结束后用一个LLM调用对当前阶段的成果进行摘要用简洁、结构化的语言替换掉冗长的原始输出。例如将10个景点的原始JSON列表摘要为“已筛选出3个符合亲子与文化主题的景点故宫4h60元、科技馆3h30元…”。向量化检索将每一步的输入输出存储在向量数据库中。当需要历史信息时不是传递全部而是根据当前步骤的查询从向量库中检索最相关的几条历史记录。这类似于给Agent增加了“工作记忆”。结构化状态设计如我们示例中的AgentState要精心设计其字段。使用字典、列表来分门别类地存储信息而不是将所有东西混在一个字符串里。这样执行节点可以精准地读取它所需的那部分状态。4.3 工具调用的可靠性与错误处理问题外部工具API可能失败、超时或返回意外格式的数据。一个步骤的失败不应导致整个Agent崩溃。对策完善的工具封装每个工具函数内部必须有完整的异常捕获try-catch。返回的结果应该是一个标准格式例如一个包含success、data、error_message字段的对象。重试机制对于网络超时等临时性错误工具层或执行器层应自动重试如最多3次并采用指数退避策略。结果验证工具返回后不要直接相信结果。增加一个“验证”环节可以是规则校验也可以是小模型调用检查结果是否基本符合预期例如搜索餐厅的结果是否包含人均价格字段价格是否为数字。验证失败则标记为工具调用“部分失败”触发反思。备选工具为关键功能设计备选工具。例如主搜索API失败后自动切换到备用搜索API。4.4 反思与重规划的触发条件与成本问题频繁反思和重规划会极大增加延迟和token消耗。但不反思又可能导致Agent在错误道路上越走越远。对策分层反思策略轻量级规则检查在每个步骤后先用简单的规则检查如“结果是否为空”“数字是否在合理范围内”。规则检查失败立即标记异常成本极低。中量级LLM检查对于规则无法判断的如“行程安排是否合理”再调用小模型如GPT-3.5进行快速评估。重量级深度反思仅当多个步骤失败或最终结果明显偏离目标时才调用大模型如GPT-4进行全面的重规划分析。设置重规划预算为整个工作流设定一个最大重规划次数例如3次。超过次数后Agent应优雅失败并向用户反馈当前遇到无法自动解决的困难可能需要用户提供更多信息或简化任务。局部重规划优先设计重规划逻辑时优先尝试只重新规划从失败点开始的后续步骤而不是推翻全部。这需要规划器有能力接受一个“部分完成的计划”作为输入。4.5 评估与持续改进问题如何知道你的Plan-and-Solve Agent是否真的比直接提示Zero-Shot的LLM更好对策建立测试集构建一个涵盖不同复杂度、不同领域的任务测试集。每个任务有明确的成功标准例如生成的行程必须满足所有约束且景点间交通时间不超过1小时。定义评估指标任务完成率有多少比例的任务被成功完成产出最终答案且通过校验约束满足率在完成的任务中平均满足了多少条用户约束步骤效率完成一个任务平均需要调用多少次工具多少次LLM人工评分随机抽样让人类评估员对结果的质量、合理性和创造性进行打分。A/B测试在相同任务集上对比Plan-and-Solve Agent和直接提示的基线模型的表现。用数据证明其价值。迭代优化根据评估结果和错误日志持续优化提示词、工具设计、重规划策略等。这是一个数据驱动的迭代过程。Plan-and-Solve不是银弹它引入了额外的复杂性和开销。但对于那些步骤清晰、约束明确、对可靠性和逻辑性要求高的任务它所提供的结构化和可控性是简单提示方法难以比拟的。它代表了当前构建复杂、可靠AI Agent的主流方向之一。 AI Agent规划与执行：Plan-and-Solve范式详解与LangGraph实战
1. 从“直接问”到“先规划”为什么Agent需要Plan-and-Solve如果你用过早期的AI助手或者直接向大语言模型LLM提过一个稍微复杂点的问题比如“帮我策划一个为期三天的北京旅游行程要包含文化古迹、特色美食和适合亲子活动的场所预算控制在人均5000元以内”你大概率会收到一份看起来“还行”的清单。景点、餐厅、活动似乎都提到了但仔细一看行程安排可能极不合理第一天上午在故宫下午就安排去了八达岭长城完全忽略了交通时间或者推荐的餐厅人均消费远超预算导致总花费超标。这种回答就是典型的“直接动手边做边想”模式——模型根据你的指令直接开始生成内容缺乏一个全局的、结构化的思考过程。这就是“Plan-and-Solve”范式要解决的核心痛点。在AI Agent智能体的开发领域我们越来越意识到让LLM直接生成最终答案对于简单任务尚可但对于多步骤、有约束条件、需要逻辑推理的复杂任务效果往往不尽人意。Agent作为能够感知环境、进行决策并执行动作的智能程序其核心能力之一就是“规划”。“Plan-and-Solve”直译过来就是“先规划再解决”它强调在执行具体动作之前先让Agent或者说驱动Agent的LLM制定一个清晰的、分步骤的计划。这听起来像是常识但在AI领域实现它却需要一套精巧的框架设计。它不仅仅是让模型在回答前加一句“让我想想”而是要求模型能明确地分解任务、识别子目标、评估资源与约束并规划出可行的执行路径。这与人类处理复杂问题的方式高度一致工程师在写代码前会画架构图项目经理在启动项目前会做WBS工作分解结构我们旅行前也会做攻略。“Plan-and-Solve”的本质是将LLM从一个“即时反应”的文本生成器升级为一个具备“前瞻性思考”和“结构化问题解决”能力的推理引擎。近年来随着LangChain、LangGraph等框架的流行以及ReActReasoning and Acting等范式的提出为Agent实现“Plan-and-Solve”提供了强大的工具链。ReAct范式鼓励模型将“思考”Reasoning和“行动”Acting交织进行在每一步行动前都输出一段推理链。而“Plan-and-Solve”可以看作是ReAct的一种前置强化和结构化延伸在开始任何“行动”之前先进行一轮集中的、宏观的“规划”Plan产出整个任务的工作流蓝图然后再按照这个蓝图去逐步“解决”Solve。这能显著提升复杂任务处理的可靠性、一致性和可解释性。2. Plan-and-Solve的核心组件与工作流拆解一个典型的、实现了Plan-and-Solve范式的Agent工作流通常包含几个关键组件它们协同工作将“规划”的思想落地。理解这些组件是设计和开发此类Agent的基础。2.1 规划器Planner任务的“总设计师”规划器是Plan-and-Solve架构的大脑通常由一个或多个LLM调用构成。它的核心职责是接收用户的初始请求或称为“目标”并输出一个结构化的计划。这个计划不是一段笼统的描述而应该是一个可操作的任务分解清单。规划器需要完成以下工作任务理解与澄清首先它需要精准理解用户意图。对于模糊的请求它可能需要发起追问例如“您说的‘高端’餐厅具体是指人均消费在什么范围”。这一步确保了规划的起点是准确的。约束与条件识别自动提取或确认任务中的限制条件如时间、预算、格式要求、可用工具API等。例如从“生成一份季度报告”中识别出需要财务数据API从“订机票”中识别出日期和预算。任务分解Decomposition将宏观目标拆解为一系列有序的、原子级的子任务。例如“策划旅游行程”可分解为1) 确定每日主题2) 为每天上午、下午、晚上筛选景点3) 为每个时间段附近的景点匹配餐厅4) 估算交通时间和方式5) 核算总预算。依赖关系分析确定子任务之间的前后顺序和依赖关系。有些任务可以并行有些必须串行。例如“查询航班信息”必须在“比较航班价格”之前而“预订酒店”和“预订景点门票”在时间确定后可以并行进行。资源分配与工具选择为每个子任务分配合适的“工具”Tool。Agent的能力来源于其可调用的工具如搜索API、计算器、代码执行器、数据库查询等。规划器需要决定“查询天气”用天气API“计算汇率”用计算器工具。一个规划良好的输出应该接近于一个项目计划表可以是JSON、YAML格式或者是一段结构极其清晰的文本。例如{ “goal”: “策划三天北京亲子游行程预算人均5000元” “constraints”: [“总时长: 3天”, “预算: ≤5000元/人”, “主题: 文化美食亲子”], “plan”: [ {“step”: 1, “task”: “确定每日核心主题和住宿区域”, “tool”: “reasoning”, “depends_on”: []}, {“step”: 2, “task”: “基于主题搜索并筛选第1天上午的文化古迹景点”, “tool”: “search_api”, “depends_on”: [1]}, {“step”: 3, “task”: “为第1天上午的景点匹配附近适合家庭的午餐餐厅并估算费用”, “tool”: “search_api calculator”, “depends_on”: [2]}, // ... 更多步骤 {“step”: N, “task”: “汇总所有信息生成格式优美的最终行程文档”, “tool”: “report_generator”, “depends_on”: [所有前置步骤]} ] }2.2 执行器Executor/ 调度器Orchestrator计划的“施工队长”规划器产出蓝图后需要有一个组件来按图施工这就是执行器在LangGraph等框架中常称为“调度器”。它的职责是工作流驱动按照规划器输出的步骤顺序依次启动每个子任务。它需要管理任务状态知道当前执行到哪一步下一步该执行什么。工具调用对于每个需要工具的子任务执行器负责准备输入参数调用对应的工具函数并获取返回结果。例如调用搜索API时需要将“北京适合亲子的博物馆”作为查询关键词传入。上下文管理这是关键所在。执行器需要维护一个“工作上下文”它包含了初始目标、已完成的步骤及其结果、当前步骤的输入输出等所有信息。这个上下文会随着执行流程逐步丰富并作为后续步骤的输入。例如步骤2搜索到的景点列表会被传递给步骤3用于寻找附近的餐厅。异常处理与重规划计划赶不上变化。当工具调用失败如API返回错误、返回结果不符合预期如搜索不到合适景点或遇到未预见的约束冲突如预算超标时执行器不能直接崩溃。它需要有能力捕获异常并将当前状态包括错误信息反馈给规划器或一个专门的“重规划”模块触发局部或全局的重新规划。LangGraph框架在这方面提供了强大的支持。它允许你用“图”来定义Agent的工作流节点Node代表状态检查或工具调用边Edge代表状态流转的条件。通过这种显式的图结构你可以非常直观地构建出包含规划、执行、条件判断、循环用于重试或重规划的复杂Agent逻辑。2.3 反思器Reflector与重规划机制系统的“质检员”一个健壮的Plan-and-Solve系统不能是“一锤子买卖”。最初的计划可能基于不完整的信息或错误的假设。因此引入“反思”环节至关重要。反思器通常在以下时机被触发每个主要步骤完成后检查结果是否满足该子任务的目标质量如何。遇到失败或意外结果时分析失败原因。所有步骤执行完毕后最终检查评估最终结果是否真正满足了用户的初始目标。反思器本身也可以是一个LLM调用它接收当前上下文和最新结果并回答诸如“当前的结果是否解决了子任务X”“预算是否仍然可控”“到目前为止的行程安排在交通时间上是否合理”等问题。如果反思器判断当前路径有问题它可以建议退回某一步进行重试或者直接触发“重规划”Replanning。重规划不一定需要推倒重来。根据问题的严重性可以是局部调整只重新规划当前出错步骤及后续受影响的步骤。全局重规划完全重新分析任务生成新计划。这通常发生在发现核心假设错误时例如用户后来补充了“其中一天要去天津”。将反思和重规划机制嵌入工作流使得Agent具备了动态适应和从错误中学习的能力大大提升了其在复杂、不确定环境中的鲁棒性。3. 实战用LangGraph构建一个Plan-and-Solve旅行规划Agent理论说得再多不如动手实现一个。下面我们以构建一个“智能旅行规划Agent”为例展示如何用LangGraph框架实现一个基本的Plan-and-Solve工作流。我们将使用Python和OpenAI的GPT-4作为规划与反思的LLM。注意以下代码为示例逻辑需要你具备基本的Python和LangChain/LangGraph知识并配置好OpenAI API密钥。3.1 定义状态与工具首先我们需要定义Agent工作流中流转的“状态”。这是一个Pydantic模型包含了所有必要的信息。from typing import TypedDict, List, Annotated, Union from langgraph.graph import StateGraph, END from langchain_core.messages import HumanMessage, SystemMessage from langchain_openai import ChatOpenAI import operator from datetime import datetime import sys # 定义工作流状态 class AgentState(TypedDict): # 用户输入 goal: str constraints: List[str] # 规划阶段产出 plan: List[dict] # 每个元素如 {step_id: int, “task”: str, “tool”: str, “status”: “pending”/“done”/“failed”} # 执行上下文 current_step: int context: dict # 存储每一步的输入输出例如 {“step_1”: {“input”: “…”, “output”: “…”}} # 最终结果 final_answer: Union[str, None] # 反思与错误信息 reflection: str needs_replan: bool接下来定义几个简单的“工具”。在实际应用中这些工具应该连接真实的API。# 模拟工具函数 def search_attractions(query: str, filters: dict None) - str: 模拟搜索景点的工具 # 这里应该是调用真实API如Google Places, 百度地图等 print(f“[工具调用] search_attractions: {query}”) # 返回模拟数据 mock_data [ {“name”: “故宫博物院”, “type”: “文化古迹”, “time_needed”: “4小时”, “ticket_price”: 60, “location”: “东城区”}, {“name”: “中国科学技术馆”, “type”: “亲子科普”, “time_needed”: “3小时”, “ticket_price”: 30, “location”: “朝阳区”}, {“name”: “颐和园”, “type”: “文化古迹”, “time_needed”: “3小时”, “ticket_price”: 30, “location”: “海淀区”}, ] return f“根据查询‘{query}’找到以下景点{mock_data}” def search_restaurants(location: str, cuisine: str None, budget_per_person: int None) - str: 模拟搜索餐厅的工具 print(f“[工具调用] search_restaurants: 位置{location}, 菜系{cuisine}, 预算{budget_per_person}”) mock_data [ {“name”: “全聚德王府井店”, “cuisine”: “北京烤鸭”, “avg_cost”: 150, “location”: “东城区”}, {“name”: “小吊梨汤”, “cuisine”: “京菜”, “avg_cost”: 80, “location”: “多家分店”}, ] return f“在{location}附近找到餐厅{mock_data}” def calculate_budget(items: List[dict]) - str: 模拟计算预算的工具 print(f“[工具调用] calculate_budget: 计算{len(items)}项花费”) total sum(item.get(‘price‘, 0) for item in items) return f“总计预算为{total}元” # 将工具封装为LangChain Tool对象此处简化实际需用tool装饰器或StructuredTool tools [search_attractions, search_restaurants, calculate_budget] tool_map {func.__name__: func for func in tools}3.2 构建规划节点规划节点接收初始状态包含用户目标调用LLM生成计划。llm ChatOpenAI(model“gpt-4-turbo”, temperature0) def planner_node(state: AgentState) - AgentState: 规划节点根据目标和约束生成计划 goal state[“goal”] constraints state.get(“constraints”, []) # 构建给LLM的提示词 system_prompt “””你是一个专业的旅行规划专家。请将用户的旅行规划请求分解为具体、可执行、有逻辑顺序的子任务步骤。 每个步骤需要说明1. 步骤ID 2. 任务描述 3. 建议使用的工具从以下工具中选择search_attractions, search_restaurants, calculate_budget, reasoning。 其中‘reasoning’工具表示仅需推理不调用外部API。 输出格式请严格遵循以下JSON示例 { “plan”: [ {“step_id”: 1, “task”: “任务描述1”, “tool”: “tool_name1”}, {“step_id”: 2, “task”: “任务描述2”, “tool”: “tool_name2”} ] } “”” user_prompt f“”” 用户目标{goal} 约束条件{‘ ‘.join(constraints) if constraints else ‘无’} 请生成规划。 “”” messages [SystemMessage(contentsystem_prompt), HumanMessage(contentuser_prompt)] response llm.invoke(messages) # 解析LLM的回复这里假设它返回了合法的JSON import json try: plan_data json.loads(response.content) plan_list plan_data.get(“plan”, []) # 为每个计划项初始化状态 for item in plan_list: item[“status”] “pending” state[“plan”] plan_list state[“current_step”] 0 # 准备从第一个步骤开始 print(f“[规划完成] 生成了{len(plan_list)}个步骤的计划。”) except json.JSONDecodeError: state[“reflection”] “规划器返回了无法解析的格式。” state[“needs_replan”] True print(“[规划错误] LLM返回非JSON格式。”) return state3.3 构建执行路由与工具调用节点我们需要一个路由函数决定当前是执行工具还是进入反思或结束。def should_continue(state: AgentState) - str: 判断下一步该去哪执行工具、反思、还是结束 if state.get(“needs_replan”): return “replan” if state[“current_step”] len(state[“plan”]): return “finalize” return “execute” def execution_node(state: AgentState) - AgentState: 执行节点运行当前步骤对应的工具 current_step_index state[“current_step”] plan_item state[“plan”][current_step_index] task_desc plan_item[“task”] tool_name plan_item[“tool”] print(f“[执行步骤 {current_step_index 1}] {task_desc} (使用工具: {tool_name})”) # 准备工具输入这里简化处理实际应根据任务描述动态提取参数 # 更复杂的实现可以用一个LLM来解析任务描述提取工具调用参数。 tool_input task_desc # 简化直接把任务描述当输入 try: if tool_name “reasoning”: # 对于推理步骤直接让LLM思考 reasoning_prompt f“请根据已有上下文进行推理分析。任务{task_desc}。当前上下文{state.get(‘context‘, {})}” result llm.invoke([HumanMessage(contentreasoning_prompt)]).content elif tool_name in tool_map: # 调用对应的工具函数 result tool_map[tool_name](tool_input) else: result f“错误未知的工具 ‘{tool_name}’” # 记录结果到上下文 step_key f“step_{current_step_index}” state.setdefault(“context”, {})[step_key] {“input”: task_desc, “output”: result, “tool”: tool_name} # 标记该步骤完成 plan_item[“status”] “done” print(f“[执行成功] 结果摘要{result[:100]}…”) # 移动到下一步 state[“current_step”] 1 except Exception as e: # 工具调用失败 plan_item[“status”] “failed” state[“reflection”] f“步骤 {current_step_index 1} 执行失败错误{str(e)}” state[“needs_replan”] True print(f“[执行失败] 错误{e}”) return state3.4 构建反思与重规划节点反思节点评估当前状态决定是否需要以及如何重规划。def reflection_node(state: AgentState) - AgentState: 反思节点分析问题决定是否及如何重规划 reflection state.get(“reflection”, “”) print(f“[进入反思] 反思信息{reflection}”) # 这里可以调用LLM进行更复杂的分析判断是局部调整还是全局重规划 # 例如询问LLM“基于当前错误‘{reflection}’我们应该修改计划中的哪些步骤还是重新生成整个计划” # 本例中我们简化处理如果失败就回到规划节点让规划器根据当前上下文重新规划。 # 首先我们需要将已完成的步骤信息提供给规划器作为上下文。 # 清空“needs_replan”标志以便重新进入规划流程。 state[“needs_replan”] False # 注意这里我们选择回到规划节点。更精细的设计可以有一个独立的‘replanner’节点。 # 为了简化我们在规划器节点中增加根据已有上下文进行重规划的逻辑。 # 这需要修改之前的planner_node使其能接收‘partial_plan’和‘context’。 # 由于篇幅我们在此仅示意触发重规划意味着回到规划节点但状态中携带了错误信息。 print(“[反思决策] 决定触发重规划。”) # 重规划时可以保留已成功步骤的结果只重新规划失败及后续步骤。 # 这里我们简单地将当前失败步骤的状态重置为pending并再次尝试执行实际可能需调整计划。 # 更优解是调用一个‘replan_node’专门处理。 return state def finalize_node(state: AgentState) - AgentState: 最终节点汇总所有结果生成最终答案 print(“[进入最终汇总]”) context state.get(“context”, {}) # 调用LLM根据所有步骤的结果生成一份友好的最终报告 summary_prompt f“”” 你是一个旅行助理。请根据以下所有步骤的执行结果整合成一份完整的、用户友好的旅行规划方案。 步骤上下文信息 {context} 用户最初的目标是{state[‘goal’]} 请直接输出最终方案。 “”” final_report llm.invoke([HumanMessage(contentsummary_prompt)]).content state[“final_answer”] final_report print(“[流程结束] 最终报告已生成。”) return state3.5 组装工作流图最后使用LangGraph将这些节点连接起来形成一个完整的工作流。# 初始化状态图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(“planner”, planner_node) workflow.add_node(“executor”, execution_node) workflow.add_node(“reflector”, reflection_node) workflow.add_node(“finalizer”, finalize_node) # 设置入口点 workflow.set_entry_point(“planner”) # 添加边定义节点间的流转逻辑 workflow.add_conditional_edges( “planner”, # 规划完成后根据状态决定下一步 lambda state: “executor” if state.get(“plan”) and not state.get(“needs_replan”) else “reflector”, {“executor”: “executor”, “reflector”: “reflector”} ) workflow.add_conditional_edges( “executor”, should_continue, # 这个函数判断下一步去哪 {“execute”: “executor”, “replan”: “reflector”, “finalize”: “finalizer”} ) workflow.add_edge(“reflector”, “planner”) # 反思后回到规划器重新规划 workflow.add_edge(“finalizer”, END) # 汇总后结束 # 编译图 app workflow.compile() # 运行Agent initial_state { “goal”: “为我规划一个为期两天的北京家庭游第一天侧重历史文化第二天侧重现代科技与亲子总预算一家三口不超过5000元。”, “constraints”: [“时间: 2天”, “人员: 3人家庭”, “总预算: ≤5000元”, “主题: 历史科技亲子”], “plan”: [], “current_step”: 0, “context”: {}, “final_answer”: None, “reflection”: “”, “needs_replan”: False } print(“开始执行旅行规划Agent...\n”) final_state app.invoke(initial_state) print(“\n” “”*50) print(“最终规划方案”) print(“”*50) print(final_state[“final_answer”])这个示例虽然简化但清晰地展示了Plan-and-Solve在LangGraph中的实现骨架规划Planner - 执行Executor - 条件判断 - 可选反思/重规划Reflector - 最终汇总Finalizer。在实际开发中你需要强化每个环节特别是规划器的提示工程、执行器的参数解析以及反思器的决策逻辑。4. 避坑指南Plan-and-Solve实战中的常见挑战与对策将Plan-and-Solve范式应用到生产环境时你会遇到许多在Demo中不曾显现的挑战。以下是我在多个项目中总结出的关键问题和应对策略。4.1 规划器的“幻觉”与不稳定性问题LLM作为规划器其生成的计划可能包含不存在的工具、逻辑上不可行的步骤顺序或者完全误解了任务约束。每次调用可能产生差异较大的计划导致系统行为不稳定。对策结构化输出与强验证必须强制LLM以严格的、可解析的格式如JSON Schema、Pydantic模型输出计划。在接收到计划后立即进行程序化验证检查工具名是否在注册列表中检查步骤依赖关系是否形成循环检查必要参数是否齐全。少样本提示Few-Shot Prompting在系统提示词中提供2-3个高质量、不同场景的规划示例。这能极大地对齐LLM的输出格式和逻辑质量。约束显式化不要依赖LLM从自然语言描述中自行提取所有约束。在提示词中明确列出所有已知约束如可用工具[A, B, C]必须遵守的格式XXX并要求规划器确认。多规划器投票Ensemble Planning对于关键任务可以并行调用多个规划器例如使用不同提示词或不同模型然后通过一个简单的投票或选择机制如选择步骤数最合理、工具使用最规范的来确定最终计划。这能提高计划的可靠性。4.2 上下文管理与信息传递的损耗问题随着执行步骤增多上下文所有历史步骤的输入输出会变得非常庞大。直接将完整的上下文扔给下一步的LLM或工具会很快耗尽令牌限制且包含大量无关信息干扰模型判断。对策摘要与提炼在每个主要阶段结束后用一个LLM调用对当前阶段的成果进行摘要用简洁、结构化的语言替换掉冗长的原始输出。例如将10个景点的原始JSON列表摘要为“已筛选出3个符合亲子与文化主题的景点故宫4h60元、科技馆3h30元…”。向量化检索将每一步的输入输出存储在向量数据库中。当需要历史信息时不是传递全部而是根据当前步骤的查询从向量库中检索最相关的几条历史记录。这类似于给Agent增加了“工作记忆”。结构化状态设计如我们示例中的AgentState要精心设计其字段。使用字典、列表来分门别类地存储信息而不是将所有东西混在一个字符串里。这样执行节点可以精准地读取它所需的那部分状态。4.3 工具调用的可靠性与错误处理问题外部工具API可能失败、超时或返回意外格式的数据。一个步骤的失败不应导致整个Agent崩溃。对策完善的工具封装每个工具函数内部必须有完整的异常捕获try-catch。返回的结果应该是一个标准格式例如一个包含success、data、error_message字段的对象。重试机制对于网络超时等临时性错误工具层或执行器层应自动重试如最多3次并采用指数退避策略。结果验证工具返回后不要直接相信结果。增加一个“验证”环节可以是规则校验也可以是小模型调用检查结果是否基本符合预期例如搜索餐厅的结果是否包含人均价格字段价格是否为数字。验证失败则标记为工具调用“部分失败”触发反思。备选工具为关键功能设计备选工具。例如主搜索API失败后自动切换到备用搜索API。4.4 反思与重规划的触发条件与成本问题频繁反思和重规划会极大增加延迟和token消耗。但不反思又可能导致Agent在错误道路上越走越远。对策分层反思策略轻量级规则检查在每个步骤后先用简单的规则检查如“结果是否为空”“数字是否在合理范围内”。规则检查失败立即标记异常成本极低。中量级LLM检查对于规则无法判断的如“行程安排是否合理”再调用小模型如GPT-3.5进行快速评估。重量级深度反思仅当多个步骤失败或最终结果明显偏离目标时才调用大模型如GPT-4进行全面的重规划分析。设置重规划预算为整个工作流设定一个最大重规划次数例如3次。超过次数后Agent应优雅失败并向用户反馈当前遇到无法自动解决的困难可能需要用户提供更多信息或简化任务。局部重规划优先设计重规划逻辑时优先尝试只重新规划从失败点开始的后续步骤而不是推翻全部。这需要规划器有能力接受一个“部分完成的计划”作为输入。4.5 评估与持续改进问题如何知道你的Plan-and-Solve Agent是否真的比直接提示Zero-Shot的LLM更好对策建立测试集构建一个涵盖不同复杂度、不同领域的任务测试集。每个任务有明确的成功标准例如生成的行程必须满足所有约束且景点间交通时间不超过1小时。定义评估指标任务完成率有多少比例的任务被成功完成产出最终答案且通过校验约束满足率在完成的任务中平均满足了多少条用户约束步骤效率完成一个任务平均需要调用多少次工具多少次LLM人工评分随机抽样让人类评估员对结果的质量、合理性和创造性进行打分。A/B测试在相同任务集上对比Plan-and-Solve Agent和直接提示的基线模型的表现。用数据证明其价值。迭代优化根据评估结果和错误日志持续优化提示词、工具设计、重规划策略等。这是一个数据驱动的迭代过程。Plan-and-Solve不是银弹它引入了额外的复杂性和开销。但对于那些步骤清晰、约束明确、对可靠性和逻辑性要求高的任务它所提供的结构化和可控性是简单提示方法难以比拟的。它代表了当前构建复杂、可靠AI Agent的主流方向之一。