Reasoning Model 自纠错机制：从 o1 到 DeepSeek-R2 的技术演进

<think>
Let me solve 17 × 23 step by step.
17 × 23 = 17 × 20 + 17 × 3
17 × 20 = 340
17 × 3 = 51
Wait, let me double-check: 17 × 3 = 17 + 17 + 17 = 51. Correct.
340 + 51 = 391

Actually, let me verify by a different method:
23 × 17 = 23 × 10 + 23 × 7 = 230 + 161 = 391 ✓
</think>

17 × 23 = 391

def self_refine(prompt, model, max_iterations=3):
    """Self-Refine 核心循环"""
    output = model.generate(prompt)

    for i in range(max_iterations):
        # Critic 阶段
        critique = model.generate(
            f"Review the following output and identify specific errors "
            f"or areas for improvement:\n\n{output}\n\n"
            f"Original task: {prompt}"
        )

        # 判断是否需要继续改进
        if "no issues found" in critique.lower():
            break

        # Refiner 阶段
        output = model.generate(
            f"Original task: {prompt}\n\n"
            f"Previous output: {output}\n\n"
            f"Feedback: {critique}\n\n"
            f"Please provide an improved version addressing the feedback."
        )

    return output

def reflexion_loop(task, model, environment, max_attempts=5):
    """Reflexion 核心循环"""
    memory = []  # 存储历史反思

    for attempt in range(max_attempts):
        # 基于历史反思生成新方案
        context = "\n".join(
            f"Attempt {m['attempt']}: {m['reflection']}"
            for m in memory
        )

        solution = model.generate(
            f"Task: {task}\n\n"
            f"Previous reflections:\n{context}\n\n"
            f"Generate a solution avoiding previous mistakes."
        )

        # 在外部环境中执行并获取反馈
        result = environment.execute(solution)

        if result.success:
            return solution

        # 生成反思并存入记忆
        reflection = model.generate(
            f"Task: {task}\n"
            f"Solution: {solution}\n"
            f"Error: {result.error}\n\n"
            f"Reflect on why this failed and how to fix it."
        )

        memory.append({
            "attempt": attempt + 1,
            "reflection": reflection
        })

    return None  # 达到最大尝试次数

def beam_search_correction(problem, model, verifier, beam_width=5):
    """Beam Search 多路径搜索"""
    # 并行生成多条推理路径
    candidates = [
        model.generate(problem, temperature=0.7)
        for _ in range(beam_width)
    ]

    # 验证器对每条路径评分
    scored = [
        (candidate, verifier.score(problem, candidate))
        for candidate in candidates
    ]

    # 返回最高分的结果
    scored.sort(key=lambda x: x[1], reverse=True)
    return scored[0][0]

def sequential_revision(problem, model, verifier, max_rounds=5):
    """Sequential Revision 逐步迭代"""
    solution = model.generate(problem)

    for round_num in range(max_rounds):
        score = verifier.score(problem, solution)

        # 达到置信度阈值则停止
        if score > 0.95:
            break

        # 识别具体错误位置
        error_analysis = verifier.locate_errors(problem, solution)

        # 针对性修正
        solution = model.generate(
            f"Problem: {problem}\n\n"
            f"Current solution: {solution}\n\n"
            f"Identified issues: {error_analysis}\n\n"
            f"Please fix only the identified issues."
        )

    return solution

import asyncio
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional


class VerificationLevel(Enum):
    SYNTAX = "syntax"
    LOGIC = "logic"
    SEMANTIC = "semantic"


@dataclass
class VerificationResult:
    passed: bool
    level: VerificationLevel
    errors: list = field(default_factory=list)
    confidence: float = 0.0


class VerificationLoop:
    """生产级分层验证循环"""

    def __init__(self, model, config):
        self.model = model
        self.max_iterations = config.get("max_iterations", 4)
        self.confidence_threshold = config.get("confidence_threshold", 0.92)
        self.verifiers = self._init_verifiers(config)

    def _init_verifiers(self, config):
        """初始化多层验证器"""
        return {
            VerificationLevel.SYNTAX: SyntaxVerifier(config),
            VerificationLevel.LOGIC: LogicVerifier(self.model, config),
            VerificationLevel.SEMANTIC: SemanticVerifier(self.model, config),
        }

    async def run(self, task: str, context: Optional[str] = None) -> dict:
        """执行完整验证循环"""
        # Phase 1: 生成初始输出
        output = await self.model.generate(task, context=context)
        history = []

        for iteration in range(self.max_iterations):
            # Phase 2: 分层验证（从快到慢）
            verification = await self._layered_verification(
                task, output
            )

            history.append({
                "iteration": iteration,
                "output_hash": hash(output),
                "verification": verification,
            })

            # Phase 3: 判断是否达标
            if verification.passed and \
               verification.confidence >= self.confidence_threshold:
                return {
                    "output": output,
                    "iterations": iteration + 1,
                    "confidence": verification.confidence,
                    "history": history,
                }

            # Phase 4: 定向修正
            output = await self._targeted_revision(
                task, output, verification
            )

        # 达到最大轮次，返回当前最优
        return {
            "output": output,
            "iterations": self.max_iterations,
            "confidence": verification.confidence,
            "history": history,
            "max_iterations_reached": True,
        }

    async def _layered_verification(
        self, task: str, output: str
    ) -> VerificationResult:
        """分层验证：语法 → 逻辑 → 语义"""
        for level in VerificationLevel:
            verifier = self.verifiers[level]
            result = await verifier.verify(task, output)

            if not result.passed:
                return result  # 快速失败，无需继续更高层验证

        # 全部通过，返回语义层的置信度
        return result

    async def _targeted_revision(
        self, task: str, output: str, verification: VerificationResult
    ) -> str:
        """基于验证结果进行定向修正"""
        error_context = "\n".join(
            f"- [{verification.level.value}] {err}"
            for err in verification.errors
        )

        revised = await self.model.generate(
            f"Task: {task}\n\n"
            f"Current output:\n{output}\n\n"
            f"Verification errors found:\n{error_context}\n\n"
            f"Fix ONLY the identified errors. Do not change other parts."
        )

        return revised

class AgentVerificationLoop(VerificationLoop):
    """Agent 场景的增强验证循环"""

    def __init__(self, model, tools, config):
        super().__init__(model, config)
        self.tools = tools  # 可用工具集合

    async def _layered_verification(self, task, output):
        """增强验证：结合工具执行"""
        # 基础验证
        base_result = await super()._layered_verification(
            task, output
        )
        if not base_result.passed:
            return base_result

        # 工具增强验证（如代码执行）
        if self._needs_execution_check(output):
            exec_result = await self.tools["code_executor"].run(
                output
            )
            if not exec_result.success:
                return VerificationResult(
                    passed=False,
                    level=VerificationLevel.SEMANTIC,
                    errors=[f"Execution failed: {exec_result.error}"],
                    confidence=0.3,
                )

        return base_result

# 使用 diff 对比纠错前后的变化量
# 如果变化过大，可能是"改崩了"
from difflib import SequenceMatcher

def is_over_correction(original, revised, threshold=0.5):
    """检测是否过度纠错"""
    similarity = SequenceMatcher(
        None, original, revised
    ).ratio()
    return similarity < threshold  # 变化超过 50% 视为过度纠错

# 必须设置多重退出条件
EXIT_CONDITIONS = {
    "max_iterations": 5,
    "confidence_threshold": 0.92,
    "no_improvement_rounds": 2,  # 连续 2 轮分数无提升则停止
    "max_latency_ms": 30000,     # 硬性时间上限
}