当前位置：首页 > news >正文

DataWhale-HelloAgents（第四章：智能体经典范式构建）

news 2025/11/15 11:08:02

前言部分

Hello-Agents 教程学习链接
github地址：https://github.com/datawhalechina/hello-agents
cookbook版本：https://book.heterocat.com.cn/

《Hello-agents》完整PDF免费下载

特别感谢本教程各位开源贡献者及文睿的支持

说实话，基础有点差，这学习进度有点太快了，跟不上大家的进度。。。。。。本章讨论智能体经典范式构建，那么直接进入代码模式

首先根据教程准备环境安装相应的依赖：

pip install openai python-dotenv

配置环境，将秘钥写入新建的env文件（冒号里面换成自己的）

# 原教材代码仅供参考
import 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_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: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# --- 客户端使用示例 ---
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)

工具类的定义与实现

针对本节设定的目标——回答关于“华为最新手机”的问题，我们需要为智能体提供一个网页搜索工具。在这里我们选用 SerpApi

pip install google-search-results

在原env文件后添加SERPAPI_API_KEY="YOUR_SERPAPI_API_KEY"，因为涉及到魔法的问题，这里就不注册演示结果，感兴趣的兄弟可以试一下，注册地址为Sign Up - SerpApi

from dotenv import load_dotenv
# 加载 .env 文件中的环境变量
load_dotenv()import os
from serpapi import SerpApiClient
from typing import Dict, Anydef 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}"from typing import Dict, Anyclass 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}' 的工具。")

ReAct 框架

先用正则表达式把 LLM 里“Thought:……Action:……”抠出来，再拿 Python 的 eval() 去动态调工具，最后用 Observation 把结果贴回 prompt。我第一次发现，让模型上网搜东西不是 ChatGPT 专用，写去吧也只要 5 行正则就能复刻。

import 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(answer="...")` 来输出最终答案。现在，请开始解决以下问题：
Question: {question}
History: {history}
"""class ReActAgent:def __init__(self, llm_client: HelloAgentsLLM, tool_executor: ToolExecutor, max_steps: int = 5):self.llm_client = llm_clientself.tool_executor = tool_executorself.max_steps = max_stepsself.history = []def run(self, question: str):self.history = []current_step = 0while current_step < self.max_steps:current_step += 1print(f"\n--- 第 {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未能返回有效响应。"); breakthought, action = self._parse_output(response_text)if thought: print(f"🤔 思考: {thought}")if not action: print("警告：未能解析出有效的Action，流程终止。"); breakif action.startswith("Finish"):final_answer = self._parse_action_input(action)print(f"🎉 最终答案: {final_answer}")return final_answertool_name, tool_input = self._parse_action(action)if not tool_name or not tool_input:self.history.append("Observation: 无效的Action格式，请检查。"); continueprint(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(f"Action: {action}")self.history.append(f"Observation: {observation}")print("已达到最大步数，流程终止。")return Nonedef _parse_output(self, text: str):thought_match = re.search(r"Thought: (.*)", text)action_match = re.search(r"Action: (.*)", text)thought = thought_match.group(1).strip() if thought_match else Noneaction = action_match.group(1).strip() if action_match else Nonereturn thought, actiondef _parse_action(self, action_text: str):match = re.match(r"(\w+)\[(.*)\]", action_text)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)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_client=llm, tool_executor=tool_executor)question = "华为最新的手机是哪一款？它的主要卖点是什么？"agent.run(question)

Plan-and-Solve框架

Plan-and-Solve 先写提纲再答题像不像我们写论文：导师先让交开题报告，通过后再回去填章节。代码实现更粗暴——把 Planner 和 Executor 做成两个独立类，中间用 Python list 传参，彻底解耦。写的时候就突然意识到，“链式思考”与“树状规划”并不是互斥，而是对应了人类大脑的两套操作系统：System1（快思考）用 ReAct 跑，System2（慢思考）用 Plan-and-Solve 跑——这一章把卡尼曼的理论落地成了可运行的 .py 文件

import os
import ast
from llm_client import HelloAgentsLLM
from dotenv import load_dotenv
from typing import List, Dict# 加载 .env 文件中的环境变量，处理文件不存在异常
try:load_dotenv()
except FileNotFoundError:print("警告：未找到 .env 文件，将使用系统环境变量。")
except Exception as e:print(f"警告：加载 .env 文件时出错: {e}")# --- 1. LLM客户端定义 ---
# 假设你已经有llm_client.py文件，里面定义了HelloAgentsLLM类# --- 2. 规划器 (Planner) 定义 ---
PLANNER_PROMPT_TEMPLATE = """
你是一个顶级的AI规划专家。你的任务是将用户提出的复杂问题分解成一个由多个简单步骤组成的行动计划。
请确保计划中的每个步骤都是一个独立的、可执行的子任务，并且严格按照逻辑顺序排列。
你的输出必须是一个Python列表，其中每个元素都是一个描述子任务的字符串。问题: {question}请严格按照以下格式输出你的计划，```python与```作为前后缀是必要的:
```python
["步骤1", "步骤2", "步骤3", ...]
```
"""class Planner:def __init__(self, llm_client: HelloAgentsLLM):self.llm_client = llm_clientdef 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 []# --- 3. 执行器 (Executor) 定义 ---
EXECUTOR_PROMPT_TEMPLATE = """
你是一位顶级的AI执行专家。你的任务是严格按照给定的计划，一步步地解决问题。
你将收到原始问题、完整的计划、以及到目前为止已经完成的步骤和结果。
请你专注于解决“当前步骤”，并仅输出该步骤的最终答案，不要输出任何额外的解释或对话。# 原始问题:
{question}# 完整计划:
{plan}# 历史步骤与结果:
{history}# 当前步骤:
{current_step}请仅输出针对“当前步骤”的回答:
"""class Executor:def __init__(self, llm_client: HelloAgentsLLM):self.llm_client = llm_clientdef execute(self, question: str, plan: list[str]) -> str:history = ""final_answer = ""print("\n--- 正在执行计划 ---")for i, step in enumerate(plan, 1):print(f"\n-> 正在执行步骤 {i}/{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}: {step}\n结果: {response_text}\n\n"final_answer = response_textprint(f"✅ 步骤 {i} 已完成，结果: {final_answer}")return final_answer# --- 4. 智能体 (Agent) 整合 ---
class PlanAndSolveAgent:def __init__(self, llm_client: HelloAgentsLLM):self.llm_client = llm_clientself.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无法生成有效的行动计划。")returnfinal_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)

Reflection架构

Reflection的灵感挺有意思的，执行→反思→优化 直接把智能体升级成“导演+演员”：

第一步、演员浪演戏，生成初版代码；
第二步、导演开喷，为智能体提供了一个内部纠错回路
第三步、演员根据吐槽改剧本，生成最终版。

书上说的好的是，这里反思代码用 “是否包含‘无需改进’四个字” 当循环终止条件，那看上去十分简单了所谓“智能体的自我进化”，原来就是 prompt 里多写一句“请指出这段代码的算法瓶颈”。大道至简，莫过于此。

from typing import List, Dict, Any
# 假设 llm_client.py 文件已存在，并从中导入 HelloAgentsLLM 类
from llm_client import HelloAgentsLLM# --- 模块 1: 记忆模块 ---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): 记录的具体内容 (例如，生成的代码或反思的反馈)。"""self.records.append({"type": record_type, "content": content})print(f"📝 记忆已更新，新增一条 '{record_type}' 记录。")def get_trajectory(self) -> str:"""将所有记忆记录格式化为一个连贯的字符串文本，用于构建提示词。"""trajectory = ""for record in self.records:if record['type'] == 'execution':trajectory += f"--- 上一轮尝试 (代码) ---\n{record['content']}\n\n"elif record['type'] == 'reflection':trajectory += f"--- 评审员反馈 ---\n{record['content']}\n\n"return trajectory.strip()def get_last_execution(self) -> str:"""获取最近一次的执行结果 (例如，最新生成的代码)。"""for record in reversed(self.records):if record['type'] == 'execution':return record['content']return None# --- 模块 2: Reflection 智能体 ---# 1. 初始执行提示词
INITIAL_PROMPT_TEMPLATE = """
你是一位资深的Python程序员。请根据以下要求，编写一个Python函数。
你的代码必须包含完整的函数签名、文档字符串，并遵循PEP 8编码规范。要求: {task}请直接输出代码，不要包含任何额外的解释。
"""# 2. 反思提示词
REFLECT_PROMPT_TEMPLATE = """
你是一位极其严格的代码评审专家和资深算法工程师，对代码的性能有极致的要求。
你的任务是审查以下Python代码，并专注于找出其在**算法效率**上的主要瓶颈。# 原始任务:
{task}# 待审查的代码:
```python
{code}
```请分析该代码的时间复杂度，并思考是否存在一种**算法上更优**的解决方案来显著提升性能。
如果存在，请清晰地指出当前算法的不足，并提出具体的、可行的改进算法建议（例如，使用筛法替代试除法）。
如果代码在算法层面已经达到最优，才能回答“无需改进”。请直接输出你的反馈，不要包含任何额外的解释。
"""# 3. 优化提示词
REFINE_PROMPT_TEMPLATE = """
你是一位资深的Python程序员。你正在根据一位代码评审专家的反馈来优化你的代码。# 原始任务:
{task}# 你上一轮尝试的代码:
```python
{last_code_attempt}
```# 评审员的反馈:
{feedback}请根据评审员的反馈，生成一个优化后的新版本代码。
你的代码必须包含完整的函数签名、文档字符串，并遵循PEP 8编码规范。
请直接输出优化后的代码，不要包含任何额外的解释。
"""class ReflectionAgent:def __init__(self, llm_client, max_iterations=3):self.llm_client = llm_clientself.memory = Memory()self.max_iterations = max_iterationsdef run(self, task: str):print(f"\n--- 开始处理任务 ---\n任务: {task}")# --- 1. 初始执行 ---print("\n--- 正在进行初始尝试 ---")initial_prompt = INITIAL_PROMPT_TEMPLATE.format(task=task)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(task=task, code=last_code)feedback = self._get_llm_response(reflect_prompt)self.memory.add_record("reflection", feedback)# b. 检查是否需要停止if "无需改进" in feedback or "no need for improvement" in feedback.lower():print("\n✅ 反思认为代码已无需改进，任务完成。")break# c. 优化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)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_codedef _get_llm_response(self, prompt: str) -> str:"""一个辅助方法，用于调用LLM并获取完整的流式响应。"""messages = [{"role": "user", "content": prompt}]# 确保能处理生成器可能返回None的情况response_text = self.llm_client.think(messages=messages) or ""return response_textif __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_iterations=2)# 3. 定义任务并运行智能体task = "编写一个Python函数，找出1到n之间所有的素数 (prime numbers)。"agent.run(task)

第四章一口气把 ReAct、Plan-and-Solve、Reflection 三种范式“徒手”撸了一遍，读下来的最大感受框架一般，还得是人操作。原pdf文档中示例代码用 200 行代码就把智能体讲明白了， Thought-Action-Observation 循环就是 prompt 模板里多打了几个换行符，规划-执行两步走，就是把同一个 LLM 先后给两种不同的提示词，“自我反思”真的就是让大模型自己和自己吵架，用第二遍输出骂第一遍输出，然后再让第三遍输出打补丁。看来智能体=提示词工程+状态机+工具回调”确实是真理啊..............

提示词是灵魂——改一行 System Message，智能体就换一副人格；
状态机是骨骼——不管用消息队列还是 Python list，只要能把历史传下去，就能“假装”有记忆；
工具调用是手脚——把 requests.get() 包一层函数，再写两行描述，LLM 就学会“上网”。

我的三点思考

① “范式”是三种“性格”：ReAct 像急性子，边想边干；Plan-and-Solve 像老谋深算，谋定而后动；Reflection 像完美主义者，交稿前必须改三遍；
② “记忆”才是下一道护城河： Memory 类只有 list.append()，但真正上线后，如何把 10 轮对话压缩成 1 条永久记忆、如何把向量库结果塞进 prompt 而不爆长度，才是拉开差距的地方；
③ “造轮子”是最高效的学习：材料中反复强调“别急着用框架”。我深以为然——把正则解析换成 JSON mode、把 requests.get() 换成异步 aiohttp、把内存记忆换成 Redis，每一次魔改都在加深对“智能体=提示词+状态+工具”这一公式的理解；而直接 pip install langchain 只会让你变成调包侠

课后习题

1. 本章介绍了三种经典的智能体范式:ReAct、Plan-and-Solve 和 Reflection。请分析: 这三种范式在"思考"与"行动"的组织方式上有什么本质区别？如果要设计一个"智能家居控制助手"（需要控制灯光、空调、窗帘等多个设备，并根据用户习惯自动调节），你会选择哪种范式作为基础架构？为什么？是否可以将这三种范式进行组合使用？若可以，请尝试设计一个混合范式的智能体架构，并说明其适用场景