Agent 学习(2) 智能体经典范式构建

一个专属的 LLM 客户端类

  为了让代码结构更清晰与易于复用，我们定义一个 LLM 客户端类。这个类将封装所有与模型服务交互的细节，让我们的主逻辑可以更专注于智能体的构建，代码如下：

import os
from openai import OpenAI
from dotenv import load_dotenv
from typing import List, Dict

# 加载 .env 文件中的环境变量
load_dotenv()

class LLM_Base:
    """
    该类用于调用任何兼容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,    # 控制输出随机性，0表示贪心解码（最确定）
                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

  使用示例如下：

if __name__ == '__main__':
    try:
        llmClient = LLM_Base()
        
        exampleMessages = [
            {"role": "system", "content": "You are a helpful assistant that writes Python code."},
            {"role": "user", "content": "你好"}
        ]
        
        print("--- 调用LLM ---")
        responseText = llmClient.think(exampleMessages)

    except ValueError as e:
        print(e)

ReAct 范式

ReAct 工作流

  ReAct 在 2022 年被提出，意为 Reason + Act，其核心思想为模仿人类解决问题的方式，将推理与行动显式的结合起来，形成一个 “思考-行动-观察” 的循环。

  在 ReAct 诞生以前，主流的方法可以分为两类，一种是纯思考，如思维链，它能引导模型进行复杂的逻辑推理，但无法与外部世界交互，容易产生事实幻觉；另一类是“纯行动”型，模型直接输出要执行的动作，但缺乏规划和纠错能力。ReAct 认识到：思考和行动是相辅相成的。思考指导行动，行动又反过来修正思考，为此，ReAct范式通过一种特殊的提示工程来引导模型，使其每一步的输出都遵循一个固定的轨迹：

• Thought(思考)：这是智能体的“内心独白”。它会分析当前情况、分解任务、制定下一步计划，或者反思上一步的结果。
• Action (行动)： 这是智能体决定采取的具体动作，通常是调用一个外部工具，例如 Search[‘华为最新款手机’]。
• Observation (观察)： 这是执行Action后从外部工具返回的结果，例如搜索结果的摘要或API的返回值。

  智能体将不断重复这个 Thought -> Action -> Observation 的循环，将新的观察结果追加到历史记录中，形成一个不断增长的上下文，直到它在Thought中认为已经找到了最终答案，然后输出结果。这个过程形成了一个强大的协同效应：推理使得行动更具目的性，而行动则为推理提供了事实依据。

  接下来我们将基于 ReAct 构建一个具备使用外部工具能力的智能体，来回答一个大语言模型仅凭自身知识库无法直接回答的问题。例如：“今天南京的天气怎么样？”

工具的定义和实现

  LLM 是智能体的大脑，工具就是其与外部世界交互的手和脚。现在我们需要为智能体提供一个网页搜索工具。在这里我们选用 SerpApi，它通过API提供结构化的 Google 搜索结果，能直接返回“答案摘要框”或精确的知识图谱信息。在获取好 Serpapi 的 API_KEY 以后，我们通过代码来定义和管理这个工具。我们先实现工具的核心功能，然后构建一个通用的工具管理器。

  (1) 实现搜索工具的核心逻辑
  一个良好定义的工具应该包含以下三个核心要素：

• 名称 (Name)： 一个简洁、唯一的标识符，供智能体在 Action 中调用，例如 Search。
• 描述 (Description)： 一段清晰的自然语言描述，说明这个工具的用途。这是整个机制中最关键的部分，因为大语言模型会依赖这段描述来判断何时使用哪个工具。
• 执行逻辑 (Execution Logic)： 真正执行任务的函数或方法。

  我们的第一个工具是 search 函数，它的作用是接收一个查询字符串然后返回搜索结果。实现代码如下：

import os
from serpapi import SerpApiClient
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}"

  (2) 构建通用的工具执行器
  当智能体需要多种工具时，我们需要一个统一的管理器来注册和调度这些工具。为此我们创建一个 ToolExecutor 类。

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()
        ])

  (3) 测试
  现在，我们将 search 工具注册到 ToolExecutor 中，并模拟一次调用，以验证整个流程是否正常工作。测试代码如下：

from tools import search
from tools import ToolExecutor

toolExecutor = ToolExecutor()
search_description = "一个网页搜索引擎，当你需要回答不在你的知识库中的信息时，使用此工具"
toolExecutor.registerTool("Search",search_description,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}' 的工具。")

ReAct 智能体的编码实现

  现在，我们把所有独立的组件组合、构建成一个完整的 ReAct 智能体。我们通过一个 ReActAgent 类来封装核心逻辑。

（1）系统提示词设计

  提示词是整个 ReAct 机制的基石，它为大语言模型提供了行动的操作指令，我们需要设计一个模板，它将动态的插入可用工具，用户问题以及中间步骤的交互历史，以下为示例：

# ReAct 提示词模板
REACT_PROMPT_TEMPLATE = """
请注意，你是一个有能力调用外部工具的智能助手。

可用工具如下:
{tools}

请严格按照以下格式进行回应:

Thought: 你的思考过程，用于分析问题、拆解任务和规划下一步行动。
Action: 你决定采取的行动，必须是以下格式之一:
- `{{tool_name}}[{{tool_input}}]`:调用一个可用工具。
- `Finish[最终答案]`:当你认为已经获得最终答案时。
- 当你收集到足够的信息，能够回答用户的最终问题时，你必须在Action:字段后使用 Finish[最终答案] 来输出最终答案。

现在，请开始解决以下问题:
Question: {question}
History: {history}
"""

  这个模板定义了智能体与LLM之间交互的规范：

• 角色定义：“你是一个有能力调用外部工具的智能助手”，设定了LLM的角色。
• 工具清单 ({tools})：告知LLM它有哪些可用的“手脚”。
• 格式规约 (Thought/Action)：这是最重要的部分，它强制LLM的输出具有结构性，使我们能通过代码精确解析其意图。
• 动态上下文 ({question}/{history})：将用户的原始问题和不断累积的交互历史注入，让LLM基于完整的上下文进行决策。

（2）核心循环的实现

  ReActAgent的核心是一个循环，其不断地“格式化提示词 -> 调用 LLM -> 执行动作 -> 整合结果”，直到任务完成或达到最大步数限制。代码如下：

def run(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 = "\n".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)

        if not response_text:
            print("错误：LLM 未能返回有效响应")

        # ...后续解析、执行、整合等

  run 方法是智能体的入口。它的 while 循环构成了 ReAct 范式的主体，max_steps 参数则是一个重要的安全阀，防止智能体陷入无限循环而耗尽资源。

（3）输出解析器的实现

  LLM 返回的是纯文本，我们需要从中精确地提取出 Thought 和 Action。这是通过几个辅助解析函数完成的，它们通常使用正则表达式来实现。代码如下：

def _parse_output(self, text: str):
    """
    解析LLM的输出，提取Thought和Action。
    """
    # Thought: 匹配到 Action: 或文本末尾
    thought_match = re.search(r"Thought:\s*(.*?)(?=\nAction:|$)", text, re.DOTALL)
    # Action: 匹配到文本末尾
    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

（4）工具调用与执行

  紧接第二步，我们解析 LLM 的输出并做出行动。代码如下：

# 3.解析 LLM 的输出
thought, action = self._parse_action(response_text)
if thought:
    print(f"思考：{thought}")

if not action:
    print("警告：未能解析出有效的 Action，流程终止。")
    break

# 4.执行Action
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_output(action)
if not tool_name or not tool_input:
    # ... 处理无效Action格式 ...
    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) # 调用真实工具

（5）观测结果的整合

  最后一步，是将 Action 本身和工具执行后的 Observation 添回历史记录中，为下一轮循环提供新的上下文，紧接在第四步后的代码如下

    print(f"👀 观察: {observation}")
    
    # 将本轮的Action和Observation添加到历史记录中
    self.history.append(f"Action: {action}")
    self.history.append(f"Observation: {observation}")

# 循环结束
print("已达到最大步数，流程终止。")
return None

（6）测试

  写出如下的测试程序：

from ReActAgent import ReActAgent
from tools import search
from tools import ToolExecutor
from LLM_base import LLM_Base

toolExecutor = ToolExecutor()
llm_base = LLM_Base()
search_description = "一个网页搜索引擎，当你需要回答不在你的知识库中的信息时，使用此工具"
toolExecutor.registerTool("Search",search_description,search)


agent = ReActAgent(llm_base=llm_base, tool_executor=toolExecutor)

agent.run("Apple最新款的手机是什么型号？")

  运行一下，可以看到终端输出了正确答案：

思考：根据搜索结果，苹果最新发布的手机是iPhone Air，于2025年4月发布，起售价RMB 7,999。同时有iPhone 17系列，但iPhone Air是最新推出的型号。
🎉 最终答案: Apple最新款的手机是iPhone Air（2025年4月发布），提供深空黑色、云白色、浅金色和天蓝色，起步存储容量256GB，起售价RMB 7,999。

总结

  至此我们就完成了一个 ReAct 智能体。接下来我们总结一下 ReAct 的优劣。

  （1）ReAct 的主要特点

• 高可解释性：ReAct 最大的优点之一就是透明。通过 Thought 链，我们可以清晰地看到智能体每一步的“心路历程”——它为什么会选择这个工具，下一步又打算做什么。这对于理解、信任和调试智能体的行为至关重要。
• 动态规划与纠错能力：与一次性生成完整计划的范式不同，ReAct 是“走一步，看一步”。它根据每一步从外部世界获得的 Observation 来动态调整后续的 Thought 和 Action。如果上一步的搜索结果不理想，它可以在下一步中修正搜索词，重新尝试。
• 工具协同能力：ReAct 范式天然地将大语言模型的推理能力与外部工具的执行能力结合起来。LLM 负责运筹帷幄（规划和推理），工具负责解决具体问题（搜索、计算），二者协同工作，突破了单一 LLM 在知识时效性、计算准确性等方面的固有局限。

  （2）ReAct 的固有局限性

• 对LLM自身能力的强依赖：ReAct 流程的成功与否，高度依赖于底层 LLM 的综合能力。如果 LLM 的逻辑推理能力、指令遵循能力或格式化输出能力不足，就很容易在 Thought 环节产生错误的规划，或者在 Action 环节生成不符合格式的指令，导致整个流程中断。
• 执行效率问题：由于其循序渐进的特性，完成一个任务通常需要多次调用 LLM。每一次调用都伴随着网络延迟和计算成本。对于需要很多步骤的复杂任务，这种串行的“思考-行动”循环可能会导致较高的总耗时和费用。
• 提示词的脆弱性：整个机制的稳定运行建立在一个精心设计的提示词模板之上。模板中的任何微小变动，甚至是用词的差异，都可能影响 LLM 的行为。此外，并非所有模型都能持续稳定地遵循预设的格式，这增加了在实际应用中的不确定性。
• 可能陷入局部最优：步进式的决策模式意味着智能体缺乏一个全局的、长远的规划。它可能会因为眼前的 Observation 而选择一个看似正确但长远来看并非最优的路径，甚至在某些情况下陷入“原地打转”的循环中。

Plan-and-solve

工作原理

  Plan-and-Solve Prompting 在 2023 年提出，其核心动机是为了解决思维链在处理多步骤、复杂问题时容易“偏离轨道”的问题。与 ReAct 将思考和行动融合在每一步不同，Plan-and-Solve 将整个流程解耦成两个核心阶段：

• 规划阶段 (Planning Phase)： 首先，智能体会接收用户的完整问题。它的第一个任务不是直接去解决问题或调用工具，而是将问题分解，并制定出一个清晰、分步骤的行动计划。这个计划本身就是一次大语言模型的调用产物。
• 执行阶段 (Solving Phase)： 在获得完整的计划后，智能体进入执行阶段。它会严格按照计划中的步骤，逐一执行。每一步的执行都可能是一次独立的 LLM 调用，或者是对上一步结果的加工处理，直到计划中的所有步骤都完成，最终得出答案。

  这种“先谋后动”的策略，使得智能体在处理需要长远规划的复杂任务时，能够保持更高的目标一致性，避免在中间步骤中迷失方向。

规划阶段

  为了凸显 Plan-and-Solve 范式在结构化推理任务上的优势，我们将不使用工具的方式，而是通过提示词的设计，完成一个推理任务。这类任务的特点是，答案无法通过单次查询或计算得出，必须先将问题分解为一系列逻辑连贯的子步骤，然后按顺序求解。这恰好能发挥 Plan-and-Solve “先规划，后执行”的核心能力。

  规划阶段的目标是让大语言模型接收原始问题，并输出一个清晰、分步骤的行动计划。这个计划必须是结构化的，以便我们的代码可以轻松解析并逐一执行。因此，我们设计的提示词需要明确地告诉模型它的角色和任务，并给出一个输出格式的范例。

PLANNER_PROMPT_TEMPLATE = """
你是一个顶级的AI规划专家。你的任务是将用户提出的复杂问题分解成一个由多个简单步骤组成的行动计划。
请确保计划中的每个步骤都是一个独立的、可执行的子任务，并且严格按照逻辑顺序排列。
你的输出必须是一个Python列表，其中每个元素都是一个描述子任务的字符串。

问题: {question}

请严格按照以下格式输出你的计划,```python与```作为前后缀是必要的:
```python
["步骤1", "步骤2", "步骤3", ...]
'```
"""

  接下来我们再封装一个 Planner 的类，这个类也是我们的规划器

class Planner:
    def __init__(self, llm_client: LLM_Base):
        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}")
        
        # 解析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 []

执行阶段

  在规划器中生成了清晰的行动蓝图后，我们就需要一个执行器来逐一完成计划中的任务。执行器不仅负责调用 LLM 来解决每个子问题，还承担着一个至关重要的角色：状态管理。它必须记录每一步的执行结果，并将其作为上下文提供给后续步骤，确保信息在整个任务链条中顺畅流动。执行器的提示词与规划器不同，它的目标不是分解问题，而是**在已有上下文的基础上，专注解决当前这一个步骤。**基于此原则，我们写下如下提示词：

EXECUTOR_PROMPT_TEMPLATE = """
你是一位顶级的AI执行专家。你的任务是严格按照给定的计划，一步步地解决问题。
你将收到原始问题、完整的计划、以及到目前为止已经完成的步骤和结果。
请你专注于解决“当前步骤”，并仅输出该步骤的最终答案，不要输出任何额外的解释或对话。

# 原始问题:
{question}

# 完整计划:
{plan}

# 历史步骤与结果:
{history}

# 当前步骤:
{current_step}

请仅输出针对“当前步骤”的回答:
"""

  接下来，我们写如下代码，实现执行器

class Executor:
    def __init__(self, llm_client: LLM_Base):
        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

整合并测试

  最终我们创建一个 PlanExecute_Agent 类，整合如上两个功能，具体代码如下

class PlanExecute_Agent:
    def __init__(self, llm_base:LLM_Base):
        self.llm_client = llm_base
        self.planner = Planner(llm_client=self.llm_client)
        self.executor = Executor(llm_client=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}")

  测试代码如下：

from PlanSolve import PlanExecute_Agent
from LLM_base import LLM_Base

llm_base = LLM_Base()

agent = PlanExecute_Agent(llm_base=llm_base)

agent.run(question="一个水果店周一卖出了15个苹果。周二卖出的苹果数量是周一的两倍。周三卖出的数量比周二少了5个。请问这三天总共卖出了多少个苹果？")

  最终生成结果如下，可以看到完整的执行计划过程。

--- 正在执行计划 ---

-> 正在执行步骤 1/3: 计算周二卖出的苹果数量：周一的两倍
🧠 正在调用 deepseek-v4-flash 模型...
✅ 大语言模型响应成功:
30
√ 步骤 1 已完成，结果: 30

-> 正在执行步骤 2/3: 计算周三卖出的苹果数量：周二减去5
🧠 正在调用 deepseek-v4-flash 模型...
✅ 大语言模型响应成功:
25
√ 步骤 2 已完成，结果: 25

-> 正在执行步骤 3/3: 计算三天总卖出数量：周一+周二+周三
🧠 正在调用 deepseek-v4-flash 模型...
✅ 大语言模型响应成功:
70
√ 步骤 3 已完成，结果: 70

--- 任务完成 ---
最终答案: 70

Reflection

核心思想

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

• 执行：首先，智能体使用我们熟悉的方法（如 ReAct 或 Plan-and-Solve）尝试完成任务，生成一个初步的解决方案或行动轨迹。这可以看作是“初稿”。

• 反思：接着，智能体进入反思阶段。它会调用一个独立的、或者带有特殊提示词的大语言模型实例，来扮演一个“评审员”的角色。这个“评审员”会审视第一步生成的“初稿”，并从多个维度进行评估，例如：事实性错误、逻辑漏洞、效率问题、遗漏信息等等。

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

  这个循环可以重复进行多次，直到反思阶段不再发现新的问题，或者达到预设的迭代次数上限。

记忆模块设计

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

  Reflection 的核心在于迭代，而迭代的前提是能够记住之前的尝试和反馈。因此一个短期记忆模块是必需品，这个记忆模块将负责存储每一次执行、反思循环的完整轨迹。完整代码如下：

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

Reflection 的编码实现

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

提示词的设计

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

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

INITIAL_PROMPT_TEMPLATE = """
你是一位资深的Python程序员。请根据以下要求，编写一个Python函数。
你的代码必须包含完整的函数签名、文档字符串，并遵循PEP 8编码规范。

要求: {task}

请直接输出代码，不要包含任何额外的解释。
"""

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

REFLECT_PROMPT_TEMPLATE = """
你是一位极其严格的代码评审专家和资深算法工程师，对代码的性能有极致的要求。
你的任务是审查以下Python代码，并专注于找出其在<strong>算法效率</strong>上的主要瓶颈。

# 原始任务:
{task}

# 待审查的代码:
```python
{code}
'```

请分析该代码的时间复杂度，并思考是否存在一种<strong>算法上更优</strong>的解决方案来显著提升性能。
如果存在，请清晰地指出当前算法的不足，并提出具体的、可行的改进算法建议（例如，使用筛法替代试除法）。
如果代码在算法层面已经达到最优，才能回答“无需改进”。

请直接输出你的反馈，不要包含任何额外的解释。
"""

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

REFINE_PROMPT_TEMPLATE = """
你是一位资深的Python程序员。你正在根据一位代码评审专家的反馈来优化你的代码。

# 原始任务:
{task}

# 你上一轮尝试的代码:
{last_code_attempt}
评审员的反馈：
{feedback}

请根据评审员的反馈，生成一个优化后的新版本代码。
你的代码必须包含完整的函数签名、文档字符串，并遵循PEP 8编码规范。
请直接输出优化后的代码，不要包含任何额外的解释。
"""

智能体的封装与实现

  接下来我们将代码封装到 ReflectionAgent 中，然后测试

class ReflectionAgent:
    def __init__(self, llm_client: LLM_Base, 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}")

        # --- 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:
                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_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

  成功！可以在终端中看到经过两轮迭代后大模型认为代码无需修改。

三种方式总结

  我们通过“亲手造轮子”的方式，从零开始编码实现了三种业界经典的智能体构建范式:ReAct、Plan-and-Solve 与 Reflection。我们不仅探索了它们的核心工作原理，还通过具体的实战案例，深入了解了各自的优势、局限与适用场景。

• ReAct:我们构建了一个能与外部世界交互的 ReAct 智能体。通过“思考-行动-观察”的动态循环，它成功地利用搜索引擎回答了自身知识库无法覆盖的实时性问题。其核心优势在于环境适应性和动态纠错能力，使其成为处理探索性、需要外部工具输入的任务的首选。
• Plan-and-Solve:我们实现了一个先规划后执行的 Plan-and-Solve 智能体，并利用它解决了需要多步推理的数学应用题。它将复杂的任务分解为清晰的步骤，然后逐一执行。其核心优势在于结构性和稳定性，特别适合处理逻辑路径确定、内部推理密集的任务。
• Reflection (自我反思与迭代):我们构建了一个具备自我优化能力的 Reflection 智能体。通过引入“执行-反思-优化”的迭代循环，它成功地将一个效率较低的初始代码方案，优化为了一个算法上更优的高性能版本。其核心价值在于能显著提升解决方案的质量，适用于对结果的准确性和可靠性有极高要求的场景。

  至此，我们已经掌握了构建单个智能体的核心技术。