pip install openai python-dotenv

# .env file
LLM_API_KEY=\"YOUR-API-KEY\"
LLM_MODEL_ID=\"YOUR-MODEL\"
LLM_BASE_URL=\"YOUR-URL\"

import os
from openai import OpenAI
from dotenv import load_dotenv
from typing import List, Dict
# 加载 .env 文件中的环境变量
load_dotenv()
classHelloAgentsLLM:
    \"\"\"
    为本书 \"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))

        ifnot all([self.model, apiKey, baseUrl]):
            raise ValueError(\"模型ID、API密钥和服务地址必须被提供或在.env文件中定义。\")
        self.client = OpenAI(api_key=apiKey, base_url=baseUrl, timeout=timeout)
    defthink(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:
                ifnot 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}\")
            returnNone
# --- 客户端使用示例 ---
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(\"\
\
--- 完整模型响应 ---\")
            print(responseText)
    except ValueError as e:
        print(e)

>>>
--- 调用LLM ---
正在调用 xxxxxx 模型...
大语言模型响应成功:
快速排序是一种非常高效的排序算法...

pip install google-search-results

# .env file
# ... (保留之前的LLM配置)
SERPAPI_API_KEY=\"YOUR_SERPAPI_API_KEY\"

from serpapi import SerpApiClient
defsearch(query: str) -> str:
    \"\"\"
    一个基于SerpApi的实战网页搜索引擎工具。
    它会智能地解析搜索结果，优先返回直接答案或知识图谱信息。
    \"\"\"
    print(f\"正在执行 [SerpApi] 网页搜索: {query}\")
    try:
        api_key = os.getenv(\"SERPAPI_API_KEY\")
        ifnot 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\"\
\".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', '')}\
{res.get('snippet', '')}\"
                for i, res in enumerate(results[\"organic_results\"][:3])
            ]
            return\"\
\
\".join(snippets)

        returnf\"对不起，没有找到关于 '{query}' 的信息。\"
    except Exception as e:
        returnf\"搜索时发生错误: {e}\"

from typing import Dict, Any
classToolExecutor:
    \"\"\"
    一个工具执行器，负责管理和执行工具。
    \"\"\"
    def__init__(self):
        self.tools: Dict[str, Dict[str, Any]] = {}
    defregisterTool(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}' 已注册。\")
    defgetTool(self, name: str) -> callable:
        \"\"\"
        根据名称获取一个工具的执行函数。
        \"\"\"
        return self.tools.get(name, {}).get(\"func\")
    defgetAvailableTools(self) -> str:
        \"\"\"
        获取所有可用工具的格式化描述字符串。
        \"\"\"
        return\"\
\".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(\"\
--- 可用的工具 ---\")
    print(toolExecutor.getAvailableTools())
    # 4. 智能体的Action调用，这次我们问一个实时性的问题
    print(\"\
--- 执行 Action: Search['英伟达最新的GPU型号是什么'] ---\")
    tool_name = \"Search\"
    tool_input = \"英伟达最新的GPU型号是什么\"
    tool_function = toolExecutor.getTool(tool_name)
    if tool_function:
        observation = tool_function(tool_input)
        print(\"--- 观察 ---\")
        print(observation)
    else:
        print(f\"错误:未找到名为 '{tool_name}' 的工具。\")

>>>
工具 'Search' 已注册。
--- 可用的工具 ---
- Search: 一个网页搜索引擎。当你需要回答关于时事、事实以及在你的知识库中找不到的信息时，应使用此工具。
--- 执行 Action: Search['英伟达最新的GPU型号是什么'] ---
正在执行 [SerpApi] 网页搜索: 英伟达最新的GPU型号是什么
--- 观察 ---
[1] GeForce RTX 50 系列显卡
GeForce RTX™ 50 系列GPU 搭载NVIDIA Blackwell 架构，为游戏玩家和创作者带来全新玩法。RTX 50 系列具备强大的AI 算力，带来升级体验和更逼真的画面。
[2] 比较GeForce 系列最新一代显卡和前代显卡
比较最新一代RTX 30 系列显卡和前代的RTX 20 系列、GTX 10 和 900 系列显卡。查看规格、功能、技术支持等内容。
[3] GeForce 显卡| NVIDIA
DRIVE AGX. 强大的车载计算能力，适用于AI 驱动的智能汽车系统 · Clara AGX. 适用于创新型医疗设备和成像的AI 计算. 游戏和创作. GeForce. 探索显卡、游戏解决方案，AI ...

# ReAct 提示词模板
REACT_PROMPT_TEMPLATE = \"\"\"
请注意，你是一个有能力调用外部工具的智能助手。
可用工具如下:
{tools}
请严格按照以下格式进行回应:
Thought: 你的思考过程，用于分析问题、拆解任务和规划下一步行动。
Action: 你决定采取的行动，必须是以下格式之一:
- `{{tool_name}}[{{tool_input}}]`:调用一个可用工具。
- `Finish[最终答案]`:当你认为已经获得最终答案时。
- 当你收集到足够的信息，能够回答用户的最终问题时，你必须在Action:字段后使用 Finish[最终答案] 来输出最终答案。
现在，请开始解决以下问题:
Question: {question}
History: {history}
\"\"\"

classReActAgent:
    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 = []
    defrun(self, question: str):
        \"\"\"
        运行ReAct智能体来回答一个问题。
        \"\"\"
        self.history = [] # 每次运行时重置历史记录
        current_step = 0
        while current_step < self.max_steps:
            current_step += 1
            print(f\"--- 第 {current_step} 步 ---\")
            # 1. 格式化提示词
            tools_desc = self.tool_executor.getAvailableTools()
            history_str = \"\
\".join(self.history)
            prompt = REACT_PROMPT_TEMPLATE.format(
                tools=tools_desc,
                question=question,
                history=history_str
            )
            # 2. 调用LLM进行思考
            messages = [{\"role\": \"user\", \"content\": prompt}]
            response_text = self.llm_client.think(messages=messages)

            ifnot response_text:
                print(\"错误:LLM未能返回有效响应。\")
                break
            # ... (后续的解析、执行、整合步骤)

# (这些方法是 ReActAgent 类的一部分)
    def_parse_output(self, text: str):
        \"\"\"解析LLM的输出，提取Thought和Action。\"\"\"
        # Thought: 匹配到 Action: 或文本末尾
        thought_match = re.search(r\"Thought:\\s*(.*?)(?=\
Action:|$)\", text, re.DOTALL)
        # Action: 匹配到文本末尾
        action_match = re.search(r\"Action:\\s*(.*?)$\", text, re.DOTALL)
        thought = thought_match.group(1).strip() if thought_match elseNone
        action = action_match.group(1).strip() if action_match elseNone
        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)
        returnNone, None

# (这段逻辑在 run 方法的 while 循环内)
            # 3. 解析LLM的输出
            thought, action = self._parse_output(response_text)

            if thought:
                print(f\"思考: {thought}\")
            ifnot action:
                print(\"警告:未能解析出有效的Action，流程终止。\")
                break
            # 4. 执行Action
            if action.startswith(\"Finish\"):
                # 如果是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)
            ifnot tool_name ornot tool_input:
                # ... 处理无效Action格式 ...
                continue
            print(f\"行动: {tool_name}[{tool_input}]\")
            tool_function = self.tool_executor.getTool(tool_name)
            ifnot tool_function:
                observation = f\"错误:未找到名为 '{tool_name}' 的工具。\"
            else:
                observation = tool_function(tool_input) # 调用真实工具

# (这段逻辑紧随工具调用之后，在 while 循环的末尾)
            print(f\"观察: {observation}\")

            # 将本轮的Action和Observation添加到历史记录中
            self.history.append(f\"Action: {action}\")
            self.history.append(f\"Observation: {observation}\")
        # 循环结束
        print(\"已达到最大步数，流程终止。\")
        returnNone

工具 'Search' 已注册。
--- 第 1 步 ---
正在调用 xxxxxx 模型...
大语言模型响应成功:
Thought: 要回答这个问题，我需要查找华为最新发布的手机型号及其主要特点。这些信息可能在我的现有知识库之外，因此需要使用搜索引擎来获取最新数据。
Action: Search[华为最新手机型号及主要卖点]
思考: 要回答这个问题，我需要查找华为最新发布的手机型号及其主要特点。这些信息可能在我的现有知识库之外，因此需要使用搜索引擎来获取最新数据。
行动: Search[华为最新手机型号及主要卖点]
正在执行 [SerpApi] 网页搜索: 华为最新手机型号及主要卖点
观察: [1] 华为手机- 华为官网
智能手机 ; Mate 系列. 非凡旗舰 · HUAWEI Mate XTs. 非凡大师 ; Pura 系列. 先锋影像 · HUAWEI Pura 80 Pro+ ; Pocket 系列. 美学新篇. HUAWEI Pocket 2 ; nova 系列. 专业人像.
[2] 2025年华为手机哪一款性价比高？华为手机推荐与市场分析 ...
现在华为手机最大的卖点只剩下鸿蒙HarmonyOS系统，以及饱受争议的品牌信仰。 这里推荐目前值得入手的几款华为系列手机，根据不同预算自行选择:. 华为目前最受欢迎，也是搭载 ...
[3] 2025年华为新款手机哪个性价比高？10款华为新款手机推荐
选华为主要还是要推荐高端手机，Mate 70和Pura 70系列是最新发布的旗舰机型。 HUAWEI Mate 70. 优点是，拍照配置依旧顶级，全焦段覆盖，适合专业摄影，做工出色，户外抗摔 ...
--- 第 2 步 ---
正在调用 xxxxxx 模型...
大语言模型响应成功:
Thought: 根据搜索结果，华为最新发布的旗舰机型包括Mate 70和Pura 80 Pro+。为了确定最新型号及其主要卖点，我将重点放在这些信息上。从提供的链接来看，Mate 70系列和Pura 80 Pro+都是近期发布的产品，但具体哪一个是“最新”还需要进一步确认。同时，我可以从这些信息中提取出它们的主要
卖点。
Action: Finish[根据最新信息，华为的最新手机可能是HUAWEI Pura 80 Pro+或HUAWEI Mate 70。其中，HUAWEI Mate 70的主要卖点包括顶级的拍照配置，全焦段覆盖，适合专业摄影，做工出色，并且具有良好的户外抗摔性能。而HUAWEI Pura 80 Pro+则强调了先锋影像技术。]
思考: 根据搜索结果，华为最新发布的旗舰机型包括Mate 70和Pura 80 Pro+。为了确定最新型号及其主要卖点，我将重点放在这些信息上。从提供的链接来看，Mate 70系列和Pura 80 Pro+都是近期发布的产品，但具体哪一个是“最新”还需要进一步确认。同时，我可以从这些信息中提取出它们的主要
卖点。
最终答案: 根据最新信息，华为的最新手机可能是HUAWEI Pura 80 Pro+或HUAWEI Mate 70。其中，HUAWEI Mate 70的主要卖点包括顶级的拍照配置，全焦段覆盖，适合专业摄影，做工出色，并且具有良好的户外抗摔性能。而HUAWEI Pura 80 Pro+则强调了先锋影像技术。

PLANNER_PROMPT_TEMPLATE = \"\"\"
你是一个顶级的AI规划专家。你的任务是将用户提出的复杂问题分解成一个由多个简单步骤组成的行动计划。
请确保计划中的每个步骤都是一个独立的、可执行的子任务，并且严格按照逻辑顺序排列。
你的输出必须是一个Python列表，其中每个元素都是一个描述子任务的字符串。
问题: {question}
请严格按照以下格式输出你的计划,```python与```作为前后缀是必要的:
```python
[\"步骤1\", \"步骤2\", \"步骤3\", ...]

该提示词通过角色设定、任务描述与格式约束（Python列表），确保输出稳定可靠。
接着，封装 `Planner` 类：
```python
# 假定 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(question=question)

        # 为了生成计划，我们构建一个简单的消息列表
        messages = [{\"role\": \"user\", \"content\": prompt}]

        print(\"--- 正在生成计划 ---\")
        # 使用流式输出来获取完整的计划
        response_text = self.llm_client.think(messages=messages) or \"\"

        print(f\"计划已生成:\
{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}
请仅输出针对“当前步骤”的回答:
\"\"\"

classExecutor:
    def__init__(self, llm_client):
        self.llm_client = llm_client
    defexecute(self, question: str, plan: list[str]) -> str:
        \"\"\"
        根据计划，逐步执行并解决问题。
        \"\"\"
        history = \"\"# 用于存储历史步骤和结果的字符串

        print(\"\
--- 正在执行计划 ---\")

        for i, step in enumerate(plan):
            print(f\"\
-> 正在执行步骤 {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}\
结果: {response_text}\
\
\"

            print(f\"步骤 {i+1} 已完成，结果: {response_text}\")
        # 循环结束后，最后一步的响应就是最终答案
        final_answer = response_text
        return final_answer

classPlanAndSolveAgent:
    def__init__(self, llm_client):
        \"\"\"
        初始化智能体，同时创建规划器和执行器实例。
        \"\"\"
        self.llm_client = llm_client
        self.planner = Planner(self.llm_client)
        self.executor = Executor(self.llm_client)
    defrun(self, question: str):
        \"\"\"
        运行智能体的完整流程:先规划，后执行。
        \"\"\"
        print(f\"\
--- 开始处理问题 ---\
问题: {question}\")

        # 1. 调用规划器生成计划
        plan = self.planner.plan(question)

        # 检查计划是否成功生成
        ifnot plan:
            print(\"\
--- 任务终止 --- \
无法生成有效的行动计划。\")
            return
        # 2. 调用执行器执行计划
        final_answer = self.executor.execute(question, plan)

        print(f\"\
--- 任务完成 ---\
最终答案: {final_answer}\")

--- 开始处理问题 ---
问题: 一个水果店周一卖出了15个苹果。周二卖出的苹果数量是周一的两倍。周三卖出的数量比周二少了5个。请问这三天总共卖出了多少个苹果？
--- 正在生成计划 ---
正在调用 xxxx 模型...
大语言模型响应成功:
```python
[\"计算周一卖出的苹果数量： 15个\", \"计算周二卖出的苹果数量： 周一数量 × 2 = 15 × 2 = 30个\", \"计算周三卖出的苹果数量： 周二数量 - 5 = 30 - 5 = 25个\", \"计算三天总销量： 周一 + 周二 + 周三 = 15 + 30 + 25 = 70个\"]

[\"计算周一卖出的苹果数量： 15个\", \"计算周二卖出的苹果数量： 周一数量 × 2 = 15 × 2 = 30个\", \"计算周三卖出的苹果数量： 周二数量 - 5 = 30 - 5 = 25个\", \"计算三天总销量： 周一 + 周二 + 周三 = 15 + 30 + 25 = 70个\"]

从上面的输出日志中，我们可以清晰地看到 Plan-and-Solve 范式的工作流程：

1. **规划阶段**： 智能体首先调用 `Planner`，成功地将复杂的应用题分解成了一个包含四个逻辑步骤的 Python 列表。这个结构化的计划为后续的执行奠定了基础。
2. **执行阶段**： `Executor` 严格按照生成的计划，一步一步地向下执行。在每一步中，它都将历史结果作为上下文，确保了信息的正确传递（例如，步骤2正确地使用了步骤1的结果“15个”，步骤3也正确使用了步骤2的结果“30个”）。
3. **结果**：整个过程逻辑清晰，步骤明确，最终智能体准确地得出了正确答案“70个”。

## 4.4 Reflection

在我们已经实现的 ReAct 和 Plan-and-Solve 范式中，智能体一旦完成了任务，其工作流程便告结束。然而，它们生成的初始答案，无论是行动轨迹还是最终结果，都可能存在谬误或有待改进之处。Reflection 机制的核心思想，正是为智能体引入一种**事后（post-hoc）的自我校正循环**，使其能够像人类一样，审视自己的工作，发现不足，并进行迭代优化。

### 4.4.1 Reflection 机制的核心思想

Reflection 机制的灵感来源于人类的学习过程：我们完成初稿后会进行校对，解出数学题后会进行验算。这一思想在多个研究中得到了体现，例如 Shinn, Noah 在2023年提出的 Reflexion 框架[3]。其核心工作流程可以概括为一个简洁的三步循环：**执行 -> 反思 -> 优化**。

1. **执行 (Execution)**：首先，智能体使用我们熟悉的方法（如 ReAct 或 Plan-and-Solve）尝试完成任务，生成一个初步的解决方案或行动轨迹。这可以看作是“初稿”。
2. **反思 (Reflection)**：接着，智能体进入反思阶段。它会调用一个独立的、或者带有特殊提示词的大语言模型实例，来扮演一个“评审员”的角色。这个“评审员”会审视第一步生成的“初稿”，并从多个维度进行评估，例如：

* **事实性错误**：是否存在与常识或已知事实相悖的内容？
* **逻辑漏洞**：推理过程是否存在不连贯或矛盾之处？
* **效率问题**：是否有更直接、更简洁的路径来完成任务？
* **遗漏信息**：是否忽略了问题的某些关键约束或方面？ 根据评估，它会生成一段结构化的**反馈 (Feedback)**，指出具体的问题所在和改进建议。

3. **优化 (Refinement)**：最后，智能体将“初稿”和“反馈”作为新的上下文，再次调用大语言模型，要求它根据反馈内容对初稿进行修正，生成一个更完善的“修订稿”。

如图4.3所示，这个循环可以重复进行多次，直到反思阶段不再发现新的问题，或者达到预设的迭代次数上限。我们可以将这个迭代优化的过程形式化地表达出来。假设  是第  次迭代产生的输出（ 为初始输出），反思模型  会生成针对  的反馈 ：

随后，优化模型  会结合原始任务、上一版输出以及反馈，生成新一版的输出 ：

![](https://mmbiz.qpic.cn/sz_mmbiz_png/iaiaehJqQJoct3je07Va3s1no2TxXm08PtY8vNMqAz8h4mbDBicrV1sya2ToKfPno1OACkgG32n4HAnj5fc9gPNfP0FkfN0M3xZ1UNcibrhe9kY/640?wx_fmt=png&from=appmsg&watermark=1#imgIndex=2)

图 4.3 Reflection 机制中的“执行-反思-优化”迭代循环

与前两种范式相比，Reflection 的价值在于：

* 它为智能体提供了一个内部纠错回路，使其不再完全依赖于外部工具的反馈（ReAct 的 Observation），从而能够修正更高层次的逻辑和策略错误。
* 它将一次性的任务执行，转变为一个持续优化的过程，显著提升了复杂任务的最终成功率和答案质量。
* 它为智能体构建了一个临时的**“短期记忆”**。整个“执行-反思-优化”的轨迹形成了一个宝贵的经验记录，智能体不仅知道最终答案，还记得自己是如何从有缺陷的初稿迭代到最终版本的。更进一步，这个记忆系统还可以是**多模态的**，允许智能体反思和修正文本以外的输出（如代码、图像等），为构建更强大的多模态智能体奠定了基础。

### 4.4.2 案例设定与记忆模块设计

为了在实战中体现 Reflection 机制，我们将引入记忆管理机制，因为reflection通常对应着信息的存储和提取，如果上下文足够长的情况，想让“评审员”直接获取所有的信息然后进行反思往往会传入很多冗余信息。这一步实践我们主要完成**代码生成与迭代优化**。

这一步的目标任务是：“编写一个Python函数，找出1到n之间所有的素数 (prime numbers)。”

这个任务是检验 Reflection 机制的绝佳场景：

1. **存在明确的优化路径**：大语言模型初次生成的代码很可能是一个简单但效率低下的递归实现。
2. **反思点清晰**：可以通过反思发现其“时间复杂度过高”或“存在重复计算”的问题。
3. **优化方向明确**：可以根据反馈，将其优化为更高效的迭代版本或使用备忘录模式的版本。

Reflection 的核心在于迭代，而迭代的前提是能够记住之前的尝试和获得的反馈。因此，一个“短期记忆”模块是实现该范式的必需品。这个记忆模块将负责存储每一次“执行-反思”循环的完整轨迹。

这个 `Memory` 类的设计比较简洁，主体是这样的：

* 使用一个列表 `records` 来按顺序存储每一次的行动和反思。
* `add_record` 方法负责向记忆中添加新的条目。
* `get_trajectory` 方法是核心，它将记忆轨迹“序列化”成一段文本，可以直接插入到后续的提示词中，为模型的反思和优化提供完整的上下文。
* `get_last_execution` 方便我们获取最新的“初稿”以供反思。

### 4.4.3 Reflection 智能体的编码实现

有了 `Memory` 模块作为基础，我们现在可以着手构建 `ReflectionAgent` 的核心逻辑。整个智能体的工作流程将围绕我们之前讨论的“执行-反思-优化”循环展开，并通过精心设计的提示词来引导大语言模型扮演不同的角色。

（1）提示词设计

与之前的范式不同，Reflection 机制需要多个不同角色的提示词来协同工作。

1. **初始执行提示词 (Execution Prompt)** ：这是智能体首次尝试解决问题的提示词，内容相对直接，只要求模型完成指定任务。

2. **反思提示词 (Reflection Prompt)** ：这个提示词是 Reflection 机制的灵魂。它指示模型扮演“代码评审员”的角色，对上一轮生成的代码进行批判性分析，并提供具体的、可操作的反馈。

{code}

3. **优化提示词 (Refinement Prompt)** ：当收到反馈后，这个提示词将引导模型根据反馈内容，对原有代码进行修正和优化。

（2）智能体封装与实现

现在，我们将这套提示词逻辑和 `Memory` 模块整合到 `ReflectionAgent` 类中。

### 4.4.4 运行实例与分析

完整的代码同样参考本书配套的代码仓库 `code` 文件夹，这里提供一个输出实例。

def find_primes(n):
    ...
    return primes

def find_primes(n):
    ...
    return primes

def find_primes(n):
    ...
    return primes

def find_primes(n):
    \"\"\"
    Finds all prime numbers between 1 and n using the Sieve of Eratosthenes algorithm.

    :param n: The upper limit of the range to find prime numbers.
    :return: A list of all prime numbers between 1 and n.
    \"\"\"
    if n < 2:
        return []

    is_prime = [True] * (n + 1)
    is_prime[0] = is_prime[1] = False

    p = 2
    while p * p <= n:
        if is_prime[p]:
            for i in range(p * p, n + 1, p):
                is_prime[i] = False
        p += 1

    primes = [num for num in range(2, n + 1) if is_prime[num]]
    return primes