第4章：推理模型专题 (Reasoning Models: O1 & DeepSeek-R1)#

Q: Roger有5个网球。他又买了2罐网球,每罐3个球。他现在有多少个网球?
A: 11个

Q: Roger有5个网球。他又买了2罐网球,每罐3个球。他现在有多少个网球?
A: 让我一步步思考:
1. Roger最初有5个球
2. 他买了2罐,每罐3个,所以是 2×3=6个球
3. 总共: 5+6=11个球
答案: 11个

class ToTSolver:
    """Tree-of-Thoughts 求解器"""
    def __init__(self, model):
        self.model = model

    def generate_thoughts(self, state, k=3):
        """生成 k 个后续步骤"""
        prompt = f"当前状态：{state}\n请给出{k}个可能的下一步："
        return self.model.generate(prompt, n=k)

    def evaluate_state(self, state):
        """评估状态价值 (0-1)"""
        prompt = f"评估以下状态解决问题的可能性(0-10分)：{state}"
        score = self.model.generate(prompt)
        return float(score) / 10

    def bfs_solve(self, initial_state):
        frontier = [(initial_state, 0)] # (状态, 分数)

        while frontier:
            # 1. 扩展 (Expansion)
            current_state, _ = frontier.pop(0)
            thoughts = self.generate_thoughts(current_state)

            # 2. 评估 (Evaluation)
            candidates = []
            for thought in thoughts:
                score = self.evaluate_state(thought)
                if score > 0.8: # 剪枝
                    candidates.append((thought, score))

            frontier.extend(candidates)

            # 3. 检查结束条件
            for cand, score in candidates:
                if self.is_solved(cand):
                    return cand
        return "无解"

import torch
import torch.nn.functional as F

def grpo_loss(policy_model, questions, group_size=4, temperature=1.0):
    """
    GRPO 损失函数实现

    Args:
        policy_model: 策略模型 (LLM)
        questions: 问题列表 [batch_size]
        group_size: 每个问题采样的输出数量
        temperature: 采样温度
    """
    batch_size = len(questions)
    all_outputs = []
    all_log_probs = []
    all_rewards = []

    # 1. 对每个问题采样 G 个输出
    for q in questions:
        outputs = policy_model.generate(
            q,
            n=group_size,
            temperature=temperature,
            return_log_probs=True  # 需要返回每个Token的log概率
        )

        # 2. 计算每个输出的奖励 (通过验证器)
        rewards = [verify_answer(q, o.text) for o in outputs]

        all_outputs.extend([o.text for o in outputs])
        all_log_probs.extend([o.log_prob for o in outputs])
        all_rewards.extend(rewards)

    # 3. 转为 Tensor
    log_probs = torch.tensor(all_log_probs, dtype=torch.float32)
    rewards = torch.tensor(all_rewards, dtype=torch.float32)

    # 4. 按组计算归一化优势 (Group Normalization)
    advantages = []
    for i in range(batch_size):
        start_idx = i * group_size
        end_idx = start_idx + group_size

        group_rewards = rewards[start_idx:end_idx]

        # 组内归一化
        mean_r = group_rewards.mean()
        std_r = group_rewards.std()

        # 避免除零
        if std_r < 1e-8:
            group_advantages = torch.zeros_like(group_rewards)
        else:
            group_advantages = (group_rewards - mean_r) / (std_r + 1e-8)

        advantages.append(group_advantages)

    advantages = torch.cat(advantages)

    # 5. Policy Gradient 损失
    # 只强化那些比组平均好的输出 (Advantage > 0)
    loss = -(log_probs * advantages).mean()

    return loss

def verify_answer(question, output):
    """
    验证器：检查答案是否正确
    返回: 1.0 (正确) 或 0.0 (错误)
    """
    # 对于数学题，可以执行代码验证
    # 对于代码题，可以运行测试用例
    # 这里简化为规则匹配
    try:
        predicted_ans = extract_answer(output)
        ground_truth = get_ground_truth(question)
        return 1.0 if predicted_ans == ground_truth else 0.0
    except:
        return 0.0

# 准备少量高质量SFT数据 (1K-10K条)
# 关键：必须包含完整的推理链，而不是简单的问答对
sft_data = [
    {
        "question": "计算 15 × 23",
        "reasoning": """让我仔细计算:

**思路**: 使用分配律简化计算

**步骤 1**: 将 23 分解为 20 + 3
  15 × 23 = 15 × (20 + 3)

**步骤 2**: 分别计算
  - 15 × 20 = 300
  - 15 × 3 = 45

**步骤 3**: 相加
  300 + 45 = 345

**验证**: 用竖式验证...
  (此处展示竖式过程)

**结论**: 答案是 345""",
        "answer": "345"
    },
    {
        "question": "证明: 根号2 是无理数",
        "reasoning": """这是一个经典的反证法证明。

**假设**: 假设 √2 是有理数

**推导 1**: 则存在互质的整数 p, q，使得 √2 = p/q

**推导 2**: 两边平方...
  (此处展开详细推理过程，包含自我质疑)

等等，我需要检查一下这里的逻辑是否严谨...
  (展示自我修正过程)

**结论**: 矛盾！因此原假设错误，√2 是无理数。""",
        "answer": "√2 是无理数（已证明）"
    }
]

# 使用标准SFT训练
# 注意：这一步不追求性能，只追求格式规范
model = train_sft(base_model, sft_data, epochs=2-3, lr=1e-5)

# 准备数学题数据集 (MATH, GSM8K, etc.)
math_problems = load_dataset("math_problems")

# GRPO训练循环
for epoch in range(100):
    for batch in math_problems:
        # 1. 采样多个输出
        outputs = model.generate(batch, n=8, temperature=0.8)

        # 2. 用验证器评分
        rewards = [verify_answer(q, o) for q, o in zip(batch, outputs)]

        # 3. 计算GRPO损失并更新
        loss = grpo_loss(model, outputs, rewards)
        loss.backward()
        optimizer.step()

def inference_with_rejection_sampling(question, n=64):
    # 生成 N 个候选答案
    candidates = model.generate(question, n=n, temperature=0.7)

    # 用验证器筛选
    verified = [c for c in candidates if verify(question, c)]

    # 返回第一个通过验证的答案
    return verified[0] if verified else candidates[0]

import torch
from transformers import AutoModelForCausalLM, AutoTokenizer

class MathReasoningSystem:
    def __init__(self, model_name="deepseek-ai/deepseek-math-7b"):
        self.model = AutoModelForCausalLM.from_pretrained(model_name)
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)

    def solve_with_cot(self, problem):
        """使用 Chain-of-Thought 解题"""
        prompt = f"""请一步步解决以下数学问题:

问题: {problem}

解答: 让我仔细思考:"""

        inputs = self.tokenizer(prompt, return_tensors="pt")
        outputs = self.model.generate(
            **inputs,
            max_length=512,
            temperature=0.7,
            do_sample=True
        )

        return self.tokenizer.decode(outputs[0])

    def solve_with_self_consistency(self, problem, n=5):
        """使用 Self-Consistency 提高准确率"""
        answers = []

        # 采样 N 次
        for _ in range(n):
            solution = self.solve_with_cot(problem)
            answer = self.extract_answer(solution)
            answers.append(answer)

        # 多数投票
        from collections import Counter
        return Counter(answers).most_common(1)[0][0]

    def extract_answer(self, solution):
        """从解答中提取最终答案"""
        # 简化实现: 提取最后一个数字
        import re
        numbers = re.findall(r'\d+', solution)
        return numbers[-1] if numbers else None

# 使用示例
system = MathReasoningSystem()
problem = "如果一个班级有 25 名学生,老师要把他们分成 5 组,每组多少人?"

# 方法1: 单次CoT
answer1 = system.solve_with_cot(problem)
print("CoT答案:", answer1)

# 方法2: Self-Consistency
answer2 = system.solve_with_self_consistency(problem, n=5)
print("Self-Consistency答案:", answer2)

# 简单任务 (闲聊、翻译)
if task_complexity == "low":
    use_direct_generation()

# 中等任务 (简单数学、代码)
elif task_complexity == "medium":
    use_cot_prompting()
    # 如果准确率不够, 加上 self_consistency()

# 复杂任务 (奥赛数学、复杂规划)
else:
    if has_verifier:
        use_best_of_n(n=64)
    else:
        use_tree_of_thoughts()

    # 终极方案: 训练自己的推理模型
    if budget_is_large:
        train_reasoning_model_with_rl()