强化学习算法 GRPO：原理与实战详解

PPO 近端策略优化 (Proximal Policy Optimization, PPO) 是一种 Actor-Critic (行动者 - 评论家) 强化学习算法，通过最大化以下目标函数来优化 LLM： $$\mathcal{J}{\text{PPO}}(\theta) = \mathbb{E}[q \sim P(Q), o \sim \pi{\theta_{\text{old}}}(O|q)] \frac{1}{|o|} \sum_{t=1}^{|o|} \min \left[ \frac{\pi_\theta(o_t|q, o_{<t})}{\pi_{\theta_{\text{old}}}(o_t|q, o_{<t})} A_t, \text{clip} \left( \frac{\pi_\theta(o_t|q, o_{<t})}{\pi_{\theta_{\text{old}}}(o_t|q, o_{<t})}, 1-\epsilon, 1+\epsilon \right) A_t \right]$$

其中，$\pi_\theta$ 和 $\pi_{\theta_{\text{old}}}$ 分别是当前和旧的策略模型，$q$ 和 $o$ 分别是从问题数据集和旧策略 $\pi_{\theta_{\text{old}}}$ 中采样的问题和输出。$\varepsilon$ 是 PPO 中引入的用于稳定训练的 clipping（裁剪）相关超参数。$A_t$ 是优势函数（Advantage），它通过应用广义优势估计（GAE）计算得出。

GRPO GRPO 放弃了价值模型，转而从分组得分中估计基线，从而显著减少了训练资源。对于每个问题 $q$，GRPO 从旧策略 $\pi_{\theta_{\text{old}}}$ 中采样一组输出 ${o_1, o_2, \cdots, o_G}$，然后通过最大化以下目标函数来优化策略模型： $$\mathcal{J}{\text{GRPO}}(\theta) = \mathbb{E}[q \sim P(Q), {o_i}{i=1}^G \sim \pi_{\theta_{\text{old}}}(O|q)] \frac{1}{G} \sum_{i=1}^G \frac{1}{|o_i|} \sum_{t=1}^{|o_i|} \left{ \min \left[ \frac{\pi_\theta(o_{i,t}|q, o_{i,<t})}{\pi_{\theta_{\text{old}}}(o_{i,t}|q, o_{i,<t})} \hat{A}{i,t}, \text{clip} \left( \frac{\pi\theta(o_{i,t}|q, o_{i,<t})}{\pi_{\theta_{\text{old}}}(o_{i,t}|q, o_{i,<t})}, 1-\epsilon, 1+\epsilon \right) \hat{A}{i,t} \right] - \beta D{KL} \left[ \pi_\theta | \pi_{\text{ref}} \right] \right}$$

其中，$\varepsilon$ 和 $\beta$ 是超参数，$\hat{A}_{i,t}$ 是基于仅在每个组内的输出的相对奖励计算的优势函数。

GRPO 通过 '组内相对比较' 和 '去价值网络' 两大创新，将强化学习从依赖绝对价值估计转变为基于群体统计的排序优化，在保持 PPO 稳定性的同时显著降低计算成本。（这一思想也可视为对 对比学习 （Contrastive Learning）和 排序学习（Learning-to-Rank）在 RLHF 领域的巧妙迁移）

三、算法原理

GRPO 的工作原理，可分为四个主要步骤：生成响应、计算优势值、估计 KL 散度、计算损失。

3.1 生成响应（Generating completions）

对于每个问题 $q$，从旧策略模型 $\pi_{\theta_{\text{old}}}$ 中采样一组输出 ${o_1, o_2, \cdots, o_G}$。

3.2 计算优势值（Computing the advantage）

针对每组生成的响应序列，通过奖励模型计算其奖励值，生成对应的 $G$ 个奖励值 $\mathbf{r} = {r_1, r_2, \cdots, r_G}$。

接着，对这些奖励进行标准化处理：

$$\hat{A}_{i,t} = \frac{r_i - \text{mean}(\mathbf{r})}{\text{std}(\mathbf{r})}$$

优势函数的作用就是计算某一个输出的 token 数值相对于平均输出的优劣势。如果某一个输出的奖励高于平均值，则结果是正的，反之低于平均值，结果是负的。这样策略模型会更倾向于生成那些具有较高奖励的输出。

这种基于组内相对比较的方法在组内引入竞争机制，推动模型生成更好的回答，也正是 GRPO 命名的由来。

3.3 估计 KL 散度（Estimating the KL divergence）

KL 散度用于衡量当前新策略 $\pi_\theta$ 和参考策略 $\pi_{\text{ref}}$ 之间的分布差异，目标是限制 $\pi_\theta$ 和 $\pi_{\text{ref}}$ 偏离太远。使用 Schulman 等人（2020 年）引入的近似器进行估计。该近似器定义如下：

$$D_{KL} \left[ \pi_\theta | \pi_{\text{ref}} \right] = \frac{\pi_{\text{ref}}(o_{i,t} | q, o_{i,<t})}{\pi_\theta(o_{i,t} | q, o_{i,<t})} - \log \frac{\pi_{\text{ref}}(o_{i,t} | q, o_{i,<t})}{\pi_\theta(o_{i,t} | q, o_{i,<t})} - 1$$

3.4 计算损失（Computing the loss）

目标是在最大化优势值的同时约束模型接近参考策略，因此损失函数定义为：

$$\mathcal{L}{\text{GRPO}}(\theta) = -\frac{1}{G} \sum{i=1}^G \frac{1}{|o_i|} \sum_{t=1}^{|o_i|} \left[ \underbrace{\min\left( \frac{\pi_\theta(o_{i,t} \mid q, o_{i,<t})}{\pi_{\theta_{\text{old}}}(o_{i,t} \mid q, o_{i,<t})} \hat{A}{i,t}, \text{clip}\left( \frac{\pi\theta(o_{i,t} \mid q, o_{i,<t})}{\pi_{\theta_{\text{old}}}(o_{i,t} \mid q, o_{i,<t})}, 1-\epsilon, 1+\epsilon \right) \hat{A}{i,t} \right)}{\text{裁剪策略梯度项}} - \underbrace{\beta D_{\text{KL}} \left[ \pi_\theta | \pi_{\text{ref}} \right]}_{\text{KL 散度惩罚项}} \right]$$

公式最左边，$t$ 对应token-level 优势，即一个句子中，每个 token 对应的优势是一样的。$G$ 表示每组生成的响应数量，最终需要取平均。$|o_i|$ 表示第 $i$ 个响应的序列长度（Token 数量），$\frac{1}{|o_i|}$ 表示对每一个生成回答的所有的 token 取平均。

裁剪策略梯度项表示缩放后的优势，其中 Clip 函数用于限制新策略与旧策略之间的比值，通过将比值限制在 $1-\epsilon, 1+\epsilon$ 之间，防止新策略相对于旧策略的数值有较大的更新变动。这种机制确保策略更新处于控制范围内，稳定性更强，收敛性更好。公式表示为： $$\text{clip} \left( \frac{\pi_\theta(o_{i,t} \mid q, o_{i,<t})}{\pi_{\theta_{\text{old}}}(o_{i,t} \mid q, o_{i,<t})}, 1-\epsilon, 1+\epsilon \right) \hat{A}_{i,t}$$

KL 散度惩罚项惩罚偏离参考策略的行为，确保策略更新的稳定性并保留通用推理能力。$\beta$ 表示的是惩罚系数。公式表示为： $$\beta D_{KL} \left( \pi_\theta | \pi_{\text{ref}} \right)$$

• 当 $D_{KL}$ 变大时，惩罚增加，迫使策略更新更加保守，不要偏离太远。
• 当 $\beta$ 变大的时候，表示在原有的优势上增加惩罚，让模型更新幅度变小，更加保守。

3.5 重要性采样*（Importance Sampling）

GRPO 算法的最终的优化目标是让新策略在旧策略的样本上，尽可能多地保留高奖励的输出。重要性采样是其核心机制之一，通过复用旧策略（参考策略）生成的样本来优化新策略，从而提升训练效率和样本利用率。

重要性采样的本质 重要性采样（Importance Sampling, IS）是一种通过从易采样的分布估计难采样分布的期望的方法。其核心思想是使用重要性权重来修正采样的偏差。

例如，我们想计算某个难采样分布下的期望： $$\mathbb{E}{x \sim p{\text{hard}}}[f(x)] = \int f(x)p_{\text{hard}}(x)dx$$ 那我们就可以从另一个易采样的分布进行采样，使用重要性采样估计，表示为： $$\mathbb{E}{x \sim p{\text{hard}}}[f(x)] = \mathbb{E}{x \sim p{\text{easy}}}\left[\frac{p_{\text{hard}}(x)}{p_{\text{easy}}(x)} f(x)\right]=\int f(x)\frac{p_{\text{hard}}(x)}{p_{\text{easy}}(x)}p_{\text{easy}}(x)dx$$

其中，$\frac{p_{\text{hard}}(x)}{p_{\text{easy}}(x)}$ 就是重要性权重，用于修正分布差异。

GRPO 中的重要性采样 在 GRPO 中，训练目标是最大化新旧策略的比值乘以 Advantage 的数值： $$\frac{\pi_\theta(o_{i,t} | q, o_{i,<t})}{\pi_{\theta_{\text{old}}}(o_{i,t} | q, o_{i,<t})} \hat{A}_{i,t}$$

其中，新旧策略的比值 $\frac{\pi_\theta}{\pi_{\theta_{\text{old}}}}$ 就是重要性采样中的重要性比值。GRPO 通过重要性采样与裁剪机制（Clipping）结合，限制策略更新幅度，从而防止过度优化导致模型崩溃。

四、代码示例

4.1 设计奖励函数

在实际使用 GRPO（Group Relative Policy Optimization）训练时，设计奖励函数的核心思路需要围绕强化学习的目标和任务特性展开，同时兼顾训练稳定性和计算效率。以下整理出一些关键设计原则：

1. 核心任务导向：奖励函数必须直接反映任务的核心目标。例如：
   • 数学推理：正确答案给予最高奖励。
   • 格式要求：对符合特定结构（如 <reasoning>...</reasoning><answer>...</answer>）的响应给予额外奖励。
   • 逻辑严谨性：通过关键词（如 'sum'、'unique'）或推理长度间接评估逻辑质量。
2. 多目标权衡：
   • 若任务需要平衡多个子目标（如正确性 + 格式 + 推理深度），可通过加权求和或分层奖励实现。
3. 稀疏 vs 稠密奖励：
   • 稀疏奖励（如仅对最终答案打分）需结合组内相对比较提取有效信号。
   • 稠密奖励（如对每一步推理打分）更适合复杂任务，但需注意奖励膨胀问题。
4. 归一化与标准化：
   • 组内奖励需归一化（如 Z-score）以消除不同提示（prompt）间的尺度差异。
   • 对原始奖励进行裁剪（如限制到 [-5, 5]）防止极端值干扰训练。
5. 可解释性：
   • 每个奖励分项（如格式、正确性、关键词）应清晰定义，便于调试和调整权重。

以下是一个示例：

import re

def reward_correct_answer(responses, answers, **kwargs) -> list[float]:
    """根据答案的正确性给予奖励。"""
    pattern = r"<answer>.*?\\boxed{([A-D])}.*?</answer>"
    rewards = []
    for resp, target in zip(responses, answers):
        match = re.search(pattern, resp, re.DOTALL)
        if match and match.group(1) == target:
            rewards.append(2.0) # 正确选项，高奖励
        else:
            rewards.append(0.0) # 错误或无法提取，无奖励
    return rewards

def reward_strict_format(responses, **kwargs) -> list[float]:
    """奖励符合严格格式 <reasoning>...</reasoning><answer>...</answer> 的回答"""
    pattern = r"^<reasoning>\n.*?\n</reasoning>\n<answer>\n\\boxed{[A-D]}\n</answer>$"
    return [1.0 if re.fullmatch(pattern, r.strip(), re.DOTALL) else 0.0 for r in responses]

def reward_soft_format(responses, **kwargs) -> list[float]:
    """奖励符合宽松格式 <reasoning>...</reasoning><answer>...</answer> 的回答"""
    pattern = r"<reasoning>.*?</reasoning>.*?<answer>.*?[A-D].*?</answer>"
    return [0.5 if re.search(pattern, r, re.DOTALL) else 0.0 for r in responses]

def count_xml(text: str) -> float:
    """根据 XML 标签的出现次数和位置给予分数"""
    score = 0.0
    if text.count("<reasoning>\n") == 1:
        score += 0.125
    if text.count("\n</reasoning>\n") == 1:
        score += 0.125
    if text.count("\n</answer>") == 1:
        score += 0.125
    score -= max(0, (len(text.split("\n</answer>")[-1]) - 1)) * 0.001
    return max(0.0, score)

def xmlcount_reward_func(responses, **kwargs) -> list[float]:
    """使用 count_xml 函数计算每个回答的分数。"""
    return [count_xml(r) for r in responses]

def reward_reasoning_length(responses, **kwargs) -> list[float]:
    """ 根据推理部分的长度给予奖励。
     - 长度小于 20 个单词，无奖励。
     - 长度在 20 到 50 个单词之间，奖励 0.25。
     - 长度在 50 到 100 个单词之间，奖励 0.5。
     - 长度超过 100 个单词，奖励 0.75。
     """
    rewards = []
    for r in responses:
        score = 0.0
        match = re.search(r"<reasoning>(.*?)</reasoning>", r, re.DOTALL)
        if match:
            content = match.group(1).strip()
            length = len(content.split())
            if length < 20:
                score += 0.0
            elif length < 50:
                score += 0.25
            elif length < 100:
                score += 0.5
            else:
                score += 0.75
        rewards.append(max(0.0, score))
    return rewards

def reward_reasoning_keywords(responses, **kwargs) -> list[float]:
    """ 根据推理部分是否包含特定关键词给予奖励。
     每命中一个关键词，奖励 0.1 分，上限为 0.5 分。
     """
    key_terms = ["row", "column", "sum", "unique", "digit", "no repeat", "clue"]
    rewards = []
    for r in responses:
        match = re.search(r"<reasoning>(.*?)</reasoning>", r, re.DOTALL)
        if not match:
            rewards.append(0.0)
            continue
        content = match.group(1).lower()
        hits = sum(1 for k in key_terms if k in content)
        rewards.append(min(hits * 0.1, 0.5)) # 命中多个关键词，奖励上限 0.5
    return rewards

# 合并所有奖励函数
def combined_reward_func(prompts, completions, response, **kwargs) -> list[float]:
    responses = completions
    pattern = r"^<reasoning>\n.*?\n</reasoning>\n<answer>\n\\boxed{([A-D])}\n</answer>$"
    answers = [re.search(pattern, answer, re.DOTALL).group(1) for answer in response]
    rca = reward_correct_answer(responses, answers)
    rstf = reward_strict_format(responses)
    rsof = reward_soft_format(responses)
    xrf = xmlcount_reward_func(responses)
    rrl = reward_reasoning_length(responses)
    rrk = reward_reasoning_keywords(responses)
    # 返回每个回答的总奖励
    return [rca[i] + rstf[i] + rsof[i] + xrf[i] + rrl[i] + rrk[i] for i in range(len(responses))]

4.2 GRPO 训练

0. 安装 trl 库

git clone https://github.com/huggingface/trl.git
cd trl/
pip install -e .

1. 加载训练数据和模型

from transformers import AutoModelForCausalLM, AutoTokenizer
from trl import GRPOConfig, GRPOTrainer
import torch
from datasets import Dataset
import json

with open("grpo_data.json", 'r') as file:
    data = json.load(file)
dataset = Dataset.from_list(data)
model_dir = "xxx"
model = AutoModelForCausalLM.from_pretrained(
    model_dir,
    torch_dtype=torch.bfloat16,
    device_map="auto",
)
tokenizer = AutoTokenizer.from_pretrained(model_dir)

2. 配置 GRPO 训练参数 关键参数说明：

• num_generations: 每个 prompt 生成 4 个候选答案（用于奖励计算）
• max_completion_length: 生成文本最大长度（8192 tokens）
• gradient_accumulation_steps: 4 步累积梯度，模拟 batch_size=4

更多参数说明详见：https://huggingface.co/docs/trl/main/en/grpo_trainer

training_args = GRPOConfig(
    learning_rate=5e-6,
    adam_beta1=0.9,
    adam_beta2=0.99,
    weight_decay=0.1,
    warmup_ratio=0.1,
    lr_scheduler_type="cosine",
    optim="paged_adamw_8bit",
    logging_steps=10,
    per_device_train_batch_size=1,
    gradient_accumulation_steps=4, # 增加到 4 以获得更平滑的训练
    num_generations=4, # 如果内存不足，请减少
    max_prompt_length=2048,
    max_completion_length=8192,
    num_train_epochs=1, # 设置为 1 以进行完整的训练
    save_steps=200,
    max_grad_norm=0.1,
    logging_dir="output_grpo/logs",
    output_dir="output_grpo",
)

3. 初始化 GRPO 训练器

trainer = GRPOTrainer(
    model=model,
    reward_funcs=combined_reward_func,
    args=training_args,
    train_dataset=dataset,
)

4. 启动训练

trainer.train()

五、总结

GRPO（群组相对策略优化）是一种高效的大语言模型强化学习微调算法，通过组内相对比较和去价值网络设计，显著降低了传统 PPO 的计算开销。其核心创新在于利用组内响应奖励的归一化比较替代绝对价值估计，结合 KL 散度惩罚和策略裁剪机制确保训练稳定性。该算法特别适合数学推理、代码生成等复杂任务，能在单卡中等规模模型上高效运行。

GRPO 的优势包括计算效率高、适配人类偏好数据特性、训练过程稳定等，但也存在对组大小敏感、长序列优化等挑战。实践建议优先保证核心目标奖励权重，结合渐进式训练策略，适用于资源受限但需要高质量微调的场景，为 RLHF 提供了一种轻量化解决方案。