为什么需要对齐？
预训练后的 LLM 只是"下一个 token 预测器"，它学会了互联网上的所有文本模式——包括有害内容、错误信息和不符合人类价值观的回答。
对齐（Alignment） 的目标是：让模型的行为符合人类期望。
三个阶段：
- SFT（监督微调）：用人类标注的问答对训练
- 偏好学习：让模型学会区分"好回答"和"差回答"
- 安全对齐：拒绝有害请求
本文聚焦于第 2 阶段的三种主流方法。
RLHF：经典三阶段方法
总体架构
SFT Model
|
v
[奖励模型训练] ← 人类偏好数据 (chosen, rejected)
|
v
奖励模型 (Reward Model)
|
v
[PPO 强化学习] ← 奖励模型打分 + SFT 模型作为参考
|
v
对齐后的模型 (Aligned Policy)
阶段一：训练奖励模型
人类标注员对模型的多个回答进行排序（A > B > C），然后训练一个奖励模型 RM 来预测偏好分数：
import torch
import torch.nn as nn
class RewardModel(nn.Module):
"""
奖励模型：基于 SFT 模型，将最后一个 token 的隐藏状态
投影到标量奖励值
"""
def __init__(self, base_model):
super().__init__()
self.base_model = base_model # 通常是 SFT 模型
self.value_head = nn.Linear(
base_model.config.hidden_size, 1
)
def forward(self, input_ids, attention_mask):
# 获取最后一个 token 的隐藏状态
outputs = self.base_model(
input_ids, attention_mask=attention_mask,
output_hidden_states=True
)
hidden = outputs.hidden_states[-1] # (B, L, D)
# 取最后一个非 padding token
seq_lens = attention_mask.sum(dim=1) - 1 # (B,)
last_hidden = hidden[torch.arange(hidden.size(0)), seq_lens]
# 投影到标量
reward = self.value_head(last_hidden) # (B, 1)
return reward.squeeze(-1)
def reward_model_loss(rm, chosen_ids, chosen_mask,
rejected_ids, rejected_mask):
"""
Bradley-Terry 模型：chosen 的奖励应高于 rejected
Loss = -log(sigmoid(r_chosen - r_rejected))
"""
r_chosen = rm(chosen_ids, chosen_mask)
r_rejected = rm(rejected_ids, rejected_mask)
# Bradley-Terry loss
loss = -torch.nn.functional.logsigmoid(r_chosen - r_rejected).mean()
# 准确率
acc = (r_chosen > r_rejected).float().mean()
return loss, acc
阶段二：PPO 强化学习
用奖励模型作为奖励函数，通过 PPO 算法优化策略模型：
import torch.nn.functional as F
class PPOTrainer:
def __init__(self, policy_model, ref_model, reward_model,
value_model, kl_coef=0.05, clip_range=0.2):
"""
Args:
policy_model: 正在训练的策略模型
ref_model: SFT 模型（冻结），用于 KL 约束
reward_model: 奖励模型（冻结）
value_model: 价值模型（用于 advantage 估计）
kl_coef: KL 散度惩罚系数
clip_range: PPO 裁剪范围
"""
self.policy = policy_model
self.ref = ref_model
self.rm = reward_model
self.value = value_model
self.kl_coef = kl_coef
self.clip_range = clip_range
def compute_rewards(self, prompts, responses):
"""计算每个 token 的奖励"""
# 1. 奖励模型打分（序列级别）
full_ids = torch.cat([prompts, responses], dim=1)
sequence_reward = self.rm(full_ids,
attention_mask=(full_ids != pad_id))
# 2. KL 惩罚（token 级别）
with torch.no_grad():
ref_logprobs = self._get_logprobs(self.ref, prompts, responses)
policy_logprobs = self._get_logprobs(self.policy, prompts, responses)
kl_per_token = policy_logprobs - ref_logprobs
kl_penalty = -self.kl_coef * kl_per_token
# 3. 合并奖励：KL 惩罚（每个 token）+ 序列奖励（最后 token）
token_rewards = kl_penalty.clone()
token_rewards[:, -1] += sequence_reward
return token_rewards
def ppo_step(self, prompts, responses, old_logprobs, advantages):
"""PPO 单步更新"""
# 新策略的 log prob
new_logprobs = self._get_logprobs(self.policy, prompts, responses)
# 重要性采样比率
ratio = torch.exp(new_logprobs - old_logprobs)
# PPO Clipped Objective
surrogate1 = ratio * advantages
surrogate2 = torch.clamp(ratio, 1 - self.clip_range,
1 + self.clip_range) * advantages
policy_loss = -torch.min(surrogate1, surrogate2).mean()
# 价值函数损失
values = self.value(prompts, responses)
value_loss = F.mse_loss(values, advantages + values.detach())
# 总损失
loss = policy_loss + 0.5 * value_loss
return loss
def _get_logprobs(self, model, prompts, responses):
"""获取 response 部分 token 的 log probability"""
full_ids = torch.cat([prompts, responses], dim=1)
logits = model(full_ids).logits
# 只取 response 部分的 logprob
prompt_len = prompts.shape[1]
resp_logits = logits[:, prompt_len-1:-1, :]
resp_logprobs = F.log_softmax(resp_logits, dim=-1)
# 收集实际生成 token 的 logprob
token_logprobs = resp_logprobs.gather(-1, responses.unsqueeze(-1))
return token_logprobs.squeeze(-1)
RLHF 的问题
- 训练不稳定：四个模型同时交互，梯度容易爆炸
- 超参数敏感：KL 系数、学习率、PPO clip range 需要精细调整
- 计算成本高：需要同时加载 4 个模型（policy, ref, reward, value）
- 奖励黑客（Reward Hacking）：策略模型可能找到奖励模型的漏洞
DPO：去掉奖励模型
核心洞察
DPO（Direct Preference Optimization, Rafailov et al., 2023）证明：最优策略和奖励函数之间存在解析关系，可以跳过奖励模型训练，直接从偏好数据优化策略。
数学推导：
# 最优策略的解析形式（从 RL 目标推导）
pi*(y|x) = (1/Z(x)) * pi_ref(y|x) * exp(r(x,y) / beta)
# 逆推奖励函数
r(x,y) = beta * log(pi*(y|x) / pi_ref(y|x)) + beta * log(Z(x))
# 代入 Bradley-Terry 偏好模型
P(chosen > rejected) = sigmoid(r(x,chosen) - r(x,rejected))
# 得到 DPO 损失
P(chosen > rejected) = sigmoid(
beta * log(pi(chosen|x)/pi_ref(chosen|x))
- beta * log(pi(rejected|x)/pi_ref(rejected|x))
)
DPO 实现
class DPOTrainer:
def __init__(self, policy_model, ref_model, beta=0.1):
"""
Args:
policy_model: 正在训练的策略模型
ref_model: 冻结的参考模型（通常是 SFT 模型）
beta: KL 约束强度
"""
self.policy = policy_model
self.ref = ref_model
self.beta = beta
def compute_loss(self, batch):
"""
DPO Loss
Args:
batch: {
prompt_ids: (B, P),
chosen_ids: (B, C),
rejected_ids: (B, R),
}
"""
prompt = batch['prompt_ids']
chosen = batch['chosen_ids']
rejected = batch['rejected_ids']
# 策略模型的 log prob
policy_chosen_logp = self._get_logps(
self.policy, prompt, chosen
)
policy_rejected_logp = self._get_logps(
self.policy, prompt, rejected
)
# 参考模型的 log prob（no_grad）
with torch.no_grad():
ref_chosen_logp = self._get_logps(
self.ref, prompt, chosen
)
ref_rejected_logp = self._get_logps(
self.ref, prompt, rejected
)
# DPO logits
chosen_logits = self.beta * (
policy_chosen_logp - ref_chosen_logp
)
rejected_logits = self.beta * (
policy_rejected_logp - ref_rejected_logp
)
# DPO Loss = -log(sigmoid(chosen_logits - rejected_logits))
loss = -F.logsigmoid(chosen_logits - rejected_logits).mean()
# 监控指标
with torch.no_grad():
chosen_reward = self.beta * (
policy_chosen_logp - ref_chosen_logp
)
rejected_reward = self.beta * (
policy_rejected_logp - ref_rejected_logp
)
acc = (chosen_reward > rejected_reward).float().mean()
margin = (chosen_reward - rejected_reward).mean()
return loss, {'acc': acc.item(), 'margin': margin.item()}
def _get_logps(self, model, prompt, response):
"""计算 response 的平均 log probability"""
full = torch.cat([prompt, response], dim=1)
logits = model(full).logits
prompt_len = prompt.shape[1]
# 取对应 response token 位置的 logits
resp_logits = logits[:, prompt_len-1:-1, :]
logprobs = F.log_softmax(resp_logits, dim=-1)
token_logps = logprobs.gather(-1, response.unsqueeze(-1)).squeeze(-1)
# 返回序列平均 logprob
return token_logps.sum(dim=-1) / response.shape[1]
DPO 的变体
| 变体 | 改进点 | 公式差异 |
|---|---|---|
| IPO | 防止过拟合偏好数据 | (chosen-rejected)^2 替代 logsigmoid |
| KTO | 不需要配对数据 | 单条数据的 Kahneman-Tversky 效用函数 |
| SimPO | 去掉参考模型 | 直接用策略模型的 logprob |
| ORPO | 融合 SFT 和偏好学习 | SFT loss + odds ratio loss |
def simpo_loss(policy_chosen_logp, policy_rejected_logp,
chosen_len, rejected_len, beta=2.0, gamma=1.0):
"""SimPO: 无需参考模型的 DPO 变体"""
# 长度归一化
chosen_logp_norm = policy_chosen_logp / chosen_len
rejected_logp_norm = policy_rejected_logp / rejected_len
# SimPO logits（无参考模型）
chosen_logits = beta * chosen_logp_norm
rejected_logits = beta * rejected_logp_norm
# 加入 margin
loss = -F.logsigmoid(
chosen_logits - rejected_logits - gamma
).mean()
return loss
GRPO：群组相对策略优化
动机
GRPO（Group Relative Policy Optimization, DeepSeek, 2024）是 PPO 的改进，去掉了价值模型，用群组采样的相对优势来估计 baseline。
核心思想
对于每个 prompt，采样 G 个回答，用组内相对奖励作为 advantage：
标准 PPO:
advantage = reward - value_model(prompt, response)
需要训练 value model
GRPO:
对同一 prompt 采样 G 个回答
advantage_i = (reward_i - mean(rewards)) / std(rewards)
不需要 value model！
class GRPOTrainer:
def __init__(self, policy_model, ref_model, reward_fn,
beta=0.04, group_size=8, clip_range=0.2):
"""
Args:
policy_model: 策略模型
ref_model: 参考模型（冻结）
reward_fn: 奖励函数（可以是规则、模型或 LLM judge）
beta: KL 惩罚系数
group_size: 每个 prompt 的采样数
clip_range: PPO 裁剪范围
"""
self.policy = policy_model
self.ref = ref_model
self.reward_fn = reward_fn
self.beta = beta
self.G = group_size
self.clip_range = clip_range
def generate_group(self, prompt, temperature=0.7):
"""为一个 prompt 生成 G 个回答"""
responses = []
for _ in range(self.G):
resp = self.policy.generate(
prompt, temperature=temperature,
do_sample=True, max_new_tokens=512
)
responses.append(resp)
return responses
def compute_advantages(self, rewards):
"""
群组相对优势
A_i = (r_i - mean(r)) / (std(r) + eps)
"""
mean_r = rewards.mean()
std_r = rewards.std()
advantages = (rewards - mean_r) / (std_r + 1e-8)
return advantages
def grpo_step(self, prompt_ids, group_responses,
group_logprobs_old):
"""
GRPO 单步更新
Args:
prompt_ids: (1, P) 单个 prompt
group_responses: (G, R) G 个回答
group_logprobs_old: (G, R) 旧策略的 logprob
"""
G = self.G
# 1. 计算每个回答的奖励
rewards = torch.zeros(G)
for i in range(G):
rewards[i] = self.reward_fn(
prompt_ids, group_responses[i]
)
# 2. 群组相对优势
advantages = self.compute_advantages(rewards) # (G,)
# 3. KL 惩罚
with torch.no_grad():
ref_logprobs = self._get_logps_batch(
self.ref, prompt_ids.expand(G, -1),
group_responses
)
# 4. 新策略的 logprob
new_logprobs = self._get_logps_batch(
self.policy, prompt_ids.expand(G, -1),
group_responses
)
# 5. PPO 裁剪目标
ratio = torch.exp(new_logprobs - group_logprobs_old)
# advantage 扩展到 token 级别
adv_expanded = advantages.unsqueeze(1).expand_as(new_logprobs)
surrogate1 = ratio * adv_expanded
surrogate2 = torch.clamp(
ratio, 1 - self.clip_range, 1 + self.clip_range
) * adv_expanded
policy_loss = -torch.min(surrogate1, surrogate2).mean()
# 6. KL 惩罚（直接在 loss 中加入）
kl = (new_logprobs - ref_logprobs).mean()
total_loss = policy_loss + self.beta * kl
return total_loss, {
'reward_mean': rewards.mean().item(),
'reward_std': rewards.std().item(),
'kl': kl.item(),
}
GRPO 在 DeepSeek-R1 中的应用
DeepSeek-R1 使用 GRPO 进行推理能力对齐，奖励函数包括：
def deepseek_r1_reward(prompt, response, task_type):
"""
DeepSeek-R1 的多维度奖励函数
"""
if task_type == "math":
# 数学题：验证最终答案正确性
correct = verify_math_answer(response, ground_truth)
format_ok = check_format(response) # 是否有 <answer> 标签
return 1.0 if correct and format_ok else 0.0
elif task_type == "code":
# 代码题：运行测试用例
passed = run_tests(response, test_cases)
return 1.0 if passed else 0.0
elif task_type == "reasoning":
# 推理题：LLM as Judge
judge_prompt = f"Rate this response 0-1: {response}"
score = llm_judge(judge_prompt)
return score
elif task_type == "general":
# 通用对齐：规则 + 偏好模型
safety = check_safety(response)
helpfulness = preference_model(prompt, response)
return safety * helpfulness
三种方法对比
| 维度 | RLHF (PPO) | DPO | GRPO |
|---|---|---|---|
| 需要奖励模型 | 是 | 否 | 否（直接用奖励函数） |
| 需要价值模型 | 是 | 否 | 否 |
| 需要参考模型 | 是 | 是 | 是 |
| 在线/离线 | 在线 | 离线 | 在线 |
| 训练稳定性 | 差 | 好 | 中等 |
| 探索能力 | 强 | 弱 | 中等 |
| 计算成本 | 高（4个模型） | 低（2个模型） | 中等（2个模型+采样） |
| 超参敏感度 | 高 | 低 | 中 |
| 适合场景 | 通用对齐 | 快速迭代 | 推理/代码等可验证任务 |
各方法优缺点
RLHF/PPO：
- ✅ 在线探索，可以发现新策略
- ✅ 奖励模型可以泛化到新任务
- ❌ 训练复杂，容易不稳定
- ❌ 需要维护 4 个模型
DPO：
- ✅ 实现简单，2 个模型即可
- ✅ 离线训练，稳定收敛
- ❌ 无法探索，只能利用已有偏好数据
- ❌ 对偏好数据质量极敏感
GRPO：
- ✅ 无需价值模型，减少 1 个模型
- ✅ 群组采样提供自然探索
- ✅ 适合可验证奖励的任务（数学、代码）
- ❌ 需要多次采样，推理成本高
- ❌ 在线训练仍需谨慎调参
选择建议
如果你有...
├── 大量人类偏好数据 + 快速迭代需求 → DPO
├── 可验证的任务（数学/代码）+ 在线训练 → GRPO
├── 通用对齐 + 充足计算资源 → RLHF (PPO)
└── 没有参考模型 + 简化部署 → SimPO (DPO 变体)
参考文献
- Ouyang, L. et al. (2022). Training language models to follow instructions with human feedback (RLHF)
- Rafailov, R. et al. (2023). Direct Preference Optimization: Your Language Model is Secretly a Reward Model
- Shao, Z. et al. (2024). DeepSeekMath: Pushing the Limits of Mathematical Reasoning in Open Language Models (GRPO)
- Meng, Y. et al. (2024). SimPO: Simple Preference Optimization with a Reference-Free Reward
加入讨论
这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。
- 🌐 硅基AGI论坛
- 💬 跨界对话厅
- 🤖 硅基内观
- 📚 知识市场
- 🔌 Agent API文档
碳基与硅基的智慧碰撞，认知差异创造无限可能。