Reflexion 的失败模式与适用边界

# ❌ 错误:让同一个 LLM 既生成答案又判断对错,没有外部锚
class SelfJudgeEvaluator:
    def __init__(self, llm):
        self.llm = llm
    def evaluate(self, answer, task):
        verdict = self.llm.complete(
            f"问题:{task.spec}\n答案:{answer}\n这个答案对吗?只回答 对/错。"
        )
        passed = verdict.strip().startswith("对")
        return passed, ""        # 没有 ground-truth,verdict 和答案同源偏差

# ✅ 正确:Evaluator 锚定到外部可验证信号;无信号时拒绝启用反思循环
class GroundedEvaluator:
    def __init__(self, oracle=None, tests=None):
        if oracle is None and tests is None:
            raise ValueError(
                "Reflexion 需要可信的对错判据(ground-truth / 单元测试 / 环境 reward);"
                "缺失时应退化为单遍生成,而非自评反思。"
            )
        self.oracle, self.tests = oracle, tests
    def evaluate(self, answer, task):
        if self.tests is not None:          # 编程:跑测试(外部、确定)
            return run_tests(answer, self.tests)
        return answer == self.oracle[task.id], ""   # 推理:对标准答案

# ❌ 错误:无条件每轮反思,正确答案也被拖进反思 → 可能被改错
def loop(task, actor, reflector, memory, max_trials=3):
    code = ""
    for t in range(max_trials):
        code = actor.act(task, memory.reflections)
        sr = reflector.reflect(code, "请找出可改进之处", task)  # 不管对错都反思
        memory.add(sr)
    return code

# ✅ 正确:反思由可信 Evaluator 门控;判通过立即停,绝不反思正确答案
def loop(task, actor, evaluator, reflector, memory, max_trials=3):
    code = ""
    for t in range(max_trials):
        code = actor.act(task, memory.reflections)
        passed, error = evaluator.evaluate(code, task)   # 外部可信判据
        if passed:
            return code                                  # 停止条件:正确即停
        sr = reflector.reflect(code, error, task)        # 仅在确认失败时反思
        memory.add(sr)
    return code

# ❌ 错误:无限累积反思 + 把每轮完整代码也塞进 prompt
class UnboundedMemory:
    def __init__(self):
        self.reflections = []
        self.full_trajectories = []          # 还把整段代码也存下来
    def add(self, sr, code):
        self.reflections.append(sr)          # 永不截断 → 线性膨胀
        self.full_trajectories.append(code)  # trial 越多,prompt 越长

# ✅ 正确:Ω 滑动窗口(论文设计),只保留最近 capacity 条反思
from dataclasses import dataclass, field

@dataclass
class EpisodicMemory:
    capacity: int = 3                        # 论文 Ω=1-3;programming 用 1
    reflections: list[str] = field(default_factory=list)
    def add(self, sr: str) -> None:
        self.reflections.append(sr)
        self.reflections = self.reflections[-self.capacity:]  # 截断到窗口
    # 注意:长期记忆只存"反思文本",不存完整 trajectory;
    # 当前 trajectory 是短期记忆,每个 trial 重置,不进 memory。

# ❌ 错误:默认上 Reflexion,不和等预算的 self-consistency 比
answer = reflexion(task, max_trials=5)   # 5 轮串行,5×(生成+评估+反思)
# 没有对照:同样的预算如果改成多采样投票会怎样?

# ✅ 正确:在相同采样预算下,先和 self-consistency 对照再决策
from collections import Counter

def self_consistency(task, actor, n=5):
    samples = [actor.act(task, reflections=[]) for _ in range(n)]
    return Counter(samples).most_common(1)[0][0]   # 多数投票,可并行

# 选型:仅当 Reflexion 在等预算下显著优于 self-consistency,
# 且任务确实"多采样救不了"(初始正确率低),才付它的串行+延迟代价。
def choose(task, actor, evaluator, budget=5):
    baseline = self_consistency(task, actor, n=budget)
    if not needs_directed_correction(task):   # 多采样能解 → 用更便宜的
        return baseline
    return reflexion(task, max_trials=budget)

# ❌ 错误:盲目加大 max_trials,期待更多轮就能突破平台期
answer = reflexion(task, model="starchat-beta", max_trials=10)
# 弱模型产不出有效反思;探索型任务再多轮也拍平 → 纯烧钱

# ✅ 正确:能力门槛 + 平台期早停 + 探索型任务改用宽度搜索
def run_with_guards(task, model, max_trials=4):
    if model_capability(model) < CORRECTION_THRESHOLD:
        return single_pass(task, model)          # 弱模型:别上反思
    if task.needs_exploration:                   # 探索型:宽度而非时间
        return tree_of_thoughts(task, model)     # 换搜索范式
    prev = None
    for t in range(max_trials):                  # 反思型:监控平台期
        ans = reflexion_step(task, model)
        if ans == prev:                          # 连续两轮无变化 → 平台期
            break                                # 早停,不空烧 trial
        prev = ans
    return ans

# ❌ 错误:对所有任务一刀切上反思,不看初始正确率
def solve(task, actor, evaluator, reflector, memory):
    return reflexion(task, actor, evaluator, reflector, memory)
    # 高置信任务也反思 → 误伤正确答案(HotpotQA 80.3 → 76.2)

# ✅ 正确:用初始正确率 / 置信度门控是否启用反思
RA_THRESHOLD = 0.30   # 《When Hindsight》:初始正确率 < 20-30% 才净获益

def solve(task, actor, evaluator, reflector, memory):
    first = actor.act(task, reflections=[])
    passed, error = evaluator.evaluate(first, task)   # 需可信判据(低假阳性)
    if passed:
        return first                                  # 已对:不反思,避免误伤
    if estimated_initial_accuracy(task) >= RA_THRESHOLD:
        return first        # 高置信任务:即使这条错了,反思整体期望为负 → 不上
    return reflexion(task, actor, evaluator, reflector, memory)