用PyTorch從零構建 DeepSeek R1：模型架構和分步訓練詳解

DeepSeek R1 的完整訓練流程核心在于，在其基礎模型 DeepSeek V3 之上，運用了多種強化學習策略。
本文將從一個可本地運行的基礎模型起步，并參照其技術報告，完全從零開始構建 DeepSeek R1，理論結合實踐，逐步深入每個訓練環節。通過可視化方式，由淺入深地解析 DeepSeek R1 的工作機制。

本文的代碼可在github上獲得，并且我將英文的注釋翻譯成了中文，項目文件結構：

 train-deepseek-r1/├── code.ipynb         # Jupyter Notebook 代碼實現├── requirements.txt   # 依賴庫列表└── r1_for_dummies.md  # 面向非技術受眾的 DeepSeek R1 解釋

環境配置

首先，克隆代碼倉庫并執行以下命令安裝必要的庫：

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

接下來，導入所需的 Python 庫：

 # 導入必要的庫
import logging
import os
import sys
import re
import math
from dataclasses import dataclass, field
from typing import List, Optional# 導入 PyTorch 與deep hub 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 deep—hub) 庫
from trl import (AutoModelForCausalLMWithValueHead,PPOConfig,PPOTrainer,GRPOTrainer,GRPOConfig,SFTTrainer
)# 導入數學相關工具庫
from latex2sympy2_extended import NormalizationConfigfrom math_verify import LatexExtractionConfig, parse, verify

訓練數據集

盡管 DeepSeek R1 的技術報告未明確指定強化學習預訓練的初始數據集，但根據其目標，我們推斷數據集應側重于推理能力。

為盡可能貼近 DeepSeek R1 的復現，本文采用以下兩個開源的 Hugging Face 推理數據集：

NuminaMath-TIR：用于 R1 Zero 階段的訓練。
Bespoke-Stratos-17k：用于 R1 階段的訓練。

NuminaMath-TIR 數據集由 DigitalLearningGmbH 發布，包含 7 萬個數學問題，

messages

列詳細記錄了解題過程的思維鏈 (Chain-of-Thought, 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 數據集由 bespokelabs 提供，包含 1.7 萬個問題，專注于數學和代碼相關的推理任務。

以下是 Bespoke-Stratos 數據集的樣本示例：

 # 從 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 ... deep hub'}]
}
##### 輸出 ####

數據集的選擇并非局限于上述兩個，您可以根據需求選用其他數據集，但需確保數據集側重于推理能力，即包含問題及其詳細的逐步解答。

DeepSeek R1 訓練流程概覽

在深入技術細節之前，先對 DeepSeek R1 的訓練流程進行簡要概述。DeepSeek R1 并非從零開始訓練，而是基于 DeepSeek 團隊已有的強大語言模型 DeepSeek-V3。為了進一步提升模型的推理能力，DeepSeek 團隊采用了強化學習方法。

強化學習 (Reinforcement Learning, RL) 的核心思想是：當語言模型在推理任務中表現出色時，給予獎勵；反之，則施以懲罰。

DeepSeek R1 的訓練并非單一的訓練過程，而是一個多階段的復雜流程，可稱之為訓練管線。首先DeepSeek 團隊進行了純粹的 強化學習 嘗試，旨在探索推理能力是否能夠自發涌現，這一階段產出了 DeepSeek-R1-Zero 模型，可視作一次探索性實驗。對于 正式的 DeepSeek-R1 模型，訓練流程被進一步細化和組織。訓練管線包含多個階段，包括預訓練數據準備、強化學習訓練、數據迭代和多輪強化學習等步驟，如同模型能力逐級提升的過程。

整個訓練流程的核心目標是顯著提升語言模型的問題分析和深入思考能力。

以上是對 DeepSeek R1 訓練流程的高度概括，后續章節將深入剖析每個訓練階段的具體細節。

基礎模型選型

DeepSeek 團隊選用 DeepSeek-V3 作為 R1 Zero 和 R1 的基礎模型。然而，DeepSeek-V3 模型規模龐大，模型體積高達 685 GB 💀，這對個人開發者而言顯然難以企及。

為降低實驗門檻，本文選用規模更小的基礎模型 Qwen/Qwen2.5–0.5B-Instruct (模型體積 0.9 GB)。若你擁有更充裕的 GPU 內存，可考慮加載更大規模的模型，如 Qwen/Qwen2.5–7B-Instruct。

以下是所選基礎模型 Qwen2.5–0.5B-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 未設置deephub，則指定 pad token 為 eos token
if tokenizer.pad_token is None:tokenizer.pad_token = tokenizer.eos_tokenprint(f"Vocabulary size: {len(tokenizer)}")
print(f"Model max length: {tokenizer.model_max_length}")
print(f"Pad token: {tokenizer.pad_token}")print(f"EOS token: {tokenizer.eos_token}")

#### 輸出 ####
Vocabulary size: 151665
Model max length: 131072
Pad token: <|endoftext|>
EOS token: <|im_end|>
#### 輸出 ####

上述代碼展示了模型的基礎信息。接下來，我們查看基礎模型的參數量：

 # 初始化基礎模型
model = AutoModelForCausalLM.from_pretrained(MODEL_NAME,trust_remote_code=True,torch_dtype=torch.bfloat16
)print(f"Model parameters: {model.num_parameters():,}")

#### 輸出 ####

Model parameters: 494,032,768
#### 輸出 ####

模型參數量約為 0.5B。為驗證模型的基本推理能力，我們測試一個簡單請求并打印模型的響應：

# 檢查 CUDA 是否可用
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print(f"Using device: {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: {test_input}")
print(f"Model Response: {response}")

#### 輸出 ####
"Test Input: how are you?
Model Response: As an AI language model I dont have feelings ..."
##### 輸出 ####

結果表明，即使是小規模的 Qwen2.5–0.5B-Instruct 模型，其輸出也具備一定的可靠性，可以作為 DeepSeek R1 仿制模型訓練的基礎模型。

強化學習 (RL) 框架中的策略模型 ?

在選定基礎模型之后，我們需要理解強化學習 (RL) 的基本框架如何應用于訓練大型語言模型 (LLM)。

DeepSeek R1 的訓練起點是 DeepSeek V3 基礎模型，而本文實踐則選用 Qwen2.5–0.5B-Instruct。此處的“起點”指的是，DeepSeek 團隊首先利用強化學習構建了 R1 Zero 的初始版本，該版本在最終 R1 版本之前存在一些缺陷。

R1 Zero 的初始版本采用強化學習進行訓練，其中 DeepSeek V3 或本文的 Qwen2.5–0.5B-Instruct 模型充當強化學習智能體 (Agent)，執行決策動作。下圖展示了其基本工作流程：

強化學習智能體 (Qwen2–0.5B) 接收到環境 (Environment) 輸入的問題，并采取行動 (Action)，即生成針對該問題的答案和推理過程。此處的“環境”即為推理任務本身。

執行行動后，環境將返回獎勵 (Reward)。獎勵信號是對智能體行動質量的反饋，告知基礎模型 (Qwen2–0.5B) 其行動的優劣程度。正向獎勵表示模型行動有效，可能得到了正確答案或進行了合理的推理。此反饋信號反向傳遞至基礎模型，幫助模型學習并調整未來的行動策略，以期獲得更高的累積獎勵。

R1 Zero 的 GRPO 算法

上一節介紹了強化學習的基本流程。現在我們要深入了解 DeepSeek R1-Zero 模型所采用的具體強化學習算法：GRPO (Gradient Reward Policy Optimization)。

目前存在多種強化學習算法，但傳統方法通常依賴于 “評論家” (Critic) 模型來輔助主決策模型（即“行動者” Actor，此處為 DeepSeek-V3/Qwen2-0.5B）。評論家模型通常與行動者模型具有相近的規模和復雜度，導致計算成本成倍增加。

DeepSeek 選擇了 GRPO 算法訓練 R1 Zero 模型。GRPO 的獨特之處在于，它能夠直接從一組行動結果中推導出基線 (Baseline)，作為評估行動優劣的參考標準。因此GRPO 算法無需額外的評論家模型，顯著降低了計算開銷，提高了訓練效率。

下圖展示了 GRPO 算法在 R1 Zero 訓練中的應用流程，隨后將對流程圖進行詳細解讀。

DeepSeek GRPO 算法與 Qwen2–0.5B 基礎模型的集成運作方式如下：

首先，問題輸入 (A) 被饋送至 Qwen 模型 (B)。Qwen 模型嘗試通過 生成補全 ? 過程，給出問題的答案。最終的 補全輸出 (D) 包含在

<think>

標簽中的推理步驟以及

<answer>

標簽中的最終答案。

隨后，問題輸入 (A) 和 標準答案 (E) 一并輸入 獎勵函數 (F)，獎勵函數充當智能評分系統的角色。這些函數將 Qwen 模型的 補全輸出 (D) 與標準答案進行比對，并從多個維度進行評估，包括：

準確性 (Accuracy)：答案是否在數學上正確？
格式 (Format)：是否規范使用了 <think>和 <answer>標簽？
推理步驟 (Reasoning Steps)：推理邏輯是否清晰可循？
余弦縮放 (Cosine Scaling)：響應內容是否精煉簡潔？
重復懲罰 (Repetition Penalty)：是否存在不必要的重復內容？

上述評估過程產生 獎勵分數 (G)，并將其傳遞給 GRPO 訓練器 (H)。訓練器利用獎勵分數，通過梯度反向傳播來調整 Qwen 模型 (B) 的參數，優化模型生成答案的方式。此過程被稱為 梯度獎勵策略優化，因為它利用梯度、獎勵反饋 和 策略調整 來優化 Qwen 模型的響應，從而最大化模型性能。

最后，經過參數更新的 Qwen 模型 (B) 會再次接受新問題的測試，通過迭代循環不斷優化自身。隨著訓練的持續進行，Qwen 模型的問題解決能力將得到持續提升。

Prompt 模板

本文沿用 DeepSeek R1 Zero 模型 GRPO 算法所采用的思考型 Prompt 模板，具體定義如下：

# 基于 GRPO 訓練的 DeepSeek 系統 Prompt
SYSTEM_PROMPT = ("A conversation between User and Assistant. The user asks a question,  \and the Assistant solves it. The assistant ""first thinks about the reasoning process in the mind and  \then deephub provides the user with the answer. The reasoning ""process and answer are enclosed within <think> </think>  \and <answer> </answer> tags, respectively, i.e., ""<think> reasoning process here </think><answer> answer here </answer>"
)

此 系統 Prompt 旨在告知基礎模型 (Qwen2–0.5B) 其角色定位為樂于助人的助手，需要在給出答案之前進行逐步推理。

<think>

** 和 **

<answer>

標簽** 的作用是規范模型輸出的結構，將內部推理過程與最終答案區分開，以便于后續的評估和獎勵計算。

訓練數據預處理

系統 Prompt 設置完成后，下一步需要依據 Prompt 模板轉換訓練數據。

首先，定義

make_conversation

函數，用于構建對話數據格式：

# 構建訓練數據結構的函數
def make_conversation(example):"""將數據集樣本轉換為對話格式。"""return {"prompt": [{"role": "system", "content": SYSTEM_PROMPT},{"role": "user", "content": example["problem"]},],}

該函數接收數據集中的

problem

列的值，并返回一個字典，其中包含系統 Prompt 和用戶提出的問題，構成一個對話樣本。接下來，定義

load_math_dataset

函數，用于加載數據集并進行預處理：

# 加載并準備數據集
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"Train set size: {len(dataset['train'])}")
print(f"Test set size: {len(dataset['test'])}")

#### 輸出 ####
Train set size: 72441
Test set size: 99
#### 輸出 ####

數據集已劃分為訓練集和測試集。在進行后續訓練步驟之前，需要驗證數據集的格式是否符合要求（例如，檢查是否存在用戶/助手對話）。定義

validate_dataset

函數進行數據校驗：

def validate_dataset(dataset):"""對數據集執行基礎驗證檢查。"""# 定義數據集所需字段required_fields = ["problem", "prompt"]# 遍歷數據集的 'train' 和 'test' 劃分for split in ['train', 'test']:print(f"\nValidating {split} split:")# 從數據集中獲取列名fields = dataset[split].column_names# 檢查是否缺少必要字段missing = [field for field in required_fields if field not in fields]if missing:print(f"Warning: Missing fields: {missing}")  # 若缺少字段，則發出警告else:print("? All required fields present")  # 確認所有必要字段均存在# 獲取數據集劃分的首個樣本sample = dataset[split][0]# 提取包含對話消息列表的 'prompt' 字段messages = sample['prompt']# 驗證 Prompt 格式:# - 至少包含兩條消息# - 首條消息 Role 為 'system'# - 次條消息 Role 為 'user'if (len(messages) >= 2 andmessages[0]['role'] == 'system' andmessages[1]['role'] == 'user'):print("? Prompt format is correct")  # 確認 Prompt 格式正確else:print("Warning: Incorrect prompt format")  # 若 Prompt 格式不正確，則發出警告# 驗證數據集
validate_dataset(dataset)

執行上述代碼，得到如下輸出：

Validating train split:? All required fields present
? Prompt format is correctValidating test split:? All required fields present
? Prompt format is correct

輸出結果顯示，訓練數據集已成功通過格式驗證 🙌，表明數據集已成功轉換為滿足訓練要求的格式。

獎勵函數

如 GRPO 算法章節所述，模型答案的評估將通過五個不同的獎勵函數進行：

準確性 (Accuracy)：答案是否在數學上正確？
格式 (Format)：是否規范使用了 <think>和 <answer>標簽？
推理步驟 (Reasoning Steps)：推理邏輯是否清晰可循？
余弦縮放 (Cosine Scaling)：響應內容是否精煉簡潔？
重復懲罰 (Repetition Penalty)：是否存在不必要的重復內容？

接下來，我們將逐一實現這五個獎勵函數。

準確性獎勵函數

準確性獎勵函數的設計思路相對直觀，但代碼實現略顯復雜。該獎勵函數的目的是驗證模型給出的答案是否在數學上等價于標準答案。

若模型答案在數學上正確，則獎勵值為 1.0；若答案錯誤，則獎勵值為 0.0。對于無法解析的標準答案，為避免不公平懲罰，將給予 0.5 的中性獎勵。

以下是準確性獎勵函數的 Python 代碼實現：

def accuracy_reward(completions, solution, **kwargs):"""獎勵函數，用于檢查模型的響應是否在數學上等價于標準答案。使用 deep hub 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.0reward = float(verify(answer_parsed, gold_parsed))else:# 若標準答案解析失敗，則給予中性獎勵 0.5reward = 0.5print("Warning: Failed to parse gold solution:", sol) # 警告：無法解析標準答案rewards.append(reward)return rewards

該函數的核心邏輯在于驗證模型響應與標準答案在數學上的 等價性。具體步驟如下：

使用 latex2sympy2工具將標準答案轉換為結構化的數學表達式。
若標準答案解析失敗，則給予中性獎勵 0.5。
提取模型輸出，并進行歸一化處理，以提高評估的魯棒性。
利用 math_verify工具，比對解析后的模型響應與解析后的標準答案是否在數學上一致。
若數學上一致，則獎勵 1，否則獎勵 0。

該獎勵函數確保了準確性評估并非基于簡單的文本相似度，而是基于 真實的數學正確性。

格式獎勵函數

格式獎勵函數旨在確保模型遵循指令，并按照預定義的結構化格式輸出結果。此前，我們已要求模型將推理過程置于 **

<think>

** 標簽內，并將最終答案置于 **

<answer>

** 標簽內。格式獎勵函數的功能正是檢查模型是否嚴格遵守了這一格式約定。

若模型正確使用了

<think>

和

<answer>

標簽，則給予獎勵值 1；若模型未能遵循格式規范，則獎勵值為 0。該獎勵機制旨在引導模型關注并遵守預設的輸出結構。

以下是格式獎勵函數的代碼實現：

# 實現格式獎勵函數
def format_reward(completions, **kwargs):"""獎勵函數，用于檢查模型輸出是否符合預定義的格式：<think>...</think>deep hub <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.0return [1.0 if match else 0.0 for match in matches]

該函數的具體實現邏輯如下：

定義正則表達式 (regex) 模式。該模式精確描述了期望的輸出格式：以 <think>開頭，<think>和 </think>標簽對之間可包含任意字符，隨后是空白字符，然后以 <answer>開頭，<answer>和 </answer>標簽對之間可包含任意字符，并以此結尾。
從每個模型的輸出結果中提取文本內容。
使用 re.match函數，逐一檢查模型輸出內容是否與定義的正則表達式模式完全匹配。re.DOTALL標志使正則表達式中的 .能夠匹配換行符，re.MULTILINE標志使 ^和 $能夠匹配整個字符串的起始和結束位置，而非僅限于行首和行尾。
對于符合格式規范的模型輸出，獎勵值設為 1；反之，設為 0。該獎勵機制對格式的正確性采取嚴格的二元評價標準。

推理步驟獎勵函數

推理步驟獎勵函數的設計更具策略性。其目標是鼓勵模型展現其 “思考過程”，即獎勵模型輸出中包含的、類似于推理步驟的成分。

該函數通過識別模型輸出中常見的逐步推理指示性詞匯和結構，例如：

“步驟 1”、“步驟 2” 等序號型步驟標識；
“1.”、“2.” 等數字編號列表；
“-” 或 “*” 等項目符號列表；
“首先”、“其次”、“然后”、“最后” 等過渡性連接詞。

模型輸出中包含的上述指示性成分越多，獲得的獎勵越高。這種獎勵機制類似于對模型“展示解題步驟”的行為進行加分。

以下是實現推理步驟獎勵函數的代碼：

def reasoning_steps_reward(completions, **kwargs):r"""獎勵函數，用于鼓勵模型進行清晰的逐步推理。該函數會檢測諸如 "Step 1:"、編號列表、項目符號以及過渡詞等模式。"""# 用于匹配推理步驟指示符的正則表達式模式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]

該函數定義了一個相對復雜的正則表達式模式，用于識別前述的推理步驟指示性成分。

函數使用

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 個推理步驟，方可獲得滿額獎勵。您可以根據實際需求調整此數值，以鼓勵模型輸出更多或更少的推理步驟。

余弦縮放獎勵函數

余弦縮放獎勵函數的設計思路更具技巧性。其核心目標是鼓勵模型在給出正確答案時盡可能簡潔，并在答案錯誤時，對較長的錯誤答案給予 相對較輕的懲罰。

其背后的邏輯是：

對于正確答案：我們更傾向于獎勵簡潔、直接的解答，而非冗長、散漫的答案。簡明扼要且正確的答案通常更佳。
對于錯誤答案：相較于嘗試進行推理的較長錯誤答案，簡短的錯誤答案可能更不可取。因此，我們希望對簡短的錯誤答案施加更重的懲罰，而對較長的錯誤答案施加相對較輕的懲罰。

以下是實現余弦縮放獎勵的代碼：

# 實現余弦縮放獎勵函數
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_correctmax_value = max_value_correctelse: # 答案錯誤min_value = max_value_wrong  # 注意此處交換了 min_value 和 max_valuemax_value = min_value_wrong# 余弦縮放公式reward = min_value + 0.5 * (max_value - min_value) * (1.0 + cosine)rewards.append(float(reward))return rewardsreturn 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

，則采用正確答案的獎勵值范圍；否則，采用錯誤答案的獎勵值范圍，但需注意交換

min_value

和

max_value

的值，以實現對較長錯誤答案施加較輕懲罰的效果。

重復懲罰獎勵函數

重復懲罰獎勵函數旨在抑制模型生成重復性內容。我們期望模型能夠生成新穎、多樣的推理過程和答案，而非簡單地重復使用相同的詞語序列。

該獎勵函數通過懲罰模型在輸出文本中過度重復使用相同的 n-gram 序列來實現上述目標。在本文示例中，我們采用 n-gram 的大小為 3 (trigrams，即三元詞組)，您可以根據需要調整 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} should not be positive")def zipngram(text: str, ngram_size: int):"""輔助函數，用于從文本中生成 n-gram。"""words = text.lower().split() # 轉換為小寫并按空格分割為單詞列表return zip(*[words[i:] for i in range(ngram_size)]) # 生成 n-gramdef 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)continueif len(completion.split()) < ngram_size: # 對于過短的輸出，不施加懲罰rewards.append(0.0)continuengrams = set() # 使用集合存儲唯一的 n-gramtotal = 0for ng in zipngram(completion, ngram_size): # 生成 n-gramngrams.add(ng) # 將 n-gram 添加到集合 (重復的 n-gram 會被自動忽略)total += 1 # 統計 n-gram 的總數量# 計算縮放因子：重復程度越高 -> 縮放因子越大scaling = 1 - len(ngrams) / totalreward = scaling * max_penalty # 基于縮放因子施加懲罰rewards.append(reward)return rewardsreturn 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)])

方法高效提取 n-gram。

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

：

# 為 GRPO 腳本參數定義 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

這兩個獎勵函數的行為，用戶可根據具體情況調整獎勵機制。

接下來，我們使用

transformers

庫提供的

TrainingArguments

類。

TrainingArguments

是一個核心配置類，幾乎控制著訓練過程的方方面面。

# 從 transformers 庫定義 TrainingArguments
training_args = TrainingArguments(output_dir=OUTPUT_DIR,          # 檢查點和日志輸出目錄overwrite_output_dir=True,num_train_epochs=1,             # 訓練的總 epoch 數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,              # 預熱步數比例weight_decay=0.01,             # 權重衰減系數，應用于除 bias 和 LayerNorm 權重外的所有層logging_steps=10,              # 日志記錄頻率 (步數)evaluation_strategy="steps",    # 評估策略：每 `eval_steps` 步進行評估eval_steps=50,                 # 評估頻率 (步數)save_strategy="steps",         # 模型保存策略：每 `save_steps` 步保存模型save_steps=50,                 # 模型保存頻率 (步數)save_total_limit=2,            # 最大 checkpoint 保存數量，超出限制則刪除舊 checkpointdataloader_num_workers=2,      # 數據加載器 worker 數量seed=42,                       # 隨機種子，用于保證實驗可復現bf16=True,                     # 啟用混合精度 BF16 訓練push_to_hub=False,             # 是否將模型推送至 Hugging Face Hubgradient_checkpointing=True,   # 啟用梯度檢查點report_to="none",              # 不使用任何報告工具
)

最后，需要定義

ModelConfig

類。

ModelConfig

用于配置與 模型自身 相關的參數，例如，指定預訓練模型名稱、數據類型 (如 bfloat16)、是否信任遠程代碼等。

@dataclass
class ModelConfig:"""模型配置類。"""model_name_or_path: str = field(default=MODEL_NAME, metadata={"help": "預訓練模型路徑或 Hugging Face Model Hub 模型標識符"})model_revision: Optional[str] = field(default="main", metadata={"help": "指定模型版本 (分支名, tag 名 或 commit id)"})torch_dtype: Optional[str] = field(default="bfloat16", metadata={"help": "覆蓋默認 torch_dtype，以指定 dtype 加載模型"})trust_remote_code: bool = field(default=True, metadata={"help": "加載模型和 tokenizer 時，信任遠程代碼"})attn_implementation: Optional[str] = field(default="flash_attention_2", metadata={"help": "選擇 Attention 實現方式, 可選 '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"

選項用于啟用 FlashAttention 2，在硬件支持的情況下，可潛在地加速訓練過程。

接下來，實例化上述配置類，以便在后續代碼中使用：

# 實例化配置對象
script_args = GRPOScriptArguments()
model_args = ModelConfig()

然后，我們需要獲取獎勵函數列表，以及在訓練過程中使用的“回調函數” (callbacks)。

回調函數類似于助手，在訓練過程的不同階段執行特定任務，例如記錄訓練進度、保存模型等。目前，我們僅使用一個簡單的日志記錄回調函數。

將獎勵函數集中管理：

# 實用函數，根據腳本參數獲取獎勵函數列表
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"Reward function '{func_name}' not found in registry.")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"Step {state.global_step}: Loss = {state.log_history[-1].get('loss', None)}, Learning Rate = {state.log_history[-1].get('learning_rate', None)}")def get_callbacks(training_args, model_args, script_args):"""返回訓練過程中使用的回調函數列表。目前僅包含 LoggingCallback。您可以擴展此列表以添加更多回調函數。"""callbacks = [LoggingCallback()] # 實例化 LoggingCallbackreturn callbacks

最后，初始化獎勵函數和回調函數：

# 獲取獎勵函數和回調函數
reward_functions = get_reward_functions(script_args)
callbacks = get_callbacks(training_args, model_args, script_args)

GRPO 訓練循環

本節將啟動 GRPO 訓練的核心引擎。初始化

GRPOTrainer

，并將之前準備好的所有組件傳入，包括模型、獎勵函數、訓練參數、數據集和回調函數。

初始化

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:dee phub__main__:Step 10: Loss = ..., Learning Rate = ...INFO:deeph ub __main__:Step 20: Loss = ..., Learning Rate = ......

訓練時長取決于硬件配置和設定的 epoch 數。由于本文示例中

num_train_epochs

僅設置為 1，且模型規模較小，因此訓練過程相對迅速。

然而，在實際的 DeepSeek R1 Zero GRPO 訓練中，通常需要進行更多 epoch 和步數的訓練。

保存 Tiny 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, # 如果模型配置需要，則設置為 Truepadding_side="right" # 確保 padding 方向一致
)# 若 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, # 如果模型架構需要，則設置為 Truetorch_dtype=torch.bfloat16 # 保持與訓練時一致的數據類型
)# 將加載的模型移至指定設備 (如有 GPU 可用，則移至 GPU)
trained_model.to(device) # 'device' 變量仍為之前定義的 CUDA 設備

進行推理測試：

# 使用訓練后的模型進行推理測試
def test_trained_model_inference(user_input: str):"""使用加載的訓練后模型和 tokenizer 進行推理測試。"""messages = [{"role": "system", "content": SYSTEM_PROMPT}, # 復用之前的系統 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)# 使用 *trained_model* 生成模型輸出outputs = trained_model.generate(**inputs,max_new_tokens=200, # 相比之前，可以生成稍長的文本do_sample=True,temperature=0.7)# 將生成的 token 解碼為文本response = tokenizer.decode(outputs[0], skip_special_tokens=True)return response

R1 Zero 的主要問題

至此Qwen2–0.5B 基礎模型（而非 DeepSeek R1 原始的 DeepSeek V3 基礎模型）完成了 R1 Zero 模型的訓練。

盡管本文的訓練模型可能無法直接體現 R1 Zero 的問題，但 DeepSeek 團隊的研究人員發現，R1 Zero 模型在推理測試中表現出色，在 AIME 2024 等任務上的得分甚至與 OpenAI-01–0912 等更先進的模型相近。

這驗證了使用強化學習 (RL) 鼓勵語言模型進行推理是一種有潛力的方向。

然而，DeepSeek 團隊也注意到 DeepSeek-R1-Zero 存在一些關鍵問題，需要解決這些問題才能使其真正應用于實際場景和更廣泛的研究領域。

DeepSeek 團隊的研究人員指出，R1 Zero 的 Prompt 模板是 有意設計得簡潔且側重于結構，避免對 推理過程本身 施加任何 內容特定的約束。例如，Prompt 模板并未明確要求：

“你必須使用逐步推理”（僅使用了 “推理過程” 這一寬泛表述，將推理方式的定義權交由模型自身）。
“你必須使用反思性推理”。
“你必須采用特定的問題解決策略”。

R1 Zero 模型的主要問題在于，**

<think>

標簽內的推理過程可讀性較差**，人類難以理解和分析模型的推理思路。

另一個問題是 語言混合。當用戶以多語言提問時，模型有時會在同一回答中混用多種語言，導致輸出結果不一致且混亂。

例如，當用戶以西班牙語提問時，模型在

<think>

標簽內的“思考”過程可能會出現 英語和西班牙語混雜 的情況，輸出質量欠佳。這些問題，即推理過程的混亂和語言的混用，成為了 R1 Zero 模型進一步發展的阻礙。

正是上述兩個主要問題，促使 DeepSeek 團隊將初始的 R1 Zero 模型迭代升級為 R1 模型。

為 SFT 準備冷啟動數據

為解決 R1 Zero 模型存在的問題，并使 DeepSeek R1 模型具備更完善的推理能力，DeepSeek 團隊開展了 冷啟動數據收集工作，并引入了監督微調 (Supervised Fine-Tuning, SFT) 技術。

可以將冷啟動數據收集理解為，在進行高強度的強化學習訓練之前，為模型奠定良好的推理基礎。其核心目標是，讓 DeepSeek-V3 Base 模型（或本文示例中的 Qwen2–0.5B 模型）學習何為高質量的推理，以及如何清晰地呈現推理過程。

基于長 CoT 的少樣本 Prompting

基于長思維鏈 (CoT) 的少樣本 Prompting 是一種有效的數據構建技術。該技術的核心思想是，向 DeepSeek-V3 Base 模型（或本文示例中的 Qwen2–0.5B 模型）展示少量問題示例，并為每個問題配備極其詳盡的、逐步分解的解答，即長思維鏈 (Long Chain-of-Thought, Long CoT)。

該技術的目的是使模型通過學習示例，模仿這種詳盡的推理風格，并逐步掌握高質量推理的模式。

以問題 “2 + 3 * 4 等于多少？” 為例，我們可以構建包含少量已解答問題的 Prompt 示例。以下 Python 代碼展示了如何實現基于長 CoT 的少樣本 Prompting：

# 加載模型和 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": "dee phub 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() # 提取助手的響應

為提出的問題定義少樣本 Prompt 示例：

# 示例問題及其解答 (使用 | special_token | 作為分隔符)
few_shot_prompt = """
Problem: dee ph ub What's the square root of 9 plus 5?
Solution: <|special_token|> First, find the square root of 9, which is 3. Then, add 5 to 3.  3 + 5 equals 8. <|special_token|> Summary: The answer is 8.Problem: Train travels at 60 mph for 2 hours, how far?
Solution: <|special_token|> Use the formula: Distance = Speed times Time. Speed is 60 mph, Time is 2 hours. Distance = 60 * 2 = 120 miles. <|special_token|> Summary: Train travels 120 miles.Problem: What is 2 + 3 * 4?
Solution:
"""

使用基礎模型進行少樣本生成：

# 使用少樣本示例，為目標問題生成響應
target_problem_prompt = few_shot_prompt + "What is 2 + 3 * 4?"
model_response_few_shot = generate_response(target_problem_prompt)print("Few-shot Prompt:")
print(target_problem_prompt)
print("\nModel Response (Few-shot CoT):")
print(model_response_few_shot)

模型輸出結果如下，呈現出結構化的數據格式：

Few-shot Prompt:
Problem: What's the square root of 9 plus 5?
Solution: <|special_token|> First, find the square root of 9,which is 3. Then, add 5 to 3.  3 + 5 equals 8.<|special_token|> Summary: The answer is 8.Problem: Train travels at 60 mph for 2 hours, how far?
Solution: <|special_token|> Use the formula: Distance = Speed times Time.Speed is 60 mph, Time is 2 hours. Distance = 60 * 2 = 120 miles.<|special_token|> Summary: Train travels 120 miles.Problem: What is 2 + 3 * 4?
Solution:Model Response (Few-shot CoT):
<|special_token|> To solve 2 + 3 * 4, we need to follow the orderof operations (PEMDAS/BODMAS). Multiplication should be performedbefore addition.
Step 1: Multiply 3 by 4, which equals 12.
Step 2: Add 2 to the result from Step 1: 2 + 12 = 14.
<|special_token|> Summary: The answer is 14.

可以看到，在學習了少量示例后，模型開始采用

<|special_token|>

分隔符來組織答案，并提供逐步推理過程，最終給出總結和最終答案。

這便是少樣本學習的強大之處，它能夠引導模型學習并生成期望的輸出格式。

直接 Prompting

直接 Prompting 是另一種數據構建方法。與少樣本 Prompting 不同，直接 Prompting 側重于直接指示模型，不僅要解決問題，還要明確地展示逐步推理過程，并對答案進行驗證。

直接 Prompting 的目標是鼓勵模型采取更審慎、更周全的問題解決策略。

以下代碼展示了如何為問題 “2 + 3 * 4 等于多少？” 構建 Prompt，并顯式要求模型進行推理和驗證：

# 直接 prompting 示例
direct_prompt_text = """
Problem: d ee p hub Solve this, show reasoning step-by-step, and verify:
What is 2 + 3 * 4?
"""model_response_direct = generate_response(direct_prompt_text)print("Direct Prompt:")
print(direct_prompt_text)
print("\nModel Response (Direct Prompting):")
print(model_response_direct)

直接 Prompting 的輸出結果清晰易懂，如下所示：

Direct Prompt:
Problem: Solve this, show reasoning step-by-step, and verify:
What is 2 + 3 * 4?Model Response (Direct Prompting):
<|special_token|> Reasoning: To solve 2 + 3 * 4, I need to followthe order of operations, which states that multiplication shouldbe done before addition.
Step 1: Multiply 3 by 4, which equals 12.
Step 2: Add 2 to the result from Step 1: 2 + 12 = 14.
Verification: d ee p hub To verify the answer, I can double-check theorder of operations and the calculations. Multiplication isindeed performed before addition, and the calculations are correct.
<|special_token|> Summary: The answer is 14.

如輸出所示，通過直接要求模型進行推理和驗證，模型生成了更為全面的輸出結果，其中包括 “Verification” (驗證) 部分。

直接 Prompting 方法能夠有效地引導模型生成用戶期望的、包含詳細推理過程的解答。

后處理優化

最后一種數據構建技術是 后處理優化。值得注意的是，DeepSeek 團隊甚至使用了已訓練完成的 R1 Zero 模型的輸出結果進行后處理優化。

即使 R1 Zero 模型存在一些問題，但其已具備一定的推理能力。因此，DeepSeek 團隊收集了 R1 Zero 模型的輸出，并由人工標注員對這些輸出進行精細化處理，使其更清晰、結構更規整，并糾正其中的錯誤。

例如，對于如下 R1 Zero 模型生成的、略顯粗糙的輸出：

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

人工標注員會將其優化為更清晰、格式更規范的輸出：

<|special_token|> Reasoning: d ee p hub To solve this, we use order of operations, doing multiplication before addition.
Step 1: Multiply 3 by 4, which is 12.
Step 2: Add 2 to the result: 2 + 12 = 14.
<|special_token|> Summary: The answer is 14.

在代碼層面，我們難以完美地模擬人工優化過程。但我們可以演示一種基本思路，即如何以編程方式對潛在的粗糙輸出進行重新格式化和結構化。

以下代碼以模擬的 “粗糙” 輸出為例，展示如何對其進行優化：

# 模擬的 R1 Zero 模型粗糙輸出
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|> Reasoning: {think_content.replace('umm...', '').strip().capitalize()}.
<|special_token|> Summary: The answer is {answer_content}."""return refined_textrefined_output_text = refine_output(messy_output)print("Messy Output (Simulated R1 Zero):")
print(messy_output)
print("\nRefined Output:")
print(refined_output_text)

代碼輸出結果如下：

Messy Output (Simulated R1 Zero):
<think>  ummm... multiply 3 and 4... get 12... then add 2...</think>
<answer> 14 </answer>Refined Output:
<|special_token|> Reasoning: Multiply 3 and 4... get 12... then add 2.
<|special_token|> Summary: The answer is 14.

示例中的

refine_output

函數僅為演示基本優化思路。真實的人工優化過程遠比代碼示例復雜，涉及到對推理步驟更細致的理解和更正。

然而，代碼示例已展現了后處理優化的核心思想：對模型的初始輸出進行質量提升和結構優化，從而構建更高質量的訓練數據。

在生成冷啟動數據之后，下一個關鍵步驟是 監督微調 (SFT)，我們將在下一節詳細探討 SFT 的訓練過程。

基于冷啟動數據的階段 1 SFT 訓練

為利用監督微調技術 (SFT) 構建 R1 模型，并生成高質量的冷啟動數據，我們需要投入專業團隊和大量的代碼開發工作。幸運的是，我們已經擁有了與冷啟動數據形式相近的數據集 (Bespoke-Stratos-17k)。

為了深入理解 SFT 的訓練機制，我們需要了解 SFT 訓練器在處理訓練數據時，其內部執行了哪些操作？

SFT 屬于監督學習范疇。這意味著，我們需要向模型提供成對的輸入和 期望的 輸出。

在本文的場景中，輸入可以是問題 Prompt，期望的輸出則是來自訓練數據集的、包含良好推理過程的逐步解答。希望這一點能夠清晰地闡釋冷啟動數據存在的必要性。

SFT 訓練器接收分詞后的訓練數據，并以批次 (batch) 的形式輸入模型。針對每個批次，訓練器會執行一系列關鍵操作。下圖展示了 SFT 訓練的內部流程：

首先，模型接收輸入，例如問題 Prompt。模型處理該輸入，并逐 token 生成其對問題解答的最佳預測，這些預測的 token 即為 預測 token。

接下來，SFT 訓練器需要評估模型預測的質量。評估過程依賴于 損失函數，通常采用交叉熵損失函數 (Cross-Entropy Loss)。損失函數從數學層面比較模型的預測 token 與訓練數據中正確的 token，計算模型答案的 “誤差”。

計算得到的 “誤差” 并不會被直接丟棄，而是作為模型學習的關鍵信號。通過名為 反向傳播 (backpropagation) 的過程，誤差信號被用于計算梯度 (gradients)。梯度類似于指南針，指示模型參數調整的方向，模型參數沿著梯度方向調整后，可有效降低預測誤差。

最后，優化器 (optimizer)，如 AdamW，利用計算得到的梯度，對模型的內部參數進行細微調整。這些調整旨在使模型在下一次預測時，輸出結果更接近標準答案。

R1 階段 1 SFT 訓練配置

R1 Zero 模型在推理過程的清晰度和語言一致性方面存在不足。SFT 訓練旨在解決這些問題。通過使用高質量、優化后的數據進行訓練，SFT 能夠引導模型：

學習清晰的推理風格：以易于理解和追溯的方式組織 “思考” 過程。
保持語言一致性：在單個回復中堅持使用單一語言，避免語言混用造成的困擾。

本文采用 Bespoke-Stratos-17k 數據集進行 SFT 訓練。如前文所述，該數據集包含 1.7 萬個數學和代碼相關的問題，其數據格式與我們的訓練需求高度契合。

回顧 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 ...'}]
}
#### 輸出 ####

該數據集包含系統 Prompt 和用戶-助手對話，非常適合用于指導模型學習如何進行包含推理過程的對話。

本文再次使用

trl

庫，該庫極大簡化了 SFT 訓練的流程。

首先，配置 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,         # 根據需要調整 epoch 數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

這兩個參數對于提升長序列 SFT 訓練的效率至關重要。

階段 1 SFT 訓練循環

加載數據集和 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,                 # Tokenizerargs=training_args,                  # 訓練參數dataset_text_field="conversations",  # 數據集中包含文本的字段 - SFT 訓練的關鍵參數packing=True,                        # 啟用數據打包max_seq_length=4096                 # 最大序列長度
)# 啟動 SFT 訓練循環
sft_train_result = sft_trainer.train()

執行上述代碼后，SFT 訓練過程將開始運行。訓練日志輸出與 GRPO 訓練類似，將在每個日志記錄步驟顯示損失值 (loss) 和學習率 (learning rate)。

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

與 GRPO 訓練類似，SFT 訓練時長取決于硬件配置和設定的 epoch 數。由于本文示例仍采用小規模模型，且僅訓練 1 個 epoch，因此訓練過程應相對快速。

保存 Tiny 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 訓練環節完成。我們已利用基礎模型，并向其展示了大量高質量推理示例，通過微調使其更擅長生成清晰、結構化的響應。

經過階段 1 SFT 訓練微調的模型，即為本文所稱的 R1 模型。

SFT 之后的訓練步驟，特別是強化學習階段和拒絕采樣策略，從零開始使用 Python 實現較為復雜。理解其背后的理論原理是把握整個訓練流程的關鍵。

面向推理的強化學習

SFT 訓練后的模型推理能力得到提升，但為進一步聚焦推理質量，并徹底解決語言混用問題，DeepSeek 團隊在后續階段再次采用了強化學習，并設計了更精細化的獎勵系統。

新的獎勵系統會檢查模型輸出的推理過程和答案是否與用戶提問時使用的語言保持一致。例如，若用戶使用英語提問，則模型整個回復（包括推理和答案）都應使用英語。這有效解決了語言混用問題。

在準確性獎勵的基礎上，DeepSeek 團隊 引入了語言一致性獎勵，以確保 SFT 模型在推理和回答時，所用語言與輸入語言保持一致。R1 Zero 階段使用的 GRPO 算法和訓練循環被復用，但獎勵信號得到改進，更精準地引導模型生成高質量推理結果和語言風格一致的輸出。

拒絕采樣

為獲得更高質量的推理數據，DeepSeek 團隊采用了 拒絕采樣 (Rejection Sampling) 策略。拒絕采樣可被視為一個過濾器，用于篩選并保留質量最佳的訓練樣本。

模型首先生成大量的推理示例，隨后，對這些示例的正確性和推理質量進行評估 (評估過程通常結合生成式獎勵模型和人工評估)。

只有質量最佳的、高質量推理示例才會被保留。DeepSeek 團隊將這些高質量推理示例與非推理數據相結合，構建了精細化的數據集，并用于 階段 2 SFT 訓練，進一步提升模型的推理能力和通用能力。

階段 2 SFT 訓練

最終的強化學習階段，目標是將模型訓練為在所有場景下都樂于助人且安全的 AI 助手，而不僅限于解決推理問題。該階段側重于模型與人類價值觀的對齊。

核心關注點：助人性和無害性獎勵

在獎勵機制設計上，不僅關注答案的準確性，還引入了以下獎勵指標：

助人性 (Helpfulness)：模型回復是否實用且信息量豐富？
無害性 (Harmlessness)：模型回復是否安全、無偏見且符合倫理道德？

訓練數據變得更加多樣化，不僅包含推理任務，還融入了人類偏好數據 (用于指示哪個輸出更佳——更助人，更無害)。

獎勵系統在準確性、助人性和無害性 之間尋求平衡。通過迭代強化學習訓練 (很可能仍采用 GRPO 算法)，模型得到持續優化，最終不僅擅長推理，更成長為一個安全、樂于助人、可廣泛應用的 AI 助手，即 DeepSeek R1 模型。

知識蒸餾

為使 DeepSeek R1 模型更易于部署和應用，DeepSeek 團隊采用了 知識蒸餾 (Distillation) 技術，將其知識遷移到規模更小的模型中。

知識蒸餾技術將大型、高性能的 “教師” 模型 (DeepSeek R1) 的知識，轉移到規模較小的 “學生” 模型。DeepSeek 團隊構建了包含大量推理示例的數據集，并將 DeepSeek R1 模型的輸出作為 目標答案。

隨后，使用監督微調 (SFT) 技術訓練小規模模型，使其模仿 DeepSeek R1 模型的輸出。最終，得到規模更小、推理速度更快的模型，這些 “學生” 模型繼承了 DeepSeek R1 模型推理能力的重要部分，使其更具實用價值，可應用于更廣泛的場景。

總結

本文深入剖析了 DeepSeek R1 模型的構建過程，從基礎模型選型到多階段訓練流程，再到關鍵技術如強化學習、拒絕采樣和知識蒸餾的應用，進行了詳盡的闡述。通過對 GRPO 算法、Prompt 模板、獎勵函數以及 SFT 訓練等核心環節的逐步解析，我們不僅了解了 DeepSeek R1 如何從零開始構建，更對其在推理能力、語言一致性以及安全助人等方面所做的努力有了更深刻的認識。希望本文能夠幫助讀者更好地理解 DeepSeek R1 的技術原理，并為相關研究和實踐提供有益的參考。

https://avoid.overfit.cn/post/ac6d4be0a234412ea00032737365638c

作者：FareedKhan