第 53 章

Reflexion：从失败中自我改进

第53章：Reflexion：从失败中自我改进

导语

ReAct 和 Plan-and-Execute 都假设 Agent 能在第一次或有限次尝试中完成任务。但现实中，某些任务——尤其是代码生成、数学推理、复杂写作——即使优秀的 LLM 也难以一次命中。Reflexion（反思）范式给出了一个优雅的答案：让 Agent 像人类专家一样，从失败中学习，通过自我反思不断改进，直到任务完成。本章深入 Reflexion 的论文原理，剖析其在 Hermes 中的实现，并提供完整的代码示例。

53.1 Reflexion 论文核心：语言强化学习

什么是 Verbal Reinforcement Learning

2023年，Noah Shinn 等人发表了 Reflexion 论文，提出了一种不需要更新模型权重的"学习"机制：通过语言反馈（verbal feedback）替代传统强化学习中的数值奖励信号。

传统强化学习的问题：

需要大量样本才能收敛
奖励函数设计困难
无法直接解释为什么失败

Reflexion 的创新：

将奖励信号转化为语言反思（reflection）
反思内容存储在**外部记忆（episodic memory）**中
下次尝试时，反思内容作为上下文注入，指导改进

flowchart TD
    subgraph Trial["单次尝试"]
        T1[任务] --> T2[Agent 执行]
        T2 --> T3{任务成功?}
    end
    
    subgraph Reflect["失败反思"]
        T3 -->|失败| R1[Evaluator\n评估失败原因]
        R1 --> R2[Self-Reflection\n生成反思文本]
        R2 --> R3[存入 Memory]
    end
    
    subgraph Retry["再次尝试"]
        R3 --> N1[新一轮 Agent]
        N1 --> N2[加载历史反思]
        N2 --> N3[执行任务\n避免之前错误]
        N3 --> T3
    end
    
    T3 -->|成功| DONE[完成]

与传统方法的对比

方法	学习机制	需要标注数据	能否解释原因	计算成本
Fine-tuning	梯度下降更新权重	是（大量）	否	极高
RLHF	人类反馈+强化学习	是（人工标注）	否	高
Reflexion	语言反思（零权重更新）	否	是	低（仅 LLM 推理）
RAG	检索外部知识	是（知识库）	部分	中

53.2 失败 → 反思 → 再试三循环

Reflexion 的核心执行循环由三个阶段构成：

阶段1：尝试执行（Trial）

Agent 使用当前策略（加上历史反思）执行任务。执行可以是 ReAct 模式或其他 Agent 模式。

阶段2：评估与反思（Reflect）

执行完成后，Evaluator 判断结果是否满足要求。如果不满足，触发 Self-Reflection：

分析失败原因（逻辑错误？信息不足？误解需求？）
总结本次尝试的教训
提出下次改进的具体建议

阶段3：记忆更新（Memory Update）

将反思内容存入外部记忆，下次尝试时注入到上下文中。

# reflexion_memory.py
from dataclasses import dataclass, field
from typing import Optional
import time

@dataclass
class TrialRecord:
    """单次尝试的完整记录"""
    trial_number: int
    task: str
    attempt: str          # Agent 的尝试内容
    result: str           # 实际输出
    evaluation: str       # 评估结果
    reflection: str       # 反思内容
    success: bool
    timestamp: float = field(default_factory=time.time)
    
    def to_context_string(self) -> str:
        """将记录格式化为上下文字符串，供下次尝试使用"""
        return f"""
[第 {self.trial_number} 次尝试 - {'成功' if self.success else '失败'}]
输出：{self.result[:300]}
评估：{self.evaluation}
反思：{self.reflection}
"""


class ReflexionMemory:
    """Reflexion 的外部记忆系统"""
    
    def __init__(self, max_trials: int = 5):
        self.trials: list[TrialRecord] = []
        self.max_trials = max_trials
    
    def add_trial(self, record: TrialRecord) -> None:
        self.trials.append(record)
    
    def get_reflections_context(self) -> str:
        """生成供下次尝试使用的反思上下文"""
        if not self.trials:
            return ""
        
        failed_trials = [t for t in self.trials if not t.success]
        if not failed_trials:
            return ""
        
        header = f"## 历史失败记录（共 {len(failed_trials)} 次）\n"
        header += "请仔细阅读以下反思，避免重复相同的错误：\n\n"
        
        records = "\n---\n".join([t.to_context_string() for t in failed_trials])
        
        footer = "\n## 改进指导\n基于以上反思，请在本次尝试中特别注意避免已识别的错误模式。"
        
        return header + records + footer
    
    def can_retry(self) -> bool:
        return len(self.trials) < self.max_trials
    
    def last_trial_succeeded(self) -> bool:
        return bool(self.trials) and self.trials[-1].success

53.3 Hermes 中的 Reflexion 实现

# reflexion_agent.py
"""
Hermes Reflexion Agent 完整实现
支持自我批评提示模板、评估器、记忆系统
"""

import asyncio
from typing import Optional, Callable
from openai import AsyncOpenAI
from reflexion_memory import ReflexionMemory, TrialRecord


# ─── 提示词模板 ───────────────────────────────────────────────

ACTOR_SYSTEM = """你是一个认真完成任务的 AI Agent。

{reflection_context}

**当前任务**：请根据以上历史失败记录（如果有）中的教训，改进你的方法并完成任务。
避免重复已知的错误，关注之前的改进建议。"""

EVALUATOR_SYSTEM = """你是一个严格的任务评估专家。
请评估 Agent 的输出是否完整、正确地完成了给定任务。

评估标准：
1. 任务完整性：是否完成了任务要求的所有内容？
2. 准确性：输出中是否存在明显错误或遗漏？
3. 质量：输出是否达到专业水准？

输出格式：
- 首先给出 PASS 或 FAIL 的判断
- 然后说明评估理由（1-3句话）
- 如果 FAIL，指出最关键的问题"""

REFLECTION_SYSTEM = """你是一个自我改进专家。
给定一个任务、Agent 的失败尝试和评估反馈，生成深刻的反思。

反思要求：
1. 分析根本原因（不是表面症状）
2. 识别具体的错误模式
3. 提出 2-3 个可操作的改进建议
4. 预测改进后应该达到的效果

反思要简洁（3-5句话），聚焦最重要的问题。"""


# ─── 评估器 ──────────────────────────────────────────────────

class Evaluator:
    """
    任务评估器：判断 Agent 的输出是否满足任务要求
    
    支持两种评估模式：
    1. LLM 评估：使用 LLM 判断输出质量
    2. 规则评估：使用自定义函数验证结果
    """
    
    def __init__(
        self,
        client: AsyncOpenAI,
        model: str,
        custom_evaluator: Optional[Callable] = None
    ):
        self.client = client
        self.model = model
        self.custom_evaluator = custom_evaluator
    
    async def evaluate(self, task: str, output: str) -> tuple[bool, str]:
        """
        评估输出质量
        返回：(是否通过, 评估说明)
        """
        # 优先使用自定义评估器（如单元测试）
        if self.custom_evaluator:
            try:
                passed = self.custom_evaluator(output)
                return passed, "自定义评估器判断通过" if passed else "自定义评估器判断失败"
            except Exception as e:
                return False, f"评估器执行异常: {str(e)}"
        
        # 使用 LLM 评估
        response = await self.client.chat.completions.create(
            model=self.model,
            messages=[
                {"role": "system", "content": EVALUATOR_SYSTEM},
                {"role": "user", "content": f"任务：{task}\n\nAgent 输出：\n{output}"}
            ],
            temperature=0.1,
            max_tokens=512
        )
        
        evaluation = response.choices[0].message.content
        passed = evaluation.upper().startswith("PASS") or "PASS" in evaluation[:50]
        
        return passed, evaluation


# ─── 自我反思器 ───────────────────────────────────────────────

class SelfReflector:
    """生成自我反思内容"""
    
    def __init__(self, client: AsyncOpenAI, model: str):
        self.client = client
        self.model = model
    
    async def reflect(
        self,
        task: str,
        attempt: str,
        result: str,
        evaluation: str
    ) -> str:
        """基于任务、尝试内容和评估生成反思"""
        
        response = await self.client.chat.completions.create(
            model=self.model,
            messages=[
                {"role": "system", "content": REFLECTION_SYSTEM},
                {
                    "role": "user",
                    "content": f"""任务：{task}

我的尝试过程摘要：
{attempt[:500]}

实际输出：
{result[:500]}

评估反馈：
{evaluation}

请生成深刻的自我反思，帮助我在下次尝试中避免相同的错误。"""
                }
            ],
            temperature=0.3,
            max_tokens=512
        )
        
        return response.choices[0].message.content


# ─── Reflexion Agent 核心 ─────────────────────────────────────

class ReflexionAgent:
    """
    Reflexion Agent 核心实现
    
    通过失败-反思-重试循环持续改进，直到任务成功或达到最大尝试次数
    """
    
    def __init__(
        self,
        model: str = "NousResearch/Hermes-3-Llama-3.1-8B",
        base_url: str = "http://localhost:8000/v1",
        api_key: str = "not-needed",
        max_trials: int = 5,
        custom_evaluator: Optional[Callable] = None
    ):
        self.client = AsyncOpenAI(base_url=base_url, api_key=api_key)
        self.model = model
        self.max_trials = max_trials
        self.memory = ReflexionMemory(max_trials=max_trials)
        self.evaluator = Evaluator(self.client, model, custom_evaluator)
        self.reflector = SelfReflector(self.client, model)
    
    async def _attempt_task(self, task: str, reflection_context: str) -> tuple[str, str]:
        """
        执行一次任务尝试
        返回：(尝试过程摘要, 最终输出)
        """
        system = ACTOR_SYSTEM.format(reflection_context=reflection_context)
        
        messages = [
            {"role": "system", "content": system},
            {"role": "user", "content": task}
        ]
        
        # 内部 ReAct 循环（简化版，实际可集成完整 ReAct）
        response = await self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            temperature=0.3,
            max_tokens=2048
        )
        
        output = response.choices[0].message.content or ""
        attempt_summary = f"直接生成输出（{len(output)} 字符）"
        
        return attempt_summary, output
    
    async def run(self, task: str) -> dict:
        """
        执行 Reflexion 循环
        
        Returns:
            dict: 包含最终结果、尝试次数、所有反思记录
        """
        print(f"\n{'='*60}")
        print(f"Reflexion Agent")
        print(f"任务：{task[:80]}...")
        print(f"最大尝试次数：{self.max_trials}")
        print(f"{'='*60}")
        
        while self.memory.can_retry():
            trial_num = len(self.memory.trials) + 1
            print(f"\n--- 第 {trial_num} 次尝试 ---")
            
            # 加载历史反思
            reflection_context = self.memory.get_reflections_context()
            if reflection_context:
                print(f"[Memory] 加载 {len([t for t in self.memory.trials if not t.success])} 条失败反思")
            
            # 执行任务
            attempt_summary, output = await self._attempt_task(task, reflection_context)
            print(f"[Actor] 完成输出（{len(output)} 字符）")
            
            # 评估输出
            passed, evaluation = await self.evaluator.evaluate(task, output)
            print(f"[Evaluator] {'PASS' if passed else 'FAIL'}: {evaluation[:100]}")
            
            if passed:
                # 任务成功
                record = TrialRecord(
                    trial_number=trial_num,
                    task=task,
                    attempt=attempt_summary,
                    result=output,
                    evaluation=evaluation,
                    reflection="",
                    success=True
                )
                self.memory.add_trial(record)
                
                print(f"\n[Reflexion] 第 {trial_num} 次尝试成功！")
                return {
                    "success": True,
                    "answer": output,
                    "trials": trial_num,
                    "reflections": [t.reflection for t in self.memory.trials if t.reflection]
                }
            
            else:
                # 任务失败，生成反思
                reflection = await self.reflector.reflect(
                    task=task,
                    attempt=attempt_summary,
                    result=output,
                    evaluation=evaluation
                )
                print(f"[Reflector] 反思：{reflection[:150]}...")
                
                record = TrialRecord(
                    trial_number=trial_num,
                    task=task,
                    attempt=attempt_summary,
                    result=output,
                    evaluation=evaluation,
                    reflection=reflection,
                    success=False
                )
                self.memory.add_trial(record)
        
        # 达到最大尝试次数
        best_trial = max(self.memory.trials, key=lambda t: len(t.result))
        
        print(f"\n[Reflexion] 达到最大尝试次数 ({self.max_trials})，返回最佳结果")
        return {
            "success": False,
            "answer": best_trial.result,
            "trials": self.max_trials,
            "reflections": [t.reflection for t in self.memory.trials if t.reflection],
            "error": f"任务未在 {self.max_trials} 次内完成"
        }


# ─── 代码生成场景示例 ─────────────────────────────────────────

def make_code_evaluator(test_cases: list[tuple]) -> Callable:
    """
    创建代码生成评估器
    通过运行测试用例验证代码正确性
    """
    def evaluate(code_output: str) -> bool:
        # 提取代码块
        import re
        code_match = re.search(r'```python\n(.*?)```', code_output, re.DOTALL)
        if not code_match:
            code_match = re.search(r'```\n(.*?)```', code_output, re.DOTALL)
        
        if not code_match:
            # 尝试直接执行输出
            code = code_output
        else:
            code = code_match.group(1)
        
        # 在沙箱中执行代码并测试
        namespace = {}
        try:
            exec(code, namespace)
        except SyntaxError as e:
            print(f"  语法错误: {e}")
            return False
        except Exception as e:
            print(f"  运行时错误: {e}")
            return False
        
        # 运行测试用例
        for input_val, expected in test_cases:
            try:
                # 假设代码定义了一个名为 'solution' 的函数
                func = namespace.get('solution') or namespace.get('solve')
                if func is None:
                    print("  未找到 solution/solve 函数")
                    return False
                
                result = func(input_val)
                if result != expected:
                    print(f"  测试失败: solution({input_val}) = {result}, 期望 {expected}")
                    return False
                else:
                    print(f"  测试通过: solution({input_val}) = {result}")
            except Exception as e:
                print(f"  测试执行异常: {e}")
                return False
        
        return True
    
    return evaluate


# ─── 使用示例 ──────────────────────────────────────────────────

async def main():
    # 场景1：代码生成（有明确的正确/错误评估）
    test_cases = [
        ([1, 2, 3, 4, 5], 15),
        ([10, -3, 7], 14),
        ([], 0),
    ]
    
    agent = ReflexionAgent(
        model="NousResearch/Hermes-3-Llama-3.1-8B",
        base_url="http://localhost:8000/v1",
        max_trials=4,
        custom_evaluator=make_code_evaluator(test_cases)
    )
    
    result = await agent.run(
        "写一个 Python 函数 `solution(nums: list) -> int`，"
        "计算列表中所有正整数的和（忽略负数和零）。"
    )
    
    print(f"\n{'='*60}")
    print(f"结果：{'成功' if result['success'] else '未成功'}")
    print(f"尝试次数：{result['trials']}")
    print(f"反思记录数：{len(result['reflections'])}")
    print(f"\n最终代码：\n{result['answer']}")
    
    # 场景2：数学推理（LLM 评估）
    agent2 = ReflexionAgent(max_trials=3)
    result2 = await agent2.run(
        "证明：对于任意正整数 n，n³ - n 总是能被 6 整除。"
        "要求给出严格的数学证明。"
    )
    print(f"\n数学证明结果：{result2['answer'][:200]}...")


if __name__ == "__main__":
    asyncio.run(main())

53.4 适用场景分析

最适合 Reflexion 的任务类型

场景	为什么适合	评估器类型
代码生成	有明确的单元测试，成功/失败判断清晰	单元测试（自动）
数学推理	答案有对错之分，可程序化验证	数学验证器
结构化数据提取	输出格式有严格要求	Schema 验证
逻辑谜题	答案唯一正确	规则检验器
SQL 生成	可通过执行结果验证	数据库验证

不适合的场景

创意写作（缺乏客观评估标准）
开放性问答（答案多样，难以定义"成功"）
单步简单任务（反思开销大于收益）

Reflexion 的局限性

Token 成本高：每次失败都需要额外的反思调用，多次尝试成本显著
评估器质量关键：LLM 评估器本身可能判断错误（误判通过或误判失败）
记忆容量限制：历史反思占用上下文空间，多次失败后反思可能被截断
不能突破模型能力边界：如果任务本身超出模型能力，反思也无法弥补

flowchart LR
    subgraph 适合["Reflexion 最有效"]
        A1[任务有明确正确答案] --> A2[评估器可自动化] --> A3[错误有规律可循]
    end
    
    subgraph 不适合["Reflexion 效果有限"]
        B1[创意任务] --> B2[主观评估] --> B3[模型能力上限问题]
    end

53.5 与强化学习循环的关系

Reflexion 可以被理解为一种"情节式"（episodic）强化学习：

状态（State）：当前任务 + 历史反思记忆
行动（Action）：Agent 的输出
奖励（Reward）：评估器的 PASS/FAIL 信号
策略更新：通过添加语言反思到记忆（而非更新权重）

与标准 RL 的关键区别：策略改进发生在推理时（inference time），而非训练时。这使得 Reflexion 无需任何训练基础设施，可以立即部署。

小结

本章深入讲解了 Reflexion 的理论与实践：

Verbal RL：用语言反思替代数值奖励，无需更新权重，成本极低但效果显著。
三循环机制：试验 → 评估 → 反思 → 再试，形成自我改进的闭环。
代码实现：完整的 Actor-Evaluator-Reflector 三组件架构，支持自定义评估器和单元测试验证。
适用场景：代码生成和数学推理是最佳场景，因为有明确的成功/失败标准。
局限性：Token 成本、评估器质量、模型能力上限是三个主要约束。

思考题

如果评估器本身不可靠（误判率 20%），Reflexion 的效果会如何？如何设计更鲁棒的评估机制？
Reflexion 的反思存储在上下文中，随着失败次数增加，反思会越来越多。如何设计反思压缩策略？
能否将多个任务的 Reflexion 记忆跨任务共享？比如"在任务 A 中学到的经验，能否帮助解决任务 B"？
Reflexion 和 Fine-tuning 各有什么优势？在什么情况下，应该从 Reflexion 切换到 Fine-tuning？

本章评分

4.9 / 5 (3 评分)