機器學習 | 從0開發大模型之DeepSeek的GRPO

作者：周末程序猿 2025-04-03 15:40:41

DeepSeek-R1的發布為國產大模型爭光了（太強了），不過 GRPO 算法源自 DeepSeekMath 7B 模型，該模型在 MATH 基準測試中取得了優異成績。GRPO 是一種在線學習算法，核心思想是通過組內相對獎勵來估計基線，從而避免使用額外的價值函數模型。

DeepSeek-R1的發布為國產大模型爭光了（太強了），不過 GRPO 算法源自 DeepSeekMath 7B 模型，該模型在 MATH 基準測試中取得了優異成績，論文發表于2024年2月份：https://huggingface.co/papers/2402.03300，以下是該論文的摘要原文：

Mathematical reasoning poses a significant challenge for language models due to its complex and structured nature. In this paper, we introduce DeepSeekMath 7B, which continues pre-training DeepSeek-Coder-Base-v1.5 7B with 120B math-related tokens sourced from Common Crawl, together with natural language and code data. DeepSeekMath 7B has achieved an impressive score of 51.7% on the competition-level MATH benchmark without relying on external toolkits and voting techniques, approaching the performance level of Gemini-Ultra and GPT-4. Self-consistency over 64 samples from DeepSeekMath 7B achieves 60.9% on MATH. The mathematical reasoning capability of DeepSeekMath is attributed to two key factors: First, we harness the significant potential of publicly available web data through a meticulously engineered data selection pipeline. Second, we introduce Group Relative Policy Optimization (GRPO), a variant of Proximal Policy Optimization (PPO), that enhances mathematical reasoning abilities while concurrently optimizing the memory usage of PPO.

翻譯如下：

數學推理對語言模型構成了重大挑戰，因為其復雜且結構化的特性。在本文中，我們介紹了DeepSeekMath 7B，它在DeepSeek-Coder-Base-v1.5 7B的基礎上進行了繼續預訓練，使用了來自Common Crawl的120B與數學相關的標記，以及自然語言和代碼數據。DeepSeekMath 7B在競爭級MATH基準測試中取得了51.7%的優異成績，且未依賴外部工具包和投票技術，接近Gemini-Ultra和GPT-4的性能水平。DeepSeekMath 7B在64個樣本上的自一致性達到了60.9%的MATH成績。DeepSeekMath的數學推理能力歸因于兩個關鍵因素：首先，我們通過精心設計的數據選擇流程，充分利用了公開可用的網絡數據的巨大潛力。其次，我們引入了群體相對策略優化（GRPO），這是一種近端策略優化（PPO）的變體，旨在增強數學推理能力，同時優化PPO的內存使用。

對比數據

1、什么是GRPO

GRPO 是一種在線學習算法，核心思想是通過組內相對獎勵來估計基線，從而避免使用額外的價值函數模型。通過在訓練期間使用受訓模型自身生成的數據來迭代改進，GRPO 旨在最大化生成補全的優勢，同時確保模型保持接近參考策略，下圖是論文中的算法流程圖：

GRPO

GRPO 是 PPO (Proximal Policy Optimization，近端策略優化，是一種強化學習算法，由OpenAI于2017年提出，旨在解決策略梯度方法中的訓練不穩定問題) 的變體，主要區別是：

GRPO 省略 value function model
GRPO 獎勵計算，改成了一個 q 生成多個 r，然后 reward 打分

GRPO算法流程：

采樣一組輸出并計算每個輸出的獎勵
對組內獎勵進行歸一化處理
使用歸一化后的獎勵計算優勢函數
通過最大化目標函數更新策略模型
迭代訓練，逐步優化策略模型

論文中的偽代碼

2、獎勵設計

huggingface 庫提供 GRPOTrainer 可以直接使用 GRPO 訓練，參數包括定義獎勵模型和函數。

2.1 獎勵模型

trainer = GRPOTrainer(
    model="Qwen/Qwen2.5-3B-Instruct",
    reward_funcs="weqweasdas/RM-Gemma-2B",
    args=training_args,
    train_dataset=dataset,
    peft_cnotallow=LoraConfig(task_type="CAUSAL_LM"),
)

這里的 reward_funcs 參數可以傳入獎勵模型。

2.2 獎勵函數

GRPOTrainer 允許用戶自定義獎勵函數，通過定義一個返回浮點數列表的函數來實現。

獎勵函數的輸入：completions（生成的補全）和 prompts（提示）
獎勵函數的輸出：返回一個浮點數列表，每個浮點數代表對應于單個補全的獎勵

（1）較長補全獎勵函數

def completion_reward(completions, **kwargs):
    '''獎勵函數，對較長的補全給予更高的分數'''
    return [float(len(completion))/100 for completion in completions]

prompts = ["The sky is", "The sun is"]
completions = [" blue.", " in the sky."]
print("completion_reward: ", completion_reward(prompts=prompts, completinotallow=completions))

（2）格式正確獎勵函數

def format_reward(completions, **kwargs):
    '''格式獎勵'''
    pattern = r"<think>.*?</think>\s*<answer>.*?</answer>"
    responses = [completion[0]["content"] for completion in completions]
    matches = [re.match(pattern, response) for response in responses]
    return [0.5if match else0.0for match in matches]

prompts = [
    [{"role": "assistant", "content": "What is the result of (1 + 2) * 4?"}],
    [{"role": "assistant", "content": "What is the result of (3 + 1) * 2?"}],
]
completions = [
    [{"role": "assistant", "content": "<think>The sum of 1 and 2 is 3, which we multiply by 4 to get 12.</think><answer>(1 + 2) * 4 = 12</answer>"}],
    [{"role": "assistant", "content": "The sum of 3 and 1 is 4, which we multiply by 2 to get 8. So (3 + 1) * 2 = 8."}],
]
print("format_reward: ", format_reward(prompts=prompts, completinotallow=completions))

根據以上的獎勵樣例，可以設計對于不同數據集的獎勵函數，如：

判斷內容中是否包含數字
判斷內容回答是否參考網頁的知識庫內容
...

然后將這些函數傳入 GRPOTrainer 即可，代碼如下：

trainer = GRPOTrainer(
    model=model,
    processing_class=tokenizer,
    reward_funcs=[
        ...
        format_reward,
        completion_reward,
    ],
    args=training_args,
    train_dataset=data,
    ...
)

3、使用 `GRPO` 訓練模型

github上已經有很多復刻 DeepSeek-R1-Zero 的方案，有興趣可以看一下這幾個開源項目（成本基本都控制在500以內）：

https://github.com/datawhalechina/unlock-deepseek
https://github.com/Jiayi-Pan/TinyZero

3.1 訓練代碼

這里為了演示如何使用 GRPO 訓練模型，本文也給出了完整的訓練代碼，其中流程如下：

使用 Qwen/Qwen2.5-3B-Instruct 作為基礎模型
使用 swulling/gsm8k_chinese 作為訓練數據集

import re
from datasets import load_dataset
from transformers import AutoModelForCausalLM, AutoTokenizer
from peft import LoraConfig
from trl import GRPOConfig, GRPOTrainer

SYSTEM_PROMPT = """
按照如下格式生成：
<think>
...
</think>
<answer>
...
</answer>
"""

def process_data(data):
    return data.map(
        lambda x: {
            "prompt": [
                {"role": "system", "content": SYSTEM_PROMPT},
                {"role": "user", "content": x["question_zh-cn"]},
            ],
            "answer": x["answer_only"],
        }
    )

def extract_answer(text):
    answer = text.split("<answer>")[-1]
    answer = answer.split("</answer>")[0]
    return answer.strip()

def correctness_reward(completions, answer, **kwargs):
    responses = [completion[0]["content"] for completion in completions]
    extracted_responses = [extract_answer(r) for r in responses]
    return [2.0if response == str(ans) else0.0for response, ans in zip(extracted_responses, answer)]

def completion_reward(completions, **kwargs):
    '''獎勵函數，對較長的補全給予更高的分數'''
    return [float(len(completion)) / 100for completion in completions]

prompts = ["The sky is", "The sun is"]
completions = [" blue.", " in the sky."]
print("completion_reward: ", completion_reward(prompts=prompts, completinotallow=completions))

def digit_reward(completions, **kwargs):
    responses = [completion[0]["content"] for completion in completions]
    extracted_responses = [extract_answer(r) for r in responses]
    return [0.5if response.isdigit() else0.0for response in extracted_responses]

def format_reward(completions, **kwargs):
    '''格式獎勵'''
    pattern = r"<think>.*?</think>\s*<answer>.*?</answer>"
    responses = [completion[0]["content"] for completion in completions]
    matches = [re.match(pattern, response) for response in responses]
    return [0.5if match else0.0for match in matches]

prompts = [
    [{"role": "assistant", "content": "What is the result of (1 + 2) * 4?"}],
    [{"role": "assistant", "content": "What is the result of (3 + 1) * 2?"}],
]
completions = [
    [{"role": "assistant", "content": "<think>The sum of 1 and 2 is 3, which we multiply by 4 to get 12.</think><answer>(1 + 2) * 4 = 12</answer>"}],
    [{"role": "assistant", "content": "The sum of 3 and 1 is 4, which we multiply by 2 to get 8. So (3 + 1) * 2 = 8."}],
]
print("format_reward: ", format_reward(prompts=prompts, completinotallow=completions))

def mark_reward(completions, **kwargs):
    '''標記獎勵（改善格式獎勵稀疏問題）'''
    def mark_num(text):
        reward = 0
        if text.count("<think>\n") == 1:
            reward += 0.125

        if text.count("</think>\n") == 1:
            reward += 0.125

        if text.count("<answer>\n") == 1:
            reward += 0.125

        if text.count("</answer>\n") == 1:
            reward += 0.125 * 2

        return reward

    responses = [completion[0]["content"] for completion in completions]
    return [mark_num(response) for response in responses]

if __name__ == "__main__":
    model_name = "Qwen/Qwen2.5-3B-Instruct"
    model = AutoModelForCausalLM.from_pretrained(model_name, cache_dir="./model")
    model.cuda()
    tokenizer = AutoTokenizer.from_pretrained(model_name)
    ds = load_dataset("swulling/gsm8k_chinese", cache_dir="./dataset")
    data = process_data(ds["train"])
    output_dir = "output"
    training_args = GRPOConfig(
        output_dir=output_dir,
        learning_rate=5e-6,
        adam_beta1=0.9,
        adam_beta2=0.99,
        weight_decay=0.1,
        warmup_ratio=0.1,
        lr_scheduler_type="cosine",
        logging_steps=1,
        bf16=True,
        per_device_train_batch_size=1,
        gradient_accumulation_steps=4,
        num_generatinotallow=16,
        max_prompt_length=256,
        max_completion_length=200,
        num_train_epochs=1,
        save_steps=100,
        max_grad_norm=0.1,
        log_on_each_node=False,
        use_vllm=False,
        report_to="tensorboard",
    )

    trainer = GRPOTrainer(
        model=model,
        processing_class=tokenizer,
        reward_funcs=[
            mark_reward,
            format_reward,
            digit_reward,
            completion_reward,
            correctness_reward,
        ],
        args=training_args,
        train_dataset=data,
        peft_cnotallow=LoraConfig(task_type="CAUSAL_LM"),
    )
    trainer.train()
    trainer.save_model(output_dir)

3.2 輸出結果

（1）Qwen/Qwen2.5-3B-Instruct 輸出：

用戶: 
使用給定的數字 [44, 19, 35]，創建一個等于 98 的方程。你可以使用基本算術運算（+、-、*、/）一次或多次，但每個數字只能使用一次。在 <think> </think> 標簽中展示你的思考過程，并在 <answer> </answer> 標簽中返回最終方程，例如 <answer> (1 + 2) / 3 </answer>。在 <think> 標簽中逐步思考。

機器人:  
<span class="think">首先我們需要找到如何將給定的數字組合起來得到 98。我們可以嘗試不同的組合，比如加法和減法。注意到 98 可以分解為 90 和 8 的和，因此我們可以考慮用 44 和 35 來構造這兩個數。</span>
<span class="think">觀察到 44 和 35 的差是 9，如果我們將 35 減去這個差，就可以得到 8。即 35 - (44 - 35) = 8。</span>
<span class="think">現在我們有了 8 和 90，可以將它們相加得到 98。所以最終的方程是 35 + (44 - 35)。</span>
<answer> 35 + (44 - 35) </answer>

（2）訓練的模型輸出：

用戶: 
使用給定的數字 [44, 19, 35]，創建一個等于 98 的方程。你可以使用基本算術運算（+、-、*、/）一次或多次，但每個數字只能使用一次。在 <think> </think> 標簽中展示你的思考過程，并在 <answer> </answer> 標簽中返回最終方程，例如 <answer> (1 + 2) / 3 </answer>。在 <think> 標簽中逐步思考。

機器人:  
<think>
首先，我需要使用數字44、19和35，通過基本算術運算（+、-、*、/）來構造一個等于98的方程，且每個數字只能使用一次。

我先考慮將較大的數字進行加法或乘法運算，以達到接近98的目標。例如，44加19等于63，再加上35可以得到98。具體計算如下：

44 + 19 = 63  
63 + 35 = 98  

因此，最終的方程可以表示為44 + 19 + 35 = 98。
</think>

<answer> 44 + 19 + 35 = 98 </answer>