Reflection反思机制-让Agent自我迭代优化
本章介绍 Reflection 范式的核心思想、提示词设计要点、完整实现，以及成本收益的权衡分析。
Reflection 反思机制：让 Agent 自我迭代优化
本章介绍 Reflection 范式的核心思想、提示词设计要点、完整实现，以及成本收益的权衡分析。
1.1 Reflection 的核心思想
Reflection 反思循环 — Actor 生成初稿，Critic 严格审查，直到满意为止
Reflection 的逻辑是：执行完不算完，还要自己审一遍，发现问题再改，直到满意为止。
以代码生成为例。LLM 生成一个找素数的函数，初版可能给出试除法：
def find_primes(n):
primes = []
for num in range(2, n + 1):
is_prime = True
for i in range(2, int(num ** 0.5) + 1):
if num % i == 0:
is_prime = False
break
if is_prime:
primes.append(num)
return primes功能是对的，但这是试除法，时间复杂度 O(n√n)（意思是当数字规模 n 增大时，运算量会成倍增长，性能很差）。数据量大了完全扛不住。如果有一个"评审员"审查一遍，发现"这个算法效率太低，改用埃拉托斯特尼筛法（一种更高效的求素数方法，通过逐步筛除合数来找出所有素数）"，再让模型改一遍，结果会好很多。
这就是 Reflection 的基本思路。
和 ReAct、Plan-and-Solve 不同，Reflection 不是在解决"怎么把任务执行完"的问题，而是在解决"怎么把任务执行好"的问题。它的目标是提升输出质量，而不是扩展任务能力。
1.2 三个提示词的设计
Reflection 的实现围绕三个提示词展开，每个的措辞都很重要。
执行提示词：让 LLM 直接完成任务，不加任何约束。
ACTOR_PROMPT = """你是一个 Python 专家。
请根据用户需求，直接给出可运行的 Python 函数，包含函数定义、必要注释和时间复杂度说明。"""反思提示词：这一步最关键，也最容易写错。如果提示词太温和，LLM 会说"这个实现很好，没有需要改进的地方"——这等于没有 Critic，白花了一次调用。提示词必须足够"严格"，明确告诉模型要找哪类问题。
CRITIC_PROMPT = """你是一个严苛的代码审查员，专门挑算法效率问题。
审查标准（必须逐项检查）：
1. 时间复杂度：是否存在更优的算法？O(n²) 能否优化到 O(n log n)？O(n√n) 能否优化到 O(n log log n)？
2. 空间复杂度：内存使用是否可以优化？
3. 边界处理：是否处理了 n=0、n=1、负数等边界情况？
4. 代码质量：变量命名是否清晰？是否有重复逻辑可以抽取？
重要规则：你的职责是找问题，不是表扬。如果你觉得代码"还不错"，说明你审查不够仔细。
请给出具体的改进建议，指出应该使用哪种算法，以及优化后的预期时间复杂度。
如果确实无法继续优化（已经是理论最优），在最后一行单独写：NO_IMPROVEMENT_NEEDED"""优化提示词：把初稿和反馈一起传给模型，让它针对性地改进。
REFINE_PROMPT = """你是一个 Python 专家。
你的任务是根据代码审查意见，改进原有代码。
原始代码：
{original_code}
审查意见：
{feedback}
请根据审查意见，给出改进后的完整代码。改动的部分需要在注释里说明改进原因。"""1.3 完整代码实现
from openai import OpenAI
from dataclasses import dataclass, field
client = OpenAI()
ACTOR_PROMPT = """你是一个 Python 专家。
请根据用户需求，直接给出可运行的 Python 函数，包含函数定义、必要注释和时间复杂度说明。"""
CRITIC_PROMPT = """你是一个严苛的代码审查员，专门挑算法效率问题。
审查标准（必须逐项检查）：
1. 时间复杂度：是否存在更优的算法？O(n²) 能否优化到 O(n log n)？O(n√n) 能否优化到 O(n log log n)？
2. 空间复杂度：内存使用是否可以优化？
3. 边界处理：是否处理了 n=0、n=1、负数等边界情况？
4. 代码质量：变量命名是否清晰？是否有重复逻辑可以抽取？
重要规则：你的职责是找问题，不是表扬。如果你觉得代码"还不错"，说明你审查不够仔细。
请给出具体的改进建议，指出应该使用哪种算法，以及优化后的预期时间复杂度。
如果确实无法继续优化（已经是理论最优），在最后一行单独写：NO_IMPROVEMENT_NEEDED"""
REFINE_PROMPT_TEMPLATE = """你是一个 Python 专家。
你的任务是根据代码审查意见，改进原有代码。
原始代码：
{original_code}
审查意见：
{feedback}
请根据审查意见，给出改进后的完整代码。改动的部分需要在注释里说明改进原因。"""
# ============================================================
# Memory：存储执行记录和反思记录
# ============================================================
@dataclass
class MemoryEntry:
version: int
code: str
feedback: str = ""
@dataclass
class Memory:
entries: list[MemoryEntry] = field(default_factory=list)
def add(self, version: int, code: str, feedback: str = ""):
self.entries.append(MemoryEntry(version=version, code=code, feedback=feedback))
def get_trajectory(self) -> str:
"""序列化成文本，用于调试和日志"""
lines = []
for entry in self.entries:
lines.append(f"=== 版本 {entry.version} ===")
lines.append(entry.code)
if entry.feedback:
lines.append(f"--- 审查意见 ---")
lines.append(entry.feedback)
lines.append("")
return "\n".join(lines)
def latest_code(self) -> str:
if self.entries:
return self.entries[-1].code
return ""
# ============================================================
# ReflectionAgent：初始执行 → 循环（反思 → 检查 → 优化）
# ============================================================
class ReflectionAgent:
def __init__(self, max_iterations: int = 3):
self.max_iterations = max_iterations
def _execute(self, task: str) -> str:
"""初次执行任务"""
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": ACTOR_PROMPT},
{"role": "user", "content": task},
],
temperature=0,
)
return response.choices[0].message.content.strip()
def _critique(self, code: str) -> tuple[bool, str]:
"""
审查代码质量。
返回：(是否需要继续改进, 反馈内容)
"""
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": CRITIC_PROMPT},
{"role": "user", "content": f"请审查以下代码：\n\n{code}"},
],
temperature=0,
)
feedback = response.choices[0].message.content.strip()
# 如果审查员认为已经无需改进，停止循环
need_improvement = "NO_IMPROVEMENT_NEEDED" not in feedback
return need_improvement, feedback
def _refine(self, original_code: str, feedback: str) -> str:
"""根据审查意见改进代码"""
prompt = REFINE_PROMPT_TEMPLATE.format(
original_code=original_code,
feedback=feedback,
)
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "user", "content": prompt},
],
temperature=0,
)
return response.choices[0].message.content.strip()
def run(self, task: str) -> tuple[str, Memory]:
memory = Memory()
# 初次执行
print("\n[初次执行]")
initial_code = self._execute(task)
print(initial_code)
memory.add(version=0, code=initial_code)
current_code = initial_code
for i in range(self.max_iterations):
print(f"\n[第 {i + 1} 轮反思]")
# 反思
need_improvement, feedback = self._critique(current_code)
print(f"审查意见：\n{feedback}")
# 检查是否需要继续改进
if not need_improvement:
print(f"\n审查员认为已无需改进，停止迭代。")
break
# 优化
print(f"\n[第 {i + 1} 轮优化]")
refined_code = self._refine(current_code, feedback)
print(refined_code)
# 更新记录
memory.entries[-1].feedback = feedback
memory.add(version=i + 1, code=refined_code)
current_code = refined_code
return current_code, memory
# ============================================================
# 测试：从试除法到埃拉托斯特尼筛法
# ============================================================
if __name__ == "__main__":
agent = ReflectionAgent(max_iterations=3)
task = "写一个函数 find_primes(n)，返回所有小于等于 n 的素数列表"
final_code, memory = agent.run(task)
print("\n" + "=" * 60)
print("最终代码：")
print(final_code)
print("\n完整迭代轨迹：")
print(memory.get_trajectory())1.4 实际效果
运行这段代码，可以看到清晰的进化过程。
版本 0（初始执行），大概率是试除法：
def find_primes(n):
"""找出所有小于等于 n 的素数，时间复杂度 O(n√n)"""
primes = []
for num in range(2, n + 1):
is_prime = True
for i in range(2, int(num ** 0.5) + 1):
if num % i == 0:
is_prime = False
break
if is_prime:
primes.append(num)
return primesCritic 的反馈会指出：O(n√n) 的试除法效率不足，应该使用埃拉托斯特尼筛法，时间复杂度可以优化到 O(n log log n)，空间换时间，用布尔数组标记合数。
版本 1（第一轮优化），Refine 后得到筛法：
def find_primes(n):
"""
埃拉托斯特尼筛法找素数。
时间复杂度：O(n log log n)，相比试除法的 O(n√n) 显著提升。
空间复杂度：O(n)
"""
if n < 2:
return []
# 初始化布尔数组，True 表示是素数
is_prime = [True] * (n + 1)
is_prime[0] = is_prime[1] = False # 0 和 1 不是素数
# 从 2 开始，把每个素数的倍数标记为合数
for p in range(2, int(n ** 0.5) + 1):
if is_prime[p]:
for multiple in range(p * p, n + 1, p):
is_prime[multiple] = False
return [num for num in range(2, n + 1) if is_prime[num]]第二轮 Critic 审查后，如果边界处理和代码质量都没问题，会输出 NO_IMPROVEMENT_NEEDED，循环提前终止。
从 O(n√n) 到 O(n log log n)，在 n=100000 的场景下，实际运行时间差了 10 倍以上。
1.5 成本收益分析
Reflection 不是免费的。
成本：每多一轮迭代，就多 2 次 LLM 调用（Critic + Refine）。如果迭代 3 轮，总共需要 7 次调用（1 次初始执行 + 3 次 Critic + 3 次 Refine）。延迟从几秒变成几十秒，token 消耗乘以 7 倍。
收益：输出质量显著提升。尤其是代码生成类任务，初稿往往功能正确但不够优化，Critic 可以发现人工 review 才能发现的问题。
适合使用 Reflection 的场景：
- 代码生成：特别是算法实现，Critic 可以专门针对时间复杂度、空间复杂度、边界情况
- 技术文档：让 Critic 检查逻辑完整性、技术准确性、表达清晰度
- 分析报告：让 Critic 检查论据是否充分、结论是否有数据支撑
- 任务不紧急、对质量要求高的场景
不适合使用 Reflection 的场景：
- 实时对话：用户在等待，加入 Reflection 让响应时间从 2 秒变成 20 秒，体验直接崩了
- 简单查询："北京今天天气怎么样"不需要 Reflection
- 任务本身有确定性答案：比如查数据库，第一次查到了就是查到了，Reflection 没有意义
还有一个需要特别注意的问题：Critic 本身也可能判断错误。如果 Critic 提示词写得不好，它可能给出错误的优化建议，导致 Refine 之后代码变得更差。解决方案是在 Critic 提示词里明确审查维度（像上面那个 CRITIC_PROMPT 一样），给它一个具体的检查清单，而不是让它泛泛地"评价代码质量"。
1.6 三种范式的定位总结
三种范式各有其适用定位：
- ReAct：适合信息检索类任务，能灵活应对意外，但缺乏全局视野。
- Plan-and-Solve：适合结构清晰的多步骤任务，方向稳定，但计划赶不上变化。
- Reflection：适合高质量输出任务，出手的东西质量有保障，但也最慢。
三者不是替代关系，而是互补。对于代码生成类任务，一个常见的组合是：Plan-and-Solve 负责整体架构设计，ReAct 负责信息收集，Reflection 负责最终的代码质量把关。把 Reflection 用在刀刃上——只在对质量要求高、时间不敏感的环节开启它。