详解如何复现DeepSeek R1:从零开始利用Python构建

DeepSeek R1 的整个训练过程，说白了就是在其基础模型（也就是 deepseek V3）之上，用各种不同的强化学习方法来“雕琢”它。

咱们从一个小小的本地运行的基础模型开始，一边跟着 DeepSeek R1 技术报告 的步骤，一边把理论知识也讲清楚，从零开始构建一切。

不管你是对 DeepSeek R1 一窍不通，还是想自己动手训练模型，这篇文章都能帮到你！🚀

更多LLM图解内容可以查看https://blog.csdn.net/qq_36603091/category_12933029.html

🧠 向所有学习者致敬！

“学习不是装满一桶水，而是点燃一把火。” —— 叶芝

---

我的博客主页： https://lizheng.blog.csdn.net

🌐 欢迎点击加入AI人工智能社区！

搭建舞台

完整代码

pip install torch torchvision torchaudio transformers datasets accelerate peft trl wandb vllm latex2sympy2_extended math_verify

现在，咱们来导入所需的库。

# 导入必要的库
import logging
import os
import sys
import re
import math
from dataclasses import dataclass, field
from typing import List, Optional

# 导入 PyTorch 和 Hugging Face Transformers
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

# 导入 TRL（Transformers Reinforcement Learning）中的库
from trl import (
    AutoModelForCausalLMWithValueHead,
    PPOConfig,
    PPOTrainer,
    GRPOTrainer,
    GRPOConfig,
    SFTTrainer
)

# 导入数学相关的工具
from latex2sympy2_extended import NormalizationConfig
from math_verify import LatexExtractionConfig, parse, verify

我们的训练数据集

虽然论文里没明确指出强化学习预训练的初始数据集到底是什么，但咱推测它应该得是跟推理相关的。

为了尽可能地贴近原版复现，咱们就用这两个开源的推理 Hugging Face 数据集：

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

AI-MO/NuminaMath-TIR 包含了 70K 道数学题，其中“messages”列展示了解题背后的 COT（推理链）。

看看它的样本长啥样：

# 从 DigitalLearningGmbH 加载 “AI-MO/NuminaMath-TIR” 数据集
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 道专注于数学和代码的题目。

它的样本看起来是这样的：

# 从 bespokelabs 加载 “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 的训练过程。DeepSeek-R1 并不是从零开始训练的，就像不是从一片空白开始的。他们一开始有一个已经很厉害的 LLM（DeepSeek-V3），但想把它打造成推理界的超级巨星。

DeepSeek R1 实现快速概览

为了达到这个目标，他们用上了强化学习（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 并带上聊天模板
tokenizer = AutoTokenizer.from_pretrained(
    MODEL_NAME,
    trust_remote_code=True,
    padding_side="right"
)

# 如果没设置 pad token，就用 eos_token 来代替
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"结束标记：{tokenizer.eos_token}")


#### 输出结果 ####
词汇表大小：151665
模型最大长度：131072
填充标记：<|endoftext|>
结束标记：<|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 参数啦，咱们先打印出一个简单的回应，然后就进入下一步。

# 检查 CUDA 是否可用
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print(f"使用设备：{device}")

# 把模型移到合适的设备上
model.to(device)

# 测试基础推理
def test_model_inference(user_input: str):
    """测试加载的模型和 tokenizer 的基础推理。"""
    messages = [
        {"role": "system", "content": "You are Qwen, a helpful assistant."},
        {"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 = "how are you?"
response = test_model_inference(test_input)
print(f"测试输入：{test_input}")
print(f"模型回应：{response}")


#### 输出结果 ####
"测试输入：how are you?
模型回应：As an AI language model I dont have feelings ..."
#### 输出结果 ####

所以，这个小模型的输出还挺靠谱的，绝对适合咱们用来训练 DeepSeek 的类似模型。

强化学习设置中的策略模型 ®

现在咱们已经选好了基础模型，接下来就得了解强化学习设置中基础模型是怎么工作的啦。

对于 DeepSeek R1 来说，他们的起点是 DeepSeek V3 基础模型，在咱们这儿就是 Qwen2.5–0.5B-Instruct。说起起点，就是说它先搞出了DeepSeek R1 Zero 初始版本，这个版本在最终版本出来之前还存在一些错误呢。

强化学习中的代理（DeepSeek V3/Qwen2–0.5B）就像是个演员，它会采取行动，也就是说，它会针对放到它的环境（也就是推理任务本身）中的问题，生成一个答案和一些推理。

采取行动之后，环境就会给出一个奖励。这个奖励就像是反馈，它告诉咱们的基础模型（DeepSeek V3/Qwen2–0.5B）它的行动好不好。要是奖励是正的，那就说明它做对了，可能是答案正确或者推理得当。这个反馈信号就会回到咱们的基础模型那儿，帮助它学习和调整以后怎么采取行动，以便以后能拿到更好的奖励。

在接下来的部分，咱们会更详细地讨论这个方法论。

GRPO 算法用于 R1 Zero

既然已经了解了强化学习的基本流程，那接下来就得学习 DeepSeek 用来训练 R1-Zero 的强化学习算法到底是什么啦。

虽然有很多强化学习算法，但传统的强化学习会用到一个叫做**“批评者”**的东西来帮助主要做决策的部分（“演员”，也就是 DeepSeek-V3/Qwen2-0.5B）。这个批评者通常和演员一样复杂，差不多大小，这就基本上把计算成本翻倍了。

不过 DeepSeek 用的是 GRPO 来训练他们的初始版本（R1 Zero），GRPO 的做法可不一样，因为它能直接从一组行动的结果里算出一个基准，一个好行动的参考点。正是因为这样，GRPO 根本就不需要一个单独的批评者模型。这就节省了大量的计算成本，让整个过程更高效啦。

接下来，咱们就画一个关于 GRPO 是怎么用于 R1 Zero 训练的流程图，然后咱们就来解读它。

DeepSeek R1 Zero 的 GRPO 流程

咱们来理解一下 DeepSeek 的 GRPO 实现是怎么和咱们的基础模型（Qwen2–0.5B）一起工作的。

首先，问题输入 (A) 被送入 Qwen 模型 (B)，Qwen 尝试通过 生成完成 © 来生成一个答案。最终结果，也就是 完成输出 (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 的是一样的，所以咱们来定义一下：

# DeepSeek 系统提示，用于基于 GRPO 的训练
SYSTEM_PROMPT = (
    "用户和助手之间的对话。用户提出问题，"
    "助手解答。助手先在脑海中思考推理过程，"
    "然后向用户提供答案。推理过程和答案分别用"
    "<think> </think> 和 <answer> </answer> 标签包裹，"
    "即：<think> 在此写出推理过程 </think><answer> 在此写出答案 </answer>"
)

这个系统提示告诉基础模型（Qwen2–0.5B）它的角色是一个会一步步推理的助手，然后再给出答案。

<think> 和 <answer> 标签用来结构化模型的回应，把它的内部推理和最终答案分开，这样就能更好地进行评估和奖励啦。

预处理训练数据

现在咱们已经有了系统提示，接下来就得按照模板来转换训练数据啦。

预处理数据集概览

咱们需要创建一个 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)

        # 如果存在，移除 'messages' 列
        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"]

    # 遍历数据集的 'train' 和 'test' 分割
    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]

        # 提取 'prompt' 字段，它包含一个消息列表
        messages = sample['prompt']

        # 验证提示格式：
        # - 至少包含两条消息
        # - 第一条消息应该是来自 'system' 角色
        # - 第二条消息应该是来自 'user' 角色
        if (len(messages) >= 2 and
            messages[0]['role'] == 'system' and
            messages[1]['role'] == 'user'):
            print("✓ 提示格式正确")  # 确认格式正确
        else:
            print("警告：提示格式不正确")  # 如果格式不正确就发出警告

# 验证数据集
validate_dataset(dataset)

它的输出结果是这样的：

验证 train 分割：

✓ 所需字段全部存在
✓ 提示格式正确

验证 test 分割：

✓ 所需字段全部存在
✓ 提示格式正确

咱们的训练数据集成功验证啦 🙌，这就意味着咱们已经成功地把数据集转换成训练所需的格式啦。

奖励函数

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

奖励函数

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

每种函数都会为每个回应计算奖励，接下来咱们就得把它们写出来啦。所以，咱们现在就开始吧。

准确率奖励

准确率奖励是最容易理解的，但代码稍微有点复杂。在这个奖励模型里，咱们想知道从数学角度来说，咱们的基础模型回应是否和正确答案等价。

准确率奖励

如果模型的答案在数学上是正确的，咱们就给它 1.0 的奖励。要是不正确，奖励就是 0.0。要是正确答案没办法解析，咱们就给一个中性的奖励 0.5，避免不公平的惩罚。

接下来，咱们来实现这个函数。

def accuracy_reward(completions, solution, **kwargs):
    """
    奖励函数，用来检查模型的回应是否在数学上和正确答案等价。
    使用 latex2sympy2 进行解析，math_verify 进行验证。
    """

    # 提取回应
    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",
            )

            # 如果正确就奖励 1.0，如果错误就奖励 0.0
            reward = float(verify(answer_parsed, gold_parsed))
        else:
            # 如果正确答案没办法解析，就给一个中性的奖励（0.5）
            reward = 0.5
            print("警告：正确答案解析失败：", sol)

        rewards.append(reward)

    return rewards

在这个函数里，咱们检查模型回应是否和正确答案等价。咱们不是简单地比较原始文本，而是：

1. 把答案转换成一个结构化的数学格式，用 latex2sympy2。
2. 如果解析失败，就给一个中性的奖励 0.5。
3. 提取模型输出，并进行规范化，提高鲁棒性。
4. 用 math_verify 来检查解析后的回应是否和解析后的答案匹配。
5. 如果正确就奖励 1，如果错误就奖励 0。

这就确保了准确率评估不仅仅看文本相似度，而是看真正的数学正确性。

格式奖励

格式奖励就是确保咱们的模型按照要求行事，把输出结构化好。咱们让它在 <think> 标签里写推理过程，在 <answer> 标签里写最终答案，对吧？这个奖励函数就会检查这个！

格式奖励

如果模型正确使用了这些标签，咱们就给它 1.0 的奖励。要是它搞砸了格式，就给 0。就这么简单！这就鼓励模型好好关注咱们想要的输出结构啦。

接下来，咱们把这个函数写出来：

# 实现格式奖励函数
def format_reward(completions, **kwargs):
    """
    奖励函数，用来检查完成内容是否有正确的格式：
    <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]

    # 如果格式正确就奖励 1.0，否则奖励 0.0
    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。这是一个严格的二进制奖励，用来检查格式是否正确。

推理步骤奖励

推理步骤奖励有点聪明。咱们想鼓励模型展示它的“思考过程”。所以，咱们会奖励它包含看起来像是推理步骤的内容。

推理步骤奖励鼓励

咱们会寻找通常出现在分步推理中的关键词和模式，比如：

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

它包含的这些内容越多，奖励就越高。这就像是给展示它的工作过程加分啦！

接下来，咱们把这个推理鼓励函数写出来：

def reasoning_steps_reward(completions, **kwargs):
    r"""
    奖励函数，用来鼓励清晰的分步推理。
    它会寻找像“第一步：”、带数字的列表、项目符号、
    以及过渡词这样的模式。
    """
    # 正则表达式模式，用来寻找推理步骤的指示
    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]

    # 奖励和推理步骤的数量成正比，最高为 1.0
    # 咱们在这里用了一个“魔法数字”3——至少有 3 个推理步骤才能得到满分
    return [min(1.0, count / 3) for count in matches]

咱们创建了一个稍微复杂点的 regex 模式。它会寻找咱们之前列出的所有推理指示内容。

咱们用 re.findall 来找出每个内容里所有匹配咱们模式的内容。len(re.findall(…)) 然后就会给出这些指示的数量。

奖励是按照 min(1.0, count / 3) 来计算的。这就意味着

• 如果它找到了 3 个或更多的推理指示（count >= 3），奖励就是 1.0（最大奖励）。
• 如果它找到的少于 3 个（比如 count = 1 或 2），它就会得到一个部分奖励（比如 1/3 或 2/3）。
• 如果它一个都没找到（count = 0），奖励就是 0.0。

这个 / 3 是一个有点随意的数字。咱们是在说**“至少要有 3 个推理步骤才能得到满分”**。你可以调整这个数字，要是你想鼓励更多或者更少的步骤的话。

余弦缩放奖励

余弦缩放奖励有点高级。它的目的是鼓励在正确答案中使用简洁的表达，并且对于错误答案，不要对长答案惩罚得太厉害。

余弦缩放概念

这么想吧：

• 对于正确答案：咱们想奖励更短、更直接的解决方案，而不是长篇大论的答案。一个简短且正确的答案通常更好。
• 对于错误答案：一个简短但错误的答案可能比一个长篇大论但错误的答案更糟糕。所以，咱们想对短的错误答案惩罚得更厉害，而对长的错误答案惩罚得少一些。

接下来，咱们看看这个巧妙的缩放代码：

# 实现余弦缩放奖励函数
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: # 假设准确率奖励对于正确答案会给出大约 1.0
                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，它就会使用正确的奖励范围，否则它会应用错误的范围，但会交换最小值和最大值，这样对长的错误答案惩罚得少一些。

重复惩罚奖励

重复惩罚奖励就是防止咱们的模型陷入循环，不断地重复自己。咱们希望它能生成新鲜、多样的推理和答案，而不是一遍又一遍地重复相同的短语！

重复惩罚的想法

这个奖励函数会惩罚模型如果它重复使用相同的词序列（n-gram）。咱们的例子中会用到大小为 3 的 n-gram（三元组），但你可以调整这个大小。

如果模型重复得太多，它就会得到一个负奖励（惩罚）。如果它更富有变化，避免了重复，惩罚就会少一些。

接下来，咱们把这个惩罚重复的函数写出来：

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

    def zipngram(text: str, ngram_size: int):
        """辅助函数，用来从文本中生成 n-gram。"""
        words = text.lower().split() # 转小写并分割成单词
        return zip(*[words[i:] for i in range(ngram_size)]) # 创建 n-gram

    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() # 使用集合来存储唯一的 n-gram
            total = 0
            for ng in zipngram(completion, ngram_size): # 生成 n-gram
                ngrams.add(ng) # 把 n-gram 添加到集合中（重复的会被忽略）
                total += 1 # 统计总 n-gram 数量

            # 计算缩放因子：重复越多，缩放因子越高
            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) 会从文本中生成 n-gram，通过把文本转成小写，分割成单词，然后用 zip(*[words[i:] for i in range(ngram_size)]) 来高效提取。

在内部，repetition_penalty_reward(...) 会计算每个完成内容的惩罚。如果它是空的或者太短，它就会得到 0.0 的奖励。

惩罚会按照 scaling = 1 - len(ngrams) / total 来计算，其中 total 是 n-gram 的数量，len(ngrams) 是唯一 n-gram 的数量。重复越多，scaling 就越接近 1，惩罚也就越高。

最终的奖励是 scaling * max_penalty，也就是说重复越少，惩罚就越小，重复越多，负奖励就越大。

咱们已经实现了这五个奖励函数啦，接下来咱们定义训练参数。

R1 Zero 的训练配置

现在咱们要写一个配置，用来微调这些 奖励函数 的工作方式。所以，咱们来定义这个配置类：

# 定义 GRPOScriptArguments，用于奖励函数参数
@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-gram 数量"},
    )
    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 是怎么工作的，让你能调整奖励是怎么给的。

接下来，咱们有 TrainingArguments，这是来自 transformers 库的 主要 配置对象，它几乎控制了训练过程的 一切 。

# 定义来自 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,            # AdamW 优化器的初始学习率
    warmup_ratio=0.1,              # 在 warmup_ratio 比例的训练步数上进行线性预热
    weight_decay=0.01,             # 对除了偏差和 LayerNorm 权重之外的所有层应用权重衰减
    logging_steps=10,              # 每 X 个更新步记录一次
    evaluation_strategy="steps",    # 每 `eval_steps` 进行评估
    eval_steps=50,                 # 评估和记录步数
    save_strategy="steps",         # 每 `save_steps` 保存一次检查点
    save_steps=50,                 # 每 X 个更新步保存一次检查点
    save_total_limit=2,            # 限制检查点的总数。会删除较旧的检查点。
    dataloader_num_workers=2,      # 数据加载的子进程数量
    seed=42,                       # 随机种子，用于复现
    bf16=True,                     # 使用混合精度 BFP16 训练
    push_to_hub=False,             # 是否将最终模型推送到 Hugging Face Hub
    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": "加载模型和 tokenizer 时信任远程代码。"}
    )
    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,  # 假设 accuracy_reward 已经在前面定义好了
        "format": format_reward,      # 假设 format_reward 已经在前面定义好了
        "reasoning_steps": reasoning_steps_reward, # 假设 reasoning_steps_reward 已经定义好了
        "cosine": get_cosine_scaled_reward( # 假设 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( # 假设 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()] # 实例化 LoggingCallback
    return callbacks

最后，初始化这些函数。

# 获取奖励函数和回调函数
reward_functions = get_reward_functions(script_args)
callbacks = get_callbacks(training_args, model_args, script_args)

GRPO 训练循环

这就是真正驱动咱们 GRPO 训练的引擎。咱们需要初始化它，把咱们准备好的所有东西都给它：咱们的模型、奖励函数、训练参数、数据集和回调函数！

接下来，咱们初始化 GRPOTrainer：

# 根据 TrainingArguments 创建 GRPOConfig
grpo_config = GRPOConfig(
    **training_args.to_dict(), # 把 TrainingArguments 转换成字典并展开
    **{
       # 在这里移除了 model_init_kwargs
       # 咱们传入的是实例化的 'model' 对象，所以 GRPOTrainer 不需要 model_init_kwargs
    }
)

grpo_trainer = GRPOTrainer(
    model=model,                      # 咱们初始化的 Qwen 模型
    reward_funcs=reward_functions,    # 前面步骤中的奖励函数列表
    args=grpo_config,                # GRPOConfig（从 TrainingArguments 创建）
    train_dataset=dataset['train'],   # 训练数据集
    eval_dataset=dataset['test'],    # 评估数据集
    callbacks=callbacks              # 回调函数列表
)

现在，咱们就可以开始 训练循环 了！这只要调用 grpo_trainer 的 train() 方法就行。

# 开始 GRPO 训练循环
train_result = grpo_trainer.train()

当你运行这个代码块的时候，训练过程就会开始。

...

INFO:__main__:步骤 10: 损失 = ..., 学习率 = ...

INFO:__main__:步骤 20: 损失 = ..., 学习率 = ...

...

训练会花一些时间，但咱们设置的是 num_train_epochs = 1，而且用的是个小模型，所以这个例子不会花太久时间。

不过对于真正的 GRPO DeepSeek R1 Zero 训练来说，你可能会训练更多轮，更多步。

保存微小的 R1 Zero LLM

一旦训练完成，咱们就可以保存咱们训练好的模型啦，这样就能用来做推理了。

# 定义训练好的模型的路径（和 OUTPUT_DIR 一样）
TRAINED_MODEL_PATH = "data/Qwen-GRPO-training"

# 保存 tokenizer
tokenizer.save_pretrained(TRAINED_MODEL_PATH)

# 保存训练好的模型
grpo_trainer.save_model(TRAINED_MODEL_PATH)

print(f"GRPO 训练好的模型已保存到 {TRAINED_MODEL_PATH}")

然后咱们就可以简单地加载训练好的模型啦：

# 加载 tokenizer - 如果需要的话，确保使用 trust_remote_code=True
tokenizer = AutoTokenizer.from_pretrained(
    TRAINED_MODEL_PATH,
    trust_remote_code=True, # 如果你的模型配置需要
    padding_side="right" # 确保填充方向一致
)

# 如果保存或加载时没有正确设置 pad token，就设置一下
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 # 保持和训练时一致的数据类型
)

# 把加载好的模型移到你的设备上（如果有 GPU 的话）
trained_model.to(device) # 'device' 仍然是咱们之前用的 CUDA 设备

为了用它来进行推理：

# 测试训练好的模型的推理
def test_trained_model_inference(user_input: str):
    """测试加载的训练好的模型和 tokenizer 的推理。"""
    messages = [
        {"role": "system", "content": SYSTEM_PROMPT}, # 重新使用咱们的系统提示
        {"role": "user", "content": user_input}
    ]

    # 使用 tokenizer 应用聊天模板
    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 基础模型）。

研究人员发现 R1 Zero 模型在推理测试上表现得相当不错，甚至在像 AIME 2024 这样的任务上，得分和更高级别的模型（比如 OpenAI-01–0912）差不多。

这表明使用强化学习（RL）来鼓励语言模型进行推理是一种很有希望的方法。

但他们也发现 DeepSeek-R1-Zero 存在一些关键问题，需要解决才能用于实际应用和更广泛的研究。

R1 Zero 的问题

DeepSeek 的研究人员表示，这个模板是 故意设计得简单且结构化的 。它 没有 对 推理过程本身 施加任何 内容特定的 约束。例如，它没有说：

• “你 必须 使用分步推理”（它只是说“推理过程”，让模型自己定义那是什么意思）。
• “你 必须 使用反思推理”
• “你 必须 使用特定的问题解决策略”

主要问题是推理过程中的 <think> 标签里的内容很难读，让人很难跟着分析。

另一个问题是语言混杂，当被问到多语言问题的时候，模型有时候会在同一个回答里混杂不同的语言，导致输出内容不一致且令人困惑。

要是你用西班牙语问它问题。突然之间，它的“思考”就会变成 英语和西班牙语 的大杂烩，一点也不精致！这些问题，混乱的推理和语言混淆，就是明显的障碍。

这些就是他们把最初的 R1 Zero 模型变成 R1 的两个主要原因

为 SFT 准备冷启动数据

为了修复 R1 Zero 的问题，真正让 DeepSeek 推理正常工作，研究人员进行了 冷启动数据收集 并且加入了 监督式微调（SFT） 。

你可以把这想象成给模型打下一个良好的推理基础，然后才是真正激烈的 RL 训练。基本上，他们想让 DeepSeek-V3 基础模型 学会什么是良好的推理，以及如何清晰地展示它。

咱们之前看到的一个冷启动数据的例子是 Bespoke-Stratos-17k，接下来咱们会用它来创建 R1，但 咱们需要了解冷数据集是怎么创建的，这样咱们才不会在实际训练中遗漏任何部分 。

长 CoT 的少样本提示

其中一个技巧是 长推理链（CoT）的少样本提示 ，咱们试着给 DeepSeek-V3 基础模型（或者在咱们这儿，是 Qwen2–0.5B）展示几个问题和超级详细的分步解决方案的例子。这就是推理链（CoT）。

长 CoT

这个方法的目标就是让模型通过例子来学习，然后开始模仿这种彻底的推理风格。

对于咱们的例子问题 “What is 2 + 3 * 4?”（2 加 3 乘 4 等于多少？），咱们可以创建包含几个已解决的问题作为例子的提示。接下来咱们看看这在 Python 里怎么实现：

# 加载模型和 tokenizer
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")

# 生成长 CoT 回答
def generate_response(prompt_text):
    messages = [
        {"role": "system", "content": "You are a helpful assistant that provides step-by-step solutions."},
        {"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 是多少？
解决方案：|特殊标记| 首先，求 9 的平方根，结果是 3。然后，把 3 加上 5。3 加 5 等于 8。|特殊标记| 总结：答案是 8。

问题：火车以 60 英里/小时的速度行驶 2 小时，它走了多远？
解决方案：|特殊标记| 使用公式：距离 = 速度 × 时间。速度是 60 英里/小时，时间是 2 小时。距离 = 60 × 2 = 120 英里。|特殊标记| 总结：火车行驶了 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 是多少？
解决方案：|特殊标记| 首先，求 9 的平方根，
结果是 3。然后，把 3 加上 5。3 加 5 等于 8。
|特殊标记| 总结：答案是 8。

问题：火车以 60 英里/小时的速度行驶 2 小时，它走了多远？
解决方案：|特殊标记| 使用公式：距离 = 速度 × 时间。
速度是 60 英里/小时，时间是 2 小时。距离 = 60 × 2 = 120 英里。
|特殊标记| 总结：火车行驶了 120 英里。

问题：2 加 3 乘 4 等于多少？
解决方案：

模型回应（少样本 CoT）：
|特殊标记| 要解决 2 加 3 乘 4，咱们需要遵循运算顺序（PEMDAS/BODMAS）。乘法应该在加法之前进行。
第一步：3 乘 4 等于 12。
第二步：把第一步的结果 12 加上 2：2 + 12 = 14。
|特殊标记| 总结：答案是 14。

看看模型在看了示例之后，是怎么开始用 |特殊标记| 分隔符来结构化它的回答，并且提供分步推理，最后得出总结和最终答案的！

这就是少样本学习的力量，它引导模型朝着咱们想要的输出格式发展。

直接提示

另一种方法是 直接提示 。在这里，咱们直接告诉模型不仅要解决问题，还要明确地展示它的推理过程，一步一步来，然后验证它的答案。

这就是鼓励一种更审慎、更有条理的问题解决方式。

基于示例的学习

接下来，咱们来写一个针对 “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 等于多少？

模型回应（直接提示）：
|特殊标记| 推理：要解决 2 加 3 乘 4，我需要遵循运算顺序，它规定乘法应该在加法之前进行。
第一步：3 乘 4 等于 12。
第二步：把第一步的结果 12 加上 2：2 + 12 = 14。
验证：为了验证答案，我可以再次检查运算顺序和计算过程。乘法确实是先于加法进行的，并且计算过程也是正确的。
|特殊标记| 总结：答案是 14。

正如你所见，通过直接要求展示推理过程并且进行验证，模型提供了一个更全面的输出，包括了一个 “验证” 部分。

这种方法直接引导模型产生咱们想要的详细推理。

后处理精炼

最后一种技巧涉及到 后处理精炼 。有趣的是，他们甚至用到了已经训练好的 R1 Zero 模型的输出来进行这个操作！

尽管 R1 Zero 存在问题，但它还是能进行一定程度的推理。所以，他们把 R1 Zero 的输出拿过来，让人类标注员对它们进行精炼，让它们变得更清晰、更有结构，并且纠正其中的错误。

处理精炼

想象一下一个混乱的 R1 Zero 输出，就像这样：

<think> 嗯……先乘 3 和 4……得到 12……然后加上 2……</think>
<answer> 14 </answer>

人类标注员会把它精炼成更清晰、更有结构的内容，就像这样：

|特殊标记| 推理：要解决这个问题，咱们使用运算顺序，先进行乘法运算。
第一步：3 乘以 4，结果是 12。
第二步：把 2 加到第一步的结果上：2 + 12 = 14。
|特殊标记| 总结：答案是 14。

尽管咱们没办法完全用代码模拟人类的精炼过程，但咱们可以用一个简单的例子来展示如何在程序中对一个可能很混乱的输出进行重新格式化和结构化。

接下来，咱们用一个模拟的 “混乱” 输出来展示这个过程：

# 模拟的混乱 R1 Zero 输出
messy_output = "<think>  嗯……先乘 3 和 4……得到 12……然后加上 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"""|特殊标记| 推理：{think_content.replace('嗯……', '').strip().capitalize()}。
|特殊标记| 总结：答案是 {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> 嗯……先乘 3 和 4……得到 12……然后加上 2……</think>
<answer> 14 </answer>

精炼后的输出：|特殊标记| 推理：先乘 3 和 4……得到 12……然后加上 2……。
|特殊标记| 总结：答案是 14。

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

不过，它展示了核心思想：把初始模型的输出拿过来，提高它们的质量和结构，从而创造出更好的训练数据。

在生成了冷启动数据之后，接下来的关键步骤就是 监督式微调（SFT） ，接下来咱们就来探索这个阶段！

SFT 阶段 1 使用冷启动数据

为了生成合适的冷启动数据，从而用监督式微调（SFT）来构建 R1，咱们显然需要一个专业的团队，以及大量的代码，但幸运的是，咱们已经有了一个和冷启动形式相似的数据集（Bespoke-Stratos-17k）。

咱们需要了解 SFT 训练器在处理咱们的训练数据时，内部到底是怎么工作的？

SFT 是一种监督式学习的形式。这意味着咱们给模型输入和 期望的输出 的配对。

在咱们这儿，输入可能是一个问题提示，而期望的输出就是咱们训练数据集中的良好推理、分步解决方案。 希望这个点能让大家清楚地看到为什么需要冷数据。

它把咱们的标记化训练数据拿过来，然后一批一批地喂给模型。对于每一批，都会发生一系列重要的操作，接下来咱们来可视化这个内部过程：

SFT 工作流程

首先，模型接收一个输入，比如一个问题提示。它处理这个输入，然后生成它对解决方案的最佳猜测，一个标记一个标记地生成。这些就是 预测的标记 。

接下来，SFT 训练器需要知道这些预测到底有多好（或者多差）。它使用一个 损失函数 ，通常是交叉熵损失函数。这个函数从数学上把模型预测的标记和咱们训练数据中的 正确 标记进行比较。可以把它想象成在计算模型答案的 “误差” 。

这个 “误差” 并不是被直接扔掉的。它是学习的关键信号。通过一个叫做 反向传播 的过程，这个误差被用来计算 梯度 。梯度就像是指南，指出参数调整的方向，这样就能减少误差。

最后，一个 优化器 ，比如 AdamW，会用这些梯度来微妙地调整模型的内部设置 —— 它的参数。这些调整是为了让模型下次的预测离正确答案更近一点。

R1 的阶段 1 SFT 训练器配置

还记得 R1 Zero 的那些问题吗，推理过程混乱，语言混杂？SFT 的目的就是修复这些问题。通过在高质量、经过精炼的数据上进行训练，咱们是在教模型：

• 清晰的推理风格 ：把它的 “思考过程” 结构化，让它变得容易阅读和理解。
• 一致的语言 ：在一个回答里坚持使用一种语言，避免令人困惑的语言混杂。

咱们要用的 Bespoke-Stratos-17k 数据集，它有 17,000 道专注于数学和代码的问题，格式看起来很适合咱们的需求。

接下来，咱们快速回顾一下 Bespoke-Stratos-17k 的一个样本：

# 从 bespokelabs 加载 “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" # SFT 模型的新输出目录
os.makedirs(OUTPUT_DIR, exist_ok=True)

# 训练参数 —— 和 GRPO 类似，但针对 SFT 进行调整
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,        # 为 SFT 调整学习率
    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 训练循环

现在，咱们来加载咱们的数据集和 tokenizer：

# 加载 Bespoke-Stratos-17k 数据集
dataset_sft = load_dataset("HuggingFaceH4/Bespoke-Stratos-17k", split='train') # 为了简单起见，只用训练集

# 初始化 tokenizer —— 和之前一样
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，然后开始训练！

# 初始化基础模型用于 SFT —— 和之前一样
model_sft = AutoModelForCausalLM.from_pretrained(
    MODEL_NAME,
    trust_remote_code=True,
    torch_dtype=torch.bfloat16
)

# 初始化 SFT 训练器
sft_trainer = SFTTrainer(
    model=model_sft,                     # 咱们初始化的 Qwen 模型
    train_dataset=dataset_sft,           # Bespoke-Stratos-17k 数据集
    tokenizer=tokenizer,                 # tokenizer
    args=training_args,                  # 训练参数
    dataset_text_field="conversations",  # 数据集中包含文本的字段 —— 对于 SFT 很重要
    packing=True,                        # 启用数据打包
    max_seq_length=4096                 # 最大序列长度
)

# 开始 SFT 训练循环
sft_train_result = sft_trainer.train()

当你运行这段代码的时候，SFT 训练过程就开始啦。它的输出看起来和 GRPO 训练输出差不多，会在每个日志记录步骤显示损失和学习率。

...
INFO:__main__:步骤 10: 损失 = ..., 学习率 = ...
INFO:__main__:步骤 20: 损失 = ..., 学习率 = ...
...

和 GRPO 一样，训练时间取决于你的硬件和选择的轮数。既然咱们还是用的小模型，而且只用了一轮，所以这个例子应该会比较快完成。

保存微小的 R1 LLM

SFT 完成之后，咱们就可以保存咱们新微调好的模型（R1）啦。

# 保存训练好的 SFT 模型
TRAINED_SFT_MODEL_PATH = "data/Qwen-SFT-training" # 和 OUTPUT_DIR 一样

# 保存 tokenizer
tokenizer.save_pretrained(TRAINED_SFT_MODEL_PATH)

# 保存训练好的模型
sft_trainer.save_model(TRAINED_SFT_MODEL_PATH)

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

SFT 部分就到这里啦！咱们已经把基础模型拿过来，给它看了很多良好推理的例子，然后微调它，让它更擅长生成清晰、有结构的回应。

经过 SFT 阶段 1 微调的这个模型就是咱们所说的 R1。

SFT 之后的阶段，尤其是强化学习阶段和拒绝采样，从头开始用 Python 实现起来挺复杂的。专注于理论理解才是关键，这样才能理解整个过程。

推理导向的强化学习

经过 SFT 之后，模型的推理能力已经有所提升，但咱们还想 进一步聚焦于推理质量 ，并且修复语言混杂的问题。这个阶段会再次使用强化学习（RL），不过这次用的是一个更聪明的奖励系统。

这个新的奖励系统会检查模型的推理和答案是否和问题的语言一致。如果你用英语提问，那么整个回答都应该用英语。这就解决了语言混杂的问题。

推理导向的强化学习

它在准确率奖励的基础上，增加了一个 语言一致性奖励 ，确保 SFT 模型的推理和答案与输入问题的语言一致。GRPO 算法和 R1 Zero 的训练循环再次被使用，不过奖励信号经过了改进，专门针对更好的推理和一致的语言输出。

拒绝采样

为了得到高质量的推理数据，DeepSeek 使用了 拒绝采样 。可以把它想象成一个过滤器，用来保留最好的例子。

拒绝采样

模型会生成大量的推理例子。这些例子随后会根据正确性和推理质量（通常使用一个生成式奖励模型以及人工检查）进行评估。

只有 质量最高 的推理例子会被保留下来。将这些经过精炼的数据与非推理数据结合起来，就形成了一个用于 SFT 阶段 2 的改进数据集，进一步提升模型的推理能力和通用能力。

SFT 阶段 2 训练

强化学习的最后阶段旨在使模型成为一个在 所有情境 下都有帮助且无害的 AI 助手，而不仅仅是解决推理问题。这关系到与人类价值观的对齐。

关键焦点：有帮助性和无害性奖励

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

• 有帮助性 ：回答是否有用且信息丰富？
• 无害性 ：回答是否安全、无偏见且符合道德？

SFT 阶段 2

训练数据变得更加多样化，包括推理任务和人类偏好数据（哪个输出更好 —— 更有帮助性，更无害？）

奖励系统现在平衡了准确率与 有帮助性和无害性 。通过迭代的强化学习训练（很可能是 GRPO 再次上场），模型被优化为不仅仅擅长推理，而且还是一个安全且有帮助的通用 AI 助手，最终诞生了 DeepSeek R1。

知识蒸馏

为了让 DeepSeek R1 更易于使用，他们将它的知识 蒸馏 到更小的模型中。

蒸馏过程

蒸馏过程会把一个大型、强大的 “教师” 模型（DeepSeek R1）的知识传递给更小的 “学生” 模型。使用大量的推理样例数据，“教师” 模型的输出被用作 目标答案 。

然后，更小的模型通过监督式微调（SFT）来模仿这些输出。这样就能得到更小、更快速的模型，它们保留了 DeepSeek R1 大部分的推理能力，让它们更适合广泛使用。

完整

引https://github.com/FareedKhan-dev/train-deepseek-r1