1. 项目概述为什么我们需要“先计划再执行”的智能体如果你已经跟着这个系列教程一路走来从基础的链Chain到简单的代理Agent再到更复杂的多代理协作可能会发现一个瓶颈当面对一个庞大、多步骤、需要深度思考的复杂任务时传统的“思考-行动-观察”循环ReAct模式有时会显得力不从心。它就像一个经验丰富的工匠面对一个复杂家具的制作虽然能一步步来但缺乏一张全局的蓝图容易在细节中迷失方向或者做出短视的决策。这就是“Plan-and-Execute”智能体代理诞生的背景。它的核心思想非常符合人类处理复杂问题的直觉谋定而后动。与其让AI模型在每一步都纠结于“我现在该做什么”不如先让它退一步花点“思考时间”为整个任务制定一个高层次的、分阶段的执行计划。然后再像一个项目经理一样严格按照这个计划调用各种工具去一步步落实。想象一下你要策划一场跨部门的线上发布会。一个传统的ReAct Agent可能会这样第一步它想到“需要确定主题”于是去搜索热门话题第二步它可能又跳去“联系讲者”第三步又回头去“设计海报”……整个过程缺乏条理容易遗漏关键环节比如预算审批、技术测试也容易产生前后矛盾。而一个Plan-and-Execute Agent则会先输出这样一份计划第一阶段策划与筹备明确发布会目标与核心信息。拟定初步预算与时间线。确定并邀请核心讲者与嘉宾。第二阶段内容与物料制作撰写新闻稿、演讲提纲。设计宣传海报、邀请函。准备演示文稿与视频素材。第三阶段执行与宣发配置直播平台与测试。通过邮件、社交媒体发布活动信息。进行会前彩排。第四阶段活动与复盘执行线上直播与互动。收集参会者反馈。活动效果分析与报告撰写。有了这份计划执行器Executor就可以按部就班地调用相应的工具如搜索工具获取行业趋势、邮件工具发送邀请、设计工具生成海报、日历工具安排会议整个过程清晰、可控且更容易回溯和调试。这不仅仅是效率的提升更是任务完成质量和可靠性的飞跃。在LangChain的生态中Plan-and-Execute模式通过planner和executor两个核心组件的分工协作来实现这一理念特别适合需要长期规划、多工具协作、有明确阶段目标的复杂任务场景比如数据分析流水线、自动化报告生成、复杂系统运维等。注意Plan-and-Execute模式会消耗更多的Token因为需要生成计划并且其效果高度依赖于规划阶段LLM的“大局观”能力。对于简单、直接的任务使用它可能有点“杀鸡用牛刀”反而增加开销和延迟。2. 核心架构拆解Planner与Executor如何协同工作要理解Plan-and-Execute Agent我们必须深入其双核心架构规划器Planner和执行器Executor。它们不是简单的先后关系而是一个带有反馈循环的精密系统。2.1 规划器任务的“总设计师”规划器的唯一职责是接收用户的初始任务描述并产出一个结构化的、可执行的行动计划。这个计划通常不是一个简单的待办列表而是一个有层次、有关联的任务树或任务列表。1. 工作原理规划器本身通常也是一个LLMChain或一个特定的Agent。它被赋予一个包含特定指令的系统提示词System Prompt这些指令会要求LLM以某种格式如JSON、Markdown列表、特定分隔符的文本来输出计划。提示词中会明确告诉LLM可用的工具、任务的最终目标并鼓励它进行任务分解、识别依赖关系、预估资源。2. 输出格式示例一个设计良好的规划器可能输出如下JSON{ plan: [ { id: 1, step: 进行市场调研获取近三个月‘智能家居’领域的社交媒体热度趋势。, tool: serpapi_search_tool, inputs: {query: 智能家居 2024 趋势 社交媒体 热度}, dependencies: [] }, { id: 2, step: 根据调研结果生成一份500字的行业分析简报。, tool: llm_chain_writing, inputs: {topic: 智能家居趋势分析, key_points: [STEP_1_RESULT]}, dependencies: [1] }, { id: 3, step: 将分析简报翻译成英文版本。, tool: translation_tool, inputs: {text: [STEP_2_RESULT]}, dependencies: [2] } ] }这种结构化的输出为执行器提供了明确的指令、工具调用参数和步骤依赖关系。3. 关键设计考量提示词工程规划器的提示词至关重要。你需要清晰地定义输出格式并举例说明。例如“请将计划输出为一个JSON列表每个条目包含‘id’、‘description’、‘tool_to_use’和‘depends_on’字段。”LLM选择规划需要较强的逻辑推理、分解和结构化输出能力。通常性能更强的模型如GPT-4、Claude-3在规划任务上表现远优于较小模型。如果使用开源模型需要特别关注其指令遵循和结构化输出能力。工具暴露规划器需要知道“有什么工具可用”。在LangChain中你需要将Tools的列表或描述提供给规划器它才能做出合理的任务分配。2.2 执行器计划的“忠实施工队”执行器接收规划器产生的计划并严格地、顺序地或根据依赖关系执行每一个步骤。它本身可以是一个简单的SequentialChain但更常见的是一个工具调用代理Agent。1. 工作流程解析计划读取当前要执行的步骤ID、描述、指定的工具和输入参数。准备上下文将上一步骤的执行结果如果有依赖填充到当前步骤的输入参数中例如上述JSON中的[STEP_1_RESULT]占位符。调用工具根据tool字段调用对应的工具并传入处理好的inputs参数。收集结果将工具执行的结果保存下来作为该步骤的输出并可能用于后续步骤或最终汇总。迭代推进移动到下一个可执行的步骤检查依赖是否已满足重复上述过程直到所有步骤完成。2. 执行器的智能性虽然叫“执行器”但它并非机械的脚本。它内置的Agent在调用工具时依然会进行必要的“思考”比如理解参数、处理工具的意外输出错误、非预期格式。更重要的是一些高级实现允许动态重规划。即当某个步骤执行失败或结果严重偏离预期时执行器可以将当前状态已完成步骤、失败信息反馈给规划器请求生成一个调整后的新计划。3. 与规划器的交互最简单的模式是“一次性规划线性执行”。但更健壮的模式是“规划-执行-监控-重规划”的闭环。这需要建立一个沟通机制让执行器能将执行状态成功、失败、部分结果反馈给规划器。在LangChain中这可以通过让规划器和执行器共享一个“状态”State对象或者通过一个协调器Orchestrator来管理。实操心得在初期我建议先从“一次性规划”开始验证整个流程。当任务复杂度极高或外部环境不确定时再引入重规划逻辑。重规划虽然强大但会显著增加复杂度和Token消耗需要谨慎设计触发条件例如仅当工具返回特定错误码或结果置信度低于阈值时才触发。3. 手把手实战构建一个智能内容创作流水线理论说得再多不如亲手搭建一个。让我们构建一个Plan-and-Execute Agent它能够完成一个相对复杂的任务“为我生成一篇关于‘城市屋顶花园’的推广博客文章并配一张相关的示意图。”这个任务涉及多个子任务信息搜集、内容撰写、图片生成或搜索需要按顺序且有依赖地执行。3.1 环境准备与工具定义首先确保你的环境已安装LangChain和相关依赖。我们需要定义几个关键工具。# 假设使用OpenAI模型和SerpAPI进行搜索 import os from langchain_openai import ChatOpenAI from langchain_community.tools import SerpAPIWrapper from langchain.agents import Tool from langchain.chains import LLMChain from langchain.prompts import PromptTemplate # 1. 初始化LLM (规划器和执行器可能使用相同或不同的模型) llm ChatOpenAI(modelgpt-4-turbo-preview, temperature0) # 规划需要更低的随机性 # 2. 定义搜索工具 (用于信息搜集) search SerpAPIWrapper() search_tool Tool( nameWebSearch, funcsearch.run, description当需要获取最新的、事实性的信息时使用此工具。输入是一个搜索查询字符串。 ) # 3. 定义一个“写作工具”本质上是一个LLMChain writing_prompt PromptTemplate( input_variables[topic, key_points], template你是一位专业的园艺与城市生活博主。请根据以下主题和关键点撰写一篇吸引人、结构清晰的博客文章正文。\n主题{topic}\n关键点{key_points}\n文章 ) writing_chain LLMChain(llmllm, promptwriting_prompt) writing_tool Tool( nameBlogWriter, funcwriting_chain.run, description当需要根据主题和要点撰写长文本内容如博客、报告时使用此工具。输入是topic和key_points。 ) # 4. 定义一个“图片建议工具”模拟实际可能是DALL-E API等 def image_suggestion_tool(theme: str) - str: # 这里模拟调用一个图片生成或搜索API返回图片描述或URL suggestion f基于主题‘{theme}’建议的图片描述一个现代城市公寓屋顶上的郁郁葱葱的绿色花园有菜圃、休闲椅和太阳能灯背景是城市天际线黄昏景色。 return suggestion image_tool Tool( nameImageSuggestor, funcimage_suggestion_tool, description当需要为文章主题生成或寻找配图建议时使用此工具。输入是文章主题或核心描述。 ) # 将所有工具放入列表 tools [search_tool, writing_tool, image_tool]3.2 构建规划器我们将创建一个简单的规划器它接收任务和工具列表输出一个分步骤的文本计划。from langchain.prompts import ChatPromptTemplate, HumanMessagePromptTemplate, SystemMessagePromptTemplate # 规划器的系统提示词 planner_system_template 你是一个高级任务规划专家。你的目标是将一个复杂的用户请求分解成一个详细的、可顺序执行的步骤计划。 你可以使用以下工具 {tools_descriptions} 请遵循以下规则来制定计划 1. 分析任务目标识别必须的子任务。 2. 为每个子任务分配最合适的工具。 3. 明确步骤之间的依赖关系。一个步骤可能需要前一个步骤的输出作为输入。 4. 输出一个清晰的、带编号的步骤列表。每个步骤的格式应为 [步骤序号]. [步骤描述] (使用工具: [工具名称])。输入依赖: [依赖的步骤号]输入参数: [参数说明]。 示例任务“查询北京今天的天气并写一首关于这个天气的打油诗。” 示例计划 1. 查询北京当前的天气情况。 (使用工具: WebSearch)。输入依赖: 无输入参数: “北京 今天 天气”。 2. 根据查询到的天气信息创作一首幽默的打油诗。 (使用工具: BlogWriter)。输入依赖: 1输入参数: topic“天气打油诗” key_points“第一步的搜索结果”。 现在请为以下任务制定计划 planner_prompt ChatPromptTemplate.from_messages([ SystemMessagePromptTemplate.from_template(planner_system_template), HumanMessagePromptTemplate.from_template(用户任务{task}) ]) # 创建规划器Chain planner_chain LLMChain(llmllm, promptplanner_prompt) def create_plan(task: str, tools: list) - str: # 将工具描述格式化 tools_descriptions \n.join([f- {tool.name}: {tool.description} for tool in tools]) # 调用规划器 plan planner_chain.run(tasktask, tools_descriptionstools_descriptions) return plan # 测试规划器 task “为我生成一篇关于‘城市屋顶花园’的推广博客文章并配一张相关的示意图。” plan_text create_plan(task, tools) print(“生成的计划”) print(plan_text)运行后你可能会得到类似这样的计划文本1. 搜索关于“城市屋顶花园”的好处、设计案例和最新趋势信息。 (使用工具: WebSearch)。输入依赖: 无输入参数: “城市屋顶花园 好处 设计案例 趋势 2024”。 2. 基于搜索到的信息撰写一篇关于城市屋顶花园的推广博客文章。 (使用工具: BlogWriter)。输入依赖: 1输入参数: topic“城市屋顶花园打造你的空中绿洲” key_points“第一步的搜索结果摘要”。 3. 为这篇关于城市屋顶花园的文章生成一个配图建议。 (使用工具: ImageSuggestor)。输入依赖: 2输入参数: theme“城市屋顶花园的推广文章”。3.3 构建执行器与协调整个流程现在我们需要一个执行器来解析并执行这个计划。由于计划是文本格式我们需要一个简单的解析器并创建一个顺序执行的工作流。import re from typing import Dict, Any def parse_plan(plan_text: str) - list: 一个简单的解析函数从文本计划中提取步骤信息。 steps [] # 使用正则表达式匹配步骤行 pattern r‘(\d)\.\s*(.?)\s*\(使用工具:\s*(.?)\)。输入依赖:\s*(.?)输入参数:\s*(.)’ matches re.findall(pattern, plan_text) for match in matches: step_id, description, tool_name, dependency_str, input_params_str match # 处理依赖可能是“无”或数字 dependencies [] if ‘无’ not in dependency_str: dependencies [int(dep.strip()) for dep in dependency_str.split(‘’) if dep.strip().isdigit()] # 简化处理将输入参数字符串作为整体实际项目需更精细解析 steps.append({ ‘id’: int(step_id), ‘description’: description.strip(), ‘tool’: tool_name.strip(), ‘dependencies’: dependencies, ‘input_params_raw’: input_params_str.strip() }) return steps def execute_plan(parsed_steps: list, tools_dict: Dict[str, Tool], initial_context: Dict[str, Any] None) - Dict[str, Any]: 顺序执行解析后的计划。 context initial_context or {} results {} # 存储每一步的结果key为步骤ID # 创建一个步骤ID到步骤对象的映射并找到执行顺序拓扑排序简化版按ID顺序假设依赖已满足 steps_by_id {s[‘id’]: s for s in parsed_steps} ordered_step_ids sorted(steps_by_id.keys()) for step_id in ordered_step_ids: step steps_by_id[step_id] # 检查依赖是否都已满足 if not all(dep_id in results for dep_id in step[‘dependencies’]): print(f“步骤 {step_id} 的依赖 {step[‘dependencies’]} 未全部完成跳过。”) continue print(f“\n 正在执行步骤 {step_id}: {step[‘description’]}”) tool_to_use tools_dict.get(step[‘tool’]) if not tool_to_use: print(f“错误未找到工具 ‘{step[‘tool’]}’”) results[step_id] f“Error: Tool {step[‘tool’]} not found.” continue # 准备输入参数这里需要根据input_params_raw和上下文动态解析此处做简化演示 # 假设 input_params_raw 是像 “topic‘城市屋顶花园’ key_points‘第一步的搜索结果摘要’” 这样的字符串 # 我们需要将“第一步的搜索结果摘要”替换为实际的结果 processed_input step[‘input_params_raw’] for dep_id in step[‘dependencies’]: placeholder f‘第{dep_id}步的搜索结果摘要’ # 这里需要根据规划器输出的模式来匹配 if placeholder in processed_input: processed_input processed_input.replace(placeholder, results[dep_id]) # 简化假设工具输入就是一个字符串参数。实际情况需要根据工具定义解析。 try: # 这里调用工具。对于不同工具可能需要不同的参数传递方式。 # 例如search_tool期望一个字符串writing_tool期望一个字典。 # 这是一个需要根据工具具体签名来适配的关键点 if tool_to_use.name “WebSearch”: output tool_to_use.run(processed_input) elif tool_to_use.name “BlogWriter”: # 简陋的参数提取仅用于演示 args {} if ‘topic’ in processed_input: args[‘topic’] processed_input.split(“topic’“)[1].split(”’“)[0] if ‘key_points’ in processed_input: args[‘key_points’] processed_input.split(“key_points’“)[1].split(”’“)[0] output tool_to_use.run(**args) elif tool_to_use.name “ImageSuggestor”: if ‘theme’ in processed_input: arg processed_input.split(“theme’“)[1].split(”’“)[0] output tool_to_use.run(arg) else: output tool_to_use.run(processed_input) else: output tool_to_use.run(processed_input) results[step_id] str(output)[:500] “...” if len(str(output)) 500 else str(output) # 截断长输出 print(f“步骤 {step_id} 执行成功。输出片段{results[step_id][:200]}...”) except Exception as e: error_msg f“执行步骤 {step_id} 时出错{e}” print(error_msg) results[step_id] error_msg return results # 主执行流程 print(“ 开始 Plan-and-Execute 流程 ”) print(“1. 规划阶段...”) plan_text create_plan(task, tools) print(plan_text) print(“\n2. 解析计划...”) parsed_steps parse_plan(plan_text) print(f“解析出 {len(parsed_steps)} 个步骤。”) print(“\n3. 执行阶段...”) tools_dict {tool.name: tool for tool in tools} execution_results execute_plan(parsed_steps, tools_dict) print(“\n 执行完成 ) print(“最终结果摘要”) for step_id, result in execution_results.items(): print(f“步骤{step_id}: {result[:150]}...”)这个示例展示了从规划到执行的基本闭环。parse_plan函数和execute_plan函数中的参数处理部分是最需要根据实际规划器输出格式和工具签名进行定制和强化的这也是实际开发中的主要工作量所在。4. 进阶探讨动态重规划与状态管理基础的Plan-and-Execute在理想情况下工作良好但现实世界充满变数。工具可能失败如搜索无结果、API超时LLM生成的内容可能不符合要求或者执行中途用户提出了修改。这时就需要引入动态重规划的能力。4.1 何时触发重规划重规划不能滥用否则会陷入无限循环。通常在这些情况下考虑工具执行失败工具返回了明确的错误如HTTP 500无效输入。结果质量不达标通过一个“验证器”可以是另一个LLM调用或规则判断步骤结果未达到预期例如搜索到的信息不相关生成的文章离题。外部干预用户主动要求修改任务目标。计划本身不可行执行过程中发现某个步骤的依赖无法满足或指定工具不存在。4.2 如何实现带状态的重规划我们需要一个状态State对象来跟踪整个流程。LangChain本身没有为Plan-and-Execute提供开箱即用的高级状态管理但我们可以设计一个简单的状态机或者利用LangGraphLangChain的另一个库用于构建有状态的、循环的代理工作流来构建更健壮的版本。简化版状态重规划流程初始化状态包含original_taskcurrent_plancompleted_steps字典id-结果failed_steps字典id-错误信息current_step_index等。执行循环从current_plan中获取下一个待执行步骤。尝试执行。如果成功将结果存入completed_steps继续。如果失败将错误存入failed_steps并触发replan标志。重规划调用当replan为真时将当前状态包括已完成步骤的结果和失败信息连同原始任务一起发送给规划器。规划器的提示词需要增强例如“这是原始任务{task}。这是已执行完成的步骤和结果{completed}。这是在步骤X失败的原因{failure}。请基于当前情况重新制定或调整剩余的计划。”规划器生成新计划更新current_plan重置replan标志然后回到执行循环。使用LangGraph的构想LangGraph非常适合建模这种带有循环和状态的工作流。你可以定义两个主要的“节点”PlanNode和ExecuteNode以及一个决定下一步是继续执行还是重新规划的ConditionalEdge。PlanNode接收状态调用规划器LLM更新状态中的计划。ExecuteNode接收状态执行当前步骤更新状态成功/失败。条件判断根据执行结果成功、失败、所有步骤完成来决定下一个节点是ExecuteNode继续执行下一步还是PlanNode需要重规划或者是结束。注意事项动态重规划极大地增加了系统的复杂性。在实现时务必为重规划设置最大次数限制避免死循环。同时要仔细设计传递给重规划器的上下文信息过多的历史信息可能导致Token超限或干扰LLM判断信息过少又可能导致规划器无法做出有效调整。一个平衡的做法是只包含最近几步的结果和关键的失败信息。5. 避坑指南与最佳实践在实际项目中应用Plan-and-Execute模式我踩过不少坑也总结了一些让项目更稳健的经验。1. 规划器的提示词是成败关键明确格式严格要求输出格式JSON, YAML, Markdown列表并在提示词中给出多个清晰、多样的示例。这能极大提高LLM输出结构的稳定性。工具描述要精准提供给规划器的工具描述description必须准确、无歧义地说明工具的用途、输入和输出。模糊的描述会导致规划器错误地分配任务。鼓励分解在提示词中明确要求“将复杂任务分解为不超过7个的原子性子任务”避免规划出过于庞大或模糊的步骤。2. 执行器的鲁棒性设计参数解析与适配如示例所示从规划器的文本输出到工具调用的参数映射是最脆弱的环节。尽量让规划器输出结构化数据如JSON并编写健壮的解析器。考虑使用Pydantic模型来验证和转换步骤数据。错误处理与重试为每个工具调用包裹完善的try-catch。对于网络超时等临时错误加入指数退避的重试机制。结果验证对于关键步骤可以添加一个轻量级的“验证”子步骤。例如写作完成后用一个简单的LLM调用检查文章是否包含核心关键词、是否符合语气要求。3. 性能与成本考量规划开销生成详细计划本身需要消耗Token。对于非常简单的任务直接使用ReAct Agent可能更经济快捷。可以通过任务复杂度判断逻辑来决定启用哪种模式。缓存中间结果执行过程中步骤的结果尤其是搜索得到的大段文本可以缓存起来避免在后续步骤或重规划时重复传递给LLM节省Token。异步执行如果步骤之间没有依赖关系可以考虑并发执行以缩短总耗时。但这需要更复杂的依赖关系分析和状态管理。4. 可观测性与调试完整日志记录规划器生成的原始计划、每个步骤的执行输入/输出、工具调用耗时和任何错误。这对于调试失败案例至关重要。可视化工作流考虑将最终的执行路径包括重规划可视化出来这能帮助非技术背景的同事理解AI的“思考过程”。人工审核点对于高风险任务如发送邮件、发布内容可以在关键步骤如执行前设置“人工审核”节点将中间结果提交给人来确认再继续执行。Plan-and-Execute Agent代表了AI智能体从“反应式”走向“前瞻式”的重要一步。它将人类项目管理的智慧编码进了AI的工作流中通过分离“战略规划”与“战术执行”让AI处理复杂任务的能力变得更加可靠和可解释。虽然实现起来比简单代理复杂但对于那些需要多步骤协作、有明确阶段目标的自动化场景它所提供的清晰度、可控性和最终效果无疑是值得投入的。