使用 Python 从零开始构建 DeepSeek R1

灵感来自于艾琳·汉森的名言

架构与训练逐步解析

免费阅读此故事：链接

DeepSeek R1 的整个训练过程无非是在其基础模型（即 deepseek V3）之上使用不同方式的强化学习。

从一个 微型基础模型 开始，该模型在本地运行，我们将 从零开始构建一切，使用 DeepSeek R1 技术报告，同时涵盖 每一步的理论。

无论你是 DeepSeek R1 的新手还是想训练自己的模型，这篇博客都能满足你的需求！🚀

本文从头到尾直观地解释了 DeepSeek R1 的工作原理。点击这里阅读：

我简化复杂主题，你可以在 Medium 上关注我！

GitHub 代码概述

本博客中显示的所有代码都可以在我的 GitHub 仓库中找到：

代码库的组织结构如下：

train-deepseek-r1/ ├── code.ipynb
├── requirements.txt
└── r1_for_dummies.md

目录

• 搭建舞台
• 我们的训练数据集
• DeepSeek R1 训练快速概述
• 选择我们的基础模型
• 在 RL 设置中的策略模型 (R)
• R1 Zero 的 GRPO 算法
• 提示模板
• 训练数据预处理
• 奖励函数
• 准确性奖励
• 格式奖励
• 推理步骤奖励
• 余弦缩放奖励
• 重复惩罚奖励
• R1 Zero 的训练配置
• GRPO 训练循环
• 保存 Tiny R1 Zero LLM
• R1 Zero 的两个主要问题
• 为 SFT 准备冷启动数据
• 使用长 CoT 的少量提示
• 直接提示
• 后处理精炼
• 使用冷启动数据的 SFT 第一阶段
• R1 的第一阶段 SFT 训练配置
• 第一阶段 STF 训练循环
• 保存 Tiny R1 LLM
• 面向推理的 RL
• 拒绝采样
• SFT 第二阶段训练
• 蒸馏

设置环境

克隆代码库并使用以下命令安装所需的库：

git clone https://github.com/FareedKhan-dev/train-deepseek-r1.git
cd train-deepseek-r1
pip install -r requirements.txt

现在，让我们导入所需的库。

import logging
import os
import sys
import re
import math
from dataclasses import dataclass, field
from typing import List, Optional

import torch
import transformers
from transformers import (
AutoModelForCausalLM,
AutoTokenizer,
HfArgumentParser,
TrainingArguments,
set_seed,
TrainerCallback,
TrainerControl,
TrainerState,
)
from transformers.trainer_utils import get_last_checkpoint

import datasets
from datasets import load_dataset

from trl import (
AutoModelForCausalLMWithValueHead,
PPOConfig,
PPOTrainer,
GRPOTrainer,
GRPOConfig,
SFTTrainer
)

from latex2sympy2_extended import NormalizationConfig
from math_verify import LatexExtractionConfig, parse, verify

我们的训练数据集

尽管论文没有具体说明 RL 预训练的初始数据集，但我们假设它应该是以推理为重点的。

因此，为了尽可能接近原始复制，我们将使用这两个开源推理 Hugging Face 数据集：

1. NuminaMath-TIR（用于 R1 零训练）
2. Bespoke-Stratos-17k（用于 R1 训练）

AI-MO/NuminaMath-TIR 包含 70K 数学问题，消息列显示了解决方案背后的 COT（思路链）推理。

看一下它的示例：

MATH_le = load_dataset(“AI-MO/NuminaMath-TIR”, “default”)

MATH_le[‘train’][0]

{ ‘problem’: ‘What is the degree of the polynomial 4 +5x^3 … ’, ‘solution’: ‘This polynomial is not written in …’, ‘messages’: [{‘from’: ‘user’, ‘value’: ‘The problem …’}] }

而 Bespoke-Stratos 包含 17K 个专注于数学和代码的问题。

它的示例如下：

bespoke_rl = load_dataset(“bespokelabs/Bespoke-Stratos-17k”, “default”)

bespoke_rl[‘train’][0]

{ ‘system’: ‘Your role as an assistant involves … ’, ‘conversations’: [{‘from’: ‘user’, ‘value’: ‘Return your …’}] }

不必仅选择这些数据集，您可以选择任何您喜欢的数据集，只要它是以推理为重点的（一个问题及其逐步解决方案）。

DeepSeek R1 训练快速概述

在深入技术实现之前，快速概述一下，DeepSeek-R1 并不是从零开始训练的。相反，他们是基于已经拥有的相当智能的 LLM DeepSeek-V3 开始的，但他们希望将其打造成推理的超级明星。

DeepSeek R1 实现快速概述（由 Fareed Khan 创建）

为了实现这一目标，他们使用了 强化学习，简称 RL，当 LLM 在推理方面做得好的时候给予奖励，否则就惩罚它。

但这并不是一次简单的训练课程。这就像是一系列步骤，他们称之为管道。他们首先尝试了纯 RL，看看推理是否会自行出现 这就是 DeepSeek-R1-Zero，有点像实验。然后对于 真正的 DeepSeek-R1，他们将其组织得更有条理，分为不同的阶段。他们给它一些起始数据以便开始，然后进行 RL，再提供更多数据，再进行更多 RL……就像一步一步地升级！

整个重点是让这些语言模型在解决问题时变得更出色。

所以，是的，这就是在我们深入每一步的疯狂细节之前的超级简短版本。

选择我们的基础模型

由于DeepSeek团队选择DeepSeek-V3作为他们创建R1 Zero和R1的基础模型，但它的大小实在太大了 685 GB 💀，显然超出了我们的能力范围。

为了简单起见，我们将使用一个小得多的基础模型 Qwen/Qwen2.5–0.5B-Instruct（大小为0.9 GB）。如果您有更高的GPU内存，可以加载未量化的LLM，您可以选择更大的模型，例如 Qwen/Qwen2.5–7B-Instruct。

让我们看看我们基础模型的一些规格：

MODEL_NAME = “Qwen/Qwen2.5-0.5B-Instruct” OUTPUT_DIR = “data/Qwen-GRPO-training”

os.makedirs(OUTPUT_DIR, exist_ok=True)

tokenizer = AutoTokenizer.from_pretrained( MODEL_NAME, trust_remote_code=True, padding_side=“right” )

if tokenizer.pad_token is None: tokenizer.pad_token = tokenizer.eos_token

print(f”词汇表大小: {len(tokenizer)}”) print(f”模型最大长度: {tokenizer.model_max_length}”) print(f”填充标记: {tokenizer.pad_token}”) print(f”EOS标记: {tokenizer.eos_token}”)

词汇表大小: 151665 模型最大长度: 131072 填充标记: <|endoftext|> EOS标记: <|im_end|>

这些是关于模型的一些基本信息，看看我们的基础模型的参数总数。

model = AutoModelForCausalLM.from_pretrained( MODEL_NAME, trust_remote_code=True, torch_dtype=torch.bfloat16 )

print(f”模型参数: {model.num_parameters():,}”)

模型参数: 494,032,768

接近0.5B的参数，让我们打印一个简单的响应，然后我们将进入下一步。

device = torch.device(“cuda” if torch.cuda.is_available() else “cpu”) print(f”使用设备: {device}”)

model.to(device)

def test_model_inference(user_input: str): """使用加载的模型和分词器测试基本模型推理。""" messages = [ {“role”: “system”, “content”: “你是Qwen，一个有帮助的助手。”}, {“role”: “user”, “content”: user_input} ]

text = tokenizer.apply\_chat\_template(
    messages,
    tokenize=False,
    add\_generation\_prompt=True
)


inputs = tokenizer(text, return\_tensors="pt").to(device)
outputs = model.generate(
    \*\*inputs,
    max\_new\_tokens=100,
    do\_sample=True,
    temperature=0.7
)

response = tokenizer.decode(outputs\[0\], skip\_special\_tokens=True)
return response

test_input = “你好吗？” response = test_model_inference(test_input) print(f”测试输入: {test_input}”) print(f”模型响应: {response}”)

“测试输入: 你好吗？ 模型响应: 作为一个AI语言模型，我没有感情…”

所以，这个小模型的输出是相当可靠的，肯定适合我们的DeepSeek类模型训练。

策略模型 (R) 在强化学习设置中

现在我们已经选择了基础模型，接下来我们需要了解基本的强化学习设置是如何运作的，以训练大型语言模型（LLM）。

对于 DeepSeek R1，他们的起点是（DeepSeek V3）基础模型，而在我们的案例中，我们从 Qwen2.5–0.5B-Instruct 开始。这里的起点是指 它创建了 DeepSeek R1 零版本，这是一个初始版本，在最终版本创建之前存在一些错误。

初始版本（R1 Zero）是使用强化学习创建的，其中（DeepSeek v3/Qwen2.5–0.5B）作为强化学习代理（执行动作的角色）。让我们先来可视化一下它是如何运作的。

Qwen 2.5 作为代理工作流程（由 Fareed Khan 创建）

强化学习代理（DeepSeek V3/Qwen2–0.5B）首先采取一个 动作，这意味着它为给定问题生成一个答案和一些推理，这个问题被放入它的 环境 中。在这种情况下，环境简单来说就是推理任务本身。

在采取动作之后，环境会返回一个 奖励。这个奖励就像反馈，它告诉我们的基础模型（DeepSeek V3/Qwen2–0.5B）它的动作有多好。正面的奖励意味着它做对了某件事，可能得到了正确的答案或进行了良好的推理。这个反馈信号随后返回给我们的基础模型，帮助它学习并调整未来采取动作的方式，以获得更好的奖励。

在下一部分中，我们将更详细地讨论这种方法。

GRPO算法用于R1 Zero

为了让我们理解基本的RL流程，现在我们需要了解DeepSeek在R1-Zero中使用的具体RL算法。

有许多RL算法可供选择，但传统的RL使用一种称为**“评论家”**的东西来帮助主要决策部分（“演员”，即DeepSeek-V3/Qwen2-0.5B）。这个评论家通常和演员本身一样庞大和复杂，这基本上使计算成本翻倍。

但DeepSeek使用GRPO来训练他们的初始模型（R1 Zero），GRPO的做法有所不同，因为它直接从一组动作的结果中找出基线，即良好动作的参考点。因此，GRPO根本不需要单独的评论家模型。这节省了大量计算，使事情更加高效。

让我们绘制一个GRPO如何用于R1 Zero训练的流程图，然后我们将解释它。

DeepSeek R1 Zero的GRPO流程（由Fareed Khan创建）

让我们理解DeepSeek的GRPO实现如何与我们的基础模型（Qwen2–0.5B）一起工作。

首先，问题输入（A）被提供给Qwen模型（B），Qwen尝试通过生成完成（C）来生成答案。最终结果称为完成输出（D），包括在<think>标签中的推理步骤和在<answer>标签中的最终解决方案。

接下来，**问题输入（A）和真实答案（E）被输入到奖励函数（F）中，作为智能评分者。这些函数比较Qwen的完成输出（D）**与正确答案，并评估不同的方面，例如：

1. 准确性（答案正确吗？）
2. 格式（<think>和<answer>标签使用正确吗？）
3. 推理步骤（逻辑清晰吗？）
4. 余弦缩放（响应简洁吗？）
5. 重复惩罚（是否有不必要的重复？）

这些评估产生奖励评分（G），然后传递给GRPO训练器（H）。训练器使用梯度来调整Qwen模型（B），微调其生成答案的方式。这个过程被称为梯度奖励策略优化，因为它使用梯度、奖励反馈和策略调整来优化Qwen的响应，以最大化性能。

最后，更新后的**Qwen模型（B）**在新问题上再次进行测试，通过重复循环不断自我完善。随着每次迭代，Qwen成为更好的问题解决者。

在接下来的部分中，我们将开始对我们的训练数据集进行预处理，以进行GRPO训练。

提示模板

我们使用与 DeepSeek 为 GRPO 算法构建 R1 Zero 相同的思维提示模板，因此让我们定义一下：

SYSTEM_PROMPT = ( “用户与助手之间的对话。用户提出一个问题， 助手进行解答。助手首先在内心思考推理过程，然后 提供用户答案。推理过程和答案分别用 <think> </think> 和 <answer> </answer> 标签包围，即 ” “<think> 推理过程在这里 </think><answer> 答案在这里 </answer>” )

这个 系统提示 告诉基础模型 (Qwen2–0.5B) 它作为一个有帮助的助手的角色，逐步推理后再进行回答。

<think> 和 <answer> 标签 用于构建模型响应，分隔其内部推理与最终答案，以便于更好的评估和奖励。

预处理训练数据

现在我们已经准备好系统提示，我们需要根据我们的模板转换训练数据。

预处理数据集概述 (由 Fareed Khan 创建)

我们需要创建 make_conversation 函数来处理对话。

def make_conversation(example): """将数据集示例转换为对话格式。""" return { “prompt”: [ {“role”: “system”, “content”: SYSTEM_PROMPT}, {“role”: “user”, “content”: example[“problem”]}, ], }

它将从我们的训练数据集中提取每个问题列的值，并返回一个包含系统提示和附加问题的字典。让我们创建这个函数来准备我们的数据集。

def load_math_dataset(): """加载并准备数学数据集。""" dataset = load_dataset( “AI-MO/NuminaMath-TIR”, name=“default”, split=[‘train’, ‘test’] )

dataset = {
    'train': dataset\[0\],
    'test': dataset\[1\]
}


for split in dataset:
    dataset\[split\] = dataset\[split\].map(make\_conversation)

    
    if "messages" in dataset\[split\].column\_names:
        dataset\[split\] = dataset\[split\].remove\_columns("messages")

return dataset

我们已经准备好了一切，让我们将训练数据转换为所需格式，并打印训练和测试的大小。

dataset = load_math_dataset()

print(f”训练集大小: {len(dataset[‘train’])}”) print(f”测试集大小: {len(dataset[‘test’])}”)

训练集大小: 72441 测试集大小: 99

现在我们已经拆分了训练数据集，我们需要在进入下一步之前验证我们的数据集（检查用户/助手对话是否存在）。

def validate_dataset(dataset): """对数据集进行基本验证检查。"""

required\_fields = \["problem", "prompt"\]


for split in \['train', 'test'\]:
    print(f"\\n验证 {split} 拆分:")

    
    fields = dataset\[split\].column\_names

    
    missing = \[field for field in required\_fields if field not in fields\]
    if missing:
        print(f"警告: 缺少字段: {missing}")  
    else:
        print("✓ 所有必需字段均存在")  

    
    sample = dataset\[split\]\[0\]

    
    messages = sample\['prompt'\]

    
    
    
    
    if (len(messages) \>\= 2 and
        messages\[0\]\['role'\] == 'system' and
        messages\[1\]\['role'\] == 'user'):
        print("✓ 提示格式正确")  
    else:
        print("警告: 提示格式不正确")  

validate_dataset(dataset)

它输出如下：

验证训练拆分:

✓ 所有必需字段均存在 ✓ 提示格式正确

验证测试拆分:

✓ 所有必需字段均存在 ✓ 提示格式正确

我们的训练数据集成功验证 🙌，这意味着我们已经成功地将数据集转换为训练格式。

奖励函数

我们已经在 GRPO 部分看到，它通过五种不同的方式评估基础模型的答案：

奖励函数（由 Fareed Khan 创建）

1. 准确性（答案正确吗？）
2. 格式（<think> 和 <answer> 标签使用得当吗？）
3. 推理步骤（逻辑清晰吗？）
4. 余弦缩放（响应简洁吗？）
5. 重复惩罚（是否有不必要的重复？）

这些函数将为每个响应计算奖励，我们需要对它们进行编码。接下来，让我们先做这个。

准确性奖励

准确性奖励是最容易理解的，但需要一些复杂的代码。在这个奖励模型中，我们想要检查我们的基础模型响应在数学上是否等同于真实的解决方案。

准确性奖励（创建者：Fareed Khan）

如果模型的答案在数学上是正确的，我们将给予1.0的奖励。如果不正确，奖励为0.0。在真实解决方案无法解析的情况下，我们给予中性奖励0.5，以避免不公平的惩罚。

现在，让我们实现这个函数。

def accuracy_reward(completions, solution, **kwargs):
    """
    Reward function to check if the model's response is mathematically 
    equivalent to the ground truth solution.
    Uses latex2sympy2 for parsing and math_verify for validation.
    """
    
    
    contents = [completion[0]["content"] for completion in completions]
    rewards = []
    
    for content, sol in zip(contents, solution):
        
        gold_parsed = parse(sol, extraction_mode="first_match", 
                            extraction_config=[LatexExtractionConfig()])
        
        if gold_parsed:  
            
            answer_parsed = parse(
                content,
                extraction_config=[
                    LatexExtractionConfig(
                        normalization_config=NormalizationConfig(
                            nits=False,
                            malformed_operators=False,
                            basic_latex=True,
                            equations=True,
                            boxed="all",
                            units=True,
                        ),
                        boxed_match_priority=0,
                        try_extract_without_anchor=False,
                    )
                ],
                extraction_mode="first_match",
            )

            
            reward = float(verify(answer_parsed, gold_parsed))
        else:
            
            reward = 0.5
            print("Warning: Failed to parse gold solution:", sol)

        rewards.append(reward)
    
    return rewards

在这个函数中，我们检查模型的响应是否与正确答案等价。我们不只是比较原始文本，而是：

1. 将解决方案转换为结构化的数学格式，使用latex2sympy2。
2. 如果解析失败，则赋予中性奖励0.5。
3. 提取模型输出并进行标准化，以提高鲁棒性。
4. 使用math_verify检查解析后的响应是否与解析后的解决方案匹配。
5. 如果正确，赋值1；如果不正确，赋值0。

这确保了准确性评估不仅仅是关于文本相似性，而是真正的数学正确性。

格式奖励

格式奖励旨在确保我们的模型遵循指令并正确构建输出。我们要求它将推理放在 <think> 标签中，将最终答案放在 <answer> 标签中，对吧？这个奖励函数正是检查这一点的！

前向奖励（由 Fareed Khan 创建）

如果模型正确使用这些标签，我们给予它 1 的奖励。如果格式错了，它就得到 0。就是这么简单！这鼓励模型关注我们想要的输出结构。

让我们来编写这个代码：

def format_reward(completions, **kwargs):
    """
    Reward function to check if the completion has the correct format:
    <think\>...</think\> <answer\>...</answer\>.
    """
    
    pattern = r"^<think\>.\*?</think\>\\s\*<answer\>.\*?</answer\>$"

    
    completion_contents = [completion[0]["content"] for completion in completions]

    
    matches = [re.match(pattern, content, re.DOTALL | re.MULTILINE)
               for content in completion_contents]

    
    return [1.0 if match else 0.0 for match in matches]

在这个函数中：

• 我们使用正则表达式（regex）定义一个模式。这个模式基本上表示“内容应该以 <think> 开头，里面有 任何内容 直到 </think>，然后是一些 空格，接着是 <answer>，里面有 任何内容 直到 </answer>，最后 结束 在那里”。
• 我们从每个模型的完成中获取实际的文本内容。
• 然后我们使用 re.match 来检查每个内容是否完美匹配我们的模式。re.DOTALL 使得 regex 中的 . 也能匹配换行符，而 re.MULTILINE 使 ^ 和 $ 匹配整个字符串的开始/结束，而不仅仅是行。
• 最后，如果它完全匹配格式，我们给予 1 的奖励，如果没有则给予 0。这是一个严格的开/关奖励，用于格式正确性。

推理步骤奖励

推理步骤奖励有点巧妙。我们希望鼓励我们的模型展示其**“思考过程”**。因此，我们将为其包含看起来像推理步骤的内容给予奖励。

推理步骤奖励鼓励（由 Fareed Khan 创建）

我们将寻找通常出现在逐步推理中的关键词和模式，例如：

• 第一步、第二步等。
• 编号列表，如 1、2
• 项目符号，如 - 或 *
• 过渡词，如 首先、其次、接下来、最后

包含这些内容越多，奖励就越高。这就像是为展示其工作过程而给分！

让我们来编写这个推理鼓励函数：

def reasoning_steps_reward(completions, **kwargs):
    r"""
    Reward function to encourage clear step-by-step reasoning.
    It looks for patterns like "Step 1:", numbered lists, bullet points,
    and transition words.
    """
    
    pattern = r"(Step \\d+:|^\\d+\\.|\\n-|\\n\\*|First,|Second,|Next,|Finally,)"

    
    completion_contents = [completion[0]["content"] for completion in completions]

    
    matches = [len(re.findall(pattern, content, re.MULTILINE))
               for content in completion_contents]

    
    
    return [min(1.0, count / 3) for count in matches]

我们创建了一个稍微复杂的正则表达式模式。它寻找我们上面列出的所有推理指示内容。

我们使用 re.findall 查找每个内容中模式的_所有_匹配项。len(re.findall(…)) 然后给我们这些指示的_计数_。

奖励的计算方式为 min(1.0, count / 3)。这意味着

• 如果找到 3 个或更多推理指示（count >= 3），奖励为 1.0（最高奖励）。
• 如果找到更少（例如，count = 1 或 2），则获得_部分_奖励（如 1/3 或 2/3）。
• 如果没有找到（count = 0），奖励为 0.0。

这里的 / 3 是一个有点神奇的数字。我们在说**“目标是大约 3 个推理步骤以获得满分”**。如果您想鼓励更多或更少的步骤，可以调整这个数字。

余弦缩放奖励

余弦缩放奖励稍微复杂一些。它旨在鼓励正确答案的_简洁性_，并对较长的不正确答案_宽容一些_。

余弦缩放概念（由 Fareed Khan 创建）

可以这样理解：

• 对于正确答案： 我们希望对_更短_、更加直接的解决方案给予更多奖励，而不是长篇大论的答案。简短的正确答案通常更好。
• 对于不正确答案： 短的错误答案可能比较长的错误答案更糟，后者至少_尝试_进行推理。因此，我们希望对短的错误答案的惩罚_更重_，而对长的错误答案的惩罚较轻。

让我们看看实现这种巧妙缩放的代码：

def get_cosine_scaled_reward( min_value_wrong: float = -0.5, max_value_wrong: float = -0.1, min_value_correct: float = 0.8, max_value_correct: float = 1.0, max_len: int = 1000, ): """ 返回一个余弦缩放奖励函数。该函数根据完成长度缩放准确度奖励。 较短的正确解决方案获得更高的奖励，较长的不正确解决方案受到较少的惩罚。 """ def cosine_scaled_reward(completions, solution, accuracy_rewards, **kwargs): """ 余弦缩放奖励函数，根据完成长度调整准确度奖励。 """ contents = [completion[0][“content”] for completion in completions] rewards = []

    for content, sol, acc\_reward in zip(contents, solution, accuracy\_rewards):
        gen\_len = len(content)  
        progress = gen\_len / max\_len 
        cosine = math.cos(progress \* math.pi) 

        if acc\_reward \> 0.5: 
            min\_value = min\_value\_correct
            max\_value = max\_value\_correct
        else: 
            min\_value = max\_value\_wrong  
            max\_value = min\_value\_wrong

        
        reward = min\_value + 0.5 \* (max\_value - min\_value) \* (1.0 + cosine)
        rewards.append(float(reward))
    return rewards
return cosine\_scaled\_reward

get_cosine_scaled_reward(...) 生成一个用于训练的奖励函数，通过 min_value_wrong/max_value_wrong（不正确答案的惩罚范围）和 min_value_correct/max_value_correct（正确答案的奖励范围）等参数自定义缩放。max_len 设置缩放的最大长度。

在内部，cosine_scaled_reward(...) 根据 completions、solution 和 accuracy_rewards 计算奖励。

它计算 gen_len，将其归一化为 progress = gen_len / max_len，并推导出一个余弦值，该值从 1（短答案）开始，逐渐减少到 -1（长答案）。

如果 acc_reward > 0.5，则使用正确的奖励范围，否则应用不正确的范围，但交换最小/最大值，以便对较长的错误答案的惩罚较轻。

重复惩罚奖励

重复惩罚奖励旨在阻止我们的模型陷入循环并重复自己。我们希望它能够生成新颖、多样的推理和答案，而不仅仅是反复复制相同的短语！

重复惩罚思想（由 Fareed Khan 创建）

这个奖励函数会惩罚模型如果它使用相同的词序列（n-grams）太多次。我们在示例中将使用大小为3的n-grams（三元组），但您可以根据需要进行调整。

如果模型重复自己很多次，它会得到一个负奖励（惩罚）。如果它更具多样性并避免重复，惩罚则会较少。

让我们实现代码来惩罚重复：

def get_repetition_penalty_reward(ngram_size: int = 3, max_penalty: float = -0.1):
    """
    返回一个重复惩罚奖励函数。惩罚生成文本中的n-grams重复。
    """
    if max_penalty > 0:
        raise ValueError(f"max_penalty {max_penalty} 不应为正数")

    def zipngram(text: str, ngram_size: int):
        """生成n-grams的辅助函数。"""
        words = text.lower().split() 
        return zip(*[words[i:] for i in range(ngram_size)]) 

    def repetition_penalty_reward(completions, **kwargs) -> float:
        """
        重复惩罚奖励函数。
        """
        contents = [completion[0]["content"] for completion in completions]
        rewards = []
        for completion in contents:
            if completion == "": 
                rewards.append(0.0)
                continue
            if len(completion.split()) < ngram_size: 
                rewards.append(0.0)
                continue

            ngrams = set() 
            total = 0
            for ng in zipngram(completion, ngram_size): 
                ngrams.add(ng) 
                total += 1 

            
            scaling = 1 - len(ngrams) / total
            reward = scaling * max_penalty 
            rewards.append(reward)
        return rewards
    return get_repetition_penalty_reward

我们的 get_repetition_penalty_reward(...) 创建了一个惩罚重复的奖励函数，包含如 ngram_size（默认值为3，用于三元组）和 max_penalty（一个负值，例如 -0.1）等参数。

辅助函数 zipngram(text, ngram_size) 通过将文本转换为小写、将其拆分为单词，并使用 zip(*[words[i:] for i in range(ngram_size)]) 进行高效提取来生成n-grams。

在内部，repetition_penalty_reward(...) 计算每个完成的惩罚。如果为空或太短，则获得0.0的奖励。

惩罚的缩放为 scaling = 1 - len(ngrams) / total，其中 total 是n-grams的数量，len(ngrams) 是唯一计数。更多的重复使得 scaling 接近1，从而增加惩罚。

最终奖励为 scaling * max_penalty，这意味着较少的重复会导致较小的惩罚，而高重复则会导致更强的负奖励。

我们已经实现了所有五个奖励函数，接下来让我们进入下一阶段，定义我们的训练参数。

R1 Zero 的训练配置

现在我们要编写一个配置，以便微调我们的 reward functions 实际上是如何工作的。让我们定义这个配置类：

@dataclass class GRPOScriptArguments: """ GRPO 训练的脚本参数，特别与奖励函数相关。 """

reward\_funcs: list\[str\] = field(
    default\_factory=lambda: \["accuracy", "format"\],
    metadata={
        "help": "奖励函数列表。可能的值：'accuracy', 'format', 'reasoning\_steps', 'cosine', 'repetition\_penalty'"
    },
)
cosine\_min\_value\_wrong: float = field(
    default=-0.5,
    metadata={"help": "错误答案的余弦缩放的最小奖励"},
)
cosine\_max\_value\_wrong: float = field(
    default=-0.1,
    metadata={"help": "错误答案的余弦缩放的最大奖励"},
)
cosine\_min\_value\_correct: float = field(
    default=0.8,
    metadata={"help": "正确答案的余弦缩放的最小奖励"},
)
cosine\_max\_value\_correct: float = field(
    default=1.0,
    metadata={"help": "正确答案的余弦缩放的最大奖励"},
)
cosine\_max\_len: int = field(
    default=1000,
    metadata={"help": "余弦缩放的最大长度"},
)

repetition\_n\_grams: int = field(
    default=3,
    metadata={"help": "重复惩罚奖励的 n-grams 数量"},
)
repetition\_max\_penalty: float = field(
    default=-0.1,
    metadata={"help": "重复惩罚奖励的最大（负）惩罚"},
)

我们的 @dataclass 装饰器使得创建一个存储数据的类变得简单。而 GRPOScriptArguments 类则保存奖励设置。

reward_funcs 列表决定了使用哪些奖励，起始值为 ["accuracy", "format"]，但你可以添加更多，如 "reasoning_steps", "cosine", "repetition_penalty"。

一些设置控制 cosine_scaled_reward 和 repetition_penalty_reward 的工作方式，让你可以调整奖励的给出方式。

接下来，我们有来自 transformers 库的 TrainingArguments。这是控制训练过程几乎 所有 方面的 主要 配置对象。

training_args = TrainingArguments( output_dir=OUTPUT_DIR,
overwrite_output_dir=True, num_train_epochs=1,
per_device_train_batch_size=8,
per_device_eval_batch_size=16,
gradient_accumulation_steps=2,
learning_rate=5e-5,
warmup_ratio=0.1,
weight_decay=0.01,
logging_steps=10,
evaluation_strategy=“steps”,
eval_steps=50,
save_strategy=“steps”,
save_steps=50,
save_total_limit=2,
dataloader_num_workers=2,
seed=42,
bf16=True,
push_to_hub=False,
gradient_checkpointing=True,
report_to=“none”,
)

最后，我们需要一个 ModelConfig。这是我们放置与 模型本身 相关的设置的地方，比如使用哪个预训练模型，使用什么数据类型（如 bfloat16），以及是否信任远程代码等。

让我们定义我们的 ModelConfig：

@dataclass class ModelConfig: """ 模型的配置。 """ model_name_or_path: str = field( default=MODEL_NAME, metadata={“help”: “预训练模型的路径或来自 huggingface.co/models 的模型标识符”} ) model_revision: Optional[str] = field( default=“main”, metadata={“help”: “要使用的特定模型版本（可以是分支名称、标签名称或提交 ID）。”} ) torch_dtype: Optional[str] = field( default=“bfloat16”, metadata={“help”: “覆盖默认的 torch_dtype 并在此数据类型下加载模型。”} ) trust_remote_code: bool = field( default=True, metadata={“help”: “加载模型和分词器时信任远程代码。”} ) attn_implementation: Optional[str] = field( default=“flash_attention_2”, metadata={“help”: “要使用的注意力实现。‘flash_attention_2’ 或 None”} )

我们的 ModelConfig 类保存了关键设置，包括 model_name_or_path，默认值为 Qwen 0.5B Instruct。我们使用 torch_dtype="bfloat16" 来提高效率，并将 trust_remote_code=True 设置为安全的远程加载。此外，如果支持，启用 attn_implementation="flash_attention_2" 以实现更快的训练。

现在我们需要实际 创建 这些配置类的实例，以便可以使用它们：

script_args = GRPOScriptArguments() model_args = ModelConfig()

接下来，我们需要获取我们的奖励函数列表和在训练期间希望使用的任何“回调”。

回调就像小助手，可以在训练过程中的不同点执行某些操作（如记录进度、保存模型等）。现在，我们将只使用一个简单的记录回调。

将我们的奖励函数集中在一个地方。

def get_reward_functions(script_args): """ 根据脚本参数返回奖励函数列表。 """ reward_funcs_list = [] reward_funcs_registry = { “accuracy”: accuracy_reward,
“format”: format_reward,
“reasoning_steps”: reasoning_steps_reward, “cosine”: get_cosine_scaled_reward( min_value_wrong=script_args.cosine_min_value_wrong, max_value_wrong=script_args.cosine_max_value_wrong, min_value_correct=script_args.cosine_min_value_correct, max_value_correct=script_args.cosine_max_value_correct, max_len=script_args.cosine_max_len, ), “repetition_penalty”: get_repetition_penalty_reward( ngram_size=script_args.repetition_n_grams, max_penalty=script_args.repetition_max_penalty, ), }

for func\_name in script\_args.reward\_funcs:
    if func\_name not in reward\_funcs\_registry:
        raise ValueError(f"奖励函数 '{func\_name}' 在注册表中未找到。")
    reward\_funcs\_list.append(reward\_funcs\_registry\[func\_name\])

return reward\_funcs\_list

我们的回调函数将跟踪损失和其他重要信息。

logger = logging.getLogger(__name__)

class LoggingCallback(TrainerCallback): """ 一个简单的回调，用于在特定步骤记录训练信息。 """ def on_step_end(self, args: TrainingArguments, state: TrainerState, control: TrainerControl, **kwargs): if state.global_step % args.logging_steps == 0: logger.info(f”步骤 {state.global_step}: 损失 = {state.log_history[-1].get(‘loss’, None)}, 学习率 = {state.log_history[-1].get(‘learning_rate’, None)}”)

def get_callbacks(training_args, model_args, script_args): """ 返回在训练期间使用的回调列表。 目前，它仅包括 LoggingCallback。你可以扩展此功能以添加更多回调。 """ callbacks = [LoggingCallback()] return callbacks

最后，初始化这些函数。

reward_functions = get_reward_functions(script_args) callbacks = get_callbacks(training_args, model_args, script_args)

GRPO 训练循环

这是实际驱动我们 GRPO 训练的引擎。我们需要初始化它，提供我们准备好的所有组件：我们的模型、奖励函数、训练参数、数据集和回调！

让我们初始化 GRPOTrainer：

grpo_config = GRPOConfig( **training_args.to_dict(), **{

}

)

grpo_trainer = GRPOTrainer( model=model,
reward_funcs=reward_functions,
args=grpo_config,
train_dataset=dataset[‘train’],
eval_dataset=dataset[‘test’],
callbacks=callbacks
)

我们现在可以开始 训练循环！这只需在我们的 grpo_trainer 上调用 train() 方法即可。

train_result = grpo_trainer.train()

当你运行这个单元时，你应该会看到训练过程开始。

…

INFO:__main__:Step 10: Loss = …, Learning Rate = …

INFO:__main__:Step 20: Loss = …, Learning Rate = …

…

训练将需要一些时间，但我们设置了 num_train_epochs = 1 并使用了一个小模型，对于这个例子来说，应该不会花费 太 长时间。

但对于实际的 GRPO DeepSeek R1 Zero 训练，你可能会训练更多的轮次和步骤。

保存 Tiny R1 Zero LLM

训练完成后，我们可以保存训练好的模型，以便进行推理。

TRAINED_MODEL_PATH = “data/Qwen-GRPO-training”

tokenizer.save_pretrained(TRAINED_MODEL_PATH)

grpo_trainer.save_model(TRAINED_MODEL_PATH)

print(f”GRPO 训练模型已保存至 {TRAINED_MODEL_PATH}”)

然后，我们可以简单地加载训练好的模型：

tokenizer = AutoTokenizer.from_pretrained( TRAINED_MODEL_PATH, trust_remote_code=True, padding_side=“right” )

if tokenizer.pad_token is None: tokenizer.pad_token = tokenizer.eos_token

trained_model = AutoModelForCausalLM.from_pretrained( TRAINED_MODEL_PATH, trust_remote_code=True, torch_dtype=torch.bfloat16 )

trained_model.to(device)

为了进行推理：

def test_trained_model_inference(user_input: str): """使用加载的训练模型和分词器进行推理测试。""" messages = [ {“role”: “system”, “content”: SYSTEM_PROMPT}, {“role”: “user”, “content”: user_input} ]

text = tokenizer.apply\_chat\_template(
    messages,
    tokenize=False,
    add\_generation\_prompt=True
)


inputs = tokenizer(text, return\_tensors="pt").to(device)


outputs = trained\_model.generate(
    \*\*inputs,
    max\_new\_tokens=200, 
    do\_sample=True,
    temperature=0.7
)


response = tokenizer.decode(outputs\[0\], skip\_special\_tokens=True)
return response

R1 Zero的两个主要问题

现在我们已经完成了使用我们的基础模型Qwen2–0.5B的R1 Zero训练方法，而不是他们的DeepSeek V3（原始基础模型）。

我们无法识别我们训练模型的问题，但DeepSeek的研究表明，R1 Zero模型在推理测试中表现非常出色，甚至在AIME 2024等任务上得分与更高级的模型如OpenAI-01–0912相似。

这表明，使用强化学习（RL）来鼓励语言模型的推理是一种有前景的方法。

但他们也注意到DeepSeek-R1-Zero存在一些关键问题，需要解决以便于实际应用和更广泛的研究。

R1 Zero的问题（由Fareed Khan创建）

DeepSeek的研究人员表示，该模板是_故意简单且结构集中_的。它_避免_对_推理过程本身_施加任何_内容特定_的限制。例如，它没有说：

• “你_必须_使用逐步推理”（它只是说“推理过程”，留给模型去定义这是什么意思）。
• “你_必须_使用反思性推理”
• “你_必须_使用特定的问题解决策略”

主要问题是**<think>**标签内的推理过程难以阅读，使得人类难以跟随和分析。

另一个问题是语言混合，当被问及多语言问题时，模型有时会在同一响应中混合语言，导致输出不一致且令人困惑。

如果你用西班牙语问它问题，突然间，它的“思考”会是英语和西班牙语的混合，不算精炼！这些问题，混乱的推理和语言困惑，显然是明显的障碍。

这就是他们将初始R1 Zero模型转变为R1的两个主要原因。

为SFT准备冷启动数据

为了修复R1零问题并真正让DeepSeek进行合理推理，研究人员进行了冷启动数据收集并包括监督微调。

可以把它看作是在真正激烈的强化学习训练之前，为模型提供良好的推理基础。基本上，他们想教会DeepSeek-V3 Base什么是良好的推理，以及如何清晰地呈现它。

冷启动数据的一个例子是我们之前看到的Bespoke-Stratos-17k，我们将用它来创建R1，但我们需要理解冷数据集是如何创建的，以便不遗漏实际训练中的任何部分。

少样本提示与长链式思维

一种技术是 少样本提示与长链式思维（CoT）， 我们尝试向 DeepSeek-V3 Base（或者在我们的案例中，Qwen2–0.5B）展示一些问题的例子，并配以超详细的逐步解决方案。这就是链式思维（CoT）。

长链式思维（由 Fareed Khan 创建）

这种方法的目标是让模型通过示例学习，并开始模仿这种全面的推理风格。

对于我们的示例问题 “2 + 3 * 4 等于多少？”，我们可以创建包含几个已解决问题作为示例的提示。让我们看看在 Python 中这是什么样的：

MODEL_NAME = “Qwen/Qwen2.5-0.5B-Instruct” tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME, trust_remote_code=True, padding_side=“right”) if tokenizer.pad_token is None: tokenizer.pad_token = tokenizer.eos_token model = AutoModelForCausalLM.from_pretrained(MODEL_NAME, trust_remote_code=True, torch_dtype=torch.bfloat16).to(“cuda” if torch.cuda.is_available() else “cpu”)

def generate_response(prompt_text): messages = [ {“role”: “system”, “content”: “你是一个提供逐步解决方案的有用助手。”}, {“role”: “user”, “content”: prompt_text} ] text = tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True) inputs = tokenizer(text, return_tensors=“pt”).to(model.device) outputs = model.generate(**inputs, max_new_tokens=200, do_sample=False) response = tokenizer.decode(outputs[0], skip_special_tokens=True) return response.split(”<|im_start|>assistant\n”)[-1].strip()

让我们为我们提出的问题相应地定义少样本示例：

few_shot_prompt = """ 问题: 9 的平方根加 5 等于多少？ 解决方案: <|special_token|> 首先，找出 9 的平方根，结果是 3。然后，将 5 加到 3 上。3 + 5 等于 8。 <|special_token|> 总结: 答案是 8。

问题: 火车以每小时 60 英里行驶 2 小时，行驶多远？ 解决方案: <|special_token|> 使用公式：距离 = 速度乘以时间。速度是每小时 60 英里，时间是 2 小时。距离 = 60 * 2 = 120 英里。 <|special_token|> 总结: 火车行驶 120 英里。

问题: 2 + 3 * 4 等于多少？ 解决方案: """

现在使用我们的基础模型，我们的示例生成看起来是这样的：

target_problem_prompt = few_shot_prompt + “2 + 3 * 4 等于多少？” model_response_few_shot = generate_response(target_problem_prompt)

print(“少样本提示:”) print(target_problem_prompt) print(“\n模型响应（少样本 CoT）：”) print(model_response_few_shot)

它输出以下结构化数据：

少样本提示: 问题: 9 的平方根加 5 等于多少？ 解决方案: <|special_token|> 首先，找出 9 的平方根， 结果是 3。然后，将 5 加到 3 上。3 + 5 等于 8。 <|special_token|> 总结: 答案是 8。

问题: 火车以每小时 60 英里行驶 2 小时，行驶多远？ 解决方案: <|special_token|> 使用公式：距离 = 速度乘以时间。 速度是每小时 60 英里，时间是 2 小时。距离 = 60 * 2 = 120 英里。 <|special_token|> 总结: 火车行驶 120 英里。

问题: 2 + 3 * 4 等于多少？ 解决方案:

模型响应（少样本 CoT）: <|special_token|> 为了解决 2 + 3 * 4，我们需要遵循运算顺序 （PEMDAS/BODMAS）。乘法应在加法之前进行。 步骤 1: 将 3 乘以 4，结果是 12。 步骤 2: 将 2 加到步骤 1 的结果上：2 + 12 = 14。 <|special_token|> 总结: 答案是 14。

看看模型在看到示例后，如何开始用 <|special_token|> 分隔符构建其答案，并提供逐步推理，最终得出总结和答案！

这就是少样本学习引导模型朝着期望输出格式的力量。

直接提示

另一种方法是直接提示。在这里，我们直接指示模型不仅要解决问题，还要明确逐步展示其推理过程，然后验证其答案。

这旨在鼓励一种更深思熟虑的问题解决方法。

基于示例的学习（由 Fareed Khan 创建）

让我们为“2 + 3 * 4 是多少？”设计一个提示，明确要求推理和验证。以下是查看其效果的 Python 代码：

direct_prompt_text = """ 问题：解决这个，逐步展示推理，并验证： 2 + 3 * 4 是多少？ """

model_response_direct = generate_response(direct_prompt_text)

print(“直接提示：”) print(direct_prompt_text) print(“\n模型响应（直接提示）：”) print(model_response_direct)

直接提示的输出非常易于理解，效果如下：

直接提示： 问题：解决这个，逐步展示推理，并验证： 2 + 3 * 4 是多少？

模型响应（直接提示）： <|special_token|> 推理：要解决 2 + 3 * 4，我需要遵循运算顺序，乘法应在加法之前进行。 步骤 1：将 3 乘以 4，结果为 12。 步骤 2：将 2 加到步骤 1 的结果上：2 + 12 = 14。 验证：为了验证答案，我可以再次检查运算顺序和计算。确实，乘法是在加法之前进行的，计算是正确的。 <|special_token|> 总结：答案是 14。

如你所见，通过直接要求推理和验证，模型提供了更全面的输出，包括“验证”部分。

这种方法直接引导模型生成我们所期望的详细推理。

后处理精炼

最后一种技术涉及后处理精炼。有趣的是，他们甚至使用了已经训练好的R1 Zero模型的输出！

尽管存在一些问题，R1 Zero仍然能够进行一定程度的推理。因此，他们将R1 Zero的输出交给人类标注者进行精炼，使其更加清晰、结构化，并纠正任何错误。

处理精炼（创建者：Fareed Khan）

想象一下这样一个杂乱的R1 Zero输出：

<think> ummm… multiply 3 and 4… get 12… then add 2…</think> <answer> 14 </answer>

人类标注者会将其精炼为更清晰、更好的格式：

<|special_token|> 推理：为了解决这个问题，我们使用运算顺序，先进行乘法再进行加法。 第1步：将3乘以4，得到12。 第2步：将2加到结果上：2 + 12 = 14。 <|special_token|> 总结：答案是14。

虽然我们无法在代码中完美模拟人类的精炼，但我们可以演示一个基本的想法，展示如何以编程方式重新格式化和结构化潜在的杂乱输出。

让我们以一个模拟的“杂乱”输出为例，展示如何进行精炼：

messy_output = “<think> ummm… multiply 3 and 4… get 12… then add 2…</think>\n<answer> 14 </answer>”

def refine_output(messy_text): think_content = messy_text.split(“<think>”)[1].split(”</think>”)[0].strip() answer_content = messy_text.split(“<answer>”)[1].split(”</answer>”)[0].strip()

refined\_text = f"""<|special\_token|\> 推理：{think\_content.replace('umm...', '').strip().capitalize()}.

<|special_token|> 总结：答案是{answer_content}。""" return refined_text

refined_output_text = refine_output(messy_output)

print(“杂乱输出（模拟R1 Zero）：”) print(messy_output) print(“\n精炼输出：”) print(refined_output_text)

这将输出：

杂乱输出（模拟R1 Zero）： <think> ummm… multiply 3 and 4… get 12… then add 2…</think> <answer> 14 </answer>

精炼输出： <|special_token|> 推理：Multiply 3 and 4… get 12… then add 2. <|special_token|> 总结：答案是14。

这个简单的refine_output函数只是一个基本示例。人类的真实精炼涉及更细致的理解和推理步骤的纠正。

然而，它展示了核心思想：获取初始模型输出并改善其质量和结构，以创建更好的训练数据。

在生成这些冷启动数据后，下一步关键步骤是监督微调（SFT），我们将在下一部分中探讨！

SFT 阶段 1 使用冷启动数据

为了生成适当的冷启动数据以使用监督微调构建 R1，我们显然需要一个合适的团队以及大量的代码，但幸运的是，我们已经拥有与冷启动形式相似的数据（Bespoke-Stratos-17k）。

我们需要了解 SFT Trainer 在处理我们的训练数据时，训练是如何进行的？

SFT 是一种监督学习。这意味着我们向模型提供输入和 期望 输出的配对。

在我们的案例中，输入可能是一个问题提示，而期望输出是来自我们训练数据集的经过充分推理的逐步解决方案。我希望这一点能清楚地说明为什么需要冷数据。

它将我们的标记化训练数据分批输入到模型中。对于每个批次，会发生一系列重要的操作，让我们来可视化这个内部过程：

SFT 工作流程（由 Fareed Khan 创建）

首先，模型接收一个输入，例如问题提示。它处理这个输入并逐个生成其最佳猜测的解决方案，这些是 预测的标记。

接下来，SFT Trainer 需要知道这些预测的好坏。它使用 损失函数，通常是交叉熵损失。这个函数在数学上比较模型的预测标记与我们训练数据中的 正确 标记。可以将其视为计算模型答案的“误差”。

这个“误差”并不是简单地被丢弃。它是学习的关键信号。通过一种称为 反向传播 的过程，这个误差用于计算 梯度。梯度就像指南，指向可以减少误差的参数调整方向。

最后，一个 优化器，如 AdamW 使用这些梯度微调模型的内部设置——其参数。这些调整旨在使模型的下一个预测更接近正确答案。

阶段 1 SFT 训练器配置 for R1

还记得我们在 R1 Zero 中遇到的混乱推理和语言混合问题吗？SFT 旨在解决这个问题。通过在高质量、精炼的数据上进行训练，我们在教模型：

• 清晰的推理风格：以易于阅读和跟随的方式构建其“思维”。
• 一致的语言：在响应中坚持使用一种语言，避免混淆的混合。

我们使用 Bespoke-Stratos-17k 数据集进行 SFT。如前所述，它包含 17,000 个专注于数学和代码的问题，格式非常适合我们的需求。

让我们快速回顾一下来自 Bespoke-Stratos-17k 的示例：

bespoke_rl = load_dataset(“bespokelabs/Bespoke-Stratos-17k”, “default”)

bespoke_rl[‘train’][0]

{ ‘system’: ‘Your role as an assistant involves … ’, ‘conversations’: [{‘from’: ‘user’, ‘value’: ‘Return your …’}] }

这个数据集，包含系统提示和用户-助手对话，非常适合向我们的模型展示推理对话应该是什么样的。

我们将再次使用 trl 库，这使得 SFT 训练变得非常简单。

首先，我们需要设置我们的配置，类似于我们为 GRPO 所做的，但这次是针对 SFT。

MODEL_NAME = “Qwen/Qwen2.5-0.5B-Instruct” OUTPUT_DIR = “data/Qwen-SFT-training” os.makedirs(OUTPUT_DIR, exist_ok=True)

training_args = TrainingArguments( output_dir=OUTPUT_DIR, overwrite_output_dir=True, num_train_epochs=1,
per_device_train_batch_size=8, per_device_eval_batch_size=16, gradient_accumulation_steps=2, learning_rate=2e-5,
warmup_ratio=0.1, weight_decay=0.01, logging_steps=10, evaluation_strategy=“no”, eval_steps=50, save_strategy=“steps”, save_steps=50, save_total_limit=2, dataloader_num_workers=2, seed=42, bf16=True, push_to_hub=False, gradient_checkpointing=True, report_to=“none”, packing=True, max_seq_length=4096 )

model_args = ModelConfig( model_name_or_path=MODEL_NAME, model_revision=“main”, torch_dtype=“bfloat16”, trust_remote_code=True, attn_implementation=“flash_attention_2” )

这些 TrainingArguments 和 ModelConfig 与我们为 GRPO 使用的相似，但有一些更适合 SFT 的调整（例如，学习率略有不同，重要的是，packing=True 和 max_seq_length=4096 以便于在更长序列上进行高效训练）。

第 1 阶段 STF 训练循环

现在，让我们加载我们的数据集和分词器：

dataset_sft = load_dataset(“HuggingFaceH4/Bespoke-Stratos-17k”, split=‘train’)

tokenizer = AutoTokenizer.from_pretrained( MODEL_NAME, trust_remote_code=True, padding_side=“right” ) if tokenizer.pad_token is None: tokenizer.pad_token = tokenizer.eos_token

最后，我们初始化 SFTTrainer 并开始训练！

model_sft = AutoModelForCausalLM.from_pretrained( MODEL_NAME, trust_remote_code=True, torch_dtype=torch.bfloat16 )

sft_trainer = SFTTrainer( model=model_sft,
train_dataset=dataset_sft,
tokenizer=tokenizer,
args=training_args,
dataset_text_field=“conversations”,
packing=True,
max_seq_length=4096
)

sft_train_result = sft_trainer.train()

当你运行这段代码时，你会看到 SFT 训练过程开始。它将类似于 GRPO 训练输出，显示每个日志步骤的损失和学习率。

… INFO:__main__:Step 10: Loss = …, Learning Rate = … INFO:__main__:Step 20: Loss = …, Learning Rate = … …

就像 GRPO 一样，训练时间将取决于你的硬件和选择的训练轮数。由于我们在这个例子中仍在使用一个小模型且仅使用 1 个训练轮，因此应该相对快速。

保存微调后的 Tiny R1 LLM

在完成 SFT 后，我们保存我们新微调的模型（R1）。

TRAINED_SFT_MODEL_PATH = “data/Qwen-SFT-training”

tokenizer.save_pretrained(TRAINED_SFT_MODEL_PATH)

sft_trainer.save_model(TRAINED_SFT_MODEL_PATH)

print(f”SFT 训练的模型已保存到 {TRAINED_SFT_MODEL_PATH}”)

这就是 SFT 部分的内容！我们已经使用基础模型，展示了许多良好推理的例子，并对其进行了微调，使其能够更好地生成清晰、结构化的响应。

这个使用 SFT 微调后的模型在 SFT 第一阶段后被称为 R1

SFT 之后的步骤，特别是 RL 阶段和拒绝采样，从头开始在 Python 中实现是复杂的。专注于理论理解是理解整个过程的关键。

面向推理的强化学习

在经过 SFT 之后，模型的推理能力有所提升，但我们希望真正关注推理质量并修复语言混用问题。这个阶段再次使用强化学习，但采用更智能的奖励系统。

这个新的奖励检查模型的推理和回答是否与问题使用相同的语言。如果你用英语提问，整个回答应该都是英语。这解决了语言混用的问题。

面向推理的强化学习（由 Fareed Khan 创建）

它在准确性基础上增加了语言一致性奖励，以确保 SFT 模型在推理和回答时使用与输入相同的语言。GRPO 算法和 R1 Zero 的训练循环被重用，但奖励信号得到了改进，专门针对更好的推理和一致的语言输出。

拒绝采样

为了获得超高质量的推理数据，DeepSeek 使用 拒绝采样。可以把它看作是一个过滤器，只保留 最佳 示例。

拒绝采样（由 Fareed Khan 创建）

模型生成许多推理示例。这些示例会被评估其正确性和推理质量（通常使用生成奖励模型和人工检查）。

只有 最佳 的高质量推理示例会被保留。结合非推理数据，这个精炼的数据集用于第二个 SFT 阶段 2，进一步提升推理和一般能力。

SFT 第二阶段训练

最后的强化学习阶段专注于使模型成为一个在 所有 情况下都能提供帮助和安全的人工智能助手，而不仅仅是解决推理问题。这是与人类价值观的一致性。

关键关注点：有用性与无害性奖励

不仅仅是准确性，奖励系统现在包括：

• 有用性： 响应是否有用且信息丰富？
• 无害性： 响应是否安全、公正且符合伦理？

SFT 第二阶段（由 Fareed Khan 创建）

训练数据变得多样化，包括推理任务和人类偏好数据（哪个输出更好——更有用，较少有害？）。

奖励系统现在平衡了准确性与 有用性和无害性。迭代的强化学习训练（可能再次是 GRPO）优化模型，使其不仅在推理方面表现良好，还能成为一个安全和有用的人工智能助手，适用于一般用途，最终形成 DeepSeek R1。

蒸馏

为了使 DeepSeek R1 更易于访问，他们将其知识 蒸馏 成更小的模型。

蒸馏过程（由 Fareed Khan 创建）

蒸馏将一个大型强大的“教师”模型（DeepSeek R1）的知识转移到更小的“学生”模型。使用大量推理示例的数据集，DeepSeek R1 的输出被用作 目标 答案。

然后训练更小的模型（SFT）以模仿这些输出。这导致了更小、更快的模型，保留了 DeepSeek R1 重要的推理能力，使其更适合广泛使用。

希望这些简化的解释和图表能使 DeepSeek R1 训练的最后阶段更加清晰，而不会陷入过多的技术细节！

祝阅读愉快！