【训练与微调篇05】RLHF/DPO/GRPO：大模型人类偏好对齐技术深度解析
🧠 RLHF/DPO/GRPO：大模型人类偏好对齐技术深度解析
GPT-4 很强，但为什么它不说脏话？为什么它拒绝"教我做炸弹"？这就是对齐（Alignment）的功劳。从 ChatGPT 震惊世界的 RLHF，到 DeepSeek R1 逆天的 GRPO，再到 ICML 2026 上最优传输奖励模型——对齐技术正在经历第三次革命。
📑 目录
为什么要对齐？模型能力之外的课题
RLHF 三阶段框架详解
奖励模型训练与2026最新突破
PPO：策略优化经典算法
DPO：直接偏好优化的优雅革命
GRPO：无Critic的群体策略优化（DeepSeek R1）
2026年对齐方法全景对比
实战：GRPO训练完整实现
实战：奖励模型过拟合与SelectiveRM
对齐的未来：2026下半年趋势
总结
一、为什么要对齐？模型能力之外的课题
1.1 对齐三要素
“你是一个有用的AI助手。”
对齐的目标是让模型同时满足三个要求：
有益（Helpful） 诚实（Honest） 无害（Harmless）
能完成任务 不说谎 不伤害他人
准确理解意图 承认不知道 拒绝危险请求
主动提供帮助 不胡编乱造 保护隐私
1.2 从预训练到对齐的完整链路
预训练（Pretrain）
模型学会了语言 → 学会了世界知识
⬇
监督微调（SFT）
学会了"问答格式"
⬇
偏好对齐（Alignment） ← 这一步决定"好用不好用"
├── 奖励建模 + RL（RLHF）
├── 直接偏好优化（DPO）
└── 群体策略优化（GRPO）
1.3 2026 年对齐方法的演进
2020: InstructGPT 提出 RLHF（PPO + Reward Model）
2023: DPO 发布 — 不需要奖励模型的革命
2024: DeepSeek R1 使用 GRPO 训练推理模型
2025: GRPO 成为开源社区首选对齐方法
2026 上半年: SelectiveRM（ICML 2026）解决奖励模型噪声问题
GRPO + KTO + ORPO 百花齐放
二、RLHF 三阶段框架详解
2.1 三阶段概览
RLHF（Reinforcement Learning from Human Feedback）由三个独立训练的阶段组成 [1]：
Stage 1: SFT（监督微调）— 让模型学会"回答问题"
├── 输入: 人工编写的"问题-答案"对
├── 训练: 标准的 Cross-Entropy Loss
└── 输出: 一个会回答问题的 base 模型
Stage 2: 奖励模型训练 — 教会模型"什么算好回答"
├── 输入: 人类对多条回复的偏好排序
├── 训练: Bradley-Terry 偏好模型
└── 输出: 一个能给回复打分的奖励模型
Stage 3: PPO 强化学习 — 让模型"学会取悦奖励模型"
├── 输入: 奖励模型的打分 + KL 惩罚
├── 训练: PPO（Proximal Policy Optimization）
└── 输出: 对齐人类偏好的最终模型
2.2 为什么需要三个阶段？
只做 SFT（不够）：
模型学会了格式，但没有"好坏"判断
容易"看起来对但实际错"（factual hallucination）
只做 SFT + RM（不够）：
奖励模型能打分，但模型不知道怎么优化
测试时可以用 RM 筛选，但训练时需要更高效的反馈
SFT + RM + PPO（完整 RLHF）：
SFT: 基础能力
RM: 偏好信号
PPO: 有效优化策略
但是牺牲: 训练复杂度 19x 增加！
三、奖励模型训练与2026最新突破
3.1 奖励模型的训练数据
奖励模型需要偏好数据，格式为：
{
“prompt”: “什么是注意力机制？”,
“chosen”: “注意力机制是一种…它通过计算…从而…”,
“rejected”: “注意力机制就是看哪里重要。”
}
关键：不需要绝对分数，只需要相对排序（chosen > rejected）。
3.2 Bradley-Terry 损失函数
奖励模型的训练目标是：最大化 chosen 回复得分 > rejected 回复得分的概率。
“”“Bradley-Terry 偏好模型损失函数”“”
import torch
import torch.nn as nn
class RewardModelLoss(nn.Module):
“”"
Bradley-Terry 偏好模型
L(θ) = -E_{(x, y_w, y_l)~D}[log σ(r_θ(x, y_w) - r_θ(x, y_l))]
其中：
- r_θ(x, y): 奖励模型对输入x和回复y的打分（一个标量）
- y_w: 被偏好的回复（chosen）
- y_l: 不被偏好的回复（rejected）
- σ: sigmoid 函数
"""
def forward(
self,
chosen_rewards: torch.Tensor, # (batch,) chosen 的奖励
rejected_rewards: torch.Tensor, # (batch,) rejected 的奖励
) -> torch.Tensor:
"""
最大化 chosen 和 rejected 的奖励差
Loss = -log(σ(chosen - rejected))
"""
# 奖励差距
reward_margin = chosen_rewards - rejected_rewards
# 最大化正确排序概率
loss = -nn.functional.logsigmoid(reward_margin).mean()
# 奖励差越大 → sigmoid 越接近 1 → loss 越小
# 奖励差为负 → sigmoid < 0.5 → loss 很大
return loss
===== 奖励模型训练完整代码 =====
class RewardModelTrainer:
“”“奖励模型训练器”“”
def __init__(self, model, tokenizer, lr=1e-5):
self.model = model
self.tokenizer = tokenizer
self.loss_fn = RewardModelLoss()
self.optimizer = torch.optim.AdamW(
model.parameters(), lr=lr, weight_decay=0.1
)
def train_step(self, batch):
"""单步训练"""
# 对 chosen 和 rejected 分别计算奖励
chosen_inputs = self.tokenizer(
batch["chosen"], return_tensors="pt", padding=True, truncation=True
).to(self.model.device)
rejected_inputs = self.tokenizer(
batch["rejected"], return_tensors="pt", padding=True, truncation=True
).to(self.model.device)
# 奖励模型输出（取 last token 的 hidden state）
chosen_rewards = self.model(**chosen_inputs).logits[:, -1]
rejected_rewards = self.model(**rejected_inputs).logits[:, -1]
# 损失
loss = self.loss_fn(chosen_rewards, rejected_rewards)
# 反向传播
self.optimizer.zero_grad()
loss.backward()
nn.utils.clip_grad_norm_(self.model.parameters(), 1.0)
self.optimizer.step()
# 准确率
accuracy = (chosen_rewards > rejected_rewards).float().mean()
return {
"loss": loss.item(),
"accuracy": accuracy.item(),
"reward_margin": (chosen_rewards - rejected_rewards).mean().item()
}
3.3 奖励模型的三大挑战
Challenge 1: Reward Hacking（奖励黑客）
策略模型学会"欺骗"奖励模型，而不是真正提升质量
例：奖励模型喜欢长回复 → 模型疯狂输出废话
Challenge 2: 偏好噪声
人类标注不一致，众包质量差
2026 发现：真实偏好数据中噪声比例可达 40-50%
Challenge 3: 分布漂移
奖励模型只在 SFT 数据上训练
PPO 阶段生成的回复分布偏移 → 奖励模型打分不准
3.4 SelectiveRM：ICML 2026 奖励模型新范式
2026 年，浙江大学 + 小红书 + 北京大学联合提出了 SelectiveRM，被 ICML 2026 接收 [2]。核心思想：
“不是所有偏好数据都值得学习——让奖励模型自主选择哪些样本是干净的。”
传统方法：
对所有标注好的偏好数据一视同仁 → 学习"好的 + 坏的" → 奖励模型带噪声
SelectiveRM（ICML 2026）：
- 计算每个样本的"语义-偏好"一致性
- 用 最优传输（Optimal Transport） 衡量联合分布差异
- 用 部分最优传输 自动排除高噪声样本
- 只对 “干净的” 偏好学习
结果：
- 奖励模型准确率提升 5-10%
- 下游 RLHF 安全性显著提升
- 在 HelpSteer, UltraFeedback, PKU-SafeRLHF 三个基准上 SOTA
核心公式：
传统 BT Loss: min L(θ) = -log σ(r_θ(x,y_w) - r_θ(x,y_l))
↓
SelectiveRM: min L(θ, π) = OT_cost(J_data, J_model)
↑ 用最优传输衡量两个联合分布的差异
其中：
J_data: 数据的经验联合分布（含噪声）
J_model: 模型学到的联合分布
OT_cost: 最优传输距离
→ 高噪声样本被自动排除（代价过高，不参与匹配）
SelectiveRM 已被集成到 TRL 和 DeepSpeed 的奖励模型训练流程中。对于动手实践奖励模型训练的工程师来说，这是 2026 年最重要的创新。
四、PPO：策略优化经典算法
4.1 PPO 在 RLHF 中的完整流程
PPO（Proximal Policy Optimization）在 RLHF 中扮演"最终优化器"的角色 [3]：
奖励模型
↑
SFT模型（冻结）→ PPO 训练 → 对齐后的模型
│ │
│ KL惩罚 │ PPO Clip
│ (防止跑偏) │ (防止更新过大)
▼ ▼
参考分布 当前分布
4.2 PPO 损失函数
“”“PPO 在 RLHF 中的完整损失函数”“”
import torch
import torch.nn as nn
import torch.nn.functional as F
class PPOInRLHF:
“”"
RLHF 中的 PPO 损失
L^PPO(θ) = -E_t[min(
r_t(θ) · A_t,
clip(r_t(θ), 1-ε, 1+ε) · A_t
)] + β · KL(π_θ || π_ref)
第一项：PPO-Clip（最大化奖励）
第二项：KL 惩罚（防止偏离原始模型）
"""
def __init__(self, epsilon=0.2, kl_coef=0.01):
self.epsilon = epsilon
self.kl_coef = kl_coef
def compute_loss(
self,
log_probs: torch.Tensor, # 当前策略的 log prob
ref_log_probs: torch.Tensor, # 参考策略（SFT）的 log prob
advantages: torch.Tensor, # 优势函数（奖励 - 基线）
old_log_probs: torch.Tensor, # 更新前的 log prob
) -> torch.Tensor:
"""
PPO 损失计算
"""
# 1. 重要性采样比率
ratio = torch.exp(log_probs - old_log_probs) # r_t(θ)
# 2. 策略梯度损失（PPO-Clip）
surr1 = ratio * advantages
surr2 = torch.clamp(
ratio,
1 - self.epsilon,
1 + self.epsilon
) * advantages
policy_loss = -torch.min(surr1, surr2).mean()
# 3. KL 散度惩罚（从 reference model 计算）
kl_div = (ref_log_probs - log_probs).mean()
# 4. 总损失
total_loss = policy_loss + self.kl_coef * kl_div
return total_loss
def compute_advantages(
self,
rewards: torch.Tensor, # 奖励模型打分
values: torch.Tensor = None, # Critic 网络估计（如使用）
gamma: float = 1.0
) -> torch.Tensor:
"""
计算优势函数
A_t = R_t - V(s_t)
如果没有 Critic（GRPO 方式），直接用奖励
"""
if values is not None:
advantages = rewards - values
else:
# 简单归一化（GRPO 方式）
advantages = (rewards - rewards.mean()) / (rewards.std() + 1e-8)
return advantages
===== RLHF + PPO 完整训练循环 =====
class RLHFTrainer:
“”"
完整的 RLHF 训练器（PPO）
“”"
def __init__(
self,
policy_model, # 策略模型（待优化）
ref_model, # 参考模型（冻结，SFT版本）
reward_model, # 奖励模型（冻结）
tokenizer,
ppo_config=None
):
self.policy = policy_model
self.ref_model = ref_model
self.reward_model = reward_model
self.tokenizer = tokenizer
self.config = ppo_config or {
"lr": 1e-6,
"kl_coef": 0.04,
"epsilon": 0.2,
"batch_size": 8,
"max_gen_len": 512,
}
self.ppo = PPOInRLHF(
epsilon=self.config["epsilon"],
kl_coef=self.config["kl_coef"]
)
self.optimizer = torch.optim.AdamW(
policy_model.parameters(), lr=self.config["lr"]
)
def train_step(self, prompts: List[str]):
"""单步 RLHF 训练"""
# ===== Step 1: 用当前策略生成回复 =====
with torch.no_grad():
inputs = self.tokenizer(prompts, return_tensors="pt", padding=True).to("cuda")
# 策略模型生成
gen_outputs = self.policy.generate(
**inputs,
max_new_tokens=self.config["max_gen_len"],
do_sample=True,
temperature=0.7,
pad_token_id=self.tokenizer.pad_token_id,
return_dict_in_generate=True,
output_scores=True,
)
generated_texts = self.tokenizer.batch_decode(
gen_outputs.sequences[:, inputs.input_ids.shape[1]:],
skip_special_tokens=True
)
# ===== Step 2: 计算奖励 =====
with torch.no_grad():
reward_inputs = self.tokenizer(
[f"{p}{g}" for p, g in zip(prompts, generated_texts)],
return_tensors="pt", padding=True
).to("cuda")
rewards = self.reward_model(**reward_inputs).logits[:, -1]
# ===== Step 3: 计算 log prob 和 KL =====
# 当前策略的 log prob
policy_logits = self.policy(**inputs).logits
policy_log_probs = F.log_softmax(policy_logits, dim=-1)
# 参考策略的 log prob
with torch.no_grad():
ref_logits = self.ref_model(**inputs).logits
ref_log_probs = F.log_softmax(ref_logits, dim=-1)
# ===== Step 4: 计算优势 =====
advantages = self.ppo.compute_advantages(
rewards, values=None, gamma=1.0
)
# ===== Step 5: PPO 损失 =====
loss = self.ppo.compute_loss(
log_probs=policy_log_probs,
ref_log_probs=ref_log_probs,
advantages=advantages,
old_log_probs=policy_log_probs.detach(), # 简化
)
# ===== Step 6: 反向传播 =====
self.optimizer.zero_grad()
loss.backward()
nn.utils.clip_grad_norm_(self.policy.parameters(), 1.0)
self.optimizer.step()
return {
"loss": loss.item(),
"reward_mean": rewards.mean().item(),
"kl": (ref_log_probs - policy_log_probs).mean().item()
}
4.3 PPO 的痛点
PPO 在 RLHF 中的问题：
-
⚠️ 需要 4 个模型同时加载
- 策略模型 (待优化) ~140GB (70B FP16)
- 参考模型 (冻结) ~140GB
- 奖励模型 ~140GB
- Critic 网络 ~140GB (如果用 value function)
→ 总计 ~560GB → 至少 4×H200
-
⚠️ 训练不稳定
- PPO 超参数敏感（epsilon, kl_coef, lr）
- 容易 collapse（reward hacking）
- 需要大量调参经验
-
⚠️ 采样效率低
- 每一步都需要重新生成
- 生成速度 = 推理速度
五、DPO：直接偏好优化的优雅革命
5.1 DPO 的核心洞察
DPO（Direct Preference Optimization）[4] 发现了一个关键等式：
RLHF 的优化目标 = 奖励模型 + PPO = 可以直接用一个偏好优化损失替代
标准 RLHF: max E[r(x,y)] - β·KL(π_θ || π_ref)
需要 奖励模型 + PPO + 参考模型
DPO 等价形式:
L_DPO = -E[log σ(β·(log π_θ(y_w|x) - log π_θ(y_l|x)))]
只需要 策略模型 + 参考模型 + 偏好数据
不需要 奖励模型！不需要 PPO！不需要 Critic！
5.2 DPO 损失函数
“”“DPO 损失函数完整实现”“”
import torch
import torch.nn as nn
import torch.nn.functional as F
class DPOLoss(nn.Module):
“”"
DPO: Direct Preference Optimization
L_DPO(θ) = -E_{(x,y_w,y_l)~D}[log σ(β·(log π_θ(y_w|x) - log π_θ(y_l|x)))]
其中：
- π_θ(y_w|x) / π_ref(y_w|x): chosen 回复的相对概率
- π_θ(y_l|x) / π_ref(y_l|x): rejected 回复的相对概率
- β: 温度参数，控制偏离参考模型的程度
"""
def __init__(self, beta: float = 0.1):
super().__init__()
self.beta = beta # 温度参数，默认 0.1
def forward(
self,
policy_chosen_logps: torch.Tensor, # log π_θ(y_w|x)
policy_rejected_logps: torch.Tensor, # log π_θ(y_l|x)
ref_chosen_logps: torch.Tensor, # log π_ref(y_w|x)
ref_rejected_logps: torch.Tensor, # log π_ref(y_l|x)
) -> torch.Tensor:
"""
DPO 损失计算
"""
# 1. 对数概率比（log odds ratio）
# log(π_θ(y_w|x) / π_ref(y_w|x)) - log(π_θ(y_l|x) / π_ref(y_l|x))
pi_logratios = policy_chosen_logps - policy_rejected_logps
ref_logratios = ref_chosen_logps - ref_rejected_logps
# 2. 隐式奖励
logits = pi_logratios - ref_logratios # β·(r(x,y_w) - r(x,y_l))
# 3. DPO 损失
loss = -F.logsigmoid(self.beta * logits).mean()
# 4. 准确率（隐式奖励判断正确的比例）
accuracy = (logits > 0).float().mean()
return loss, accuracy
===== 完整 DPO 训练器 =====
class DPOTrainer:
“”"
DPO 训练器（不需要奖励模型！）
“”"
def __init__(self, model, ref_model, tokenizer, beta=0.1, lr=1e-6):
self.model = model # 待训练的策略模型
self.ref_model = ref_model # 冻结的参考模型
self.tokenizer = tokenizer
self.beta = beta
self.loss_fn = DPOLoss(beta=beta)
self.optimizer = torch.optim.AdamW(
model.parameters(), lr=lr, weight_decay=0.01
)
def _get_log_probs(
self, model, input_ids, attention_mask
) -> torch.Tensor:
"""
计算序列中 assistant 部分的平均 log prob
"""
outputs = model(input_ids, attention_mask=attention_mask)
logits = outputs.logits
# Shift
shift_logits = logits[:, :-1, :].contiguous()
shift_input_ids = input_ids[:, 1:].contiguous()
# 只计算 assistant 回复部分的 log prob
# 假设 assistant 部分从某个特殊 token 开始
log_probs = F.log_softmax(shift_logits, dim=-1)
token_log_probs = log_probs.gather(
dim=-1, index=shift_input_ids.unsqueeze(-1)
).squeeze(-1)
return token_log_probs.sum(dim=-1)
def train_step(self, batch):
"""
单步 DPO 训练
batch: {
"chosen_input_ids": ...,
"chosen_attention_mask": ...,
"rejected_input_ids": ...,
"rejected_attention_mask": ...,
}
"""
# 当前策略的 log prob
policy_chosen_logps = self._get_log_probs(
self.model,
batch["chosen_input_ids"],
batch["chosen_attention_mask"]
)
policy_rejected_logps = self._get_log_probs(
self.model,
batch["rejected_input_ids"],
batch["rejected_attention_mask"]
)
# 参考模型的 log prob
with torch.no_grad():
ref_chosen_logps = self._get_log_probs(
self.ref_model,
batch["chosen_input_ids"],
batch["chosen_attention_mask"]
)
ref_rejected_logps = self._get_log_probs(
self.ref_model,
batch["rejected_input_ids"],
batch["rejected_attention_mask"]
)
# DPO 损失
loss, accuracy = self.loss_fn(
policy_chosen_logps,
policy_rejected_logps,
ref_chosen_logps,
ref_rejected_logps,
)
# 反向传播
self.optimizer.zero_grad()
loss.backward()
nn.utils.clip_grad_norm_(self.model.parameters(), 1.0)
self.optimizer.step()
return {
"loss": loss.item(),
"accuracy": accuracy.item(),
"reward_margin": (
(policy_chosen_logps - policy_rejected_logps).mean().item()
)
}
5.3 DPO vs PPO 的核心区别
维度 PPO-RLHF DPO
需要奖励模型 ✅ 是 ❌ 否（隐式）
需要参考模型 ✅ 是 ✅ 是
需要采样生成 ✅ 每步生成 ❌ 离线训练
训练复杂度 高（4个模型） 低（2个模型）
训练速度 慢（生成耗时） 快（纯梯度）
稳定性 敏感 稳定
显存需求 560GB (70B) 280GB (70B)
理论基础 PPO 偏好优化闭式解
主流框架 TRL + transformers TRL + PEFT
💡 面试加分点：DPO 的数学核心是——偏好优化问题存在闭式解（closed-form solution），RLHF 的 PPO 阶段实际上是冗余的。DPO 通过重新参数化，将 PPO 的两步(奖励模型训练 + RL)合并为一步优化。
5.4 DPO 的变体
变体 核心改进 适用场景
DPO 原始版本 标准偏好对齐
IPO 解决过拟合 数据噪声大
KTO 只有好/坏标签，不需要配对 只有点赞/踩数据
ORPO 在 SFT 阶段混合偏好优化 端到端训练
SimPO 用生成概率替代隐式奖励 推理时间对齐
CPO 对比式偏好优化 翻译任务
六、GRPO：无Critic的群体策略优化（DeepSeek R1）
6.1 GRPO 的诞生
GRPO（Group Relative Policy Optimization）由 DeepSeek 团队提出 [5]，核心思想：
不要 Critic 网络（Value Function），用同一 prompt 生成的多条回复的相对表现来估计优势。
PPO: A_t = R_t - V(s_t) ← 需要 Critic 网络（额外 140GB）
GRPO: A_i = (R_i - mean®) / std® ← 只需要一个 group 的奖励！
GRPO 的优势：
-
不需要 Critic 网络 → 2个模型就够了（策略 + 奖励）
-
隐式提供基线 → 降低方差
-
训练更稳定
6.2 GRPO 的核心公式
GRPO 对每个 prompt x，采样 G 个回复 {y_1, …, y_G}： -
计算每个回复的奖励：r_i = R(y_i|x)
-
计算优势：A_i = (r_i - μ_r) / σ_r
（组内相对优势，而不是依赖 Critic） -
优化目标：
max E[ 1/G · Σ(min(π_θ(y_i|x)/π_old(y_i|x) · A_i,
clip(…, 1-ε, 1+ε) · A_i))
- β · KL(π_θ || π_ref) ]
6.3 GRPO 完整实现
“”“GRPO 完整实现（DeepSeek R1 风格）”“”
import torch
import torch.nn as nn
import torch.nn.functional as F
class GRPOLoss(nn.Module):
“”"
GRPO: Group Relative Policy Optimization
对每个 prompt，采样 G 个回复，用组内相对奖励估计优势
"""
def __init__(self, beta: float = 0.04, epsilon: float = 0.2):
super().__init__()
self.beta = beta # KL 惩罚系数
self.epsilon = epsilon # PPO-Clip 阈值
def forward(
self,
old_log_probs: torch.Tensor, # (batch, group, seq_len) 旧策略log prob
new_log_probs: torch.Tensor, # (batch, group, seq_len) 新策略log prob
ref_log_probs: torch.Tensor, # (batch, group, seq_len) 参考log prob
rewards: torch.Tensor, # (batch, group) 每条回复的奖励
mask: torch.Tensor, # (batch, group, seq_len) token mask
) -> torch.Tensor:
"""
GRPO 损失
"""
batch, group, seq_len = old_log_probs.shape
# 1. 计算每个 token 的重要性采样比率
log_ratio = new_log_probs - old_log_probs # log(π_new / π_old)
ratio = torch.exp(log_ratio) # π_new / π_old
# 2. 组内优势归一化（无需 Critic！）
# 对每个 prompt，计算组内均值、标准差
mean_rewards = rewards.mean(dim=-1, keepdim=True) # (batch, 1)
std_rewards = rewards.std(dim=-1, keepdim=True) + 1e-8
advantages = (rewards - mean_rewards) / std_rewards # (batch, group)
advantages = advantages.unsqueeze(-1) # (batch, group, 1)
# 3. PPO-Clip 策略梯度
surr1 = ratio * advantages
surr2 = torch.clamp(ratio, 1 - self.epsilon, 1 + self.epsilon) * advantages
# 4. 策略损失（只计算有效 token）
policy_loss = -torch.min(surr1, surr2) * mask.float()
policy_loss = policy_loss.sum() / mask.float().sum()
# 5. KL 散度惩罚（从参考模型）
kl_div = (ref_log_probs - new_log_probs) * mask.float()
kl_div = kl_div.sum() / mask.float().sum()
# 6. 总损失
total_loss = policy_loss + self.beta * kl_div
return total_loss
===== 完整 GRPO 训练器 =====
class GRPOTrainer:
“”"
GRPO 训练器（不需要 Critic 网络！）
DeepSeek R1 使用的训练方法
“”"
def __init__(
self,
policy_model, # 策略模型（待优化）
ref_model, # 参考模型（冻结）
reward_model, # 奖励模型（冻结，可选）
tokenizer,
beta: float = 0.04,
epsilon: float = 0.2,
group_size: int = 8, # 每组采样 8 条回复
max_gen_len: int = 1024,
):
self.policy = policy_model
self.ref_model = ref_model
self.reward_model = reward_model
self.tokenizer = tokenizer
self.group_size = group_size
self.max_gen_len = max_gen_len
self.loss_fn = GRPOLoss(beta=beta, epsilon=epsilon)
self.optimizer = torch.optim.AdamW(
policy_model.parameters(), lr=1e-6
)
@torch.no_grad()
def _generate_group(self, prompt: str):
"""
对单个 prompt 生成 G 条回复
使用不同的随机种子/temperature
DeepSeek R1 的做法：
- temperature = 0.6 (训练时)
- top_p = 0.95
- 生成 G = 8 条
"""
inputs = self.tokenizer(prompt, return_tensors="pt").to("cuda")
# 扩展为 G 份
input_ids = inputs.input_ids.repeat(self.group_size, 1)
attention_mask = inputs.attention_mask.repeat(self.group_size, 1)
# 生成（每个 prompt 生成 8 条不同的回复）
outputs = self.policy.generate(
input_ids=input_ids,
attention_mask=attention_mask,
max_new_tokens=self.max_gen_len,
do_sample=True,
temperature=0.6, # DeepSeek R1 风格
top_p=0.95,
pad_token_id=self.tokenizer.pad_token_id,
)
return outputs, input_ids
@torch.no_grad()
def _compute_rewards(self, prompt: str, responses: List[str]):
"""
计算组内每条回复的奖励
可以只用奖励模型，也可以用规则奖励（DeepSeek R1 的方式）
"""
rewards = []
for response in responses:
full_text = prompt + response
# 奖励模型打分
inputs = self.tokenizer(full_text, return_tensors="pt").to("cuda")
rm_score = self.reward_model(**inputs).logits[:, -1].item()
# DeepSeek R1 也使用规则奖励
# 如：格式奖励（是否在 <think> 标签内）
# 语言奖励、长度奖励等
reward = rm_score
rewards.append(reward)
return torch.tensor(rewards, device="cuda")
def train_step(self, prompts: List[str]):
"""
单步 GRPO 训练
"""
all_losses = []
for prompt in prompts:
# ===== Step 1: 生成 G 条回复 =====
gen_outputs, input_ids = self._generate_group(prompt)
responses = self.tokenizer.batch_decode(
gen_outputs[:, input_ids.shape[1]:],
skip_special_tokens=True,
)
# ===== Step 2: 计算奖励 =====
rewards = self._compute_rewards(prompt, responses)
# ===== Step 3: 计算各组件的 log prob =====
# 当前策略
new_logits = self.policy(gen_outputs).logits
new_log_probs = F.log_softmax(new_logits, dim=-1)
# 参考策略
with torch.no_grad():
ref_logits = self.ref_model(gen_outputs).logits
ref_log_probs = F.log_softmax(ref_logits, dim=-1)
# 需要注意的是旧策略log prob（简化处理）
old_log_probs = new_log_probs.detach()
# ===== Step 4: GRPO 损失 =====
loss = self.loss_fn(
old_log_probs=old_log_probs,
new_log_probs=new_log_probs,
ref_log_probs=ref_log_probs,
rewards=rewards,
mask=gen_outputs != self.tokenizer.pad_token_id,
)
all_losses.append(loss)
# ===== Step 5: 更新 =====
total_loss = torch.stack(all_losses).mean()
self.optimizer.zero_grad()
total_loss.backward()
nn.utils.clip_grad_norm_(self.policy.parameters(), 1.0)
self.optimizer.step()
return {"loss": total_loss.item()}
6.4 DeepSeek R1 的 GRPO 实践
DeepSeek R1 在训练中使用了混合奖励策略[5]：
GRPO 训练中的奖励信号：
-
规则奖励（Rule-based Reward）—— 占主要
├── 格式奖励：是否使用 标签
├── 语言奖励：是否使用目标语言
├── 长度奖励：回复长度是否适中
└── 关键词奖励：是否包含关键推理步骤 -
模型奖励（Model-based Reward）—— 辅助
├── 奖励模型打分
└── 用于开放域任务
规则奖励的好处：
- 零标注成本
- 完全可控
- 不会被 reward hacking
DeepSeek R1 证明了 GRPO + 规则奖励可以训练出强大的推理模型。GRPO 正在成为 2026 年开源社区的首选对齐方法 5。
七、2026年对齐方法全景对比
7.1 方法对比总表
方法 年份 奖励模型 参考模型 Critic 内存(70B) 训练速度 效果 代表模型
RLHF + PPO 2020 ✅ ✅ ✅ 560GB 慢 ⭐⭐⭐⭐⭐ GPT-4, Claude
DPO 2023 ❌ ✅ ❌ 280GB 快 ⭐⭐⭐⭐ LLaMA 3
GRPO 2024 可选 ✅ ❌ 280GB 中 ⭐⭐⭐⭐⭐ DeepSeek R1
KTO 2024 ❌ ✅ ❌ 280GB 快 ⭐⭐⭐ 社区试验
ORPO 2024 ❌ ❌ ❌ 140GB 极快 ⭐⭐⭐ 小型模型
SimPO 2025 ❌ ❌ ❌ 140GB 极快 ⭐⭐⭐ 推理优化
SelectiveRM 2026 ✅ ✅ ✅ 560GB 慢 ⭐⭐⭐⭐⭐+ 研究阶段
7.2 选型决策树
我的资源？
├── 有 8×H200 + 3周时间 → RLHF + PPO（效果最好）
├── 有 4×H200 + 1周时间 → DPO（效果好 + 简单）
├── 有 4×H200 + 训练推理模型 → GRPO（DeepSeek R1 路线）⭐
├── 有 单卡 5090 + 几天时间 → ORPO / SimPO
├── 需要训练奖励模型 → SelectiveRM (ICML 2026) ⭐
└── 只有"好/坏"标签，无配对数据 → KTO
2026 年推荐：
⭐ 推理模型 → GRPO
⭐ 对话模型 → DPO 或 RLHF + PPO
⭐ 极致效率 → ORPO
⭐ 奖励模型训练 → SelectiveRM
7.3 各方法的 KL-奖励前沿
KL 散度（偏离程度）
↑
高 | PPO
| GRPO
| DPO
低 | ORPO
±-----------------------→ 奖励（对齐程度）
高
解释：
PPO: 能获得最高奖励，但 KL 也最高（偏离原始模型最远）
DPO: 中间位置，平衡
GRPO: 接近 PPO 的效果，KL 更低
ORPO: KL 最低，但奖励上限也低
八、实战：GRPO训练完整实现
8.1 使用 TRL 的 GRPOTrainer（2026推荐）
“”“使用 TRL 的 GRPOTrainer（2026 年最简方式）”“”
from trl import GRPOTrainer
from transformers import AutoModelForCausalLM, AutoTokenizer
from datasets import load_dataset
===== 1. 加载模型 =====
model = AutoModelForCausalLM.from_pretrained(
“Qwen/Qwen3-7B”,
torch_dtype=torch.bfloat16,
attn_implementation=“flash_attention_2”,
device_map=“auto”,
)
ref_model = AutoModelForCausalLM.from_pretrained(
“Qwen/Qwen3-7B”,
torch_dtype=torch.bfloat16,
device_map=“auto”,
)
tokenizer = AutoTokenizer.from_pretrained(“Qwen/Qwen3-7B”)
tokenizer.pad_token = tokenizer.eos_token
===== 2. 配置 GRPO =====
from trl import GRPOConfig
training_args = GRPOConfig(
output_dir=“qwen3-grpo”,
# GRPO 特有参数
beta=0.04, # KL 惩罚系数
epsilon=0.2, # PPO-Clip 阈值
group_size=8, # 每组 8 条回复（与 R1 一致）
# 训练参数
per_device_train_batch_size=4,
gradient_accumulation_steps=4,
num_train_epochs=1,
learning_rate=1e-6,
warmup_ratio=0.1,
bf16=True,
logging_steps=1,
save_strategy="steps",
save_steps=100,
# 生成参数
max_new_tokens=1024,
temperature=0.6, # DeepSeek R1 风格
top_p=0.95,
do_sample=True,
)
===== 3. 定义奖励函数（混合奖励） =====
def reward_function(prompts, responses):
“”"
混合奖励函数
1. 格式奖励：是否包含推理步骤
2. 长度奖励
3. 奖励模型打分（可选）
“”"
rewards = []
for prompt, response in zip(prompts, responses):
score = 0.0
# 格式奖励：是否有关键词/格式
if "因此" in response or "所以" in response:
score += 0.5
if len(response) > 100:
score += 0.3
if len(response) > 500:
score += 0.2
# 长度惩罚（太长不好）
if len(response) > 2000:
score -= 0.5
rewards.append(score)
return rewards
===== 4. 启动 GRPO 训练 =====
trainer = GRPOTrainer(
model=model,
ref_model=ref_model,
reward_function=reward_function,
args=training_args,
train_dataset=dataset,
tokenizer=tokenizer,
)
trainer.train()
8.2 规则奖励设计
“”“DeepSeek R1 风格的规则奖励系统”“”
class RuleBasedReward:
“”“规则奖励——GRPO 训练的奖励信号”“”
@staticmethod
def format_reward(response: str) -> float:
"""
格式奖励：鼓励使用推理格式
DeepSeek R1 的 <think> 标签奖励
"""
score = 0.0
# 是否包含推理标志
if "### 推理过程" in response or "思考：" in response:
score += 0.3
if "### 回答" in response or "答案：" in response:
score += 0.3
# 结构完整
if response.count("\n") >= 5:
score += 0.2
if response.count(".") >= 3:
score += 0.2
return min(score, 1.0)
@staticmethod
def length_reward(response: str, min_length: int = 50, max_length: int = 2048) -> float:
"""
长度奖励：惩罚过长或过短的回复
"""
length = len(response)
if length < min_length:
return -1.0 # 太短，严厉惩罚
elif length > max_length:
return -0.5 # 太长，轻微惩罚
elif min_length <= length <= min_length * 2:
return 1.0 # 适中最优
else:
return 0.5
@staticmethod
def language_reward(response: str, target_lang: str = "zh") -> float:
"""
语言奖励：确保使用目标语言
"""
if target_lang == "zh":
chinese_chars = sum(1 for c in response if '\u4e00' <= c <= '\u9fff')
ratio = chinese_chars / max(len(response), 1)
if ratio > 0.3:
return 1.0
elif ratio > 0.1:
return 0.0
else:
return -1.0
return 0.0
@staticmethod
def combined_reward(
response: str,
rm_score: float = None
) -> float:
"""
综合奖励
"""
total = 0.0
total += RuleBasedReward.format_reward(response) * 0.3
total += RuleBasedReward.length_reward(response) * 0.2
total += RuleBasedReward.language_reward(response, "zh") * 0.2
if rm_score is not None:
total += rm_score * 0.3
return total
九、实战：奖励模型过拟合与SelectiveRM
9.1 SelectiveRM 的核心实现
“”“SelectiveRM 的核心逻辑（简化实现）”“”
import torch
import torch.nn as nn
from scipy.optimize import linear_sum_assignment # OT 求解
import numpy as np
class SelectiveRMLoss(nn.Module):
“”"
SelectiveRM: 基于最优传输的选择性奖励模型训练
核心思想：
1. 不强制拟合所有偏好数据
2. 用最优传输衡量"语义-偏好"联合分布的一致性
3. 自动排除噪声偏好样本
"""
def __init__(self, noise_rejection_ratio: float = 0.2):
super().__init__()
self.rejection_ratio = noise_rejection_ratio # 自动排除 20% 高噪声样本
def compute_semantic_similarity(
self, chosen_reps: torch.Tensor, rejected_reps: torch.Tensor
) -> torch.Tensor:
"""
1. 计算语义距离
使用 reward model 中间层的 hidden state 计算
"""
# 归一化
chosen_reps = F.normalize(chosen_reps, dim=-1)
rejected_reps = F.normalize(rejected_reps, dim=-1)
# 余弦相似度
similarity = (chosen_reps @ rejected_reps.T)
return similarity # 较大的值 = 语义相似
def partial_ot_loss(
self,
chosen_rewards: torch.Tensor,
rejected_rewards: torch.Tensor,
semantic_sim: torch.Tensor,
chosen_reps: torch.Tensor,
rejected_reps: torch.Tensor,
) -> torch.Tensor:
"""
2. 部分最优传输（Partial OT）
构建传输代价矩阵 C
C[i,j] = ||h_chosen_i - h_rejected_j||² + λ·|r_chosen_i - r_rejected_j|²
然后用部分 OT 自动排除高噪声样本
"""
batch = chosen_rewards.shape[0]
# 语义距离矩阵
semantic_dist = 1 - semantic_sim # (B, B)
# 偏好距离矩阵
reward_dist = torch.abs(
chosen_rewards.unsqueeze(1) - rejected_rewards.unsqueeze(0)
)
# 组合代价
C = semantic_dist + 0.5 * reward_dist # 联合代价
# 部分传输：只传输 (1 - rejection_ratio) 的质量
mass = 1.0 - self.rejection_ratio
# 使用 Sinkhorn 算法求解部分 OT
transport_plan = self._partial_sinkhorn(C, mass)
# OT 损失 = 传输代价
ot_loss = (transport_plan * C).sum()
# 计算被排除的样本（高噪声）
excluded_mask = transport_plan.sum(dim=1) < 1e-6
return ot_loss, excluded_mask, transport_plan
def _partial_sinkhorn(self, C, mass=0.8, reg=0.01, num_iters=20):
"""
带质量的 Sinkhorn 算法（Partial OT）
"""
batch = C.shape[0]
# 初始分配
a = torch.ones(batch, device=C.device) * mass
b = torch.ones(batch, device=C.device) * mass
K = torch.exp(-C / reg)
u = torch.ones_like(a)
for _ in range(num_iters):
v = b / (K.T @ u + 1e-8)
u = a / (K @ v + 1e-8)
transport = u.unsqueeze(1) * K * v.unsqueeze(0)
return transport
def forward(
self,
chosen_rewards: torch.Tensor,
rejected_rewards: torch.Tensor,
chosen_hidden: torch.Tensor, # 中间层表示
rejected_hidden: torch.Tensor,
) -> torch.Tensor:
"""
SelectiveRM 损失
"""
# 1. 语义相似度
sem_sim = self.compute_semantic_similarity(
chosen_hidden.mean(dim=1), # pool
rejected_hidden.mean(dim=1)
)
# 2. 部分 OT 损失
ot_loss, excluded, plan = self.partial_ot_loss(
chosen_rewards, rejected_rewards, sem_sim,
chosen_hidden, rejected_hidden,
)
# 3. 在保留的"干净"样本上计算 BT Loss
kept_mask = ~excluded
if kept_mask.sum() > 0:
bt_loss = -F.logsigmoid(
chosen_rewards[kept_mask] - rejected_rewards[kept_mask]
).mean()
else:
bt_loss = 0.0
# 总损失 = OT损失 + BT损失（只对干净样本）
total_loss = ot_loss * 0.1 + bt_loss
return total_loss, excluded
9.2 过拟合检测
“”“DPO 过拟合检测”“”
def detect_dpo_overfitting(
dpo_trainer,
eval_prompts: List[str],
eval_threshold: float = 0.9
):
“”"
检测 DPO 训练是否过拟合
- chosen 和 rejected 的 log prob 差距过大 → 过拟合
- 准确率接近 100% → 过拟合
“”"
model = dpo_trainer.model
tokenizer = dpo_trainer.tokenizer
with torch.no_grad():
# 计算 log prob 差距
chosen_logps = dpo_trainer._get_log_probs(
model, batch["chosen_ids"], batch["chosen_mask"]
)
rejected_logps = dpo_trainer._get_log_probs(
model, batch["rejected_ids"], batch["rejected_mask"]
)
margin = chosen_logps - rejected_logps
accuracy = (margin > 0).float().mean()
# 预警信号
warnings = []
if accuracy > 0.97:
warnings.append("⚠️ 准确率 > 97%，可能过拟合")
if margin.mean() > 5.0:
warnings.append(f"⚠️ 奖励差距过大: {margin.mean():.2f}")
# 检查 log prob 是否异常高
if chosen_logps.mean() > 0:
warnings.append(f"⚠️ chosen log prob 为正: {chosen_logps.mean():.2f}")
return {
"accuracy": accuracy.item(),
"reward_margin": margin.mean().item(),
"warnings": warnings,
"overfitting": len(warnings) > 0
}
十、对齐的未来：2026下半年趋势
10.1 趋势一览
趋势 状态 说明
GRPO 成为默认对齐方法 ✅ 已确认 DeepSeek R1 效应，TRL 0.10+ 集成
奖励模型噪音问题得到重视 ✅ 已确认 SelectiveRM / ICML 2026
奖励vs策略联合训练 ⚠️ 发展中 避免 reward hacking
自奖励模型（Self-Rewarding） ✅ 部分应用 LLaMA 3 等方法
离线对齐 > 在线对齐 ✅ 趋势 DPO 系优于 PPO 系
Scalable Oversight ✅ 研究热点 观点学习、辩论
Constitutional AI ✅ Claude 4+ 使用 Anthropic 推动
10.2 2026 最佳实践
训练一个高质量的对话模型：
Step 1: 预训练 (25T tokens)
→ Qwen3 级别的基础能力
Step 2: SFT (15K-50K 样本)
→ 学会对话格式
→ 用 DataFlow 风格的高质量数据
Step 3: DPO (50K-100K 偏好对) ⭐ 推荐
→ 对齐人类偏好
→ 稳定、高效、效果好
Step 4 (可选): GRPO 推理优化
→ 针对推理任务进一步优化
→ DeepSeek R1 路线
Step 4 (可选): 奖励模型训练 (SelectiveRM)
→ 有资源的话 +5-10% 效果
面试高频问题
Q: DPO 和 PPO 的核心区别是什么？
A: PPO 需要 4 个模型（策略+参考+奖励+Critic），通过强化学习逐步优化；DPO 通过闭式解将偏好优化转化为监督学习，只需要 2 个模型（策略+参考）。DPO 更简单稳定，但 PPO 在复杂任务上仍能获得更高奖励上限。
Q: GRPO 相比 PPO 的优势是什么？
A: GRPO 不需要 Critic 网络（Value Function），用组内相对奖励替代。对每个 prompt 采样 G 条回复，用组内均值和标准差归一化来估计优势。优点是减少一个模型（140GB节省）+ 训练更稳定。DeepSeek R1 的成功实践证明了 GRPO 的有效性。
Q: 奖励模型的过拟合问题如何解决？
A: 2026年最佳方案是 SelectiveRM（ICML 2026），用最优传输衡量"语义-偏好"一致性，自动排除高噪声样本。此外还有：数据增强、正则化、集成奖励模型、Reward Model Ensemble 等方法。
Q: DeepSeek R1 为什么用 GRPO 而不是 PPO？
A: 1) GRPO 不需要 Critic，训练更稳定 2) 推理任务中"推理正确性"本身就可以作为奖励信号（规则奖励），不需要复杂的奖励模型 3) GRPO 的组内归一化让训练更高效
参考资料
[1] Ouyang et al. Training language models to follow instructions with human feedback. arXiv:2203.02155 (InstructGPT)
[2] Pan et al. SelectiveRM: Selective Reward Modeling via Optimal Transport. ICML 2026
[3] Schulman et al. Proximal Policy Optimization Algorithms. arXiv:1707.06347
[4] Rafailov et al. Direct Preference Optimization: Your Language Model is Secretly a Reward Model. NeurIPS 2023
[5] Shao et al. DeepSeek-R1: Incentivizing Reasoning Capability in LLMs via Reinforcement Learning. arXiv:2501.12948
[6] DeepSeek-AI. DeepSeek-V4 Technical Report. 2026
欢迎加入DeepSeek 技术社区。在这里，你可以找到志同道合的朋友，共同探索AI技术的奥秘。
更多推荐
4
0- 0
已为社区贡献4条内容
所有评论(0)