AI–智能体Reflection架构

0.前言

在前面实现的两种架构中，智能体一旦完成任务后，工作流程便会停止，由于函数的概率性，其所生成的结果并不百分百完全正确，依旧可能存在错误或者需要改进的地方，所以我们需要思考如何降低这种概率。

Reflection 机制本质是融入了我们反思的架构，我们希望智能体像我们人类一样，审视自己的工作，反思并进行迭代优化

（学习内容主要参考来源：https://github.com/datawhalechina/hello-agents）

1.Reflection架构的工作原理

Reflection架构的核心工作流程可以概括为简单的三步循环：执行 -> 反思 -> 优化

执行：首先，智能体使用我们熟悉的方法（如 ReAct 或 Plan-and-Solve）尝试完成任务，生成一个初步的解决方案或行动轨迹。这可以看作是“初稿”。
反思：接着，智能体进入反思阶段。它会调用一个独立的、或者带有特殊提示词的大语言模型实例，来扮演一个“评审员”的角色。这个“评审员”会审视第一步生成的“初稿”，并从多个维度进行评估，例如：
- 事实性错误：是否存在与常识或已知事实相悖的内容？
- 逻辑漏洞：推理过程是否存在不连贯或矛盾之处？
- 效率问题：是否有更直接、更简洁的路径来完成任务？
- 遗漏信息：是否忽略了问题的某些关键约束或方面？根据评估，它会生成一段结构化的**反馈 (Feedback)**，指出具体的问题所在和改进建议。
优化：最后，智能体将“初稿”和“反馈”作为新的上下文，再次调用大语言模型，要求它根据反馈内容对初稿进行修正，生成一个更完善的“修订稿”。

与前两种架构相比，Reflection 架构的优势在于：

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

2.Reflection 架构的记忆模块设计

首先明确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

3.Reflection 架构的提示词工程

在构建好记忆模块后，开始着手构建核心逻辑，一如既往的是以提示词作为开始，由于存在执行-反思-优化的逻辑，所以该架构的提示词也分成了三种用来扮演不同的角色

初始执行提示词

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

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

要求: {task}

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

反思提示词

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

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

# 原始任务:
{task}

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

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

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

优化提示词

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

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

# 原始任务:
{task}

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

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

4.Reflection 架构的代码封装

# 假设 llm_client.py 和 memory.py 已定义
# from llm_client import HelloAgentsLLM
# from memory import Memory

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

完整代码

import os
#为后续调用llm下载函数
from openai import OpenAI
#把.env 文件里的键值对读进当前进程的环境变量
from dotenv import load_dotenv


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

#定义一个类，类名是HelloAgentsLLM，类里面有三个方法：__init__、think、__main__
class HelloAgentsLLM:
    """
    为本书 "Hello Agents" 定制的LLM客户端。
    它用于调用任何兼容OpenAI接口的服务，并默认使用流式响应。
    """
    #等价于 def __init__(self, model=None, apiKey=None, baseUrl=None, timeout=None):
    def __init__(self, model: str | None = None, apiKey: str | None = None, baseUrl: str | None = None, timeout: int | None = 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 | None:
        """
        调用大语言模型进行思考，并返回其响应。
        """
        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: # 如果没有选择，跳过,因为有些大语言模型会返回空的选择，比如OpenAI 会在所有正常 chunk 推完之后，再单独推一个特殊的收尾 chunk
                    continue
                # 从当前 chunk 中提取内容
                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)

通过运行可以看到，智能体在接收到明确的反馈后，于优化阶段成功地实现了更高效的筛法，将算法复杂度降至 O(n log log n)，完成了第一次有意义的自我迭代。充分证明，一个设计良好的 Reflection 机制，其价值不仅在于修复错误，更在于驱动解决方案在质量和效率上实现阶梯式的提升，这使其成为构建复杂、高质量智能体的关键技术之一。

5.Reflection 架构的成本分析

尽管Reflection 机制在提升任务解决质量上表现出色，但是在实际应用中，我们需要权衡其带来的收益与相应的成本。

5.1主要成本

模型调用开销增加：

这是最直接的成本。每进行一轮迭代，至少需要额外调用两次大语言模型（一次用于反思，一次用于优化）。如果迭代多轮，API 调用成本和计算资源消耗将成倍增加。

任务延迟显著提高：

Reflection 是一个串行过程，每一轮的优化都必须等待上一轮的反思完成。这使得任务的总耗时显著延长，不适合对实时性要求高的场景。

提示工程复杂度上升：

如我们的案例所示，Reflection 的成功在很大程度上依赖于高质量、有针对性的提示词。为“执行”、“反思”、“优化”等不同阶段设计和调试有效的提示词，需要投入更多的开发精力。

5.2核心收益

解决方案质量的跃迁：

最大的收益在于，它能将一个“合格”的初始方案，迭代优化成一个“优秀”的最终方案。这种从功能正确到性能高效、从逻辑粗糙到逻辑严谨的提升，在很多关键任务中是至关重要的。

鲁棒性与可靠性增强：

通过内部的自我纠错循环，智能体能够发现并修复初始方案中可能存在的逻辑漏洞、事实性错误或边界情况处理不当等问题，从而大大提高了最终结果的可靠性。

6.Reflection 架构的场景分析

综上所述，Reflection 机制是一种典型的“以成本换质量”的策略。它非常适合那些对最终结果的质量、准确性和可靠性有极高要求，且对任务完成的实时性要求相对宽松的场景。例如:

生成关键的业务代码或技术报告。
在科学研究中进行复杂的逻辑推演。
需要深度分析和规划的决策支持系统。

AI--智能体Reflection架构

https://one-null-pointer.github.io/2026/03/15/AI --智能体Reflection架构/

Author

liaoyue

Posted on

March 15, 2026

传送口

AI--智能体Plan-and-Solve架构 Next

AI--智能体Reflection架构

AI–智能体Reflection架构

0.前言

1.Reflection架构的工作原理

2.Reflection 架构的记忆模块设计

3.Reflection 架构的提示词工程

初始执行提示词

反思提示词

优化提示词

4.Reflection 架构的代码封装

5.Reflection 架构的成本分析

5.1主要成本

模型调用开销增加：

任务延迟显著提高：

提示工程复杂度上升：

5.2核心收益

解决方案质量的跃迁：

鲁棒性与可靠性增强：

6.Reflection 架构的场景分析