Reflection 机制的灵感来源于人类的学习过程—— 执行->反思->优化。它为智能体引入了一种事后(post-hoc)的自我校正循环。

  1. 执行(Execution)：智能体使用 ReAct 或 Plan-and-Solve 尝试完成任务，生成一个初步的行动轨迹
  2. 反思(Reflection)：智能体调用独立的（带有定制prompt）扮演“评审员”角色的大模型，从事实错误、逻辑漏洞、效率问题等多个维度对行动轨迹进行评估。
  3. 优化(Refinement)：智能体将初步行动轨迹和评审结果作为新的上下文，再次调用大模型。

这个循环可以重复进行多次，直到反思阶段不再发现新的问题，或者达到预设的迭代次数上限。我们可以将这个迭代优化的过程形式化地表达出来。假设 $O_i$ 是第 ( i ) 次迭代产生的输出 $O_0$ 为初始输出），反思模型 $\pi_{\text{reflect}}$ 会生成针对 $O_i$ 的反馈 $F_i$ ：

$F_i = \pi_{\text{reflect}}(\text{Task}, O_i)$
随后，优化模型 $\pi_{\text{refine}}$ 会结合原始任务、上一版输出以及反馈，生成新一版的输出 $O_{i+1}$ ：

$$ O_{i+1} = \pi_{\text{refine}}(\text{Task}, O_i, F_i) $$

image-20260810105352513

Reflection实现

Reflection 通常对应着信息的存储和提取，为了在实战中体现Reflection机制，需要引入记忆管理机制。这里以一个代码生成与迭代优化任务——使用python找到[1,n]之间的全部质数为例，有以下原因：

  • 大模型初次生成的代码很可能是简单且效率低下的递归实现。
  • 通过反思能够发现重复计算的问题

提示词设计

初始任务提示词

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

要求：{task}

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

反思提示词设计

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

# 原始任务：
{task}

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

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

优化提示词设计

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

# 原始任务：
{task}

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

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

记忆管理

以下实现通过records存储行动和反思，并使用 get_trajectory将记忆序列化为上下文。

from typing import List, Any, Dict, Optional


class Memory:
    def __init__(self):
        self.records: List[Dict[str, Any]] = []

    def add_record(self, record_type: str, content: str):
        record = {"type": record_type, "content":content}
        self.records.append(record)

    # 将记忆列表格式化为可构建上下文的字符串
    def get_trajectory(self)-> str:
        trajectory_parts = []
        for record in self.records:
            if record['type'] == 'execution':
                trajectory_parts.append(self, f"---上一轮尝试---\n {record['content']}")
            elif record['type'] == 'reflection':
                trajectory_parts.append(f"---评审员反馈---\n{record['content']}")
                
        return "\n\n".join(trajectory_parts)
    
    def get_last_excution(self)->Optional[str]:
        for record in reversed(self.records):
            if record['type'] == 'execution':
                return record['content']
        return None

Agent 内核实现


class ReflectionAgent:
    def __init__(self, llm_client, max_iterations):
        self.llm_client = llm_client
        self.memory = Memory()
        self.max_iterations = max_iterations

    def run(self, task: str):
        initial_prompt = INITIAL_PROMPT_TEMPLATE.format(task = task)
        initial_code = self._get_llm_response(initial_prompt)
        self.memory.add_record("excution", initial_code)

        for i in range(self.max_iterations):
            last_code = self.memory.get_last_excution()
            reflect_prompt = REFLECT_PROMPT_TEMPLATE.format(task = task, last_code = last_code)
            feedback = self._get_llm_response(reflect_prompt)

            if "无需改进" in feedback:
                break

            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("excution", refined_code)

    def _get_llm_response(self, prompt: str)-> str:
        messages = [{"role": "user", "content": prompt}]
        response_text = self.llm_client.think(messages = messages) or ""
        return response_text

Reflection的特点

  • 成本增加 ：模型调用开销增加、任务延迟增加、提示词复杂度增加
  • 质量跃迁：反思和优化极大的提高了解决方案的质量和可靠性。