1. 项目概述当AI学会“先谋后动”在AI应用开发尤其是基于大语言模型LLM构建智能体Agent的领域里我们常常遇到一个瓶颈面对一个稍微复杂点的任务比如“帮我分析一下上个月的销售数据写一份报告并给出下个季度的三条营销建议”直接丢给一个ChatGPT式的对话模型结果往往不尽如人意。它可能会生成一份结构混乱的报告或者干脆遗漏掉“分析数据”这个关键步骤直接开始编建议。问题的核心在于传统的、单一调用的LLM就像是一个才华横溢但缺乏条理的“天才员工”你给他一个复杂问题他倾向于一次性给出所有答案而缺乏拆解、规划和分步执行的系统性思维。这就是“Plan-and-Execute”模式要解决的核心痛点。这个模式的核心思想非常符合人类处理复杂任务的方式先计划再执行。它让AI智能体不再是一个“一次性反应”的黑盒而是一个拥有“项目经理”思维的协同系统。具体来说它会将用户提出的复杂、开放式任务首先交给一个“规划者”Planner角色去分析。这个规划者通常也是一个LLM它的职责是理解任务全局并将其拆解成一个清晰的、可顺序或并行执行的子任务列表计划。然后这个计划会被交给一个“执行者”Executor角色执行者会严格按计划调用相应的工具如搜索、计算、代码执行、数据库查询等或能力一步步完成每个子任务并最终汇总结果。我之所以花时间深入实践并分享这个模式是因为它在实际项目中带来的提升是颠覆性的。它不仅仅是让输出更规整更重要的是极大地提升了复杂任务处理的可靠性、可解释性和可控性。你可以清晰地看到AI的“思考过程”即计划在哪个环节出了问题也能快速定位。无论是构建一个自动化的数据分析流水线还是一个能够处理多步骤客户咨询的客服机器人Plan-and-Execute都是将LLM从“玩具”升级为“生产级工具”的关键架构之一。接下来我将结合LangChain这个目前最流行的LLM应用框架带你从零开始手把手实现一个功能强大的Plan-and-Execute Agent并分享我在实践中趟过的坑和积累的经验。2. 核心架构与组件深度解析Plan-and-Execute模式的成功依赖于几个核心组件的精密配合。理解每个组件的职责和它们之间的数据流是灵活运用和定制该模式的基础。2.1 规划者Planner任务的“总设计师”规划者是整个系统的“大脑”它负责战略层面的思考。其输入是用户的原始请求输出是一个结构化的计划。1. 核心职责任务理解与拆解深度理解用户意图识别任务中的隐含需求和依赖关系。例如用户说“总结A公司的最新财报并评估其股价影响”规划者需要识别出“获取财报内容”和“查询股价数据”是两个并行或先后依赖的子任务。计划生成将拆解出的子任务组织成一个有序的步骤列表。这个计划需要是具体、可操作的。糟糕的计划是“1. 分析数据2. 生成报告”。好的计划是“1. 使用‘网络搜索’工具关键词‘A公司 2024 Q1 财报摘要’。2. 使用‘股票查询’工具获取A公司近一个月股价数据。3. 调用‘总结分析’工具对比财报亮点与股价波动。4. 使用‘报告撰写’工具整合以上信息生成Markdown格式报告。”2. 技术实现要点在LangChain中规划者本身通常就是一个LLMChain或ChatPromptTemplate。关键在于设计它的提示词Prompt。一个强大的规划提示词需要包含角色定义明确告诉LLM它现在是一个项目规划专家。工具目录清晰列出执行者可用的所有工具及其功能描述。这是规划可行的前提规划者不能建议执行者去用它没有的工具。输出格式规范严格要求LLM以指定的格式如JSON、YAML或带编号的列表输出计划以便程序能自动解析。例如要求输出为[{step: 1, action: tool_name, action_input: {query: ...}}, ...]实操心得规划者的质量直接决定了整个Agent的天花板。实践中我发现使用更强大的LLM如GPT-4作为规划者而用成本更低的模型如GPT-3.5-Turbo作为执行者是一个性价比很高的架构。因为规划需要更强的推理和全局观而执行更多是遵循指令和调用工具。2.2 执行者Executor可靠的“行动派”执行者是系统的“四肢”它负责战术层面的实施。其输入是规划者生成的单个子任务步骤输出是该步骤的执行结果。1. 核心职责工具调用根据计划步骤中的action字段准确选择并调用对应的工具。例如action是web_search它就调用封装好的搜索引擎API。参数传递将action_input中的参数正确地传递给工具。结果处理与传递捕获工具的执行结果成功或失败并将其格式化传递给下一个步骤或作为最终输出的一部分。2. 技术实现要点在LangChain中执行者通常通过AgentExecutor来实现但这里的Agent是“工具使用代理”。我们需要为它配备一个与规划者提示词中描述完全一致的工具集Tools。这个工具集可以包括SerpAPIWrapper用于网络搜索。WikipediaQueryRun用于查询维基百科。PythonREPLTool用于执行Python代码需在安全沙箱中。SqlDatabaseToolkit用于查询数据库。自定义工具通过tool装饰器封装任何API或函数。规划者与执行者的协作流程可以类比为一个高效的“导演-演员”剧组用户提出需求剧本创意。规划者导演研读剧本分解出分镜头列表计划详细到每个场景需要哪位演员工具、什么道具输入、达成什么效果。执行者演员/剧组拿到第一个分镜头指令找到指定的演员调用工具使用道具传入参数完成表演执行并将拍摄结果输出反馈给导演。规划者根据上一个镜头的拍摄结果决定是继续下一个镜头还是需要重拍或调整后续计划处理依赖和错误。这个过程循环直到所有分镜头完成。最终导演或一个汇总模块将所有镜头素材剪辑成片生成最终答案给用户。2.3 状态管理与工作流引擎系统的“调度中心”简单的Plan-and-Execute可以是一个顺序执行的循环。但对于更复杂的、带有条件分支、循环或并行任务的应用就需要引入状态管理和工作流引擎。这就是LangChain的另一个强大模块——LangGraph的用武之地。1. 为什么需要LangGraph想象一下这个任务“监控社交媒体上关于我公司新产品的讨论如果发现负面情绪帖子超过10条则自动生成一份危机预警报告如果是正面帖子则收集起来用作宣传素材。”这个任务包含了“监控”、“条件判断if-else”、“循环”和“不同执行路径”。用简单的线性计划-执行循环很难优雅地实现。2. LangGraph的核心概念图Graph将整个Agent的工作流定义为一个由节点Nodes和边Edges组成的有向图。节点代表一个功能单元比如“规划节点”、“执行节点”、“判断节点”。边定义了节点之间的流转条件。例如从“判断节点”可以引出两条边一条边条件是“情绪为负面”指向“生成预警报告节点”另一条边条件是“情绪为正面”指向“收集素材节点”。状态State一个在所有节点间共享和传递的数据结构通常是一个字典包含了当前的任务描述、已生成的计划、已执行的结果、中间变量等。3. 在Plan-and-Execute中的应用我们可以构建一个两层的图外层图负责整个任务的流程控制。节点包括plan_node调用规划者、execute_node调用执行者处理一个子任务、should_continue_node判断所有子任务是否完成。状态流转plan_node将生成的计划存入状态。execute_node从状态中取出当前要执行的子任务执行后更新结果列表。should_continue_node检查是否还有未执行的子任务并决定下一个节点是execute_node还是结束。通过LangGraph我们实现了非线性的、带状态的Plan-and-Execute Agent能够处理现实世界中绝大多数复杂的业务流程。3. 基于LangChain的实战构建理论说得再多不如一行代码。让我们从一个基础版本开始逐步构建一个功能完整的Plan-and-Execute Agent。这里我们以“获取某科技公司最新动态并分析其对我方业务的影响”为例。3.1 基础环境搭建与工具定义首先确保你的环境已安装必要库并设置好LLM的API密钥这里以OpenAI为例你也可以轻松替换为其他兼容API的模型。pip install langchain langchain-openai langchain-communityimport os from langchain_openai import ChatOpenAI from langchain.agents import Tool, AgentExecutor from langchain.agents.format_scratchpad import format_log_to_str from langchain.agents.output_parsers import ReActSingleInputOutputParser from langchain.tools import Tool from langchain_community.tools import WikipediaQueryRun, DuckDuckGoSearchRun from langchain_community.utilities import WikipediaAPIWrapper # 1. 初始化LLM # 规划者使用更强的模型执行者使用更经济的模型 planner_llm ChatOpenAI(modelgpt-4, temperature0, api_keyos.getenv(OPENAI_API_KEY)) executor_llm ChatOpenAI(modelgpt-3.5-turbo, temperature0, api_keyos.getenv(OPENAI_API_KEY)) # 2. 定义执行者可以使用的工具集 # 工具一网络搜索 search DuckDuckGoSearchRun() search_tool Tool( nameWebSearch, funcsearch.run, descriptionUseful for searching the internet for current events, company news, or general information. Input should be a search query string. ) # 工具二维基百科查询 wiki WikipediaQueryRun(api_wrapperWikipediaAPIWrapper()) wiki_tool Tool( nameWikipedia, funcwiki.run, descriptionUseful for getting factual, encyclopedic information about historical events, concepts, companies, or people. Input should be a clear query topic. ) # 工具三文本总结与分析一个自定义的简单工具 from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser summarize_prompt ChatPromptTemplate.from_template(请对以下文本进行关键点总结和分析\n\n{text}) summarize_chain summarize_prompt | executor_llm | StrOutputParser() def summarize_text(text: str) - str: 对长文本进行总结分析。 return summarize_chain.invoke({text: text}) summarize_tool Tool( nameTextSummarizer, funcsummarize_text, descriptionUseful for summarizing long articles or reports into key points. Input should be the text content as a string. ) # 将所有工具放入列表供后续使用 executor_tools [search_tool, wiki_tool, summarize_tool]注意事项工具的描述description至关重要规划者LLM完全依赖这些描述来决定在什么情况下使用哪个工具。描述要准确、具体说明工具的用途和输入格式。模糊的描述会导致规划者做出错误的选择。3.2 构建规划者Planner接下来我们创建规划者。它的核心是一个精心设计的提示词模板。from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_core.messages import HumanMessage, SystemMessage # 规划者提示词模板 planner_prompt ChatPromptTemplate.from_messages([ SystemMessage(content你是一个顶尖的项目规划专家。你的任务是将用户复杂的请求分解成一个清晰、可执行的分步计划。 你可以指挥一个拥有以下工具的执行团队 {tools_descriptions} 请遵循以下规则生成计划 1. 仔细分析用户请求确保理解所有隐含需求。 2. 将任务分解为具体的、顺序执行的步骤。后一步骤可以使用前一步骤的结果。 3. 每个步骤必须指定使用的工具只能是上面列出的工具之一和该工具所需的精确输入。 4. 如果任务无法用现有工具完成请说明缺少什么工具。 5. 输出格式必须是严格的JSON列表每个元素是一个步骤对象。 示例格式 [ {{step: 1, action: 工具名称, action_input: {{query: 搜索关键词}} }}, {{step: 2, action: 工具名称, action_input: {{text: 待总结的文本}} }} ] 现在开始为以下用户请求制定计划), MessagesPlaceholder(variable_namechat_history), HumanMessage(content{input}) ]) # 构建规划者链 planner_chain planner_prompt | planner_llm这里的关键是{tools_descriptions}我们需要在运行时将工具列表格式化成字符串传入。MessagesPlaceholder用于支持多轮对话让规划者能考虑之前的聊天历史。3.3 构建执行者Executor与基础循环执行者就是一个标准的LangChain Agent我们使用ReAct模式来构建它。from langchain.agents import create_react_agent # 为执行者创建提示词ReAct格式 executor_prompt ChatPromptTemplate.from_messages([ (system, 你是一个可靠的执行者严格按计划步骤行动。你有权使用以下工具\n{tools}\n\n请严格遵循以下格式\nThought: 思考当前步骤该做什么\nAction: 要使用的工具名\nAction Input: 工具的输入\nObservation: 工具返回的结果\n... (这个循环可以重复) \nFinal Answer: 当前步骤的最终输出), MessagesPlaceholder(variable_namechat_history), (human, {input}), MessagesPlaceholder(variable_nameagent_scratchpad), ]) # 创建ReAct Agent executor_agent create_react_agent(executor_llm, executor_tools, executor_prompt) # 创建Agent执行器 executor AgentExecutor(agentexecutor_agent, toolsexecutor_tools, verboseTrue, handle_parsing_errorsTrue)现在我们可以编写一个最基础的顺序执行循环将规划者和执行者串联起来import json import re def basic_plan_and_execute(user_query: str): 基础版的Plan-and-Execute执行函数 print(f用户请求: {user_query}) # 步骤1生成计划 tools_desc \n.join([f- {tool.name}: {tool.description} for tool in executor_tools]) plan_response planner_chain.invoke({ input: user_query, tools_descriptions: tools_desc, chat_history: [] # 初始无历史 }) plan_text plan_response.content print(f生成的计划:\n{plan_text}) # 步骤2解析计划这里需要简单的JSON解析实际中需更健壮的错误处理 try: # 尝试从响应中提取JSON部分 json_match re.search(r\[.*\], plan_text, re.DOTALL) if json_match: plan json.loads(json_match.group()) else: # 如果LLM没有严格按格式输出这里可能失败 plan json.loads(plan_text) except json.JSONDecodeError as e: print(f解析计划失败: {e}) return 抱歉规划阶段出现了错误。 # 步骤3顺序执行计划 final_results [] for step in plan: step_num step.get(step, N/A) action step.get(action) action_input step.get(action_input) if not action or action not in [t.name for t in executor_tools]: print(f步骤{step_num}: 无效的工具指令 - {action}) continue print(f\n--- 执行步骤 {step_num}: {action} ---) # 将action_input字典转换为执行者能理解的字符串格式 # 这里简化处理假设输入是字典我们将其主要值拼接 input_str str(action_input) if isinstance(action_input, dict): # 取第一个值或根据工具需求定制 input_str next(iter(action_input.values()), str(action_input)) try: step_result executor.invoke({input: f请使用工具{action}输入是{input_str}}) final_output step_result.get(output, 无输出) print(f步骤结果: {final_output[:200]}...) # 打印前200字符 final_results.append(f步骤{step_num}结果: {final_output}) except Exception as e: error_msg f步骤{step_num}执行出错: {e} print(error_msg) final_results.append(error_msg) # 步骤4汇总结果这里可以引入另一个LLM来总结或直接返回 print(\n 所有步骤执行完毕 ) return \n\n.join(final_results) # 测试运行 if __name__ __main__: query 查询苹果公司Apple Inc.最新的产品发布会信息并总结其主要亮点。 result basic_plan_and_execute(query) print(\n最终返回给用户的内容) print(result)这个基础版本演示了最核心的“计划-执行”循环。但它非常脆弱比如规划者输出格式不标准、执行步骤间无法传递数据、错误处理简单等。接下来我们就用LangGraph来构建一个更健壮、更强大的版本。4. 使用LangGraph构建健壮的工作流LangGraph允许我们以“图”的形式定义Agent的工作流清晰地管理状态和流程控制。4.1 定义状态与图节点首先我们定义一个共享状态的结构并创建图中各个节点对应的函数。from typing import TypedDict, Annotated, List, Union from langgraph.graph import StateGraph, END import operator # 1. 定义状态结构 class AgentState(TypedDict): Agent工作流的共享状态 input: str # 用户原始输入 plan: List[dict] # 规划步骤列表 completed_steps: List[dict] # 已完成的步骤及结果 current_step: int # 当前要执行的步骤索引 output: str # 最终输出 chat_history: List # 对话历史用于支持多轮 # 2. 创建图构建器 workflow StateGraph(AgentState) # 3. 定义节点函数 def plan_node(state: AgentState) - dict: 规划节点生成任务计划 print([Plan Node] 正在生成计划...) tools_desc \n.join([f- {tool.name}: {tool.description} for tool in executor_tools]) plan_response planner_chain.invoke({ input: state[input], tools_descriptions: tools_desc, chat_history: state.get(chat_history, []) }) # 解析计划这里简化实际应用需要更健壮的解析器 import json, re plan_text plan_response.content try: json_match re.search(r\[.*\], plan_text, re.DOTALL) plan json.loads(json_match.group()) if json_match else [] except: plan [] print(警告计划解析失败使用空计划) # 更新状态 return {plan: plan, current_step: 0, completed_steps: []} def execute_node(state: AgentState) - dict: 执行节点执行当前步骤 step_idx state[current_step] plan state[plan] if step_idx len(plan): return {output: 错误当前步骤索引超出计划范围。} step plan[step_idx] action step.get(action) action_input step.get(action_input, {}) print(f[Execute Node] 正在执行步骤 {step_idx1}: {action}) # 查找对应工具 tool_to_use None for tool in executor_tools: if tool.name action: tool_to_use tool break if not tool_to_use: result f错误未找到工具 {action} else: # 执行工具 try: # 将action_input字典传递给工具函数 if isinstance(action_input, dict): # 这里假设工具函数接受字典参数或需要转换 # 根据工具实际定义调整 result tool_to_use.invoke(action_input) else: result tool_to_use.invoke(str(action_input)) except Exception as e: result f工具执行出错: {e} # 记录完成步骤 completed_step { step_index: step_idx, action: action, input: action_input, result: str(result)[:500] # 截断长结果 } completed_steps state[completed_steps] [completed_step] # 更新状态步骤索引1记录结果 return { completed_steps: completed_steps, current_step: step_idx 1 } def should_continue_node(state: AgentState) - str: 判断节点决定下一步是继续执行还是结束 current state[current_step] total len(state[plan]) print(f[Decision Node] 已完成 {current}/{total} 步) if current total: # 还有步骤未执行回到执行节点 return continue_to_execute else: # 所有步骤完成进入汇总节点 return proceed_to_summarize def summarize_node(state: AgentState) - dict: 汇总节点基于所有步骤结果生成最终答案 print([Summarize Node] 正在生成最终报告...) completed state[completed_steps] original_input state[input] # 将所有步骤的结果拼接成上下文 context_parts [] for step in completed: context_parts.append(f步骤{step[step_index]1}({step[action]})结果: {step[result]}) full_context \n\n.join(context_parts) # 使用LLM生成最终总结性回答 summarizer_prompt ChatPromptTemplate.from_messages([ (system, 你是一个业务分析助理。请根据以下分步执行的结果整合信息直接回应用户最初的请求。回答应专业、简洁、全面。), (human, f用户最初的问题是{original_input}\n\n分步执行收集到的信息如下\n{full_context}) ]) summarizer_chain summarizer_prompt | executor_llm | StrOutputParser() final_answer summarizer_chain.invoke({}) return {output: final_answer}4.2 组装工作流图并运行定义了节点函数后我们将它们添加到图中并设置边流转条件。# 添加节点 workflow.add_node(planner, plan_node) workflow.add_node(executor, execute_node) workflow.add_node(summarizer, summarize_node) # 设置边的起点 workflow.set_entry_point(planner) # 添加边定义节点间的流转 workflow.add_edge(planner, executor) # 计划完成后开始执行第一步 # 执行完一步后需要判断是否继续 workflow.add_conditional_edges( executor, should_continue_node, # 这个函数返回下一个节点的“键名” { continue_to_execute: executor, # 回到执行节点处理下一步 proceed_to_summarize: summarizer # 去汇总节点 } ) workflow.add_edge(summarizer, END) # 汇总完成后结束 # 编译图 app workflow.compile() # 运行Agent工作流 def run_agent_with_graph(user_input: str): 使用LangGraph工作流运行Plan-and-Execute Agent initial_state AgentState( inputuser_input, plan[], completed_steps[], current_step0, output, chat_history[] ) print(*50) print(f开始处理请求: {user_input}) print(*50) # 执行图 final_state app.invoke(initial_state) print(\n *50) print(处理完成) print(*50) return final_state[output] # 测试更复杂的任务 complex_query 请帮我完成一个竞争分析简报 1. 搜索特斯拉Tesla最近一个季度的主要财务数据或新闻。 2. 搜索其竞争对手比如比亚迪近期的市场动态。 3. 基于以上信息总结当前电动汽车市场的竞争格局。 result run_agent_with_graph(complex_query) print(\n生成的竞争分析简报) print(result)通过LangGraph我们构建了一个状态完整、逻辑清晰的工作流。它自动处理了循环执行、状态传递和最终汇总代码结构也比之前的基础循环更加清晰和可维护。你可以通过修改图的结构轻松实现并行执行、条件分支等复杂逻辑。5. 高级技巧与生产级考量将Plan-and-Execute Agent从Demo推向生产环境还需要考虑更多因素。5.1 提升规划质量的策略规划者是系统的瓶颈提升其质量能直接提升整个Agent的水平。少样本示例Few-Shot提示在规划者的提示词中提供2-3个高质量的任务拆解示例。这能极大地引导LLM输出符合格式和逻辑的计划。planner_prompt ChatPromptTemplate.from_messages([ SystemMessage(content你是一个规划专家...), HumanMessage(content示例任务1: 查询北京今天的天气和空气质量。), AIMessage(content[{step:1, action:WebSearch, action_input:{query:北京 今日 天气 空气质量指数}}]), HumanMessage(content示例任务2: 解释什么是量子计算并找一个相关的科普视频。), AIMessage(content[{step:1, action:Wikipedia, action_input:{query:量子计算}}, {step:2, action:WebSearch, action_input:{query:量子计算 科普视频 中文}}]), HumanMessage(content{input}) ])计划验证与重试在规划节点后可以增加一个“计划验证”节点。用另一个LLM或规则检查生成的计划是否合理、工具是否可用。如果不合理则重新规划或提示用户澄清。子目标细化对于极其复杂的任务可以采用分层规划。第一层规划者生成高级目标如“市场调研”、“技术评估”第二层规划者再将每个高级目标细化为具体操作步骤。5.2 执行阶段的稳定性保障执行阶段是出错的高发区需要完善的错误处理。工具调用异常处理网络工具可能超时API可能限流。必须在每个工具调用外包裹try...except并设计重试机制如tenacity库。在状态中记录错误并让规划者或一个专门的“错误处理节点”决定是重试、跳过还是终止。结果验证与过滤执行工具返回的结果可能包含无关信息或错误。可以设计一个“结果清洗”节点使用LLM或规则判断结果是否有效、是否回答了子问题。无效结果可以触发重新执行该步骤。上下文长度管理随着步骤增多所有中间结果都会累积在上下文中可能超过LLM的令牌限制。需要设计策略如只保留最近N个步骤的详细结果更早的结果则用高度概括的摘要代替。5.3 自定义工具的开发与集成LangChain的强大之处在于能轻松集成任何功能作为工具。使用tool装饰器这是创建自定义工具最快捷的方式。from langchain.tools import tool import requests tool def get_weather(city: str) - str: 获取指定城市的当前天气。输入是城市名。 # 这里调用一个模拟的天气API try: # 示例实际请替换为真实API # response requests.get(fhttps://api.weather.com/{city}) # return response.json()[weather] return f{city}的天气是晴朗25摄氏度。 except Exception as e: return f获取天气失败: {e} # 将其加入到工具列表 executor_tools.append(get_weather)集成内部系统你可以将企业内部CRM、ERP的查询接口或者数据库的查询函数封装成工具让Agent能够操作企业内部数据实现真正的业务流程自动化。5.4 性能监控与评估在生产环境中你需要知道你的Agent表现如何。关键指标Metrics任务完成率用户提出的复杂任务有多少被成功、正确地完成。步骤成功率每个子步骤的执行成功率。平均步骤数/耗时处理不同类型任务的平均复杂度与时间。Token消耗区分规划、执行、汇总各阶段的消耗用于成本核算。评估方法人工评估抽样检查黄金标准。基于LLM的自动评估使用一个更强的LLM如GPT-4作为“裁判”根据任务要求和Agent输出从“相关性”、“正确性”、“完整性”等维度打分。端到端测试集构建一个涵盖常见任务类型的测试集定期运行监控指标变化。6. 常见问题与实战排坑指南在实际开发和部署中我遇到了不少典型问题以下是它们的解决方案。问题1规划者生成的计划格式不稳定无法解析。现象LLM有时输出带解释的文本有时JSON格式错误。解决方案强化提示词约束在提示词中明确强调“只输出JSON不要有任何额外解释”。使用输出解析器LangChain提供了JsonOutputParser、PydanticOutputParser等能强制LLM按指定格式输出并在解析失败时进行重试。后处理清洗用正则表达式或简单的字符串处理从响应中提取可能的JSON部分。降级方案如果解析完全失败可以回退到让执行者直接阅读规划者的自然语言描述但这会降低可靠性。问题2执行者错误地选择了工具或输入格式不对。现象规划者指示用WebSearch但执行者却调用了Wikipedia或者给工具传递的输入参数类型错误。解决方案优化工具描述确保描述清晰指出工具的精确用途和输入格式。例如“输入应为单个搜索关键词字符串”。工具选择提示词优化在执行者的Agent提示词中明确要求其“严格按计划中的工具名称选择工具”。输入格式转换在execute_node函数中根据工具的定义将action_input转换为正确的格式。例如如果工具函数接收字符串但action_input是{query: xxx}则需要提取出“xxx”。问题3多步骤任务中后续步骤无法使用前序步骤的结果。现象步骤2需要步骤1的结果作为输入但在我们的基础循环中步骤2的输入是固定的。解决方案这是状态管理的核心。需要在状态中维护一个变量存储区。规划者在生成计划时可以使用特殊标记如{{step1_result}}。执行节点在完成后将结果以step1_result为键存入状态。在执行后续步骤前由一个“变量替换”节点或直接在execute_node中将计划中的标记替换为实际的值。问题4任务无限循环或卡住。现象Agent在某些步骤上反复执行无法推进。解决方案设置最大步数限制在状态中增加steps_taken计数器在should_continue_node中检查是否超过阈值如20步超过则强制跳转到汇总或失败处理节点。检测重复操作在状态中记录最近几步的操作和输入哈希如果检测到完全相同的操作-输入对重复出现则判定为循环触发错误处理。超时控制对整个工作流的执行设置总超时时间。问题5处理用户模糊或不可能完成的任务。现象用户提问“预测明天股市涨跌”这超出了工具能力范围。解决方案规划阶段识别在规划者提示词中要求其“如果任务无法用现有工具完成请说明缺少什么工具”。当规划者返回此类信息时工作流可以直接跳转到向用户澄清的节点。执行阶段反馈当某个工具多次执行失败或返回“无法处理”时将错误信息反馈给一个“重新规划”或“求助用户”节点修改后续计划或直接向用户提问。构建一个成熟的Plan-and-Execute Agent是一个迭代的过程。从最简单的循环开始逐步引入状态管理、错误处理、评估反馈最终形成一个稳定、可靠、能处理真实业务场景的智能系统。这个模式真正释放了LLM在复杂序列决策任务中的潜力是通往更高级别AI自主智能的关键一步。