DeepSeek-R1 深度解析：推理能力涌现与纯 RL 的范式突破

1
class MultiHeadLatentAttention(nn.Module):
2
    """
3
    MLA: 将 KV 压缩到低秩 latent space
4
    """
5
    def __init__(self, d_model=7168, n_heads=128, n_kv_heads=128, latent_dim=512):
6
        super().__init__()
7
        # 压缩后的 latent
8
        self.q_proj = nn.Linear(d_model, d_model)
9
        self.kv_latent = nn.Linear(d_model, latent_dim)  # 压缩
10
        self.kv_output = nn.Linear(latent_dim, d_model * 2)
11

12
    def forward(self, x):
13
        q = self.q_proj(x)
14
        # 压缩的 KV
15
        kv = self.kv_latent(x)              # (B, L, latent_dim)
16
        k, v = self.kv_output(kv).chunk(2, dim=-1)
17
        # ...

1
DeepSeek-R1-Zero 训练流程：
2

3
DeepSeek-V3 Base (无 SFT)
4
        │
5
        ↓
6
    GRPO 训练
7
    ├─ 每条 prompt 采样 G=16 个回答
8
    ├─ 用规则奖励评分（答案对错）
9
    ├─ 组内归一化优势
10
    └─ PPO-style 更新
11
        │
12
        ↓
13
  训练 数千步...
14
        │
15
        ↓
16
  涌现能力自动出现：
17
    - 长思维链（数千 tokens）
18
    - 自我反思（"等等，让我重新检查..."）
19
    - 尝试多种方法
20
    - 回溯与修正

1
def compute_reward_ZERO(
2
    problem: dict,
3
    response: str,
4
    answer_format_check: bool = True,
5
) -> float:
6
    """
7
    DeepSeek-R1-Zero 的奖励函数。
8
    全部基于规则，无需训练。
9
    """
10
    total_reward = 0.0
11

12
    # === 奖励 1: 答案正确性 ===
13
    if problem["type"] == "math":
14
        ground_truth = extract_ground_truth(problem)
15
        model_answer = extract_final_answer(response)
16

17
        if model_answer is not None and model_answer == ground_truth:
18
            total_reward += 1.0
19
        else:
20
            total_reward -= 0.5  # 轻微惩罚错误答案
21

22
    elif problem["type"] == "code":
23
        # 执行测试用例
24
        test_result = execute_tests(response, problem["tests"])
25
        total_reward += test_result.pass_rate
26

27
    # === 奖励 2: 格式正确性 ===
28
    # 检查回答是否包含最终答案标记
29
    has_answer_tag = (
30
        "The answer is" in response or
31
        "答案是" in response or
32
        "\\boxed{" in response or
33
        "Final Answer:" in response
34
    )
35
    if has_answer_tag:
36
        total_reward += 0.1  # 鼓励明确标记答案
37

38
    # === 奖励 3: 思考过程 ===
39
    # 鼓励模型展示推理过程（而非直接给答案）
40
    has_thinking = (
41
        "Let me" in response or
42
        "First" in response or
43
        "Step" in response or
44
        "分析" in response or
45
        len(response) > 500  # 回答足够长
46
    )
47
    if has_thinking:
48
        total_reward += 0.05
49

50
    return total_reward
51

52

53
def extract_ground_truth(problem: dict) -> str:
54
    """
55
    从问题中提取标准答案。
56
    支持：
57
      - 数学表达式的数值结果
58
      - 证明题的逻辑验证
59
      - 选择题的正确答案
60
    """
61
    gt = problem.get("answer", "").strip()
62

63
    # 归一化比较
64
    gt_normalized = normalize_answer(gt)
65
    return gt_normalized

1
训练步数 vs 行为变化（AIME 2024 pass@1）：
2

3
Step 0 (Base):
4
  表现: 26.7%
5
  行为: 直接输出答案，无推理过程
6
  输出示例: "答案是 42。"
7

8
Step 200:
9
  表现: 40.0%
10
  行为: 开始出现短推理（1-2 句）
11
  输出示例: "让我计算... 答案是 42。"
12

13
Step 500:
14
  表现: 52.3%
15
  行为: 推理变长（5-10 句），但结构散乱
16
  输出示例: "首先... 其次... 因此... 答案是 42。"
17

18
Step 1000:
19
  表现: 68.7%
20
  行为: 出现结构化推理，开始有"步骤感"
21
  输出示例: "Step 1: ... Step 2: ... 验证: ... 最终答案是 42。"
22

23
Step 2000 (关键涌现点):
24
  表现: 73.3%
25
  行为:
26
    - 出现自我反思："等等，我需要检查第一步"
27
    - 尝试多种方法："另一种思路是..."
28
    - 回溯修正："我发现之前的计算有误，修正后..."
29
  输出示例: "我先用方法 A... 得到 ... 但检查后发现不对。
30
             让我用方法 B... 验证后正确。
31
             因此答案是 42。"
32

33
Step 5000:
34
  表现: 79.8%
35
  行为:
36
    - 长推理链（2000+ tokens）
37
    - 主动验证中间步骤
38
    - 在困难问题上"停下来思考"
39
    - 复杂的反思与修正循环

1
def deepseek_r1_grpo_loss(
2
    policy_logps: torch.Tensor,     # (B,)
3
    ref_logps: torch.Tensor,        # (B,)
4
    rewards: torch.Tensor,          # (B,) 规则奖励
5
    group_size: int = 16,
6
    beta: float = 0.04,
7
    clip_ratio: float = 0.2,
8
    entropy_coef: float = 0.02,
9
    response_masks: torch.Tensor = None,  # 思维链部分 mask
10
) -> tuple[torch.Tensor, dict]:
11
    """
12
    DeepSeek-R1-Zero 使用的 GRPO 损失。
13
    与标准 GRPO 的区别：
14
      1. 使用 self-consistency 增强（多答案采样）
15
      2. 显式鼓励长回答（长回答通常包含更多思考）
16
      3. 更高的熵正则系数
17
    """
18
    num_groups = rewards.size(0) // group_size
19

20
    # === 1. 组内归一化优势 ===
21
    rewards = rewards.view(num_groups, group_size)
22
    mean_r = rewards.mean(dim=-1, keepdim=True)
23
    std_r = rewards.std(dim=-1, keepdim=True) + 1e-8
24
    advantages = (rewards - mean_r) / std_r  # (num_groups, G)
25

26
    # === 2. PPO-style Clip Loss ===
27
    # 展平处理
28
    advantages_flat = advantages.flatten()
29

30
    ratio = torch.exp(policy_logps - ref_logps)
31
    surr1 = ratio * advantages_flat
32
    surr2 = torch.clamp(ratio, 1 - clip_ratio, 1 + clip_ratio) * advantages_flat
33
    policy_loss = -torch.min(surr1, surr2).mean()
34

35
    # === 3. KL 散度 ===
36
    kl_penalty = (policy_logps - ref_logps).mean()
37

38
    # === 4. 长度鼓励（鼓励长思维链）===
39
    # 如果最终答案正确，奖励更长的思考过程
40
    with torch.no_grad():
41
        is_correct = rewards > 0.5  # 正确答案
42
        avg_length = response_masks.sum(dim=-1).float().mean() if response_masks is not None else 0
43

44
    # === 5. 熵正则 ===
45
    # R1-Zero 使用更高的熵系数防止坍缩
46
    entropy_loss = -entropy_coef * compute_response_entropy(policy_logps)
47

48
    total_loss = policy_loss + beta * kl_penalty + entropy_loss
49

50
    # 监控指标
51
    metrics = {
52
        "policy_loss": policy_loss.item(),
53
        "kl_penalty": kl_penalty.item(),
54
        "entropy": entropy_loss.item(),
55
        "mean_reward": rewards.mean().item(),
56
        "correct_rate": is_correct.float().mean().item(),
57
        "mean_length": avg_length.item(),
58
    }
59

60
    return total_loss, metrics

1
┌─────────────────────────────────────────────────────────────────┐
2
│                    DeepSeek-R1 四阶段训练                          │
3
├─────────────────────────────────────────────────────────────────┤
4
│                                                                  │
5
│  阶段 0: DeepSeek-V3 Base                                       │
6
│     236B 参数，MoE 架构，无 SFT                                  │
7
│              ↓                                                   │
8
│  阶段 1: GRPO for Reasoning (冷启动数据)                          │
9
│     ├─ 加入少量高质量冷启动数据（数千条）                          │
10
│     ├─ 规则奖励：数学/代码/逻辑                                   │
11
│     ├─ 输出: S1_Checkpoint                                       │
12
│     └─ 目的: 建立初步的推理格式                                   │
13
│              ↓                                                   │
14
│  阶段 2: Reinforced Fine-Tuning (RFT)                            │
15
│     ├─ 用 S1 生成 60 万条思维链数据                                │
16
│     ├─ SFT 微调                                                  │
17
│     ├─ 输出: S2_Checkpoint                                       │
18
│     └─ 目的: 将 GRPO 知识压缩为快速推理能力                        │
19
│              ↓                                                   │
20
│  阶段 3: 拒绝采样 + Supervised Fine-Tuning                        │
21
│     ├─ 用 S2 生成 80 万条多样化数据                               │
22
│     ├─ 拒绝采样：过滤低质量回答                                   │
23
│     ├─ 混合写作/角色扮演/指令等通用数据                           │
24
│     ├─ SFT 微调                                                  │
25
│     ├─ 输出: S3_Checkpoint                                       │
26
│     └─ 目的: 扩展能力边界到所有场景                               │
27
│              ↓                                                   │
28
│  阶段 4: GRPO for All Scenarios                                  │
29
│     ├─ 混合奖励：规则奖励 + RM + 安全                            │
30
│     ├─ 通用能力对齐                                              │
31
│     └─ 输出: DeepSeek-R1                                         │
32
│                                                                  │
33
└─────────────────────────────────────────────────────────────────┘

1
def generate_cold_start_data(
2
    num_samples: int = 5000,
3
    domain: str = "math",
4
) -> list[dict]:
5
    """
6
    生成冷启动数据。
7
    注意：只提供格式引导，不提供推理内容。
8
    """
9
    cold_start_prompts = [
10
        "你是一个擅长数学推理的助手。请在回答中包含："
11
        "【思考过程】...详细的推理步骤...\n"
12
        "【最终答案】...简洁的答案..."
13
    ]
14

15
    # 收集问题（带标准答案）
16
    problems = load_verifiable_problems(domain)
17

18
    # 用少量示例 + few-shot 引导格式
19
    data = []
20
    for problem in problems[:num_samples]:
21
        data.append({
22
            "prompt": cold_start_prompts[0] + "\n\n问题：" + problem["question"],
23
            "response": "",  # 让模型自己推理
24
            "format_hint": "请使用【思考过程】和【最终答案】的格式。",
25
        })
26

27
    return data

1
def rft_stage2(
2
    grpo_model,        # 阶段 1 输出的 GRPO 模型
3
    base_model,        # DeepSeek-V3 Base
4
    num_prompts: int = 600000,
5
    sft_epochs: int = 2,
6
):
7
    """
8
    RFT: Reinforced Fine-Tuning
9

10
    核心思想：
11
      GRPO 学到了"如何推理"
12
      RFT 把这个知识压缩成"快速推理"
13
    """
14
    # === 1. 生成高质量思维链数据 ===
15
    print("Generating reasoning chains...")
16
    reasoning_data = []
17

18
    prompts = sample_prompts(num_prompts)
19
    for batch in tqdm(prompts):
20
        # 用 GRPO 模型生成（允许探索）
21
        with torch.no_grad():
22
            responses = grpo_model.generate(
23
                batch,
24
                max_tokens=8192,
25
                temperature=0.7,
26
                do_sample=True,
27
            )
28

29
        for prompt, response in zip(batch, responses):
30
            # 评分（只保留正确答案）
31
            reward = rule_based_reward(prompt, response)
32
            if reward > 0.5:  # 正确答案
33
                reasoning_data.append({
34
                    "prompt": prompt,
35
                    "response": response,
36
                    "reward": reward,
37
                })
38

39
    print(f"Generated {len(reasoning_data)} high-quality reasoning samples")
40

41
    # === 2. SFT 微调 ===
42
    print("SFT fine-tuning...")
43
    sft_model = copy.deepcopy(base_model)
44
    optimizer = torch.optim.AdamW(sft_model.parameters(), lr=1e-5)
45

46
    dataloader = DataLoader(reasoning_data, batch_size=64, shuffle=True)
47

48
    for epoch in range(sft_epochs):
49
        for batch in dataloader:
50
            loss = compute_sft_loss(
51
                sft_model,
52
                batch["prompt"],
53
                batch["response"],
54
            )
55
            loss.backward()
56
            optimizer.step()
57
            optimizer.zero_grad()
58

59
    return sft_model

1
def rejection_sampling(
2
    model,
3
    prompts: list[dict],
4
    num_generations: int = 100,
5
    reward_threshold: float = 0.8,
6
) -> list[dict]:
7
    """
8
    拒绝采样：从多次生成中选择最好的回答。
9
    """
10
    selected = []
11

12
    for prompt_data in prompts:
13
        prompt = prompt_data["prompt"]
14
        domain = prompt_data.get("domain", "general")
15

16
        # 生成多次
17
        candidates = []
18
        for _ in range(num_generations):
19
            with torch.no_grad():
20
                response = model.generate(prompt, max_tokens=4096)
21

22
            if domain in ["math", "code"]:
23
                # 规则奖励
24
                reward = rule_based_reward(prompt, response)
25
            else:
26
                # RM 奖励
27
                reward = reward_model.score(prompt, response)
28

29
            candidates.append({
30
                "prompt": prompt,
31
                "response": response,
32
                "reward": reward,
33
            })
34

35
        # 拒绝采样：只保留最高质量的
36
        best = max(candidates, key=lambda x: x["reward"])
37
        if best["reward"] >= reward_threshold:
38
            selected.append(best)
39

40
    return selected
41

42

43
def build_s3_dataset():
44
    """
45
    构建阶段 3 的训练数据。
46
    """
47
    dataset = []
48

49
    # 1. 数学/代码（拒绝采样）
50
    math_data = rejection_sampling(
51
        s2_model,
52
        load_math_prompts(200000),
53
        num_generations=100,
54
        reward_threshold=0.9,
55
    )
56
    dataset.extend(math_data)
57

58
    # 2. 通用写作/总结
59
    writing_data = load_filtered_data(
60
        source="general_writing",
61
        filter_fn=lambda x: x["quality_score"] > 0.85,
62
    )
63
    dataset.extend(writing_data)
64

65
    # 3. 角色扮演/对话
66
    chat_data = load_filtered_data(
67
        source="roleplay",
68
        filter_fn=lambda x: x["engagement_score"] > 0.8,
69
    )
70
    dataset.extend(chat_data)
71

72
    # 4. 指令遵循
73
    instruction_data = load_filtered_data(
74
        source="instruction",
75
        filter_fn=lambda x: x["ifeval_score"] > 0.85,
76
    )
77
    dataset.extend(instruction_data)
78

79
    return dataset
80
    # 总计约 80 万条

1
def multi_reward(
2
    prompt: str,
3
    response: str,
4
    domain: str,
5
    reward_model=None,
6
) -> float:
7
    """
8
    混合奖励函数。
9
    """
10
    reward = 0.0
11

12
    # 1. 准确性奖励（最重要）
13
    if domain in ["math", "code", "logic"]:
14
        reward += 0.5 * rule_accuracy_reward(prompt, response)
15

16
    # 2. 帮助性奖励
17
    if reward_model is not None:
18
        reward += 0.3 * reward_model.score(prompt, response)
19

20
    # 3. 格式奖励
21
    reward += 0.1 * format_reward(response)
22

23
    # 4. 安全奖励
24
    reward += 0.1 * safety_reward(response)
25

26
    return reward
27

28

29
def format_reward(response: str) -> float:
30
    """
31
    格式奖励：鼓励清晰的结构。
32
    """
33
    score = 0.0
34

35
    if "｜reasoning｜" in response or "【思考过程】" in response:
36
        score += 0.3
37
    if "｜answer｜" in response or "【最终答案】" in response:
38
        score += 0.3
39
    if response.count("\n") > 3:  # 有清晰的段落
40
        score += 0.2
41
    if not response.startswith(" ") and not response.startswith("\n"):
42
        score += 0.2  # 不以空白开头
43

44
    return min(score, 1.0)
45

46

47
def safety_reward(response: str) -> float:
48
    """
49
    安全奖励：惩罚有害内容。
50
    """
51
    if contains_harmful_content(response):
52
        return -1.0
53
    return 0.0

1
def generate_distillation_data(
2
    teacher_model,      # DeepSeek-R1
3
    student_prompts: list[str],
4
    num_samples: int = 1000000,
5
) -> list[dict]:
6
    """
7
    用 R1 生成蒸馏数据。
8
    """
9
    distillation_data = []
10

11
    for prompt in tqdm(student_prompts[:num_samples]):
12
        with torch.no_grad():
13
            response = teacher_model.generate(
14
                prompt,
15
                max_tokens=8192,
16
                temperature=0.6,
17
                do_sample=True,
18
            )
19

20
        distillation_data.append({
21
            "prompt": prompt,
22
            "response": response,
23
        })
24

25
    return distillation_data

1
1.5B 模型不同训练方式对比（AIME 2024）：
2

3
从头训练 + GRPO (无蒸馏):
4
  32.5%  ← 需要大量计算
5

6
直接蒸馏（用 R1 数据 SFT）:
7
  38.7%  ← 仅用 1/100 计算量
8

9
蒸馏 + GRPO 精调:
10
  42.8%  ← 蒸馏 + RL 进一步提升
11

12
结论：
13
  小模型无法仅靠 RL 学会推理（计算量不够）
14
  但可以蒸馏大模型的推理能力
15
  蒸馏 + 轻量 RL = 最佳性价比

1
def merge_with_recent_checkpoint(
2
    model,
3
    recent_ckpt_path: str,
4
    merge_alpha: float = 0.8,
5
) -> torch.nn.Module:
6
    """
7
    与最近的 checkpoint 合并，防止灾难性遗忘。
8
    """
9
    recent_ckpt = torch.load(recent_ckpt_path)
10

11
    merged_state = {}
12
    for key in model.state_dict():
13
        if key in recent_ckpt:
14
            # 线性插值
15
            merged_state[key] = (
16
                merge_alpha * model.state_dict()[key]
17
                + (1 - merge_alpha) * recent_ckpt[key]
18
            )
19
        else:
20
            merged_state[key] = model.state_dict()[key]
21

22
    model.load_state_dict(merged_state)
23
    return model

1
def test_time_scaling(
2
    model,
3
    prompt: str,
4
    num_samples: int = 64,
5
    majority_vote: bool = True,
6
) -> str:
7
    """
8
    测试时 Scaling：
9
    生成多个回答，用投票或 RM 选择最好的。
10

11
    这与预训练的 Scaling Law 正交——
12
    推理时可以独立增加计算量。
13
    """
14
    responses = []
15
    for _ in range(num_samples):
16
        with torch.no_grad():
17
            response = model.generate(
18
                prompt,
19
                max_tokens=4096,
20
                temperature=0.8,
21
            )
22
        responses.append(response)
23

24
    if majority_vote:
25
        # 多数投票
26
        return majority_vote_answers(responses)
27
    else:
28
        # RM 选择
29
        best_response = max(
30
            responses,
31
            key=lambda r: rule_based_reward(prompt, r)
32
        )
33
        return best_response
34

35

36
# 测试时 Scaling 效果（AIME 2024）：
37
#  sample=1:   71.3%
38
#  sample=4:   75.8%
39
#  sample=16:  78.2%
40
#  sample=64:  79.8%   ← DeepSeek-R1 报告的数字

1
┌──────────────────────────────────────────────────────────────┐
2
│                    DeepSeek-R1 之后的研究方向                  │
3
├──────────────────────────────────────────────────────────────┤
4
│                                                              │
5
│  1. 更强的推理基础                                           │
6
│     → 更大规模的纯 RL 训练                                   │
7
│     → 结合搜索树（MCTS）的推理                               │
8
│                                                              │
9
│  2. 自我改进的极限                                           │
10
│     → 模型能否自己生成更难的问题？                            │
11
│     → Constitutional AI + R1 的结合                           │
12
│                                                              │
13
│  3. 多模态推理                                              │
14
│     → R1 的推理能力能否迁移到视觉/音频？                      │
15
│     → MathVista, MMMU 上的突破                               │
16
│                                                              │
17
│  4. 高效推理                                                │
18
│     → 蒸馏 + 量化 + Speculative Decoding                    │
19
│     → 让 R1 级别的推理在手机/边缘设备上运行                   │
20
│                                                              │
21
│  5. 安全与对齐                                              │
22
│     → 如何确保"长思考"不产生有害内容？                        │
23
│     → R1 的"反思"能力能否用于安全对齐？                       │
24
│                                                              │
25
└──────────────────────────────────────────────────────────────┘

1
┌─────────────────────────────────────────────────────────────┐
2
│               DeepSeek-R1 的三点核心贡献                      │
3
├─────────────────────────────────────────────────────────────┤
4
│                                                              │
5
│  1. 证明了"纯 RL 可以涌现复杂推理"                             │
6
│     → 不需要人类写的思维链数据                                │
7
│     → 推理能力可以通过"试错-奖励"自学                         │
8
│     → 对 AI 发展的路径选择有深远影响                          │
9
│                                                              │
10
│  2. 提出了高效的后训练范式                                     │
11
│     → GRPO + RFT + 拒绝采样 + 多阶段 GRPO                    │
12
│     → 降低了 SOTA 对齐的训练成本                              │
13
│                                                              │
14
│  3. 开创了知识蒸馏的新方法                                     │
15
│     → 大模型 RL 探索 → 小模型蒸馏                             │
16
│     → 让 SOTA 推理能力普惠化                                  │
17
│                                                              │
18
└─────────────────────────────────────────────────────────────┘