GRPO 强化学习训练指南

技能文档

---
name: majiayu000-grpo
title: GRPO 强化学习训练指南
category: 学术研究
description: 面向大模型人类反馈强化学习（RLHF）的 GRPO 训练指南：覆盖 GRPOTrainer 配置、奖励函数设计（含基于 token 的思考感知奖励模式）、策略优化与 KL 散度约束的稳定训练，以及奖励劫持、训练不稳定等问题的排查。当用户需要进行 GRPO/RLHF 训练、设计奖励函数、微调推理模型，或提到 GRPOTrainer、策略优化、KL 惩罚时触发。触发词：GRPO、RLHF 训练、奖励函数、强化学习对齐。
---

# Group Relative Policy Optimization（GRPO）

## 概述

GRPO 是一种用于大模型对齐的强化学习方法。它为每个提示词生成多个补全，用奖励函数打分，再通过相对策略梯度优化策略，使其偏好高奖励的回复。本技能同时包含训练思考型/推理型模型的奖励模式。

## 速查表

| 组件 | 用途 |
|------|------|
| `GRPOTrainer` | 用于策略优化的强化学习训练器 |
| `GRPOConfig` | 训练超参数 |
| `reward_funcs` | 用于打分的奖励函数 |
| `completion_ids` | 传入奖励函数的 token ID（无需重新分词） |
| `beta` | KL 惩罚系数（典型值 0.1） |
| `num_generations` | 每个提示词的补全数量（2-4） |
| `learning_rate` | 1e-5（比 SFT 低 10 倍） |
| Token ID 151668 | Qwen3-Thinking 模型的 `</think>` 边界 |

## 关键环境设置

```python
import os
from dotenv import load_dotenv
load_dotenv()

# Force text-based progress in Jupyter
os.environ["TQDM_NOTEBOOK"] = "false"

# CRITICAL: Set BEFORE importing unsloth/TRL
os.environ['ACCELERATE_MIXED_PRECISION'] = 'bf16'
```

## 关键导入顺序

```python
# CRITICAL: Import unsloth FIRST for proper TRL patching
import unsloth
from unsloth import FastLanguageModel, is_bf16_supported

# Then TRL imports
from trl import GRPOConfig, GRPOTrainer
from datasets import Dataset
import torch
```

**警告**：在导入之后再设置 `ACCELERATE_MIXED_PRECISION` 可能导致训练问题。

## GRPO 核心概念

### GRPO 的工作方式

1. 为每个提示词生成多个补全
2. 用奖励函数为补全打分
3. 在每个组内计算相对优势
4. 更新策略使其偏好高奖励补全
5. 施加 KL 惩罚以防止偏离参考策略

### 与 PPO 的关键区别

| 维度 | GRPO | PPO |
|------|------|-----|
| 基线 | 组内相对 | 价值函数 |
| Critic | 不需要 | 必需 |
| 显存 | 更低 | 更高 |
| 稳定性 | 良好 | 可能不稳定 |

---

## 技能工作流

### 步骤1：环境与模型准备

按「关键环境设置」与「关键导入顺序」初始化环境，加载模型并设置 pad token（GRPO 必需），然后挂载 LoRA。

```python
from unsloth import FastLanguageModel

# 标准模型
model, tokenizer = FastLanguageModel.from_pretrained(
    "unsloth/Qwen3-4B-unsloth-bnb-4bit",
    max_seq_length=512,
    load_in_4bit=True,
)

# 思考型模型（用于推理任务）
model, tokenizer = FastLanguageModel.from_pretrained(
    "unsloth/Qwen3-4B-Thinking-2507-unsloth-bnb-4bit",
    max_seq_length=1024,  # 思考内容需要更长上下文
    load_in_4bit=True,
)

# 设置 pad token（GRPO 必需）
if tokenizer.pad_token is None:
    tokenizer.pad_token = tokenizer.eos_token
    tokenizer.pad_token_id = tokenizer.eos_token_id
```

应用 LoRA：

```python
model = FastLanguageModel.get_peft_model(
    model,
    r=16,
    lora_alpha=16,
    lora_dropout=0,
    target_modules=["q_proj", "k_proj", "v_proj", "o_proj",
                    "gate_proj", "up_proj", "down_proj"],
    use_gradient_checkpointing="unsloth",
)
```

### 步骤2：准备数据集

GRPO 只需要提示词（补全在训练中生成）：

```python
# GRPO requires prompts only (completions generated during training)
dataset = Dataset.from_dict({
    "prompt": [
        tokenizer.apply_chat_template(
            [{"role": "user", "content": "What is recursion?"}],
            tokenize=False, add_generation_prompt=True
        ),
        # ... more prompts
    ]
})
```

### 步骤3：设计奖励函数

**简单奖励函数**：

```python
def length_reward(completions, prompts=None):
    """Reward based on response length."""
    rewards = []
    for completion in completions:
        length = len(completion.split())
        if length < 5:
            rewards.append(-1.0)
        elif length < 50:
            rewards.append(1.0)
        else:
            rewards.append(0.5)
    return rewards
```

**LLM 评审奖励**：

```python
def llm_judge_reward(completions, prompts):
    """Use another LLM to score responses."""
    rewards = []
    for prompt, completion in zip(prompts, completions):
        score = judge_model.evaluate(prompt, completion)
        rewards.append(score)
    return rewards
```

**规则式奖励**：

```python
def format_reward(completions, prompts=None):
    """Reward proper formatting."""
    rewards = []
    for completion in completions:
        score = 0.0
        if completion.endswith("."):
            score += 0.5
        if not completion.startswith(" "):
            score += 0.5
        rewards.append(score)
    return rewards
```

**复合奖励**：

```python
def combined_reward(completions, prompts):
    """Combine multiple reward signals."""
    length_scores = length_reward(completions)
    format_scores = format_reward(completions)
    return [0.5 * l + 0.5 * f for l, f in zip(length_scores, format_scores)]
```

**思考感知奖励函数（基于 token）** — 使用 TRL 提供的 `completion_ids` 做高效的 token 级解析：

```python
THINK_END_TOKEN_ID = 151668  # </think> token for Qwen3-Thinking models

def thinking_reward_fn(completions, prompts=None, completion_ids=None, **kwargs):
    """
    Token-based reward function using completion_ids provided by TRL.

    Benefits over string matching:
    - No re-tokenization overhead (faster training)
    - Exact token boundaries (no regex edge cases)
    - Consistent with inference code pattern

    Scoring:
    - No </think> token: -1.0 (strongly penalized)
    - Short thinking (<10 tokens): 0.3
    - Medium thinking (10-30 tokens): 0.7
    - Long thinking (>30 tokens): 1.0
    - Bonus +0.1 for self-questioning (contains '?')
    """
    rewards = []

    for completion, comp_ids in zip(completions, completion_ids):
        # Token-based detection using </think> token ID
        if THINK_END_TOKEN_ID in comp_ids:
            end_idx = comp_ids.index(THINK_END_TOKEN_ID)
            thinking_length = end_idx  # Token count before </think>

            # String-based content analysis for question detection
            thinking_content = completion.split('</think>')[0]
            has_self_questions = '?' in thinking_content

            # Score based on thinking token count
            if thinking_length < 10:
                reward = 0.3  # Minimal thinking
            elif thinking_length < 30:
                reward = 0.7 + (0.1 if has_self_questions else 0)
            else:
                reward = 1.0 + (0.1 if has_self_questions else 0)
        else:
            reward = -1.0  # No </think> token found

        rewards.append(reward)

    return rewards
```

**关键洞见**：TRL 会把 `completion_ids` 直接传给奖励函数，免去重新分词的开销。

**多目标思考奖励（基于 token）**：

```python
THINK_END_TOKEN_ID = 151668  # </think> token for Qwen3-Thinking models

def comprehensive_thinking_reward(completions, prompts=None, completion_ids=None, **kwargs):
    """
    Evaluate multiple aspects of thinking quality using token IDs.

    Scoring breakdown:
    - Has </think> token: +0.3
    - Thinking depth (20+ tokens): +0.3
    - Structured sentences: +0.2
    - Self-questioning: +0.1
    - Step-by-step reasoning: +0.1
    """
    rewards = []

    for completion, comp_ids in zip(completions, completion_ids):
        score = 0.0

        # Token-based boundary detection
        if THINK_END_TOKEN_ID in comp_ids:
            score += 0.3  # Has proper </think> token
            end_idx = comp_ids.index(THINK_END_TOKEN_ID)
            thinking_length = end_idx  # Token count

            # Extract thinking content for text analysis
            thinking = completion.split('</think>')[0]

            # Depth (token count from IDs)
            if thinking_length >= 20:
                score += 0.3
            elif thinking_length >= 10:
                score += 0.2

            # Structure (sentences in text)
            sentences = thinking.count('.') + thinking.count('!')
            if sentences >= 2:
                score += 0.2

            # Self-questioning
            if '?' in thinking:
                score += 0.1

            # Step-by-step reasoning
            if any(w in thinking.lower() for w in ['first', 'then', 'next', 'finally']):
                score += 0.1
        else:
            score = -0.5  # Penalize missing </think> token

        rewards.append(score)

    return rewards
```

### 步骤4：配置 GRPOTrainer

**基础配置**：

```python
from trl import GRPOConfig

grpo_config = GRPOConfig(
    output_dir="./grpo_output",
    per_device_train_batch_size=1,
    gradient_accumulation_steps=4,
    max_steps=100,
    learning_rate=1e-5,
    fp16=not is_bf16_supported(),
    bf16=is_bf16_supported(),
    optim="adamw_8bit",
    max_completion_length=128,
    num_generations=4,
    beta=0.1,
)
```

**关键参数**：

| 参数 | 典型取值 | 作用 |
|------|----------|------|
| `beta` | 0.01-0.1 | KL 惩罚强度 |
| `num_generations` | 2-8 | 每个提示词的补全数量 |
| `max_completion_length` | 64-256 | 生成长度 |
| `learning_rate` | 1e-6 至 1e-5 | 低于 SFT 的学习率 |

### 步骤5：启动训练

**基础训练循环**：

```python
from trl import GRPOTrainer

trainer = GRPOTrainer(
    model=model,
    args=grpo_config,
    train_dataset=dataset,
    processing_class=tokenizer,
    reward_funcs=length_reward,
)

trainer.train()
```

**多奖励函数**：

```python
trainer = GRPOTrainer(
    model=model,
    args=grpo_config,
    train_dataset=dataset,
    processing_class=tokenizer,
    reward_funcs=[length_reward, format_reward],
    reward_weights=[0.5, 0.5],
)
```

### 步骤6：训练后排查与收尾

遇到问题时按下方「故障排查」处理；在 Jupyter 中训练结束后释放 GPU 显存：

```python
import IPython
print("Shutting down kernel to release GPU memory...")
app = IPython.Application.instance()
app.kernel.do_shutdown(restart=False)
```

**重要**：在训练笔记本结束、切换到其他模型之前，务必执行此操作。

---

## 故障排查

### 奖励劫持

**症状**：模型钻奖励函数的空子（例如总是输出同一长度）

**修复**：
- 增加多样性惩罚
- 使用多路奖励信号
- 限制最大奖励值

### KL 散度过高

**症状**：策略过度偏离参考策略

**修复**：
- 调大 `beta`（更强的 KL 惩罚）
- 降低 `learning_rate`
- 减少训练步数

### 训练不稳定

**症状**：损失尖峰或 NaN

**修复**：
- 把 `learning_rate` 降到 5e-6
- 把 `num_generations` 降到 2
- 检查奖励量级（应大致在 -1 到 1 之间）

### 显存不足

**症状**：多路生成时 OOM

**修复**：
- 把 `num_generations` 降到 2
- 启用梯度检查点
- 减小 `max_completion_length`

## 适用时机

在以下场景使用本技能：

- 让模型与人类偏好对齐
- 针对特定行为做优化
- SFT 之后的精调
- 构建奖励驱动的系统
- 作为 PPO 的更简替代

使用说明

# GRPO 强化学习训练指南

面向大模型 RLHF 训练的 GRPO 实操指南：GRPOTrainer 配置、奖励函数设计（含思考感知奖励模式）、KL 散度约束与常见训练故障排查。

## 使用

```text
我要用 GRPO 对 Qwen3-4B 做推理能力对齐，
奖励怎么设计？beta 和学习率怎么设？
```

技能按「环境与模型准备 → 数据集 → 奖励函数 → GRPOTrainer 配置 → 训练 → 排查收尾」六步给出可直接运行的代码与参数建议。

## 工作原理

- **组内相对优化**：每提示词生成多个补全，组内计算相对优势，免 Critic、显存更低
- **奖励函数**：支持规则式、LLM 评审、复合奖励；思考型模型可用 `completion_ids` 做 token 级思考奖励
- **稳定训练**：KL 惩罚（beta 0.01-0.1）+ 低学习率（1e-5 量级），覆盖奖励劫持、KL 过高、OOM 等排查路径

如何安装此技能？

访问技能市场，点击「安装」按钮，按提示将技能包放入 AI 编程助手的 skills 目录即可。