Reflexion：把失败教训存下来

旅游助手今天犯了一个错：用户说"轻松步行"，它理解成"加更多景点"。Maker-Checker 能修当前这一轮。但明天用户又提了同样的需求，它还会犯同样的错。

Reflexion 更进一步：失败后写一条教训，存进记忆，下次同类任务取出来用。

一句话

Reflexion 把"失败就重试"变成"失败 → 写教训 → 存记忆 → 下次带着教训重试"，让重复任务不从零开始。

它修什么失败

问题	表面看起来	实际风险
每次任务独立	简单	同一个错误反复犯
反馈停在一次运行里	当次能修	下次忘了
教训太笼统	听着反思了	无法检索也无法执行

它引入什么复杂度

谁	负责什么
模型	回答、失败后写教训、带教训重试
验证器	判断 pass/fail
记忆存储	存教训
Python	控制轮次、读写记忆、trace

教训应该短、具体、可执行："轻松步行"要求时，景点不超过 3 个，除非用户明确说要更多。

和其他模式的关系

flowchart LR
  MC["Maker-Checker"] -.->|"只管当前轮"| REF["Reflexion"]
  REF -->|"教训存到哪里"| MEM["Agent Memory"]
  REF -.->|"事实验证"| COVE["CoVe"]

Maker-Checker：修当前这一轮的草稿；Reflexion 跨任务积累经验。两者可以叠加：Checker 发现问题 → Reflexion 写教训。
Agent Memory：Reflexion 的教训是 episodic memory 的一种。Agent Memory 页面讲四种记忆的分类和存储。
CoVe：验证器可以用 CoVe 实现，逐条检查事实再判断 pass/fail。

完整实现：旅游助手学会"轻松步行"

"""reflexion_travel.py — 从失败中学习"""
from __future__ import annotations

import json
from dataclasses import dataclass
from typing import Any, Callable


# ── 极简运行时 ──────────────────────────────────────────

@dataclass(frozen=True)
class Message:
    role: str
    content: str

class MockLLM:
    def __init__(self, responses: list[str]):
        self._responses = list(responses)
        self._idx = 0
    def complete(self, messages: list[Message]) -> str:
        if self._idx >= len(self._responses):
            raise RuntimeError("MockLLM 回复用完了")
        out = self._responses[self._idx]
        self._idx += 1
        return out


class InMemoryKV:
    """简单的键值记忆存储。"""
    def __init__(self):
        self._data: dict[str, Any] = {}
    def get(self, key: str) -> Any:
        if key not in self._data:
            raise KeyError(key)
        return self._data[key]
    def set(self, key: str, value: Any) -> None:
        self._data[key] = value


@dataclass(frozen=True)
class VerificationResult:
    ok: bool
    feedback: str


# ── Reflexion 核心逻辑 ──────────────────────────────────

def reflexion(
    model: MockLLM,
    *,
    task: str,
    verify: Callable[[str], VerificationResult],
    memory: InMemoryKV,
    memory_key: str = "reflexion.lessons",
    max_rounds: int = 3,
) -> str:
    """
    1) 读已有教训
    2) 带教训回答
    3) 验证
    4) 没通过 → 写教训存回记忆 → 重试
    """
    # 读已有教训
    lessons: list[str] = []
    try:
        existing = memory.get(memory_key)
        if isinstance(existing, list):
            lessons = [str(x) for x in existing]
    except KeyError:
        pass

    if lessons:
        print(f"[记忆] 已有 {len(lessons)} 条教训：")
        for l in lessons:
            print(f"  - {l}")

    last_answer = ""
    for r in range(max_rounds):
        # 构建 prompt
        prompt = task
        if lessons:
            lessons_text = "\n".join(f"- {l}" for l in lessons)
            prompt = f"任务：\n{task}\n\n历史教训（务必遵守）：\n{lessons_text}"

        last_answer = model.complete([
            Message(role="system", content="完成任务。遵守所有教训。"),
            Message(role="user", content=prompt),
        ])
        print(f"[轮次 {r + 1}] 回答：{last_answer[:80]}...")

        # 验证
        v = verify(last_answer)
        print(f"[轮次 {r + 1}] 验证：ok={v.ok}, feedback={v.feedback}")

        if v.ok:
            return last_answer

        # 写教训
        raw_lesson = model.complete([
            Message(
                role="system",
                content="写一条简短、具体、可执行的教训，避免下次犯同样的错。只返回 JSON。",
            ),
            Message(
                role="user",
                content=f"任务：\n{task}\n\n回答：\n{last_answer}\n\n失败反馈：\n{v.feedback}",
            ),
        ])
        lesson = json.loads(raw_lesson)["lesson"]
        lessons.append(lesson)
        memory.set(memory_key, lessons)
        print(f"[轮次 {r + 1}] 新教训：{lesson}")

    print(f"[警告] {max_rounds} 轮后仍未通过")
    return last_answer


# ── 运行 ────────────────────────────────────────────────

def main() -> None:
    task = "为用户规划杭州一日游。用户要求：轻松步行，不赶路。"

    def verify(answer: str) -> VerificationResult:
        """简单规则验证：景点不超过 3 个算轻松。"""
        # 粗略计数：看有几个 "→" 分隔符
        stops = answer.count("→") + 1
        if stops > 3:
            return VerificationResult(
                ok=False,
                feedback=f"行程有 {stops} 个景点，轻松步行不应超过 3 个。"
            )
        if "步行" not in answer and "走" not in answer:
            return VerificationResult(
                ok=False,
                feedback="行程没有提到步行，和用户要求不匹配。"
            )
        return VerificationResult(ok=True, feedback="")

    kv = InMemoryKV()

    model = MockLLM([
        # 轮次 1：回答太多景点
        "西湖 → 灵隐寺 → 茶叶博物馆 → 龙井村 → 河坊街（全程步行）",
        # 轮次 1：写教训
        json.dumps({
            "lesson": "轻松步行要求时，景点不超过 3 个，除非用户明确说要更多。"
        }, ensure_ascii=False),
        # 轮次 2：遵守教训，减少景点
        "上午西湖断桥（步行 40 分钟）→ 中午茶叶博物馆（步行 30 分钟）→ 下午龙井村采茶（步行 1 小时）",
    ])

    result = reflexion(
        model,
        task=task,
        verify=verify,
        memory=kv,
        max_rounds=3,
    )
    print(f"\n最终行程：\n{result}")

    # 验证教训被存下来了
    print(f"\n记忆中的教训：{kv.get('reflexion.lessons')}")


if __name__ == "__main__":
    main()

运行日志

[轮次 1] 回答：西湖 → 灵隐寺 → 茶叶博物馆 → 龙井村 → 河坊街（全程步行）...
[轮次 1] 验证：ok=False, feedback=行程有 5 个景点，轻松步行不应超过 3 个。
[轮次 1] 新教训：轻松步行要求时，景点不超过 3 个，除非用户明确说要更多。
[轮次 2] 回答：上午西湖断桥（步行 40 分钟）→ 中午茶叶博物馆（步行 30 分钟）→ 下午龙井村采茶（步行 1 小时）...
[轮次 2] 验证：ok=True, feedback=

最终行程：
上午西湖断桥（步行 40 分钟）→ 中午茶叶博物馆（步行 30 分钟）→ 下午龙井村采茶（步行 1 小时）

记忆中的教训：['轻松步行要求时，景点不超过 3 个，除非用户明确说要更多。']

第一轮 5 个景点太多，验证不通过。模型写了一条教训，第二轮带着教训重试，减少到 3 个，通过。教训被持久化到 KV 存储，下次同类任务会自动取出。

调试：trace 里看什么

要看的	怎么看	说明什么
`round_index`	第几轮通过	太多轮说明验证器或教训不够精确
教训内容	trace 里的 lesson	"小心点"这种 = 没用，要求具体
教训是否被使用	第二轮 prompt 里有没有教训	教训写了但没送进 prompt = bug
验证反馈	`v.feedback`	反馈太模糊 → 教训也会模糊
记忆增长	`len(lessons)`	太多教训会污染 prompt

工程笔记

成本公式

每轮 = 1（回答）+ 1（验证）+ 1（写教训，仅失败时）
通过轮 = 1（回答）+ 1（验证）
最好 = 2（第 1 轮就通过）
最坏 = max_rounds × 3

常见坑

坑	现象	修法
教训太笼统	"下次注意"	要求写具体检查条件
记忆污染	旧教训误导新任务	加审核、过期、命名空间
检索不到教训	教训存了但没用上	加标签/任务键
过度依赖记忆	旧教训扭曲新任务	验证器始终在循环里
验证器不可靠	假通过或假失败	用规则+工具组合验证

什么时候用

类似任务会反复出现。
有可靠的验证信号。
教训可以写成短检查条件。
你能维护记忆质量。

什么时候别用

任务是一次性的。
没有可靠的验证信号。
教训可能误导未来任务。
高风险记忆没有审核流程。

读完以后

Reflexion 适合从可验证失败中学习的系统。

如果只需要修当前这一轮，读 Maker-Checker。如果想了解记忆的完整分类和存储，读 Agent Memory。