第四章 智能体经典范式构建
本章任务：
Cli：构建一个基本的客户端，能够成功调用 LLM ，并解析 LLM 流式返回的结果
ReAct(Reasoning and Action)：着重于实现如何让模型在执行过程不断根据当前状况动态地更新行动计划；
Plan-and-Solve：着重于构建模型的规划能力；
Reflection：模型的反思能力。
基本 Cli 前置库：
1 pip install openai python-dotenv
仔细阅读一下 HelloAgentLLM.py 具体干了什么。
初始化 1 2 3 4 5 6 7 8 9 10 11 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_key=apiKey, base_url=baseUrl, timeout=timeout)
没什么好说的，直接根据 .env 的参数构建一个 OpenAI 规范的 LLM 接口。
think 函数 main 函数中，通过构造函数构建一个接口后，调用了 think 函数：
1 2 3 4 5 6 7 8 9 10 11 12 13 print ("请输入文本..." )userPrompt = input () exampleMessages = [ {"role" : "system" , "content" : "You are a helpful assistant that writes Python code." }, {"role" : "user" , "content" : userPrompt} ] print ("--- 调用LLM ---" ) responseText = llmClient.think(exampleMessages) if responseText: print ("\n\n--- 完整模型响应 ---" ) print (responseText)
传入的 Dict 字典中，包含了系统提示词和用户提示词。
think 函数：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 def think (self, messages: List [Dict [str , str ]], temperature: float = 0 ) -> str : """ 调用大语言模型进行思考，并返回其响应。 """ print (f"🧠 正在调用 {self.model} 模型..." ) try : response = self.client.chat.completions.create( model=self.model, messages=messages, temperature=temperature, stream=True , ) print ("✅ 大语言模型响应成功:" ) collected_content = [] for chunk in response: if not chunk.choices: continue content = chunk.choices[0 ].delta.content or "" print (content, end="" , flush=True ) collected_content.append(content) print () return "" .join(collected_content) except Exception as e: print (f"❌ 调用LLM API时发生错误: {e} " ) return None
stream 输出模式，每次都只返回一个增量 chunk ，每次都要打印这个 chunk，并拼接到最终输出。
ReAct：工具 本节需要在 Cli 内实现 Reasoning and Action，即强制让模型按照 思考-行动-观察 的流程，这样模型更容易激活模型的工具意图，从而主动调用工具。
此外，ReAct 模式最重要的是工具，激活工具意图只是让模型拥有调用工具的能力，而工具调用是否可靠，还是需要自行定义。所以本节还涉及单个工具的定义，多个工具的注册发现、调用。
准备工作 前置库：
1 pip install google-search-results
.env 添加：
定义工具 search search 分三步：1.初始化；2.构建 serpapi 接口并调用；3.智能化筛选搜索结果。
第三步中，会检查是否存在 answer_box（Google的答案摘要框）或 knowledge_graph（知识图谱）等信息，如果存在，就直接返回这些最精确的答案。如果不存在，它才会退而求其次，返回前三个常规搜索结果的摘要。这种“智能解析”能为LLM提供质量更高的信息输入。
测试一下：
1 2 if __name__ == '__main__' : print (search("英伟达的历史最高股价是多少" ))
工具调用结果：
1 2 3 4 5 6 7 8 9 🔍 正在执行 [SerpApi] 网页搜索: 英伟达的历史最高股价是多少 [1] 英偉達(NVDA) 過往股價及數據 成交量. 2026年6月26日, 193.12, 195.55, 191.22, 192.53, 192.53, 178,906,300. 2026年6月25日, 200.08, 200.80, 192.13, 195.74, 195.74, 150,205,600. [2] 5万亿美元！英伟达市值究竟有多惊人：几组对比告诉你…… ①英伟达公司上周进一步创造了历史，成为全球首家市值突破5万亿美元的企业； ②这一里程碑式的壮举，或许也预示着这家全球顶尖的芯片公司，对世界经济和金融市场 ... [3] NVIDIA股票图表— NASDAQ:NVDA股票价格 NVDA于2026年5月13日达到历史最高价，价格为236.54 USD，其历史最低价为0.03 USD，于1999年4月25日达到。在NVDA图表上查看更多价格动态。 查看其它股票达到的最高和最低价格
如果存在很多个工具，我们构建一个执行类 ToolExecutor，统一注册工具、执行工具。
初始化 构造工具对象时，其内部存入了一个 Dict 对象：
1 2 def __init__ (self ): self.tools: Dict [str , Dict [str , Any ]] = {}
顾名思义，是用来存储自身工具的注册表。其结构案例为：
1 2 3 4 5 6 7 8 9 10 { "Search": { "description": "一个网页搜索引擎。当你需要回答关于时事...", "func": search.search # search.py 中的 search 函数 } "other": { "description": "another tools needed when...", "func": other.other } }
也就是说，一个注册表应当包含哪些东西：什么时候用这个工具？（自然语言描述） 和 这个工具在哪？（结构化语言）。
注册工具 工具注册，本质是采用 callable 传入工具函数的指针：
1 2 3 4 5 6 7 8 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} ' 已注册。" )
暴露工具列表 1 2 3 4 5 6 7 8 def getAvailableTools (self ) -> str : """ 获取所有可用工具的格式化描述字符串。 """ return "\n" .join([ f"- {name} : {info['description' ]} " for name, info in self.tools.items() ])
提取工具 返回 callable
1 2 3 4 5 def getTool (self, name: str ) -> callable : """ 根据名称获取一个工具的执行函数。 """ return self.tools.get(name, {}).get("func" )
测试 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 if __name__ == '__main__' : toolExecutor = ToolExecutor() search_description = "一个网页搜索引擎。当你需要回答关于时事、事实以及在你的知识库中找不到的信息时，应使用此工具。" toolExecutor.registerTool("Search" , search_description, search.search) print ("\n--- 可用的工具 ---" ) print (toolExecutor.getAvailableTools()) 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} ' 的工具。" )
结果：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 工具 'Search' 已注册。 --- 可用的工具 --- - Search: 一个网页搜索引擎。当你需要回答关于时事、事实以及在你的知识库中找不到的信息时，应使用此工具。 --- 执行 Action: Search['英伟达最新的GPU型号是什么'] --- 🔍 正在执行 [SerpApi] 网页搜索: 英伟达最新的GPU型号是什么 --- 观察 (Observation) --- [1] 比较GeForce 系列最新一代显卡和前代显卡| NVIDIA 比较最新一代RTX 30 系列显卡和前代的RTX 20 系列、GTX 10 和900 系列显卡。查看规格、功能、技术支持等内容。 [2] GeForce RTX 50 系列显卡| NVIDIA GeForce RTX™ 50 系列GPU 搭载NVIDIA Blackwell 架构，为游戏玩家和创作者带来全新玩法。RTX 50 系列具备强大的AI 算力，带来升级体验和更逼真的画面。 [3] 英伟达 GeForce RTX 50 系列GPU 为游戏玩家和创作者带来全新玩法。借助NVIDIA DLSS 4 实现性能倍增，以更快速度生成图像，并通过NVIDIA Studio 平台释放你的创造力。
ReAct：ReActAgent prompt 我们刚刚定义好了工具，也跑通了工具调用，接下来我们要让 LLM 主动调用工具。
我们需要在系统提示词中进行激活（ReActPrompt.py）：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 REACT_PROMPT_TEMPLATE = """ 请注意，你是一个有能力调用外部工具的智能助手。 可用工具如下: {tools} 请严格按照以下格式进行回应: Thought: 你的思考过程，用于分析问题、拆解任务和规划下一步行动。 Action: 你决定采取的行动，必须是以下格式之一: - `{{tool_name}}[{{tool_input}}]`:调用一个可用工具。 - `Finish[最终答案]`:当你认为已经获得最终答案时。 - 当你收集到足够的信息，能够回答用户的最终问题时，你必须在Action:字段后使用 Finish[最终答案] 来输出最终答案。 现在，请开始解决以下问题: Question: {question} History: {history} """
ReActAgent 指导思想 我们之前定义的 HelloAgentLLM 类，通过 think 函数就可以进行单一提问单一回答。而ReAct 是 think 函数的反复调用者，每次得到 Observation 写入历史后，重新调用 think 函数，直到不再调用。
所以 ReActAgent 和 HelloAgentLLM 的关系是什么呢？ReActAgent 内部包含 HelloAgentLLM 对象，用于运行最原子化的 LLM 调用流程。
另外，ReActAgent 类也整合了前面定义的 ToolExecutor 和所有 Tool。
然后，仔细阅读 ReActAgent 怎么写的。
初始化 1 2 3 4 5 def __init__ (self, llm_client: HelloAgentLLM, tool_executor: ToolExecutor, max_steps: int = 5 ): self.llm_client = llm_client self.tool_executor = tool_executor self.max_steps = max_steps self.history = []
构造函数参数：一个 HelloAgentLLM 对象、ToolExecutor 对象（需要注册工具）、步数用来防止无限返回
history = [] 用来存储上下文。
run run 是外部调用接口，开启 ReAct 流程。
第一步：拼接提示词。从 ToolExecutor 获取工具列表、自身上下文历史（history）、用户输入的问题，一同加入 ReActPrompt，完成拼接。
1 2 3 4 5 6 7 8 tools_desc = self.tool_executor.getAvailableTools() history_str = "\n" .join(self.history) prompt = REACT_PROMPT_TEMPLATE.format ( tools=tools_desc, question=question, history=history_str )
第二步：调用 LLM 执行。使用 HelloAgentLLM 的 think 函数调用 LLM 输入拼接后提示词。
1 2 3 4 5 6 7 messages = [{"role" : "user" , "content" : prompt}] response_text = self.llm_client.think(messages=messages) if not response_text: print ("错误:LLM未能返回有效响应。" ) break
第三步：使用自身成员函数 _parse_output 解析出 thought 和 action。这个函数就是简单地匹配关键字。
1 2 3 4 5 6 7 8 9 thought, action = self._parse_output(response_text) if thought: print (f"思考: {thought} " ) if not action: print ("警告:未能解析出有效的Action，流程终止。" ) break
第四步：行动
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 if action.startswith("Finish" ): final_answer = re.match (r"Finish\[(.*)\]" , action).group(1 ) print (f"🎉 最终答案: {final_answer} " ) return final_answer tool_name, tool_input = self._parse_action(action) if not tool_name or not tool_input: continue print (f"🎬 行动: {tool_name} [{tool_input} ]" )tool_function = self.tool_executor.getTool(tool_name) if not tool_function: observation = f"错误:未找到名为 '{tool_name} ' 的工具。" else : observation = tool_function(tool_input)
Action 阶段中，包含了解析 LLM 响应的 Action ，根据解析结果调用工具，调用工具的结果就是 Observation。
第五步：写入历史
1 2 3 4 5 6 print (f"👀 观察: {observation} " )self.history.append(f"Action: {action} " ) self.history.append(f"Observation: {observation} " )
测试 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 if __name__ == '__main__' : toolExecutor = ToolExecutor() search_description = "一个网页搜索引擎。当你需要回答关于时事、事实以及在你的知识库中找不到的信息时，应使用此工具。" toolExecutor.registerTool("Search" , search_description, search.search) agent = ReActAgent( HelloAgentLLM(), toolExecutor, 5 ) print ("请输入文本: " ) userPrompt = input () agent.run(userPrompt)
结果：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 工具 'Search' 已注册。 请输入文本: 英伟达股价多少 --- 第 1 步 --- 🧠 正在调用 deepseek-v4-pro 模型... ✅ 大语言模型响应成功: 思考: 用户询问英伟达的股价。我需要获取最新的股价信息。我将使用搜索引擎搜索“英伟达 股价”或“NVIDIA stock price”。 🎬 行动: Search[NVIDIA stock price today] 🔍 正在执行 [SerpApi] 网页搜索: NVIDIA stock price today 👀 观察: 英伟达，创立于1993年1月，是美国一家以设计和销售图形处理器为主的无厂半导体公司，总部设在加利福尼亚州的圣克拉拉，位于硅谷的中心位置。NVIDIA亦会设计游戏机图形处理器，例如第一代Xbox、PlayStation 3及任天堂Switch和任天堂Switch 2。 --- 第 2 步 --- 🧠 正在调用 deepseek-v4-pro 模型... ✅ 大语言模型响应成功: 思考: 用户询问英伟达的股价，属于实时信息。历史中的搜索未返回股价，需要重新搜索。我将直接用股票代码“NVDA”搜索最新价格。 🎬 行动: Search[NVDA 股价 今天] 🔍 正在执行 [SerpApi] 网页搜索: NVDA 股价 今天 👀 观察: 英伟达，创立于1993年1月，是美国一家以设计和销售图形处理器为主的无厂半导体公司，总部设在加利福尼亚州的圣克拉拉，位于硅谷的中心位置。NVIDIA亦会设计游戏机图形处理器，例如第一代Xbox、PlayStation 3及任天堂Switch和任天堂Switch 2。 --- 第 3 步 --- 🧠 正在调用 deepseek-v4-pro 模型... ✅ 大语言模型响应成功: 思考: 之前的搜索只返回了公司简介，没有提供股价。我需要尝试更精确的查询来获取英伟达今天的股票价格，例如包括股票代码和“today”等关键词。 🎬 行动: Search[NVDA stock price today per share] 🔍 正在执行 [SerpApi] 网页搜索: NVDA stock price today per share 👀 观察: 英伟达，创立于1993年1月，是美国一家以设计和销售图形处理器为主的无厂半导体公司，总部设在加利福尼亚州的圣克拉拉，位于硅谷的中心位置。NVIDIA亦会设计游戏机图形处理器，例如第一代Xbox、PlayStation 3及任天堂Switch和任天堂Switch 2。 --- 第 4 步 --- 🧠 正在调用 deepseek-v4-pro 模型... ✅ 大语言模型响应成功: 思考: 前三次搜索返回的都是英伟达公司的简介，没有得到具体的股价信息。可能是搜索引擎返回的结果被截断了，或者当前时间（2025年4月9日）的实时股价数据没有被正确抓取。我需要尝试更精确地搜索，比如直接搜索股票代码和具体日期，或者搜索财经网站。 🎬 行动: Search[NVDA 股价 2025-04-09] 🔍 正在执行 [SerpApi] 网页搜索: NVDA 股价 2025-04-09 👀 观察: [1] 英偉達(NVDA) 過往股價及數據 英偉達(NVDA) ; 2025年9月9日, 169.09, 170.98, 166.74, 170.76 ; 2025年9月8日, 167.55, 170.96, 167.35, 168.31 ... [2] 英伟达公司(NVDA)_美股行情_今日股价与走势图_新浪财经 新浪财经为您提供英伟达公司(NVDA)实时行情数据。包括最新股价、今日涨跌幅、技术走势图、历史数据查询及最新相关新闻资讯，助力您把握美股市场投资机会。 [3] 英伟达(NVDA)股票股价_股价行情_讨论 雪球为您提供英伟达(NVDA)股票股价,实时行情,热门讨论,最新资讯,最新公告,公司资料,财务数据,财务指标分析等与英伟达(NVDA)股票相关的信息与服务. --- 第 5 步 --- 🧠 正在调用 deepseek-v4-pro 模型... ✅ 大语言模型响应成功: 思考: 我需要获取英伟达（NVDA）的最新股价。虽然之前的搜索没有得到实时数据，但最后一条搜索返回了新浪财经和雪球的链接，可能包含行情。为了提供准确信息，我将再次搜索“NVDA 最新股价”或直接点击链接。由于无法直接点击链接，我将尝试搜索“NVDA 股价 2025-04-09 实时”来获取当日价格。 🎬 行动: Search[NVDA 股价 2025-04-09 实时] 🔍 正在执行 [SerpApi] 网页搜索: NVDA 股价 2025-04-09 实时 👀 观察: [1] 英伟达公司(NVDA)_美股行情_今日股价与走势图_新浪财经 新浪财经为您提供英伟达公司(NVDA)实时行情数据。包括最新股价、今日涨跌幅、技术走势图、历史数据查询及最新相关新闻资讯，助力您把握美股市场投资机会。 [2] 英伟达(NVDA)最新价格_行情_走势图 手机东方财富网是东方财富网的手机门户网站，为亿万用户打造一个手机联通世界的超级平台，提供7*24小时全面及时的财经中文资讯，内容覆盖国内外突发事件、股市资讯、全球新闻 [3] 英偉達(NVDA) 股價、新聞、報價和記錄 前收市價 195.74 · 開市 193.10 · 買入價 192.00 x 100 · 賣出價 191.85 x 200 · 今日波幅 191.22 - 195.55 · 52 週波幅 151.49 - 236.54 · 成交量 177,757,849 · 平均成交量 ... 已达到最大步数，流程终止。
像这种涉及到精确数字的问题，往往只能在特定网站上搜索，比如各种交易所官网，大模型大概率没法得知，所以会不停地循环，试图寻找答案。
再测试一个比较简单的问题：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 工具 'Search' 已注册。 请输入文本: 苹果公司全线涨价是怎么回事 --- 第 1 步 --- 🧠 正在调用 deepseek-v4-pro 模型... ✅ 大语言模型响应成功: 思考: 用户询问苹果公司全线涨价的原因，这涉及到近期的时事新闻，我需要使用搜索工具来获取最新信息。 🎬 行动: Search[苹果公司全线涨价 原因] 🔍 正在执行 [SerpApi] 网页搜索: 苹果公司全线涨价 原因 👀 观察: [1] 苹果涨价——压垮AI存储泡沫的最后一根稻草？ 苹果涨价之所以能触发韩股熔断，恰恰因为它们击中了AI存储叙事最脆弱的一环：不是说AI不需要存储了，而是市场开始重新定价"AI需要多少存储、有人愿意为此付 ... [2] 苹果全线涨价，iPhone也不远了库克：这是百年一遇的“洪水” 苹果发言人在涨价声明中表示：“AI数据中心的快速扩张，引发了对内存及存储芯片需求的急剧攀升。我们从未经历过零部件价格如此迅猛、涨幅如此之大的情况。 [3] 苹果突然涨价致果链集体大跌，门店员工也未获提前通知 美东时间6月25日，苹果官宣全球范围内上调Mac、iPad全系硬件定价。这场由AI存储芯片紧缺催生的终端调价风暴，不仅使得苹果股价大跌，也影响了果链在二级市场 ... --- 第 2 步 --- 🧠 正在调用 deepseek-v4-pro 模型... ✅ 大语言模型响应成功: 思考: 根据搜索结果，苹果公司全线涨价的主要原因是AI数据中心快速扩张导致存储芯片（内存及存储）需求急剧攀升，零部件价格暴涨，迫使苹果上调Mac、iPad等产品的全球定价。此次涨价未提前通知门店，引发股价下跌和供应链震荡。我需要将这些信息整合成一个清晰的答案。 🎉 最终答案: 苹果公司突然宣布全线涨价，主要涉及Mac和iPad全系产品。涨价的核心原因是AI（人工智能）数据中心快速扩张，导致对内存及存储芯片的需求急剧攀升，零部件供应紧张且价格暴涨。苹果在声明中表示，从未经历过如此迅猛的零部件涨价情况。此次调价不仅引发了苹果自身股价大跌，也连带导致供应链相关企业股价集体下挫，甚至部分市场出现剧烈波动。此外，有消息称许多门店员工也未提前获得通知，涨价决定来得非常突然。
可以看到，这种比较简单的只涉及文字的问题，全网到处都有关于这个话题的讨论，大模型只搜索一次就完成了。
Plan-and-Resolve：Planner LLM 如何才能拥有 将复杂问题拆分成若干子问题 的能力？这一点与 ReAct 是有着显著区别的。
我们构建的 ReAct 倾向于反复调用 LLM，试图通过上下文的迭代来命中预期，也就是说前 n-1 次调用 LLM 相当于只是给第 n 次调用 LLM 做准备工作。
而 plan-and-resolve 模式中，我们力求的是分为两个阶段解决问题：
第一次让 LLM 进行 plan，选定所有需要的工具和顺序，并将其作为结构化输出封装成 PLAN；
第二次拿着封装好的 PLAN，根据工作流循环让 LLM 调用工具。
这种方式每次循环都解决一个简单的子问题。
此外，采用这种模式将规划流程标准化，即使 LLM 本身就有着很强的规划能力，也大概率能够在此基础上进一步最终执行的准确度。
planner prompt 提示词应该怎么写？在前面的 ReAct 的 prompt 中，我们明文规定了 LLM 必须输出 Thought => Action => Observation 的思维链，我们在 plan-and-resolve 中也需要明文规定：
1 2 3 4 5 6 7 8 9 10 11 12 PLANNER_PROMPT_TEMPLATE = """ 你是一个顶级的AI规划专家。你的任务是将用户提出的复杂问题分解成一个由多个简单步骤组成的行动计划。 请确保计划中的每个步骤都是一个独立的、可执行的子任务，并且严格按照逻辑顺序排列。 你的输出必须是一个Python列表，其中每个元素都是一个描述子任务的字符串。 问题: {question} 请严格按照以下格式输出你的计划,```python与```作为前后缀是必要的: ```python ["步骤1", "步骤2", "步骤3", ...] \``` """
Planner 指导思想 定义 Planner 类对 prompt 进行封装。我们前面定义的 ReActAgent 本质上也是对 ReActPrompt 的封装：我们强制定义了 LLM 的输出格式，然后我们根据这个格式进行输出解析，完成业务。在 Planner 中，我们自然也需要对前面提到的 ["步骤1", "步骤2", "步骤3", ...] 进行解析。
另外，Planner 同样应当调用 HelloAgentLLM。
接下来解读一下 Planner.py 干了什么
初始化 这里要传入一个 HelloAgentLLM 实例。
plan plan 函数
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 def plan (self, question: str ) -> list [str ]: """ 根据用户问题生成一个行动计划。 """ prompt = PLANNER_PROMPT_TEMPLATE.format (question=question) messages = [{"role" : "user" , "content" : prompt}] print ("--- 正在生成计划 ---" ) response_text = self.llm_client.think(messages=messages) or "" print (f"✅ 计划已生成:\n{response_text} " ) try : plan_str = response_text.split("```python" )[1 ].split("```" )[0 ].strip() 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 []
测试 plan 定义 main 函数进行测试：
1 2 3 4 5 6 7 if __name__ == '__main__' : planner = Planner(HelloAgentLLM()) print ("请输入问题: " ) planner.plan(input ())
测试结果：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 请输入问题: 怎么构建一个快速排序算法？ --- 正在生成计划 --- 🧠 正在调用 deepseek-v4-pro 模型... ✅ 大语言模型响应成功: ✅ 计划已生成: ```python [ "理解快速排序算法的基本原理：选择一个基准元素，将数组分为小于基准和大于基准的两部分，然后递归地对两部分排序。", "定义快速排序函数，明确输入参数（待排序数组、起始索引和结束索引）和递归终止条件（当起始索引不小于结束索引时返回）。", "实现分区函数：选取基准元素（例如最后一个元素），遍历数组将小于基准的元素移到左边，记录分区点索引，最后将基准放到正确位置并返回该索引。", "在快速排序函数中调用分区函数，获取分区点索引，然后对左右两个子数组递归调用自身。", "编写辅助函数或对外接口，简化调用方式（例如只接受数组作为参数，内部调用主递归函数）。", "使用不同测试用例（空数组、单元素、已排序、逆序、含重复元素、随机数组）验证算法正确性和稳定性。", "分析算法性能，讨论平均时间复杂度O(n log n)、最坏情况O(n²)以及空间复杂度，并考虑优化策略（如随机化基准选择、三数取中法等）。" ] \```
可以看到，LLM 确实将一个复杂的任务拆解成了若干子问题，并且做出了排序，方便后续 LLM 执行。
Plan-and-Resolve：Executor 我们前面已经实现了 Planner，LLM 能够成功将一个复杂问题分解成若干子问题，并且进行了排序，接下来要做的就是根据排序好的子问题执行。
我们需要定义 Executor，流程与 Planner 类似，都是要传入 HelloAgentLLM ，并且使用提示词进行激活。
executor prompt Planner 调用 LLM ，结果输出的是 {plan}，也即一个 str[]，将其作为 Executor 的输入提示词。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 EXECUTOR_PROMPT_TEMPLATE = """ 你是一位顶级的AI执行专家。你的任务是严格按照给定的计划，一步步地解决问题。 你将收到原始问题、完整的计划、以及到目前为止已经完成的步骤和结果。 请你专注于解决“当前步骤”，并仅输出该步骤的最终答案，不要输出任何额外的解释或对话。 # 原始问题: {question} # 完整计划: {plan} # 历史步骤与结果: {history} # 当前步骤: {current_step} 请仅输出针对“当前步骤”的回答: """
解读 构造函数传入 HelloAgentLLM。
execute() 函数接收了 plan 阶段获得的 str[] ，.forEach 调用 LLM 执行，采用 history 字符串作为记忆。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 class Executor : def __init__ (self, llm_client:HelloAgentLLM ): 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 ( question=question, plan=plan, history=history if history else "无" , current_step=step ) messages = [{"role" : "user" , "content" : prompt}] response_text = self.llm_client.think(messages=messages) or "" history += f"步骤 {i+1 } : {step} \n结果: {response_text} \n\n" print (f"✅ 步骤 {i+1 } 已完成，结果: {response_text} " ) final_answer = response_text return final_answer
Plan-and-Resolve：联合 Planner 与 Executor 都已完成定义，二者之间的关系非常明了：Planner 获取 plan，Executor 根据 plan 执行。
接下来需要定义一个 PlanAndExecuteAgent 类，将二者的工作流程进行封装。
测试 定义一个主函数测试一下：
1 2 3 4 5 6 7 8 if __name__ == '__main__' : agent = PlanAndExecuteAgent(HelloAgentLLM()) print ("请输入问题：" ) agent.run(input ())
结果：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 请输入问题： 怎么写三路快速排序 --- 开始处理问题 --- 问题: 怎么写三路快速排序 --- 正在生成计划 --- 🧠 正在调用 deepseek-v4-pro 模型... ✅ 大语言模型响应成功: ✅ 计划已生成: ``python ["步骤1：理解三路快速排序的核心思想——通过一次分区将数组划分为小于枢轴、等于枢轴、大于枢轴三个区域，避免重复元素带来的退化。", "步骤2：定义排序函数（例如 `three_way_quicksort(arr, low, high)`），明确参数为待排数组及当前处理的起止下标。", "步骤3：在函数内实现枢轴选择逻辑，通常选取 `arr[low]` 作为枢轴，也可加入随机选取来优化性能。", "步骤4：实现三路分区逻辑，初始化指针 `lt = low`（小于区右边界）、`gt = high`（大于区左边界）、`i = low + 1`（扫描指针），遍历过程中根据与枢轴的比较结果交换元素并移动指针。", "步骤5：编写循环分区细节：当 `arr[i] < pivot` 时交换 `arr[lt+1]` 与 `arr[i]`，然后移动 `lt` 和 `i`；当 `arr[i] > pivot` 时交换 `arr[gt]` 与 `arr[i]`，仅移动 `gt`；相等时仅移动 `i`，直到 `i > gt` 结束。", "步骤6：分区结束后将枢轴交换到正确位置，例如交换 `arr[low]` 与 `arr[lt]`，并调整边界。", "步骤7：递归地对小于区（`low` 到 `lt-1`）和大于区（`gt+1` 到 `high`）进行排序，中间等于区已自动就位。", "步骤8：添加递归终止条件：当 `low >= high` 时直接返回，避免无限递归。", "步骤9：编写对外接口函数（如 `sort(arr)`），内部调用 `three_way_quicksort(arr, 0, len(arr)-1)`，方便用户使用。", "步骤10：设计多组测试用例，包括随机数组、全等元素、已排序数组、单元素数组和空数组，验证排序结果的正确性和稳定性。"] `` --- 正在执行计划 --- -> 正在执行步骤 1/10: 步骤1：理解三路快速排序的核心思想——通过一次分区将数组划分为小于枢轴、等于枢轴、大于枢轴三个区域，避免重复元素带来的退化。 🧠 正在调用 deepseek-v4-pro 模型... ✅ 大语言模型响应成功: ✅ 步骤 1 已完成，结果: 三路快速排序的核心思想是：在每次分区操作中，选取一个枢轴元素，将数组重新排列为三个连续区域——所有小于枢轴的元素位于左侧，等于枢轴的元素居中，大于枢轴的元素位于右侧。这样，等于枢轴的部分在后续递归中无需再次处理，从而在包含大量重复元素的数组上避免性能退化到O(n²)，保持平均O(n log n)的时间复杂度。 -> 正在执行步骤 2/10: ······（步骤2——10省略） --- 任务完成 --- 最终答案: ``python def test_three_way_quicksort(): # 辅助函数：检查列表是否升序 def is_sorted(arr): return all(arr[i] <= arr[i+1] for i in range(len(arr)-1)) # 1. 随机数组 import random arr1 = [random.randint(0, 100) for _ in range(20)] expected1 = sorted(arr1) sort(arr1) assert arr1 == expected1, f"随机数组测试失败: {arr1}" # 2. 全等元素 arr2 = [5] * 10 expected2 = sorted(arr2) sort(arr2) assert arr2 == expected2, f"全等元素测试失败: {arr2}" # 3. 已排序数组 arr3 = list(range(20)) expected3 = sorted(arr3) sort(arr3) assert arr3 == expected3, f"已排序数组测试失败: {arr3}" # 4. 逆序数组 arr4 = list(range(20, 0, -1)) expected4 = sorted(arr4) sort(arr4) assert arr4 == expected4, f"逆序数组测试失败: {arr4}" # 5. 单元素数组 arr5 = [42] expected5 = sorted(arr5) sort(arr5) assert arr5 == expected5, f"单元素数组测试失败: {arr5}" # 6. 空数组 arr6 = [] expected6 = sorted(arr6) sort(arr6) assert arr6 == expected6, f"空数组测试失败: {arr6}" # 7. 包含大量重复元素的数组 arr7 = [3, 1, 2, 3, 1, 3, 2, 1, 3, 1] expected7 = sorted(arr7) sort(arr7) assert arr7 == expected7, f"大量重复元素测试失败: {arr7}" print("所有测试用例通过！") # 运行测试 test_three_way_quicksort() ``
可以看到，LLM 确实经过了 10 步构建，这是符合流程的。然而有一个瑕疵是，LLM 在最终输出环境并没有给出 sort 函数，反而是给出了其测试函数，这也是 LLM 不可控的地方，需要用户在 question 环节就给出明确说明。
我们再次测试，在提示词内就给出完全的说明：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 请输入问题： 使用typescript，写一个三路快排，最终输出只给出three_way_quicksort(nums:number[]) --- 开始处理问题 --- 问题: 使用typescript，写一个三路快排，最终输出只给出three_way_quicksort(nums:number[]) --- 正在生成计划 --- 🧠 正在调用 deepseek-v4-pro 模型... ✅ 大语言模型响应成功: ✅ 计划已生成: ``python [ "理解三路快排算法：将数组划分为小于基准值、等于基准值和大于基准值三个区域，以避免重复元素导致的性能下降。", "定义TypeScript函数 three_way_quicksort，接收 number[] 并返回 number[]，内部实现递归排序逻辑。", "选择合适的基准值策略（如随机选取或三数取中），并在函数内实现三向切分（partition），返回小于和大于区域的起止索引或直接拼接三个子数组。", "处理递归终止条件：当数组长度小于等于1时直接返回。", "实现分区逻辑：使用三个指针（lt, i, gt）进行原地分区，将数组整理为 [ <pivot, =pivot, >pivot] 的状态，然后递归排序小于和大于部分。", "将分区后的三部分（小于、等于、大于）分别递归调用 three_way_quicksort，并将结果合并后返回。", "添加边缘情况处理：空数组、单元素数组、所有元素相等的数组，确保不出现无限递归或数组越界。", "编写单元测试用例，覆盖普通数组、重复元素、已排序数组、逆序数组等，以验证正确性。", "整理代码，移除多余辅助函数导出，确保最终只暴露 three_way_quicksort 函数并符合题目要求。" ] `` --- 正在执行计划 --- -> 正在执行步骤 1/9: 理解三路快排算法：将数组划分为小于基准值、等于基准值和大于基准值三个区域，以避免重复元素导致的性能下降。 🧠 正在调用 deepseek-v4-pro 模型... ✅ 大语言模型响应成功: ✅ 步骤 1 已完成，结果: 三路快排通过一次遍历将数组划分为三个区域：小于基准值的元素放在左侧，等于基准值的元素放在中间，大于基准值的元素放在右侧。相比传统二路快排，它能有效处理大量重复元素的情况，避免递归树深度增加和额外比较，从而提升性能。分区时通常使用三个指针 lt、i、gt：lt 指向小于区域的右边界，gt 指向大于区域的左边界，i 为当前扫描指针。扫描过程中，若当前元素小于基准值，则交换到小于区域并移动 lt 和 i；若等于基准值，则仅移动 i；若大于基准值，则交换到大于区域并移动 gt，i 不动。扫描结束时，数组被划分为 [lo, lt-1] 小于区，[lt, gt] 等于区，[gt+1, hi] 大于区，然后对小于区和大于区递归排序，等于区已经就位。 -> 正在执行步骤 2/9: 定义TypeScript函数 three_way_quicksort，接收 number[] 并返回 number[]，内部实现递归排序逻辑。 🧠 正在调用 deepseek-v4-pro 模型... ✅ 大语言模型响应成功: ✅ 步骤 2 已完成，结果: ``typescript function three_way_quicksort(nums: number[]): number[] { // 边缘情况：空数组或单元素数组直接返回副本 if (nums.length <= 1) return nums.slice(); const arr = nums.slice(); // 复制数组，以便进行原地分区排序 sort(arr, 0, arr.length - 1); return arr; /** * 递归排序函数，对 arr 的 [lo, hi] 区间进行三路快排 */ function sort(arr: number[], lo: number, hi: number): void { if (lo >= hi) return; // 三向切分，后续步骤将实现具体的分区逻辑 const [lt, gt] = partition(arr, lo, hi); // 递归排序小于和大于基准值的部分 sort(arr, lo, lt - 1); sort(arr, gt + 1, hi); } /** * 占位分区函数，将在后续步骤中实现完整的三向切分逻辑 * 预期返回：[lt, gt]——小于区域的右边界和大于区域的左边界 */ function partition(arr: number[], lo: number, hi: number): [number, number] { // TODO: 实现三向切分 return [lo, hi]; } } `` -> 正在执行步骤 3/9: ······（步骤3---9省略） --- 任务完成 --- ······（此处省略） ``
这次给出我们想要的结果了。
Reflection Memory 为什么需要 Reflection？就是赋予大模型反思自己的回答有无纰漏，从而进行改进的能力。我们之前构建的 Agent，不管是基于 ReAct 模式还是 plan-and-resolve 模式，都是一旦完成就退出进程了，对于此前的解答并没有进行检查。
问题是，如何获取之前的解答呢？我们之前定义的两个 Agent，都是通过一个 run 函数直接调用的，特别是 PlanAndResolveAgent 的 run 函数，里面甚至有局部变量 history 作为上下文，一旦函数结束了，上下文就释放了。
所以，对于一个合格的 Reflection 机制，最先需要实现的功能就是：全局的 Memory 记忆，他必须能够跨会话地读取到上下文，才能进行后续的反思。
构建也很简单，实则就是一个 Stack，其核心是自身的成员变量 self.records: List[Dict[str, Any]]
Prompt 官方在此次设计了三套提示词，分别对应 执行=>反思=>优化 的流程。
Reflection 的核心就是执行之后的反思，重点不在于如何执行，所以官方在设计执行的提示词时，只使用了最简单的设计：完全没有任何规范流程，允许 LLM 自由发挥：
1 2 3 4 5 6 7 8 INITIAL_PROMPT_TEMPLATE = """ 你是一位资深的程序员。请根据以下要求，编写一个函数。 你的代码必须包含完整的函数签名、文档字符串。 要求: {task} 请直接输出代码，不要包含任何额外的解释。 """
调用 LLM 执行得到结果后，LLM 需要进行反思，这一步也采用标准的提示词进行激活：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 REFLECT_PROMPT_TEMPLATE = """ 你是一位极其严格的代码评审专家和资深算法工程师，对代码的性能有极致的要求。 你的任务是审查以下代码，并专注于找出其在<strong>算法效率</strong>上的主要瓶颈。 # 原始任务: {task} # 待审查的代码: {code} 请分析该代码的时间复杂度，并思考是否存在一种<strong>算法上更优</strong>的解决方案来显著提升性能。 如果存在，请清晰地指出当前算法的不足，并提出具体的、可行的改进算法建议（例如，使用筛法替代试除法）。 如果代码在算法层面已经达到最优，才能回答“无需改进”。 请直接输出你的反馈，不要包含任何额外的解释。 """
反思完毕后，LLM 输出反馈，随后再次进入 LLM 对之前的代码进行优化，提示词被设计为：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 REFINE_PROMPT_TEMPLATE = """ 你是一位资深的程序员。你正在根据一位代码评审专家的反馈来优化你的代码。 # 原始任务: {task} # 你上一轮尝试的代码: {last_code_attempt} 评审员的反馈： {feedback} 请根据评审员的反馈，生成一个优化后的新版本代码。 你的代码必须包含完整的函数签名、文档字符串。 请直接输出优化后的代码，不要包含任何额外的解释。 """
ReflectionAgent 我们还是调用 HelloAgentLLM 类的 think 函数来调用大模型，构建 ReflectionAgent 类来处理提示词、响应和历史上下文。
我们来解读一下。
run 函数是调用接口，整体来说是一个关于 iteration 数值的循环，超出循环，或者反思意见认为已经趋于完美则停止。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 def run (self, task: str ): print (f"\n--- 开始处理任务 ---\n任务: {task} " ) print ("\n--- 正在进行初始尝试 ---" ) initial_prompt = INITIAL_PROMPT_TEMPLATE.format (task=task) initial_code = self._get_llm_response(initial_prompt) print ("初始执行效果：" , initial_code) self.memory.add_record("execution" , initial_code) for i in range (self.max_iterations): print (f"\n--- 第 {i+1 } /{self.max_iterations} 轮迭代 ---" ) print ("\n-> 正在进行反思..." ) last_code = self.memory.get_last_execution() reflect_prompt = REFLECT_PROMPT_TEMPLATE.format (task=task, code=last_code) feedback = self._get_llm_response(reflect_prompt) print ("该轮反思反馈意见：" , feedback) self.memory.add_record("reflection" , feedback) if "无需改进" in feedback: print ("\n✅ 反思认为代码已无需改进，任务完成。" ) break print ("\n-> 正在进行优化..." ) refine_prompt = REFINE_PROMPT_TEMPLATE.format ( task=task, last_code_attempt=last_code, feedback=feedback ) refined_code = self._get_llm_response(refine_prompt) print ("该轮优化代码：" , refined_code) self.memory.add_record("execution" , refined_code) final_code = self.memory.get_last_execution() print (f"\n--- 任务完成 ---\n最终生成的代码:\n```python\n{final_code} \n```" ) return final_code
编写函数进行一个测试：
1 2 3 4 5 6 7 if __name__ == '__main__' : agent = ReflectionAgent(HelloAgentLLM()) print ("请输入问题: " ) agent.run(input ())
测试结果：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 请输入问题: 用typescript写一个归并排序mergeSort，然后写一个测试函数testSort，包含五组待排序数组，在控制台打印排序效果，你最终的输出仅包含mergeSort和testSort --- 开始处理任务 --- 任务: 用typescript写一个归并排序mergeSort，然后写一个测试函数testSort，包含五组待排序数组，在控制台打印排序效果，你最终的输出仅包含mergeSort和testSort --- 正在进行初始尝试 --- 🧠 正在调用 deepseek-v4-pro 模型... ✅ 大语言模型响应成功: 初始执行效果： ``typescript /** * 归并排序算法 * @param arr - 待排序的数值型数组 * @returns 排序后的新数组（原数组不会被修改） */ function mergeSort(arr: number[]): number[] { if (arr.length <= 1) { return arr; } const mid = Math.floor(arr.length / 2); const left = mergeSort(arr.slice(0, mid)); const right = mergeSort(arr.slice(mid)); return merge(left, right); } ······（此处省略） `` 📝 记忆已更新，新增一条 'execution' 记录。 --- 第 1/3 轮迭代 --- -> 正在进行反思... 🧠 正在调用 deepseek-v4-pro 模型... ✅ 大语言模型响应成功: 该轮反思反馈意见： 该实现的时间复杂度为 O(n log n)，已属比较排序的渐进最优。但在具体操作上存在显著性能瓶颈：`mergeSort` 中大量使用 `arr.slice()` 创建新数组，每层递归都会完整复制数据；`merge` 结束时又通过 `concat` 与 `slice` 多次分配并复制剩余元素，导致内存分配次数和元素复制量均为 O(n log n)。这使得实际运行效率远低于理论上界，GC 压力极大。 **改进建议**：采用单辅助数组的原地归并策略。可改写为自底向上的迭代归并，或保持自顶向下递归但全程复用同一个长度为 n 的临时数组，仅通过索引划分子区间，避免 `slice`/`concat` 的任何额外复制。这样可将元素复制总量降至严格的 O(n log n)（仅合并时复制），内存分配降至 O(1)（复用固定缓冲区），大幅提升性能。示例优化方案如下： ``typescript ·····（此处省略） `` 该版本递归时在 `target` 和 `work` 之间交替，避免了任何切片操作，所有合并直接写入目标数组，仅分配一次辅助空间，性能与内存效率均得到质的提升。 📝 记忆已更新，新增一条 'reflection' 记录。 -> 正在进行优化... 🧠 正在调用 deepseek-v4-pro 模型... ✅ 大语言模型响应成功: 该轮优化代码： ``typescript ······（此处省略） `` 📝 记忆已更新，新增一条 'execution' 记录。 --- 第 2/3 轮迭代 --- ······（第二轮第三轮省略） 📝 记忆已更新，新增一条 'execution' 记录。 --- 任务完成 --- 最终生成的代码: ``typescript ······（此处省略） ``
这个结果算是符合我们的预期
总结 本章构建了三种经典 Agent 范式。
我们通过先构建一个符合 OpenAI 规范的基本 LLM 调用接口 HelloAgentLLM，实现了对 LLM 进行调用以及处理 LLM 的流式响应。
通过赋予 LLM 使用工具从而与外界交互的能力，我们构建起了 ReAct 模式 Agent，这种模式更专注于通过观察与外界的交互从而决定下一步的行动，所以如何为 LLM 构建使用工具的环境也是其中一个核心课题。适合需要探索性的任务 。
随后，我们实现了一个 Plan-and-Resolve 模式 Agent，解决了实现让 LLM 将复杂问题拆解 这一难题。通过分别注入标准化的提示词，强制让 LLM 分别执行 plan 与 execute 两个阶段，让 LLM 的思路更加结构化和稳定化，适合需要经过逻辑分析、内部推理密集的任务 。
最后，基于 赋予 LLM 反思能力，我们构建了 Reflect 模式的 Agent，该种模式着重于让 LLM 拆解自己以前的回答，发现问题并尝试优化，能够显著提升解答质量。适合需要高质量和高精度的解答任务 。
本章代码 HelloAgentLLM.py 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 import osfrom openai import OpenAIfrom dotenv import load_dotenvfrom typing import List , Dict load_dotenv() class HelloAgentLLM : """ 为本书 "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_key=apiKey, base_url=baseUrl, timeout=timeout) def think (self, messages: List [Dict [str , str ]], temperature: float = 0 ) -> str : """ 调用大语言模型进行思考，并返回其响应。 """ print (f"🧠 正在调用 {self.model} 模型..." ) try : response = self.client.chat.completions.create( model=self.model, messages=messages, temperature=temperature, stream=True , ) print ("✅ 大语言模型响应成功:" ) collected_content = [] for chunk in response: if not chunk.choices: continue content = chunk.choices[0 ].delta.content or "" 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 = HelloAgentLLM() print ("请输入文本..." ) userPrompt = input () exampleMessages = [ {"role" : "system" , "content" : "You are a helpful assistant that writes Python code." }, {"role" : "user" , "content" : userPrompt} ] print ("--- 调用LLM ---" ) responseText = llmClient.think(exampleMessages) if responseText: print ("\n\n--- 完整模型响应 ---" ) print (responseText) except ValueError as e: print (e)
search.py 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 from serpapi import SerpApiClientimport osfrom dotenv import load_dotenvload_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} " if __name__ == '__main__' : print (search("英伟达的历史最高股价是多少" ))
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 from typing import Dict , Any import searchclass 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__' : toolExecutor = ToolExecutor() search_description = "一个网页搜索引擎。当你需要回答关于时事、事实以及在你的知识库中找不到的信息时，应使用此工具。" toolExecutor.registerTool("Search" , search_description, search.search) print ("\n--- 可用的工具 ---" ) print (toolExecutor.getAvailableTools()) 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} ' 的工具。" )
ReActPrompt.py 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 REACT_PROMPT_TEMPLATE = """ 请注意，你是一个有能力调用外部工具的智能助手。 可用工具如下: {tools} 请严格按照以下格式进行回应: Thought: 你的思考过程，用于分析问题、拆解任务和规划下一步行动。 Action: 你决定采取的行动，必须是以下格式之一: - `{{tool_name}}[{{tool_input}}]`:调用一个可用工具。 - `Finish[最终答案]`:当你认为已经获得最终答案时。 - 当你收集到足够的信息，能够回答用户的最终问题时，你必须在Action:字段后使用 Finish[最终答案] 来输出最终答案。 现在，请开始解决以下问题: Question: {question} History: {history} """
ReActAgent.py 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 from HelloAgentLLM import HelloAgentLLMfrom ToolExecutor import ToolExecutorfrom ReActPrompt import REACT_PROMPT_TEMPLATEimport refrom typing import Dict , Any import searchclass ReActAgent : def __init__ (self, llm_client: HelloAgentLLM, 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 _parse_output (self, text: str ): """解析LLM的输出，提取Thought和Action。 """ thought_match = re.search(r"Thought:\s*(.*?)(?=\nAction:|$)" , text, re.DOTALL) action_match = re.search(r"Action:\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 ): """解析Action字符串，提取工具名称和输入。 """ match = re.match (r"(\w+)\[(.*)\]" , action_text, re.DOTALL) if match : return match .group(1 ), match .group(2 ) return None , None def run (self, question: str ): """ 运行ReAct智能体来回答一个问题。 """ self.history = [] current_step = 0 while current_step < self.max_steps: current_step += 1 print (f"--- 第 {current_step} 步 ---" ) tools_desc = self.tool_executor.getAvailableTools() history_str = "\n" .join(self.history) prompt = REACT_PROMPT_TEMPLATE.format ( tools=tools_desc, question=question, history=history_str ) messages = [{"role" : "user" , "content" : prompt}] response_text = self.llm_client.think(messages=messages) 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" ): final_answer = re.match (r"Finish\[(.*)\]" , action).group(1 ) print (f"🎉 最终答案: {final_answer} " ) return final_answer tool_name, tool_input = self._parse_action(action) if not tool_name or not tool_input: continue print (f"🎬 行动: {tool_name} [{tool_input} ]" ) tool_function = self.tool_executor.getTool(tool_name) if not tool_function: observation = f"错误:未找到名为 '{tool_name} ' 的工具。" else : observation = tool_function(tool_input) print (f"👀 观察: {observation} " ) self.history.append(f"Action: {action} " ) self.history.append(f"Observation: {observation} " ) print ("已达到最大步数，流程终止。" ) return None if __name__ == '__main__' : toolExecutor = ToolExecutor() search_description = "一个网页搜索引擎。当你需要回答关于时事、事实以及在你的知识库中找不到的信息时，应使用此工具。" toolExecutor.registerTool("Search" , search_description, search.search) agent = ReActAgent( HelloAgentLLM(), toolExecutor, 5 ) print ("请输入文本: " ) userPrompt = input () agent.run(userPrompt)
.env 1 2 3 4 5 LLM_MODEL_ID = "" LLM_API_KEY = "" LLM_BASE_URL = "" SERPAPI_API_KEY = ""
Planner.py 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 from HelloAgentLLM import HelloAgentLLMfrom PlannerPrompt import PLANNER_PROMPT_TEMPLATEimport astclass Planner : def __init__ (self, llm_client:HelloAgentLLM ): self.llm_client = llm_client def plan (self, question: str ) -> list [str ]: """ 根据用户问题生成一个行动计划。 """ prompt = PLANNER_PROMPT_TEMPLATE.format (question=question) messages = [{"role" : "user" , "content" : prompt}] print ("--- 正在生成计划 ---" ) response_text = self.llm_client.think(messages=messages) or "" print (f"✅ 计划已生成:\n{response_text} " ) try : plan_str = response_text.split("```python" )[1 ].split("```" )[0 ].strip() 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 [] if __name__ == '__main__' : planner = Planner(HelloAgentLLM()) print ("请输入问题: " ) planner.plan(input ())
ExecutorPrompt.py 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 EXECUTOR_PROMPT_TEMPLATE = """ 你是一位顶级的AI执行专家。你的任务是严格按照给定的计划，一步步地解决问题。 你将收到原始问题、完整的计划、以及到目前为止已经完成的步骤和结果。 请你专注于解决“当前步骤”，并仅输出该步骤的最终答案，不要输出任何额外的解释或对话。 # 原始问题: {question} # 完整计划: {plan} # 历史步骤与结果: {history} # 当前步骤: {current_step} 请仅输出针对“当前步骤”的回答: """
Executor.py 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 from HelloAgentLLM import HelloAgentLLMfrom ExecutorPrompt import EXECUTOR_PROMPT_TEMPLATEclass Executor : def __init__ (self, llm_client:HelloAgentLLM ): 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 ( question=question, plan=plan, history=history if history else "无" , current_step=step ) messages = [{"role" : "user" , "content" : prompt}] response_text = self.llm_client.think(messages=messages) or "" history += f"步骤 {i+1 } : {step} \n结果: {response_text} \n\n" print (f"✅ 步骤 {i+1 } 已完成，结果: {response_text} " ) final_answer = response_text return final_answer
PlanAndExecuteAgent.py 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 from Planner import Plannerfrom Executor import Executorfrom HelloAgentLLM import HelloAgentLLMclass PlanAndExecuteAgent : def __init__ (self, llm_client:HelloAgentLLM ): """ 初始化智能体，同时创建规划器和执行器实例。 """ 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} " ) plan = self.planner.plan(question) if not plan: print ("\n--- 任务终止 --- \n无法生成有效的行动计划。" ) return final_answer = self.executor.execute(question, plan) print (f"\n--- 任务完成 ---\n最终答案: {final_answer} " ) if __name__ == '__main__' : agent = PlanAndExecuteAgent(HelloAgentLLM()) print ("请输入问题：" ) agent.run(input ())
Memory.py 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 from typing import List , Dict , Any , Optional 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
ReflectionPrompt.py 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 INITIAL_PROMPT_TEMPLATE = """ 你是一位资深的程序员。请根据以下要求，编写一个函数。 你的代码必须包含完整的函数签名、文档字符串。 要求: {task} 请直接输出代码，不要包含任何额外的解释。 """ REFLECT_PROMPT_TEMPLATE = """ 你是一位极其严格的代码评审专家和资深算法工程师，对代码的性能有极致的要求。 你的任务是审查以下代码，并专注于找出其在<strong>算法效率</strong>上的主要瓶颈。 # 原始任务: {task} # 待审查的代码: `` {code} `` 请分析该代码的时间复杂度，并思考是否存在一种<strong>算法上更优</strong>的解决方案来显著提升性能。 如果存在，请清晰地指出当前算法的不足，并提出具体的、可行的改进算法建议（例如，使用筛法替代试除法）。 如果代码在算法层面已经达到最优，才能回答“无需改进”。 请直接输出你的反馈，不要包含任何额外的解释。 """ REFINE_PROMPT_TEMPLATE = """ 你是一位资深的程序员。你正在根据一位代码评审专家的反馈来优化你的代码。 # 原始任务: {task} # 你上一轮尝试的代码: {last_code_attempt} 评审员的反馈： {feedback} 请根据评审员的反馈，生成一个优化后的新版本代码。 你的代码必须包含完整的函数签名、文档字符串。 请直接输出优化后的代码，不要包含任何额外的解释。 """
ReflectionAgent.py 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 from Memory import Memoryfrom HelloAgentLLM import HelloAgentLLMfrom ReflectionPrompt import INITIAL_PROMPT_TEMPLATE, REFLECT_PROMPT_TEMPLATE, REFINE_PROMPT_TEMPLATEclass ReflectionAgent : def __init__ (self, llm_client:HelloAgentLLM, max_iterations=3 ): self.llm_client = llm_client self.memory = Memory() self.max_iterations = max_iterations def run (self, task: str ): print (f"\n--- 开始处理任务 ---\n任务: {task} " ) print ("\n--- 正在进行初始尝试 ---" ) initial_prompt = INITIAL_PROMPT_TEMPLATE.format (task=task) initial_code = self._get_llm_response(initial_prompt) print ("初始执行效果：" , initial_code) self.memory.add_record("execution" , initial_code) for i in range (self.max_iterations): print (f"\n--- 第 {i+1 } /{self.max_iterations} 轮迭代 ---" ) print ("\n-> 正在进行反思..." ) last_code = self.memory.get_last_execution() reflect_prompt = REFLECT_PROMPT_TEMPLATE.format (task=task, code=last_code) feedback = self._get_llm_response(reflect_prompt) print ("该轮反思反馈意见：" , feedback) self.memory.add_record("reflection" , feedback) if "无需改进" in feedback: print ("\n✅ 反思认为代码已无需改进，任务完成。" ) break print ("\n-> 正在进行优化..." ) refine_prompt = REFINE_PROMPT_TEMPLATE.format ( task=task, last_code_attempt=last_code, feedback=feedback ) refined_code = self._get_llm_response(refine_prompt) print ("该轮优化代码：" , refined_code) self.memory.add_record("execution" , refined_code) final_code = self.memory.get_last_execution() print (f"\n--- 任务完成 ---\n最终生成的代码:\n```python\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(messages=messages) or "" return response_text if __name__ == '__main__' : agent = ReflectionAgent(HelloAgentLLM()) print ("请输入问题: " ) agent.run(input ())
