1. Agent为什么需要不同设计模式随着大语言模型的发展LLM已经具备了强大的理解和生成能力。但是单纯调用LLM仍然存在一些问题无法主动调用外部工具面对复杂任务容易迷失缺少任务规划能力输出结果缺少自我检查因此出现了Agent。简单来说Agent LLM Tools Memory Workflow不同任务需要不同Agent设计模式。本文基于一个简单Agent Demo实现三种经典范式ReActPlan-and-SolveReflection项目结构demo1 ├── .env ├── llm_client.py ├── prompt.py ├── tools.py ├── ReAct.py ├── Plan_and_solve.py └── Reflection.py其中llm_client.py负责大模型调用封装tools.py提供Agent可调用工具prompt.py管理不同Agent提示词llm_client.pyimport os from openai import OpenAI from dotenv import load_dotenv from typing import List, Dict # 加载 .env 文件中的环境变量 load_dotenv() class HelloAgentsLLM: 为本书 Hello Agents 定制的LLM客户端。 它用于调用任何兼容OpenAI接口的服务并默认使用流式响应。 def __init__(self, model: str None, apiKey: str None, baseUrl: str None, timeout: int None): 初始化客户端。优先使用传入参数如果未提供则从环境变量加载。 self.model model or os.getenv(LLM_MODEL_ID) apiKey apiKey or os.getenv(LLM_API_KEY) baseUrl baseUrl or os.getenv(LLM_BASE_URL) timeout timeout or int(os.getenv(LLM_TIMEOUT, 60)) if not all([self.model, apiKey, baseUrl]): raise ValueError(模型ID、API密钥和服务地址必须被提供或在.env文件中定义。) self.client OpenAI(api_keyapiKey, base_urlbaseUrl, timeouttimeout) def think(self, messages: List[Dict[str, str]], temperature: float 0) - str: 调用大语言模型进行思考并返回其响应。 print(f 正在调用 {self.model} 模型...) try: response self.client.chat.completions.create( modelself.model, messagesmessages, temperaturetemperature, streamTrue, ) # 处理流式响应 print(✅ 大语言模型响应成功:) collected_content [] for chunk in response: if not chunk.choices: continue content chunk.choices[0].delta.content or print(content, end, flushTrue) collected_content.append(content) print() # 在流式输出结束后换行 return .join(collected_content) except Exception as e: print(f❌ 调用LLM API时发生错误: {e}) return None # --- 客户端使用示例 --- if __name__ __main__: try: llmClient HelloAgentsLLM() exampleMessages [ {role: system, content: You are a helpful assistant that writes Python code.}, {role: user, content: 写一个快速排序算法} ] print(--- 调用LLM ---) responseText llmClient.think(exampleMessages) if responseText: print(\n\n--- 完整模型响应 ---) print(responseText) except ValueError as e: print(e) # # --- 调用LLM - -- # 正在调用 # xxxxxx # 模型... # ✅ 大语言模型响应成功: # 快速排序是一种非常高效的排序算法...tools.pyfrom serpapi import SerpApiClient from typing import Dict, Any import os from dotenv import load_dotenv load_dotenv() def search(query: str) - str: 一个基于SerpApi的实战网页搜索引擎工具。 它会智能地解析搜索结果优先返回直接答案或知识图谱信息。 print(f 正在执行 [SerpApi] 网页搜索: {query}) try: api_key os.getenv(SERPAPI_API_KEY) if not api_key: return 错误:SERPAPI_API_KEY 未在 .env 文件中配置。 params { engine: google, q: query, api_key: api_key, gl: cn, # 国家代码 hl: zh-cn, # 语言代码 } client SerpApiClient(params) results client.get_dict() # 智能解析:优先寻找最直接的答案 if answer_box_list in results: return \n.join(results[answer_box_list]) if answer_box in results and answer in results[answer_box]: return results[answer_box][answer] if knowledge_graph in results and description in results[knowledge_graph]: return results[knowledge_graph][description] if organic_results in results and results[organic_results]: # 如果没有直接答案则返回前三个有机结果的摘要 snippets [ f[{i 1}] {res.get(title, )}\n{res.get(snippet, )} for i, res in enumerate(results[organic_results][:3]) ] return \n\n.join(snippets) return f对不起没有找到关于 {query} 的信息。 except Exception as e: return f搜索时发生错误: {e} class ToolExecutor: 一个工具执行器负责管理和执行工具。 def __init__(self): self.tools: Dict[str, Dict[str, Any]] {} def registerTool(self, name: str, description: str, func: callable): 向工具箱中注册一个新工具。 if name in self.tools: print(f警告:工具 {name} 已存在将被覆盖。) self.tools[name] {description: description, func: func} print(f工具 {name} 已注册。) def getTool(self, name: str) - callable: 根据名称获取一个工具的执行函数。 return self.tools.get(name, {}).get(func) def getAvailableTools(self) - str: 获取所有可用工具的格式化描述字符串。 return \n.join([ f- {name}: {info[description]} for name, info in self.tools.items() ]) # # --- 工具初始化与使用示例 --- # if __name__ __main__: # # 1. 初始化工具执行器 # toolExecutor ToolExecutor() # # # 2. 注册我们的实战搜索工具 # search_description 一个网页搜索引擎。当你需要回答关于时事、事实以及在你的知识库中找不到的信息时应使用此工具。 # toolExecutor.registerTool(Search, search_description, search) # # # 3. 打印可用的工具 # print(\n--- 可用的工具 ---) # print(toolExecutor.getAvailableTools()) # # # 4. 智能体的Action调用这次我们问一个实时性的问题 # print(\n--- 执行 Action: Search[英伟达最新的GPU型号是什么] ---) # tool_name Search # tool_input 英伟达最新的GPU型号是什么 # # tool_function toolExecutor.getTool(tool_name) # if tool_function: # observation tool_function(tool_input) # print(--- 观察 (Observation) ---) # print(observation) # else: # print(f错误:未找到名为 {tool_name} 的工具。)prompt.py# ReAct 提示词模板 REACT_PROMPT_TEMPLATE 请注意你是一个有能力调用外部工具的智能助手。 可用工具如下: {tools} 请严格按照以下格式进行回应: Thought: 你的思考过程用于分析问题、拆解任务和规划下一步行动。 Action: 你决定采取的行动必须是以下格式之一: - {{tool_name}}[{{tool_input}}]:调用一个可用工具。 - Finish[最终答案]:当你认为已经获得最终答案时。 - 当你收集到足够的信息能够回答用户的最终问题时你必须在Action:字段后使用 Finish[最终答案] 来输出最终答案。 现在请开始解决以下问题: Question: {question} History: {history} 下面主要介绍三个Agent核心实现。2. ReAct Agent让模型边思考边行动2.1 ReAct是什么ReActReasoning Acting核心思想让LLM模拟人类解决问题过程思考 ↓ 行动 ↓ 观察结果 ↓ 继续思考例如用户查询今天北京天气Agent第一步Thought: 我不知道实时天气需要查询工具 Action: WeatherTool(北京)工具返回Observation: 北京今天晴25℃然后Thought: 已经获得结果可以回答用户 Finish: 北京今天晴天2.2 ReAct执行流程整体流程User Question ↓ LLM ↓ Thought ↓ Action ↓ Tool ↓ Observation ↓ 继续推理 ↓ Final Answer2.3 ReAct核心代码ReAct.pyimport re from llm_client import HelloAgentsLLM from tools import ToolExecutor, search # (此处省略 REACT_PROMPT_TEMPLATE 的定义) REACT_PROMPT_TEMPLATE 请注意你是一个有能力调用外部工具的智能助手。 可用工具如下 {tools} 请严格按照以下格式进行回应 Thought: 你的思考过程用于分析问题、拆解任务和规划下一步行动。 Action: 你决定采取的行动必须是以下格式之一 - {{tool_name}}[{{tool_input}}]调用一个可用工具。 - Finish[最终答案]当你认为已经获得最终答案时。 - 当你收集到足够的信息能够回答用户的最终问题时你必须在Action:字段后使用 Finish[最终答案] 来输出最终答案。 现在请开始解决以下问题 Question: {question} History: {history} class ReActAgent: def __init__(self, llm_client: HelloAgentsLLM, tool_executor: ToolExecutor, max_steps: int 5): self.llm_client llm_client self.tool_executor tool_executor self.max_steps max_steps self.history [] def run(self, question: str): self.history [] current_step 0 while current_step self.max_steps: current_step 1 print(f\n--- 第 {current_step} 步 ---) tools_desc self.tool_executor.getAvailableTools() history_str \n.join(self.history) prompt REACT_PROMPT_TEMPLATE.format(toolstools_desc, questionquestion, historyhistory_str) messages [{role: user, content: prompt}] response_text self.llm_client.think(messagesmessages) if not response_text: print(错误LLM未能返回有效响应。); break thought, action self._parse_output(response_text) if thought: print(f 思考: {thought}) if not action: print(警告未能解析出有效的Action流程终止。); break if action.startswith(Finish): # 如果是Finish指令提取最终答案并结束 final_answer self._parse_action_input(action) print(f 最终答案: {final_answer}) return final_answer tool_name, tool_input self._parse_action(action) if not tool_name or not tool_input: self.history.append(Observation: 无效的Action格式请检查。); continue print(f 行动: {tool_name}[{tool_input}]) tool_function self.tool_executor.getTool(tool_name) observation tool_function(tool_input) if tool_function else f错误未找到名为 {tool_name} 的工具。 print(f 观察: {observation}) self.history.append(fAction: {action}) self.history.append(fObservation: {observation}) print(已达到最大步数流程终止。) return None def _parse_output(self, text: str): # Thought: 匹配到 Action: 或文本末尾 thought_match re.search(rThought:\s*(.*?)(?\nAction:|$), text, re.DOTALL) # Action: 匹配到文本末尾 action_match re.search(rAction:\s*(.*?)$, text, re.DOTALL) thought thought_match.group(1).strip() if thought_match else None action action_match.group(1).strip() if action_match else None return thought, action def _parse_action(self, action_text: str): match re.match(r(\w)\[(.*)\], action_text, re.DOTALL) return (match.group(1), match.group(2)) if match else (None, None) def _parse_action_input(self, action_text: str): match re.match(r\w\[(.*)\], action_text, re.DOTALL) return match.group(1) if match else if __name__ __main__: llm HelloAgentsLLM() tool_executor ToolExecutor() search_desc 一个网页搜索引擎。当你需要回答关于时事、事实以及在你的知识库中找不到的信息时应使用此工具。 tool_executor.registerTool(Search, search_desc, search) agent ReActAgent(llm_clientllm, tool_executortool_executor) question 华为最新的手机是哪一款它的主要卖点是什么 agent.run(question)ReAct最核心的就是这个循环Thought → Action → Observation2.4 ReAct特点优点动态性强Agent可以根据工具返回结果调整下一步行为。适合搜索数据查询API调用缺点任务规划能力有限面对复杂任务例如写一份行业分析报告可能出现步骤混乱重复执行因此出现下一种模式。3. Plan-and-Solve Agent先规划再执行3.1 为什么需要Plan-and-SolveReAct边想边做但是复杂任务通常需要先设计路线 再开始执行例如生成一份市场分析报告。不能直接写第一段 查资料 继续写更合理先拆分1. 收集行业信息 2. 分析市场规模 3. 总结发展趋势 4. 输出报告这就是Plan-and-Solve。3.2 核心架构Plan-and-Solve包含两个核心组件用户问题 | ↓ Planner规划器 | 任务计划 | ↓ Executor执行器 | 最终结果3.3 Plan-and-Solve代码Planner负责把复杂问题拆解。例如输入计算水果店三天销量输出[ 计算周二销量, 计算周三销量, 计算总销量 ]Executor负责按照计划逐步执行。Plan_and_solve.pyimport ast PLANNER_PROMPT_TEMPLATE 你是一个顶级的AI规划专家。你的任务是将用户提出的复杂问题分解成一个由多个简单步骤组成的行动计划。 请确保计划中的每个步骤都是一个独立的、可执行的子任务并且严格按照逻辑顺序排列。 你的输出必须是一个Python列表其中每个元素都是一个描述子任务的字符串。 问题: {question} 请严格按照以下格式输出你的计划,python与作为前后缀是必要的: python [步骤1, 步骤2, 步骤3, ...] # 假定 llm_client.py 中的 HelloAgentsLLM 类已经定义好 from llm_client import HelloAgentsLLM class Planner: def __init__(self, llm_client): self.llm_client llm_client def plan(self, question: str) - list[str]: 根据用户问题生成一个行动计划。 prompt PLANNER_PROMPT_TEMPLATE.format(questionquestion) # 为了生成计划我们构建一个简单的消息列表 messages [{role: user, content: prompt}] print(--- 正在生成计划 ---) # 使用流式输出来获取完整的计划 response_text self.llm_client.think(messagesmessages) or print(f✅ 计划已生成:\n{response_text}) # 解析LLM输出的列表字符串 try: # 找到python和之间的内容 plan_str response_text.split(python)[1].split()[0].strip() # 使用ast.literal_eval来安全地执行字符串将其转换为Python列表 plan ast.literal_eval(plan_str) return plan if isinstance(plan, list) else [] except (ValueError, SyntaxError, IndexError) as e: print(f❌ 解析计划时出错: {e}) print(f原始响应: {response_text}) return [] except Exception as e: print(f❌ 解析计划时发生未知错误: {e}) return [] EXECUTOR_PROMPT_TEMPLATE 你是一位顶级的AI执行专家。你的任务是严格按照给定的计划一步步地解决问题。 你将收到原始问题、完整的计划、以及到目前为止已经完成的步骤和结果。 请你专注于解决“当前步骤”并仅输出该步骤的最终答案不要输出任何额外的解释或对话。 # 原始问题: {question} # 完整计划: {plan} # 历史步骤与结果: {history} # 当前步骤: {current_step} 请仅输出针对“当前步骤”的回答: class Executor: def __init__(self, llm_client): self.llm_client llm_client def execute(self, question: str, plan: list[str]) - str: 根据计划逐步执行并解决问题。 history # 用于存储历史步骤和结果的字符串 print(\n--- 正在执行计划 ---) for i, step in enumerate(plan): print(f\n- 正在执行步骤 {i 1}/{len(plan)}: {step}) prompt EXECUTOR_PROMPT_TEMPLATE.format( questionquestion, planplan, historyhistory if history else 无, # 如果是第一步则历史为空 current_stepstep ) messages [{role: user, content: prompt}] response_text self.llm_client.think(messagesmessages) or # 更新历史记录为下一步做准备 history f步骤 {i 1}: {step}\n结果: {response_text}\n\n print(f✅ 步骤 {i 1} 已完成结果: {response_text}) # 循环结束后最后一步的响应就是最终答案 final_answer response_text return final_answer class PlanAndSolveAgent: def __init__(self, llm_client): 初始化智能体同时创建规划器和执行器实例。 self.llm_client llm_client self.planner Planner(self.llm_client) self.executor Executor(self.llm_client) def run(self, question: str): 运行智能体的完整流程:先规划后执行。 print(f\n--- 开始处理问题 ---\n问题: {question}) # 1. 调用规划器生成计划 plan self.planner.plan(question) # 检查计划是否成功生成 if not plan: print(\n--- 任务终止 --- \n无法生成有效的行动计划。) return # 2. 调用执行器执行计划 final_answer self.executor.execute(question, plan) print(f\n--- 任务完成 ---\n最终答案: {final_answer}) # --- 5. 主函数入口 --- if __name__ __main__: try: llm_client HelloAgentsLLM() agent PlanAndSolveAgent(llm_client) question 一个水果店周一卖出了15个苹果。周二卖出的苹果数量是周一的两倍。周三卖出的数量比周二少了5个。请问这三天总共卖出了多少个苹果 agent.run(question) except ValueError as e: print(e)3.4 Plan-and-Solve特点优势任务结构更加清晰适合长文本生成数据分析代码生成多步骤推理不足计划错误会影响后续执行如果Planner拆错后面的Executor也会受到影响。4. Reflection Agent让Agent自我优化4.1 为什么需要Reflection前面的Agent完成任务后就结束。但是第一次结果一定正确吗例如LLM生成代码第一次普通遍历判断素数虽然正确但是效率低。Reflection可以让Agent生成 ↓ 检查 ↓ 优化 ↓ 重新生成4.2 Reflection核心流程三个阶段Execution 生成初稿 ↓ Reflection 发现问题 ↓ Refinement 优化结果4.3 Memory设计Reflection需要记录历史。例如records [ { type:execution, content:第一次生成代码 }, { type:reflection, content:时间复杂度过高 } ]作用让模型知道上一次做了什么为什么需要修改4.4 Reflection Agent代码Reflection.pyfrom typing import List, Dict, Any, Optional from demo1.llm_client import HelloAgentsLLM class Memory: 一个简单的短期记忆模块用于存储智能体的行动与反思轨迹。 def __init__(self): 初始化一个空列表来存储所有记录。 self.records: List[Dict[str, Any]] [] def add_record(self, record_type: str, content: str): 向记忆中添加一条新记录。 参数: - record_type (str): 记录的类型 (execution 或 reflection)。 - content (str): 记录的具体内容 (例如生成的代码或反思的反馈)。 record {type: record_type, content: content} self.records.append(record) print(f 记忆已更新新增一条 {record_type} 记录。) def get_trajectory(self) - str: 将所有记忆记录格式化为一个连贯的字符串文本用于构建提示词。 trajectory_parts [] for record in self.records: if record[type] execution: trajectory_parts.append(f--- 上一轮尝试 (代码) ---\n{record[content]}) elif record[type] reflection: trajectory_parts.append(f--- 评审员反馈 ---\n{record[content]}) return \n\n.join(trajectory_parts) def get_last_execution(self) - Optional[str]: 获取最近一次的执行结果 (例如最新生成的代码)。 如果不存在则返回 None。 for record in reversed(self.records): if record[type] execution: return record[content] return None INITIAL_PROMPT_TEMPLATE 你是一位资深的Python程序员。请根据以下要求编写一个Python函数。 你的代码必须包含完整的函数签名、文档字符串并遵循PEP 8编码规范。 要求: {task} 请直接输出代码不要包含任何额外的解释。 REFLECT_PROMPT_TEMPLATE 你是一位极其严格的代码评审专家和资深算法工程师对代码的性能有极致的要求。 你的任务是审查以下Python代码并专注于找出其在strong算法效率/strong上的主要瓶颈。 # 原始任务: {task} # 待审查的代码: python {code} 请分析该代码的时间复杂度并思考是否存在一种strong算法上更优/strong的解决方案来显著提升性能。 如果存在请清晰地指出当前算法的不足并提出具体的、可行的改进算法建议例如使用筛法替代试除法。 如果代码在算法层面已经达到最优才能回答“无需改进”。 请直接输出你的反馈不要包含任何额外的解释。 REFINE_PROMPT_TEMPLATE 你是一位资深的Python程序员。你正在根据一位代码评审专家的反馈来优化你的代码。 # 原始任务: {task} # 你上一轮尝试的代码: {last_code_attempt} 评审员的反馈 {feedback} 请根据评审员的反馈生成一个优化后的新版本代码。 你的代码必须包含完整的函数签名、文档字符串并遵循PEP 8编码规范。 请直接输出优化后的代码不要包含任何额外的解释。 # 假设 llm_client.py 和 memory.py 已定义 # from llm_client import HelloAgentsLLM # from memory import Memory class ReflectionAgent: def __init__(self, llm_client, max_iterations3): self.llm_client llm_client self.memory Memory() self.max_iterations max_iterations def run(self, task: str): print(f\n--- 开始处理任务 ---\n任务: {task}) # --- 1. 初始执行 --- print(\n--- 正在进行初始尝试 ---) initial_prompt INITIAL_PROMPT_TEMPLATE.format(tasktask) initial_code self._get_llm_response(initial_prompt) self.memory.add_record(execution, initial_code) # --- 2. 迭代循环:反思与优化 --- for i in range(self.max_iterations): print(f\n--- 第 {i 1}/{self.max_iterations} 轮迭代 ---) # a. 反思 print(\n- 正在进行反思...) last_code self.memory.get_last_execution() reflect_prompt REFLECT_PROMPT_TEMPLATE.format(tasktask, codelast_code) feedback self._get_llm_response(reflect_prompt) self.memory.add_record(reflection, feedback) # b. 检查是否需要停止 if 无需改进 in feedback: print(\n✅ 反思认为代码已无需改进任务完成。) break # c. 优化 print(\n- 正在进行优化...) refine_prompt REFINE_PROMPT_TEMPLATE.format( tasktask, last_code_attemptlast_code, feedbackfeedback ) refined_code self._get_llm_response(refine_prompt) self.memory.add_record(execution, refined_code) final_code self.memory.get_last_execution() print(f\n--- 任务完成 ---\n最终生成的代码:\npython\n{final_code}\n) return final_code def _get_llm_response(self, prompt: str) - str: 一个辅助方法用于调用LLM并获取完整的流式响应。 messages [{role: user, content: prompt}] response_text self.llm_client.think(messagesmessages) or return response_text if __name__ __main__: # 1. 初始化LLM客户端 (请确保你的 .env 和 llm_client.py 文件配置正确) try: llm_client HelloAgentsLLM() except Exception as e: print(f初始化LLM客户端时出错: {e}) exit() # 2. 初始化 Reflection 智能体设置最多迭代2轮 agent ReflectionAgent(llm_client, max_iterations2) # 3. 定义任务并运行智能体 task 编写一个Python函数找出1到n之间所有的素数 (prime numbers)。 agent.run(task)4.5 Reflection运行效果任务生成1-n之间所有素数第一次for循环判断Reflection发现时间复杂度O(n√n) 可以使用埃拉托斯特尼筛法优化优化筛法实现再次检查无需改进最终得到优化代码。5. 三种Agent模式对比模式核心思想适合场景ReAct边思考边行动工具调用、搜索Plan-and-Solve先规划后执行复杂任务Reflection执行后自我优化代码生成、内容优化6. 实际项目中的应用实际开发Agent时并不是三选一。更多情况是组合例如AI客服用户问题 ↓ Planner拆任务 ↓ ReAct调用订单工具 ↓ Reflection检查回答质量 ↓ 返回结果7. 总结本文通过一个简单Demo实现了三种经典Agent范式ReAct解决Agent如何调用工具完成任务Plan-and-Solve解决Agent如何处理复杂任务Reflection解决Agent如何发现问题并优化自己这三种模式也是目前Agent框架设计中的重要思想。