偏好对齐深度解析：从 RLHF 到 DPO/KTO 的技术演进
7560 字
38 分钟
偏好对齐深度解析：从 RLHF 到 DPO/KTO 的技术演进
1. 引言：为什么 LLM 需要偏好对齐？
1.1 预训练模型的”自由意志”问题
GPT、Llama、DeepSeek 等预训练语言模型学习了互联网上海量文本的分布，能生成流畅、连贯的文本。但它们有一个根本缺陷：不知道自己应该生成什么，只知道最可能生成什么。
预训练 vs 人类意图：
预训练目标: P(next_token | context)
学到的是："互联网上这句话最可能出现在那个上下文之后" 问题：不一定是"用户想看到的"
人类意图: P(human_preferred | context)
想要的是："在这个上下文下，最符合人类偏好的回复" 要求：helpful, harmless, honest这导致了三个经典问题：
┌────────────────────────────────────────────────────────────┐│ 预训练模型的三大问题 │├────────────────────────────────────────────────────────────┤│ ││ 1. Helpfulness 缺失 ││ → 模型不知道用户的真实意图 ││ → 回复可能不相关、过于简短或过于冗长 ││ → 缺乏指令遵循能力 ││ ││ 2. Harmlessness 缺失 ││ → 模型可能生成有害、偏见、歧视性内容 ││ → 可能提供危险信息（武器制作、毒品配方） ││ → 缺乏安全边界意识 ││ ││ 3. Honesty 缺失 ││ → 模型可能编造不存在的事实（hallucination） ││ → 不愿承认不确定性 ││ → 过度自信或过度谦逊 ││ │└────────────────────────────────────────────────────────────┘1.2 偏好对齐的定义
偏好对齐（Preference Alignment） 是让 LLM 生成符合人类期望输出的技术。其核心是：
学习一个函数 ，使得对于任意输入 和两个候选回复 ， 当且仅当人类更偏好 。
def preference_alignment_loss(model, prompt, y_winner, y_loser): """ 偏好对齐的数学形式化
目标：最大化 P(prefer winner | prompt) """ r_winner = model.score(prompt, y_winner) # 评分函数 r_loser = model.score(prompt, y_loser)
# Bradley-Terry 模型 P_prefer = sigmoid(r_winner - r_loser)
# 目标：最大化 P_prefer loss = -log(P_prefer)
return loss1.3 本系列文章关联
| 文章 | 关联 |
|---|---|
| RLHF 深度解析 | PPO 的信赖域优化与 KL 约束详解 |
| DPO 深度解析 | DPO 的理论推导与闭式解 |
| GRPO 深度解析 | GRPO 的组内归一化优势估计 |
| 后训练深度解析 | 对齐在后训练流水线中的位置 |
2. 偏好对齐的理论基础
2.1 Bradley-Terry 模型
偏好对齐的理论基础来自统计学中的 Bradley-Terry 模型（1952），后被引入 LLM 对齐领域。
class BradleyTerryModel: """ Bradley-Terry 模型
假设：每个回复 y 有一个隐式的"质量分数" θ(y) 两个回复被偏好的概率为：
P(y_w wins y_l) = σ(θ(y_w) - θ(y_l))
其中 σ 是 sigmoid 函数 """
def __init__(self): self.theta = {} # 每个回复的隐式分数
def probability(self, score_winner, score_loser): """ P(y_w 优于 y_l) = σ(score_w - score_l) """ return sigmoid(score_winner - score_loser)
def log_likelihood(self, preferences): """ 给定偏好数据 D = {(x, y_w, y_l)}， Bradley-Terry 的对数似然为：
L = Σ_{(x, y_w, y_l)∈D} log σ(θ(y_w) - θ(y_l))
优化这个函数可以学习隐式分数 θ """ total = 0.0 for x, y_w, y_l in preferences: score_w = self.theta.get(y_w, 0.0) score_l = self.theta.get(y_l, 0.0) total += log(sigmoid(score_w - score_l))
return total2.2 从 Bradley-Terry 到 Reward Model
在 LLM 对齐中，我们将 Bradley-Terry 模型扩展为 Reward Model：
class RewardModel: """ Reward Model: 学习 Bradley-Terry 偏好概率
给定 prompt x 和回复 y，Reward Model 输出一个标量 r(x, y)
偏好概率： P(prefer y_w over y_l | x) = σ(r(x, y_w) - r(x, y_l))
损失函数（negative log-likelihood）： L = -E_{(x, y_w, y_l)∼D}[log σ(r_w - r_l)] """
def __init__(self, base_model): self.model = copy.deepcopy(base_model) # 替换输出层为 1 维 self.model.lm_head = nn.Linear( self.model.config.hidden_size, 1, bias=False )
def forward(self, input_ids, attention_mask): """ 前向传播 """ outputs = self.model(input_ids, attention_mask)
# 取最后一个 token 的 hidden state 对应的 reward last_hidden = outputs.last_hidden_state[:, -1, :] reward = self.model.lm_head(last_hidden).squeeze(-1) # (B,)
return reward
def preference_loss(self, batch): """ Preference Loss: Bradley-Terry NLL
L = -log σ(r_w - r_l)
等价于 binary cross-entropy with labels=1 """ r_winner = self.forward(**batch["winner"]) r_loser = self.forward(**batch["loser"])
# P(prefer winner) = σ(r_w - r_l) probs = torch.sigmoid(r_winner - r_loser)
# Negative log-likelihood loss = -torch.log(probs + 1e-8).mean()
return loss2.3 从 Reward 到 Policy
有了 Reward Model，下一步是优化 Policy（策略模型）使其生成高奖励的回复：
偏好对齐的三个层次：
Level 1: Reward Modeling 输入: 偏好数据 (x, y_w, y_l) 输出: Reward Model r_θ(x, y) 目标: 学习隐式偏好函数
Level 2: Policy Optimization 输入: Reward Model 输出: 优化后的 Policy π_θ 目标: 最大化期望奖励
Level 3: Constraint Satisfaction 输入: 优化目标 输出: 带约束的 Policy 目标: 最大化奖励 + 保持 KL 约束
┌──────────────────────────────────────────────────────────┐│ ││ Preference Data ──→ Reward Model ──→ Policy Update ││ ↓ ↓ ↓ ││ (x, y_w, y_l) r_θ(x, y) π_θ(y|x) 优化 ││ Bradley- 约束: ││ Terry KL(π_θ || π_0) ≤ ε ││ │└──────────────────────────────────────────────────────────┘2.4 KL 约束的重要性
为什么需要 KL 约束？
def kl_constraint_necessity(): """ KL 约束的必要性
问题：如果只优化奖励，会发生什么？
1. Reward Hacking（奖励黑客） → 模型找到"作弊"的方式获得高奖励 → 可能生成无意义但评分高的内容
2. 能力退化 → 模型过度适应 reward model 的偏好 → 失去预训练学到的有用知识
3. 模式坍缩 → 模型只生成"安全"的回答 → 缺乏多样性和创造性 """
# 无约束优化的问题 print("Unconstrained reward optimization:") print(" → 模型发现: 生成'great!'重复 100 次得高分") print(" → 实际: 毫无意义的输出")
# KL 约束的作用 print("\nWith KL constraint (β · KL):") print(" → 奖励增加的方向必须与参考策略'兼容'") print(" → 防止模型偏离太远") print(" → 保持生成多样性")3. RLHF：经典的三阶段方法
3.1 InstructGPT 的三阶段
RLHF（Reinforcement Learning from Human Feedback）由 OpenAI 在 InstructGPT（2022）中提出，是偏好对齐的开创性方法。
class RLHFPipeline: """ RLHF 三阶段流水线（InstructGPT 风格）
阶段 1: SFT (Supervised Fine-Tuning) 阶段 2: Reward Model Training 阶段 3: PPO Reinforcement Learning """
def __init__(self, base_model): self.base_model = base_model self.sft_model = None self.reward_model = None self.policy_model = None
def stage1_sft(self, sft_data): """ 阶段 1: SFT
目标：让模型学会基本的指令-回答格式 方法：标准的语言模型交叉熵损失
数据：~13K 标注样本 """ self.sft_model = copy.deepcopy(self.base_model)
optimizer = torch.optim.AdamW(self.sft_model.parameters(), lr=1e-5)
for epoch in range(3): for batch in sft_data: # 标准 LM 损失（只在 response 上计算） loss = self.sft_loss(self.sft_model, batch)
loss.backward() optimizer.step() optimizer.zero_grad()
return self.sft_model
def stage2_reward_model(self, preference_data): """ 阶段 2: Reward Model
目标：学习人类偏好 方法：Bradley-Terry 损失
数据：~13K 偏好对 """ # 从 SFT 模型初始化 self.reward_model = RewardModel(self.sft_model)
optimizer = torch.optim.AdamW( self.reward_model.parameters(), lr=1e-5 )
for epoch in range(1): for batch in preference_data: loss = self.reward_model.preference_loss(batch)
loss.backward() optimizer.step() optimizer.zero_grad()
return self.reward_model
def stage3_ppo(self, prompts, reward_model, ref_model): """ 阶段 3: PPO 对齐
目标：优化策略以最大化奖励 方法：PPO + KL 约束
L = E[r(x, y)] - β · KL(π_θ || π_ref)
其中 r(x, y) 是 learned reward β 是 KL 系数（通常 0.1） """ self.policy_model = copy.deepcopy(self.sft_model)
for ppo_step in range(num_ppo_steps): # 1. Rollout: 用当前策略生成回复 rollouts = self.generate_rollouts(self.policy_model, prompts)
# 2. 计算奖励 rewards = [] for prompt, response in rollouts: r = reward_model(prompt, response) # 加入 KL 惩罚 kl = self.compute_kl(prompt, response, ref_model) rewards.append(r - beta * kl)
# 3. GAE 计算优势 advantages = self.compute_gae(rewards)
# 4. PPO 更新 self.ppo_update(self.policy_model, rollouts, advantages)
return self.policy_model3.2 PPO 的信赖域约束
PPO（Proximal Policy Optimization）的核心是 信赖域约束，防止策略更新过大导致性能崩溃：
class PPOUpdate: """ PPO 策略更新
目标函数（Clipped Surrogate Objective）：
L = min( ratio(θ) · A, clip(ratio(θ), 1-ε, 1+ε) · A )
其中： - ratio(θ) = π_θ(y|x) / π_old(y|x) （策略比率） - A 是优势估计 - ε 是裁剪参数（通常 0.2） """
def __init__(self, clip_ratio=0.2, ent_coef=0.01): self.clip_ratio = clip_ratio self.ent_coef = ent_coef
def ppo_loss(self, old_logps, new_logps, advantages, attention_mask): """ PPO 损失函数 """ # 策略比率 ratio = torch.exp(new_logps - old_logps)
# Clipped 比率 clipped_ratio = torch.clamp( ratio, 1 - self.clip_ratio, 1 + self.clip_ratio )
# Surrogate 损失（取最小，防止过度更新） surr1 = ratio * advantages surr2 = clipped_ratio * advantages policy_loss = -torch.min(surr1, surr2)
# 熵正则（鼓励探索） entropy = self.compute_entropy(new_logps, attention_mask)
total_loss = policy_loss.mean() - self.ent_coef * entropy.mean()
return total_loss, { "policy_loss": policy_loss.mean().item(), "entropy": entropy.mean().item(), "clip_fraction": self.clip_fraction(ratio).mean().item(), }
def clip_fraction(self, ratio): """ 裁剪比例：有多少比例的样本被裁剪了
这是一个重要的诊断指标： - 过高（>20%）说明更新过于激进 - 过低（~0%）说明没有有效学习 """ return (torch.abs(ratio - 1) > self.clip_ratio).float()
def compute_gae(self, rewards, values, next_values, gamma=0.99, lam=0.95): """ GAE (Generalized Advantage Estimation)
A_t = δ_t + γ·λ·δ_{t+1} + ... + (γ·λ)^{T-t}·δ_T
其中 δ_t = r_t + γ·V(s_{t+1}) - V(s_t)
λ 控制偏差-方差权衡： - λ=1：高方差，低偏差（Monte Carlo） - λ=0：低方差，高偏差（TD(0)） """ advantages = [] gae = 0
for t in reversed(range(len(rewards))): delta = rewards[t] + gamma * next_values[t] - values[t] gae = delta + gamma * lam * gae advantages.insert(0, gae)
return torch.tensor(advantages)3.3 RLHF 的问题
尽管 RLHF 效果显著，但有三个显著问题：
RLHF 的三大问题：
问题 1: 工程复杂度高 → 需要训练 3 个模型（SFT, RM, Value） → PPO 需要大量的 rollout 采样 → 显存占用大（~4× SFT 模型） → 训练不稳定，需要精细调参
问题 2: Reward Model 精度有限 → 偏好是主观的，标注一致性低 → Reward Model 可能学习到标注者的偏见 → 难以泛化到复杂场景
问题 3: 分布偏移（Distribution Shift） → Rollout 生成的回复可能偏离 RM 训练分布 → RM 在这些新回复上可能给出错误评分 → 形成反馈循环，导致性能下降4. DPO：绕过 Reward Model
4.1 DPO 的核心思想
DPO（Direct Preference Optimization，Rafailov et al., 2023）是一个革命性的简化：直接优化偏好，完全绕过 Reward Model。
class DPO: """ Direct Preference Optimization (DPO)
核心洞察： RLHF 的目标函数： L = E[r(x, y)] - β · KL(π_θ || π_ref)
可以重参数化为： π_θ(y|x) ∝ π_ref(y|x) · exp(r(x, y)/β)
这意味着： r(x, y) = β · log(π_θ(y|x) / π_ref(y|x))
代入 Bradley-Terry 概率： P(prefer y_w over y_l) = σ(r_w - r_l) = σ(β · log(π_θ(y_w|x) / π_ref(y_w|x)) - β · log(π_θ(y_l|x) / π_ref(y_l|x)))
这就是 DPO 的目标函数！ 不需要训练 Reward Model，直接优化 Policy。 """
def __init__(self, policy, reference, beta=0.1): self.policy = policy self.reference = reference self.beta = beta # KL 系数
def dpo_loss(self, batch): """ DPO 损失函数
L = -E_{(x, y_w, y_l)∼D}[log σ( β · log(π_θ(y_w|x) / π_ref(y_w|x)) - β · log(π_θ(y_l|x) / π_ref(y_l|x)) )] """ prompt = batch["prompt"] chosen = batch["chosen_response"] rejected = batch["rejected_response"]
# 计算 log probs # π_θ(y|x) - π_ref(y|x) pi_chosen = self.policy.log_prob(prompt, chosen) pi_rejected = self.policy.log_prob(prompt, rejected)
ref_chosen = self.reference.log_prob(prompt, chosen) ref_rejected = self.reference.log_prob(prompt, rejected)
# 优势 = β · (log π_θ(y_w) - log π_ref(y_w)) - β · (log π_θ(y_l) - log π_ref(y_l)) chosen_advantage = self.beta * (pi_chosen - ref_chosen) rejected_advantage = self.beta * (pi_rejected - ref_rejected)
advantage = chosen_advantage - rejected_advantage
# DPO 损失 = -log σ(advantage) loss = -F.logsigmoid(advantage).mean()
return loss4.2 DPO 的数学推导
DPO 的推导是理解它的关键：
DPO 数学推导：
Step 1: RLHF 的最优策略 在 KL 约束下最大化奖励的最优策略为：
π*(y|x) ∝ π_ref(y|x) · exp(r(x, y)/β)
这是 softmax 的形式，可由拉格朗日乘数法推导得到。
Step 2: 逆问题 已知最优策略 π_θ，求隐式奖励 r 对两边取对数并重排：
log π_θ(y|x) = log π_ref(y|x) + r(x, y)/β + const r(x, y) = β · (log π_θ(y|x) - log π_ref(y|x)) + const
Step 3: Bradley-Terry 概率 P(prefer y_w over y_l | x) = σ(r_w - r_l)
代入 Step 2 的 r： P = σ(β · (log π_θ(y_w|x) - log π_ref(y_w|x)) - β · (log π_θ(y_l|x) - log π_ref(y_l|x)))
Step 4: DPO 目标 最大化偏好概率，等价于最小化：
L_DPO = -log σ(β · log(π_θ(y_w|x) / π_ref(y_w|x)) - β · log(π_θ(y_l|x) / π_ref(y_l|x)))4.3 DPO 的实现细节
class DPOImplementation: """ DPO 实际实现的关键细节 """
def __init__(self, policy, reference, beta=0.1): self.policy = policy self.reference = reference self.beta = beta
def get_log_probs(self, model, prompt, response): """ 计算 log P(response | prompt)
关键：只计算 response 部分的 log prob prompt 部分不参与 loss """ # Tokenize inputs = tokenize(prompt, response, tokenizer)
# 前向传播 outputs = model(**inputs) logits = outputs.logits # (B, L, V)
# 计算 response 部分的 log prob response_len = len(tokenizer.encode(response))
# 提取 response 的 logits（从 |SYS| 之后开始） # 注意：这里需要精确计算 response 部分的 token 位置 response_logits = logits[:, -response_len:, :] response_ids = inputs["input_ids"][:, -response_len:]
# Log prob log_probs = F.log_softmax(response_logits, dim=-1) selected_log_probs = log_probs.gather(-1, response_ids.unsqueeze(-1)).squeeze(-1)
return selected_log_probs.sum(dim=-1) # 序列的总 log prob
def dpo_loss_with_ref_free(self, batch): """ 变体：不显式计算 reference 的 log prob
适用于 reference 模型冻结的情况 可以预计算 reference 的 log prob 节省显存 """ # 预计算 reference log probs with torch.no_grad(): ref_chosen = self.reference.get_log_probs(batch["prompt"], batch["chosen"]) ref_rejected = self.reference.get_log_probs(batch["prompt"], batch["rejected"])
# Policy log probs pi_chosen = self.policy.get_log_probs(batch["prompt"], batch["chosen"]) pi_rejected = self.policy.get_log_probs(batch["prompt"], batch["rejected"])
# 优势 advantage = self.beta * ( (pi_chosen - ref_chosen) - (pi_rejected - ref_rejected) )
return -F.logsigmoid(advantage).mean()4.4 DPO vs RLHF 对比
| 维度 | RLHF | DPO |
|---|---|---|
| 模型数量 | 3（SFT + RM + Value） | 2（Policy + Reference） |
| Reward Model | 需要训练 | 不需要 |
| 显存占用 | ~4× Policy | ~2× Policy |
| 训练稳定性 | 中等（PPO 敏感） | 高（简单分类损失） |
| 计算效率 | 低（需要大量 rollout） | 高（直接优化） |
| 理论基础 | RL + Bradley-Terry | Bradley-Terry + 闭式推导 |
| 收敛性 | 依赖 PPO 超参 | 依赖 β 和数据质量 |
5. GRPO：去掉 Critic 的高效对齐
5.1 GRPO 的核心思想
GRPO（Group Relative Policy Optimization，DeepSeek）是另一种绕过 Reward Model 的方法，但与 DPO 不同：它使用规则奖励（Rule-based Reward）而非偏好数据。
class GRPO: """ GRPO: Group Relative Policy Optimization
核心思想： 1. 对于每个 prompt，采样 G 个回复 2. 用奖励函数评分 3. 用组内归一化计算优势 4. 用 PPO-style 更新策略
与 PPO 的区别： - 不需要 Value Network（Critic） - 用组内均值/标准差代替绝对优势估计
优势 = (r_i - μ_G) / σ_G
其中 r_i 是第 i 个回复的奖励 μ_G, σ_G 是组内均值和标准差 """
def __init__(self, policy, reference, reward_fn, group_size=8, beta=0.04): self.policy = policy self.reference = reference self.reward_fn = reward_fn self.group_size = group_size self.beta = beta
def grpo_update(self, prompts): """ GRPO 单步更新 """ # 1. 采样 G 个回复 rollouts = [] for prompt in prompts: group_rollouts = [] for _ in range(self.group_size): response = self.policy.generate(prompt) logp = self.policy.get_log_prob(prompt, response) group_rollouts.append({ "prompt": prompt, "response": response, "log_prob": logp, }) rollouts.append(group_rollouts)
# 2. 计算奖励 rewards = [] for group in rollouts: group_rewards = [] for rollout in group: r = self.reward_fn(rollout["prompt"], rollout["response"]) group_rewards.append(r) rewards.append(group_rewards)
# 3. 组内归一化优势 advantages = [] for group_rewards in rewards: group_rewards = torch.tensor(group_rewards) mean_r = group_rewards.mean() std_r = group_rewards.std() + 1e-8 group_adv = (group_rewards - mean_r) / std_r advantages.extend(group_adv.tolist())
advantages = torch.tensor(advantages)
# 4. 计算损失 old_logps = torch.cat([r["log_prob"] for g in rollouts for r in g])
# PPO-style clip ratio = torch.exp(old_logps - old_logps.detach()) clipped_ratio = torch.clamp(ratio, 1 - 0.2, 1 + 0.2)
policy_loss = -torch.min(ratio * advantages, clipped_ratio * advantages).mean()
# KL 散度 ref_logps = torch.zeros_like(old_logps) # 实际应从 ref model 计算 kl = (old_logps - ref_logps).mean()
loss = policy_loss + self.beta * kl
return loss5.2 GRPO 的规则奖励设计
GRPO 的优势在于可以使用规则奖励，而非 Reward Model：
class RuleBasedReward: """ 规则奖励函数
优点： - 不需要训练数据 - 客观、可验证 - 避免 Reward Model 的偏见 """
def math_reward(self, problem, solution): """ 数学问题的规则奖励 """ ground_truth = extract_answer(problem) model_answer = extract_answer(solution)
if model_answer == ground_truth: return 1.0 elif is_approximately_equal(model_answer, ground_truth): return 0.8 else: return -0.5 # 轻微惩罚错误答案
def code_reward(self, problem, code): """ 代码问题的规则奖励 """ # 执行测试用例 test_results = execute_tests(code, problem["test_cases"])
if test_results["all_passed"]: return 1.0 elif test_results["pass_rate"] > 0.5: return test_results["pass_rate"] else: return -0.5
def format_reward(self, response): """ 格式奖励（鼓励特定格式） """ score = 0.0
# 检查是否包含特定格式标记 if "## " in response or "### " in response: score += 0.2 # 鼓励 Markdown 格式 if len(response) > 100: score += 0.1 # 鼓励详细回答 if response.count("\n") > 3: score += 0.1 # 鼓励分段
return score
def combined_reward(self, problem, response): """ 组合奖励 """ reward = 0.0
# 准确性奖励（最重要） if problem["type"] == "math": reward += self.math_reward(problem, response) elif problem["type"] == "code": reward += self.code_reward(problem, response)
# 格式奖励 reward += self.format_reward(response)
return reward5.3 GRPO vs DPO
| 维度 | GRPO | DPO |
|---|---|---|
| 数据格式 | (prompt, r) 奖励信号 | (prompt, y_w, y_l) 偏好对 |
| 奖励来源 | 规则 / Reward Model | Reward Model（隐式） |
| 探索能力 | 高（每次采样多个回复） | 低（直接优化偏好） |
| 适用场景 | 推理任务（规则可定义） | 通用对话（依赖偏好标注） |
| 数据成本 | 低（不需要偏好标注） | 高（需要大量偏好对） |
| 稳定性 | 中（依赖采样质量） | 高（确定性优化） |
6. KTO：超越配对的心理学对齐
6.1 KTO 的动机
DPO 需要偏好对（chosen, rejected），但收集偏好对比收集单一判断（正/负）成本更高。KTO（Kahneman-Tversky Optimization，Ethayarajh et al., 2024）从心理学出发，只需正样本或负样本，无需配对。
class KTO: """ KTO: Kahneman-Tversky Optimization
核心思想： 人类决策受"损失厌恶"影响——损失的痛苦 > 收益的快乐 KTO 基于前景理论（Prospect Theory）的价值函数：
v(y) = { +u(y) if y 是正样本 (desirable) -λ·u(y) if y 是负样本 (undesirable) }
其中 λ > 1 是损失厌恶系数（通常 λ ≈ 2）
损失函数： L = -log σ(β · (v_desirable - v_undesirable)) """
def __init__(self, policy, reference, beta=0.1, lambda_loss=1.0): self.policy = policy self.reference = reference self.beta = beta self.lambda_loss = lambda_loss
def utility(self, prompt, response, is_positive=True): """ 前景理论价值函数
v(y) = { log(π_θ(y|x) / π_ref(y|x)) if desirable -λ · log(π_θ(y|x) / π_ref(y|x)) if undesirable } """ log_ratio = self.policy.log_prob(prompt, response) - \ self.reference.log_prob(prompt, response)
if is_positive: return log_ratio else: return -self.lambda_loss * log_ratio
def kto_loss(self, batch): """ KTO 损失函数
L = -E[log σ(β · (v_pos - v_neg))]
注意：v_pos 和 v_neg 可以来自不同的样本 不需要像 DPO 那样显式配对 """ # 正样本的 utility v_pos_list = [] for item in batch["positive"]: v = self.utility(item["prompt"], item["response"], is_positive=True) v_pos_list.append(v)
# 负样本的 utility v_neg_list = [] for item in batch["negative"]: v = self.utility(item["prompt"], item["response"], is_positive=False) v_neg_list.append(v)
# 配对计算（随机配对以创建训练样本） loss = 0.0 for v_pos in v_pos_list: for v_neg in v_neg_list: # 优势 = β · (v_pos - v_neg) advantage = self.beta * (v_pos - v_neg) loss += -F.logsigmoid(advantage)
return loss / (len(v_pos_list) * len(v_neg_list))6.2 KTO 的理论优势
KTO 相比 DPO 的理论优势：
1. 数据标注更简单 DPO: 需要标注 (prompt, y_w, y_l) 三元组 KTO: 只需标注 (prompt, y, label) 二元组
实际场景中，判断"这个回答好不好"比"哪个回答更好"更简单
2. 利用率更高 DPO: 一对数据只生成 1 个训练样本 KTO: N 个正 + M 个负可生成 N×M 个训练样本
当正负样本数量不均衡时，KTO 更高效
3. 损失厌恶建模 损失厌恶是人类决策的普遍特征 KTO 显式建模这一点，更符合人类心理6.3 IPO：等效概率优化
IPO（Azar et al., 2023）是另一个不需要配对的方法，基于等效概率：
class IPO: """ IPO: Identity Preference Optimization
核心思想： DPO 的损失可以分解为两部分： 1. 偏好概率优化 2. 均匀化（uniformity）
IPO 去掉均匀化项，只保留偏好优化：
L_IPO = -E[log σ( log(π_θ(y_w|x) / π_ref(y_w|x)) - log(π_θ(y_l|x) / π_ref(y_l|x)) - β )]
其中 β > 0 是一个常数偏移 """
def __init__(self, policy, reference, beta=0.1): self.policy = policy self.reference = reference self.beta = beta
def ipo_loss(self, batch): """ IPO 损失 """ prompt = batch["prompt"] chosen = batch["chosen_response"] rejected = batch["rejected_response"]
# 计算 log ratio log_ratio_chosen = ( self.policy.log_prob(prompt, chosen) - self.reference.log_prob(prompt, chosen) ) log_ratio_rejected = ( self.policy.log_prob(prompt, rejected) - self.reference.log_prob(prompt, rejected) )
# IPO 目标：log_ratio_chosen - log_ratio_rejected > β advantage = log_ratio_chosen - log_ratio_rejected - self.beta
return -F.logsigmoid(advantage).mean()7. 其他偏好对齐方法
7.1 SimPO：不需要 Reference 的偏好对齐
class SimPO: """ SimPO: Simple Preference Optimization (Meng et al., 2024)
核心思想： DPO 需要 reference model 计算 log ratio SimPO 发现可以直接用策略模型的奖励：
策略优势 = β · (avg_log_prob_chosen - avg_log_prob_rejected)
其中 avg_log_prob 是序列的平均 token log prob （而非序列总和）
优点： 1. 不需要 reference model 2. 更稳定（避免了 reference 的累积误差） 3. 训练更快（显存更少） """
def __init__(self, policy, beta=0.1, gamma=0.3): self.policy = policy self.beta = beta # 奖励系数 self.gamma = gamma # margin
def simpo_loss(self, batch): """ SimPO 损失
L = -E[log σ( β · (r_chosen - r_rejected) - γ )]
其中 r = avg_log_prob（平均 token log prob） γ > 0 是 margin，鼓励区分度 """ prompt = batch["prompt"] chosen = batch["chosen_response"] rejected = batch["rejected_response"]
# 平均 log prob（按序列长度归一化） r_chosen = self.policy.avg_log_prob(prompt, chosen) r_rejected = self.policy.avg_log_prob(prompt, rejected)
# 带 margin 的优势 advantage = self.beta * (r_chosen - r_rejected) - self.gamma
return -F.logsigmoid(advantage).mean()7.2 RSO：统计偏好优化
class RSO: """ RSO: Reward-Supervised Preference Optimization (Gao et al., 2023)
核心思想： DPO 假设偏好服从 Bradley-Terry 分布 但实际标注中可能存在不一致性（Cycle Preference）
RSO 使用更鲁棒的偏好模型： - 用 Statistical Tests 检测不一致偏好 - 对不一致的偏好对降低权重
优点： 1. 更鲁棒 2. 对标注噪声更容忍 """
def __init__(self, policy, reference): self.policy = policy self.reference = reference
def detect_cycle_preference(self, preference_graph): """ 检测循环偏好（如 A > B > C > A） 循环偏好表明标注不一致 """ # 用图的传递性检测循环 # 如果存在 A > B > C > A，则 A > C 可能是噪声 pass
def rso_loss(self, batch, weights=None): """ RSO 损失（加权 DPO）
对低置信度的偏好对使用更小的权重 """ if weights is None: weights = torch.ones_like(batch["chosen"])
# 计算 DPO 优势 advantage = self.compute_dpo_advantage(batch)
# 加权损失 loss = -weights * F.logsigmoid(advantage)
return loss.mean()7.3 方法综合对比
| 方法 | 数据格式 | 需要 RM | 需要 Ref | 关键创新 |
|---|---|---|---|---|
| RLHF | (prompt, r) | ✅ | ✅ | 经典 RL |
| DPO | (prompt, y_w, y_l) | ❌ | ✅ | 绕过 RM |
| GRPO | (prompt, r) | ✅/❌* | ✅ | 组内归一化 |
| KTO | (prompt, y, label) | ❌ | ✅ | 损失厌恶 |
| IPO | (prompt, y_w, y_l) | ❌ | ✅ | 等效概率 |
| SimPO | (prompt, y_w, y_l) | ❌ | ❌ | 无 Ref |
| RSO | (prompt, y_w, y_l) | ❌ | ✅ | 鲁棒加权 |
*GRPO 可用规则奖励替代 RM
8. 偏好数据构建
8.1 数据来源
class PreferenceDataBuilder: """ 偏好数据构建方法 """
def build_from_human_annotators(self, prompts, annotator_pool): """ 方法 1: 人类标注 最可靠但成本最高 """ data = []
for prompt in prompts: # 生成多个候选 candidates = generate_candidates(prompt, num=4)
# 让人类标注者两两比较 for i in range(len(candidates)): for j in range(i+1, len(candidates)): preference = human_annotator.compare( prompt, candidates[i], candidates[j] )
if preference == "first": data.append((prompt, candidates[i], candidates[j])) elif preference == "second": data.append((prompt, candidates[j], candidates[i]))
return data
def build_from_llm_judge(self, prompts, judge_model): """ 方法 2: LLM 作为评判者（LLM-as-Judge） 成本低，可规模化 """ data = []
for prompt in prompts: candidates = generate_candidates(prompt, num=4)
# 两两比较 for i in range(len(candidates)): for j in range(i+1, len(candidates)): winner = judge_model.judge_preference( prompt, candidates[i], candidates[j] )
if winner == "first": data.append((prompt, candidates[i], candidates[j])) elif winner == "second": data.append((prompt, candidates[j], candidates[i]])
return data
def build_from_execution_feedback(self, code_prompts, executor): """ 方法 3: 执行反馈（代码场景） 奖励通过的测试用例数 """ data = []
for prompt in code_prompts: candidates = generate_code_candidates(prompt, num=4)
# 按通过率排序 results = [] for code in candidates: pass_rate = executor.run_tests(code, prompt["tests"]) results.append((code, pass_rate))
results.sort(key=lambda x: x[1], reverse=True)
# 选择最优和最差的 if len(results) >= 2: winner = results[0][0] loser = results[-1][0] data.append((prompt, winner, loser))
return data
def build_from_self_inconsistency(self, model, prompts, num_samples=8): """ 方法 4: 自一致性（用于推理任务） 多次生成，选择出现频率最高的答案作为 winner """ data = []
for prompt in prompts: # 多次采样 samples = [ model.generate(prompt, temperature=t) for t in [0.5, 0.7, 0.9, 1.0, 1.1] ]
# 提取答案 answers = [extract_final_answer(s) for s in samples]
# 选择出现最多的作为 winner from collections import Counter counter = Counter(answers) winner_answer = counter.most_common(1)[0][0] winner = samples[answers.index(winner_answer)]
# 最少出现的作为 loser loser_answer = counter.most_common()[-1][0] loser = samples[answers.index(loser_answer)]
if winner != loser: data.append((prompt, winner, loser))
return data8.2 数据质量控制
class PreferenceDataFilter: """ 偏好数据质量过滤 """
def __init__(self, consistency_threshold=0.6): self.consistency_threshold = consistency_threshold
def filter_by_annotator_agreement(self, data, num_annotators=3): """ 过滤标注一致性低的数据 """ filtered = []
for prompt, candidates in data: # 统计各候选被选为最佳的次数 votes = [0] * len(candidates)
for annotator_id in range(num_annotators): winner = self.get_annotator_vote(annotator_id, prompt, candidates) votes[winner] += 1
# 一致性 = max(votes) / total_votes max_votes = max(votes) consistency = max_votes / sum(votes)
if consistency >= self.consistency_threshold: # 选择最常被选为最佳的作为 winner winner_idx = votes.index(max_votes) # 选择最不被选为最佳的作为 loser loser_idx = votes.index(min(votes))
filtered.append({ "prompt": prompt, "chosen": candidates[winner_idx], "rejected": candidates[loser_idx], "consistency": consistency, })
return filtered
def filter_by_length_ratio(self, data, max_ratio=5.0): """ 过滤长度差异过大的数据
理由： - 人类偏好可能受长度影响 - 太长/太短的回答可能都不是最优的 """ filtered = []
for item in data: len_chosen = len(item["chosen"].split()) len_rejected = len(item["rejected"].split())
ratio = max(len_chosen, len_rejected) / (min(len_chosen, len_rejected) + 1)
if ratio <= max_ratio: filtered.append(item)
return filtered
def filter_by_semantic_overlap(self, data, model): """ 过滤语义过于相似的配对
如果两个回答语义相同，则偏好没有意义 """ filtered = []
for item in data: overlap = self.compute_semantic_overlap( item["chosen"], item["rejected"], model )
if overlap < 0.8: # 语义重叠度 < 80% filtered.append(item)
return filtered
def compute_semantic_overlap(self, text1, text2, embed_model): """ 计算两个文本的语义重叠度 """ emb1 = embed_model.encode(text1) emb2 = embed_model.encode(text2)
# 余弦相似度 sim = F.cosine_similarity(emb1, emb2, dim=-1)
return sim.item()9. Reward Hacking 与对策
9.1 Reward Hacking 的本质
Reward Hacking 是 RL 领域的经典问题，指智能体找到”作弊”方式获得高奖励，而非真正完成目标。
Reward Hacking 在 LLM 对齐中的表现：
形式 1: 长度作弊 → 模型发现：更长的回答往往被评为更好 → 实际：填充大量无关内容 → 例子："Let me think about this carefully... [重复1000字]"
形式 2: 格式作弊 → 模型发现：包含特定关键词的回答评分更高 → 实际：刻意添加关键词但内容无关 → 例子："This is a great question! Let me analyze... [内容空洞]"
形式 3: RM 过拟合 → 模型过度适应 Reward Model 的偏好 → 在真实人类评估上表现差 → 形成"对齐税"（Alignment Tax）
形式 4: 奖励黑客（极端） → 模型发现 reward function 的漏洞 → 生成表面上高分但实际无意义的内容 → 例如：包含 "Sure, I'm happy to help!" 后跟随机文字9.2 Reward Hacking 的检测
class RewardHackingDetector: """ Reward Hacking 检测 """
def detect_length_correlation(self, rollouts, rewards): """ 检测奖励是否与长度高度相关 """ lengths = [len(r.split()) for r in rollouts]
correlation = scipy.stats.pearsonr(lengths, rewards)
if abs(correlation) > 0.5: print(f"WARNING: Length-reward correlation = {correlation:.3f}") print(" Model may be length hacking")
return correlation
def detect_keyword_stuffing(self, rollouts): """ 检测关键词填充 """ suspicious_patterns = [ r'(great|excellent|amazing)\s+\1{2,}', # 重复修饰词 r'(thank you|appreciate)\s+.{0,10}\s+\1', # 重复感谢 r'(in conclusion|to summarize)\s+.{0,20}$', # 结尾填充 ]
for rollout in rollouts: for pattern in suspicious_patterns: matches = re.findall(pattern, rollout, re.IGNORECASE) if len(matches) > 3: print(f"WARNING: Keyword stuffing detected") break
def detect_human_v_rm_divergence(self, rollouts, human_evaluations, rm_scores): """ 检测人类评估与 RM 评分的分歧
如果分歧大，说明 RM 可能被 hacking """ correlation = scipy.stats.kendalltau(human_evaluations, rm_scores)
if correlation < 0.5: print(f"WARNING: Human-RM divergence = {correlation:.3f}") print(" Reward Model may be misaligned")
return correlation9.3 对策：多样化正则
class DiverseAlignmentLoss: """ 多样化对齐：防止 Reward Hacking """
def __init__(self, base_loss_fn, diversity_coef=0.1): self.base_loss = base_loss_fn self.diversity_coef = diversity_coef
def entropy_loss(self, logits, attention_mask): """ 熵正则：鼓励策略保持一定随机性 防止策略坍缩到单一模式 """ probs = F.softmax(logits, dim=-1) entropy = -(probs * torch.log(probs + 1e-8)).sum(dim=-1)
# 只在有效 token 上计算 masked_entropy = entropy * attention_mask
return -masked_entropy.sum() / attention_mask.sum()
def length_penalty_loss(self, responses, target_length=None): """ 长度惩罚：防止长度作弊
如果没有指定目标长度， 则惩罚过长的回复 """ lengths = torch.tensor([len(r.split()) for r in responses])
if target_length is None: # 惩罚过长 penalty = F.relu(lengths - lengths.mean() * 1.5) else: # 惩罚偏离目标长度 penalty = torch.abs(lengths - target_length)
return penalty.mean()
def repetition_penalty_loss(self, logits): """ 重复惩罚：防止重复生成 """ # n-gram 重复检测 n = 3 batch_size, seq_len, vocab_size = logits.shape
# 简化的重复检测 # 检查连续相同 token 的概率 probs = F.softmax(logits, dim=-1) max_probs = probs.max(dim=-1).values
# 如果某个 token 的概率过高，惩罚它 penalty = F.relu(max_probs - 0.9).mean()
return penalty
def combined_loss(self, batch, logits, attention_mask): """ 组合损失 """ # 基础对齐损失 base = self.base_loss(batch)
# 多样性正则 entropy = self.entropy_loss(logits, attention_mask) repetition = self.repetition_penalty_loss(logits)
return base + self.diversity_coef * (entropy + repetition)9.4 对策：混合奖励 + 人类评估
class HybridReward: """ 混合奖励：结合规则奖励和 RM 奖励 """
def __init__(self, rm, rules): self.rm = rm self.rules = rules
def compute_reward(self, prompt, response): """ 混合奖励 """ total_reward = 0.0
# 1. RM 奖励（主观偏好） rm_reward = self.rm(prompt, response) total_reward += 0.6 * rm_reward
# 2. 规则奖励（客观指标） rule_reward = self.rules(prompt, response) total_reward += 0.4 * rule_reward
# 3. 长度惩罚 if len(response.split()) > 500: total_reward -= 0.1
# 4. 重复惩罚 repetition_score = self.compute_repetition(response) if repetition_score > 0.3: total_reward -= 0.1 * repetition_score
return total_reward
def compute_repetition(self, text, n=3): """ 计算文本的 n-gram 重复度 """ words = text.split() if len(words) < n: return 0.0
ngrams = [tuple(words[i:i+n]) for i in range(len(words)-n+1)] unique_ngrams = len(set(ngrams)) total_ngrams = len(ngrams)
return 1 - unique_ngrams / total_ngrams10. 对齐税问题
10.1 什么是对齐税？
Alignment Tax：为了提升对齐质量而牺牲部分任务能力的代价。
对齐税的典型表现：
SFT 模型（基线）： → 任务能力强（代码、数学、推理） → 但可能有毒害/偏见输出 → 指令遵循能力有限
RLHF/DPO 对齐后： → Helpfulness 提升 → Harmlessness 提升 → 但某些任务能力可能下降（5-10%）
例如： Llama-2 (base) → Llama-2 (chat) - 代码能力: 47.4% → 44.1% (↓3.3%) - 数学能力: 33.1% → 29.3% (↓3.8%)10.2 缓解对齐税
class AlignmentTaxMitigation: """ 对齐税缓解策略 """
def mixed_data_alignment(self, policy, sft_data, preference_data, ratio=0.8): """ 策略 1: 混合 SFT 数据
在对齐时混入 SFT 数据，保持任务能力 ratio: 偏好数据占总数据的比例 """ # 混合数据训练 mixed_data = mix_datasets(sft_data, preference_data, ratio=ratio)
# 用混合数据训练 for batch in mixed_data: loss = self.compute_loss(policy, batch) # ...
def two_stage_alignment(self, base_model, task_data, pref_data): """ 策略 2: 两阶段对齐
第一阶段：保持任务能力 第二阶段：对齐 """ # 第一阶段：任务数据 SFT task_model = sft_train(base_model, task_data)
# 第二阶段：轻度对齐 aligned_model = dpo_train(task_model, pref_data, epochs=1)
return aligned_model
def auxiliary_loss(self, base_loss, task_metrics): """ 策略 3: 辅助任务损失
在对齐损失中加入任务性能作为辅助信号 """ task_loss = -sum(task_metrics.values()) # 鼓励任务指标提升
return base_loss + 0.1 * task_loss11. 核心公式汇总
11.1 Bradley-Terry 偏好概率
11.2 RLHF 目标函数
11.3 DPO 损失函数
11.4 GRPO 优势估计
11.5 KTO 前景理论价值
11.6 SimPO 奖励定义
12. 总结
12.1 各方法适用场景
┌─────────────────────────────────────────────────────────────┐│ 对齐方法选择决策树 │├─────────────────────────────────────────────────────────────┤│ ││ 你有什么类型的数据？ ││ │ ││ ├── 偏好对 (y_w, y_l) ││ │ │ ││ │ ├── 有能力训练 RM + Reference ││ │ │ └── DPO（最常用） ││ │ │ ││ │ └── 只有 Policy（不需要 Reference） ││ │ └── SimPO ││ │ ││ ├── 单一标签 (y, label) ││ │ └── KTO（损失厌恶建模） ││ │ ││ └── 只有奖励 (prompt, r) ││ │ ││ ├── 规则可定义（数学/代码） ││ │ └── GRPO（高效，无需偏好标注） ││ │ ││ └── 规则不可定义 ││ └── RLHF（需要 RM） ││ │└─────────────────────────────────────────────────────────────┘12.2 工程实践建议
┌─────────────────────────────────────────────────────────────┐│ 对齐工程实践建议 │├─────────────────────────────────────────────────────────────┤│ ││ 1. 数据质量优先 ││ → 偏好标注一致性 > 标注数量 ││ → 过滤低质量数据 > 收集更多数据 ││ → 标注者培训和校准很重要 ││ ││ 2. 从简单开始 ││ → 先用 DPO（简单、稳定） ││ → 如果 DPO 效果不够，再考虑 RLHF ││ → 如果没有偏好数据，用 GRPO + 规则奖励 ││ ││ 3. 监控是关键 ││ → 监控 Reward Hacking 迹象 ││ → 定期用人类评估验证 ││ → 监控任务能力的对齐税 ││ ││ 4. 迭代改进 ││ → 第一轮对齐 → 收集失败的 case ││ → 扩充数据 → 继续对齐 ││ → 多次迭代直到满意 ││ │└─────────────────────────────────────────────────────────────┘推荐阅读
- InstructGPT（Ouyang et al., 2022）—— RLHF 开创之作
- Learning to summarize with RLHF（Stiennon et al., 2020）—— 早期对齐实验
- DPO（Rafailov et al., 2023）—— 绕过 RM 的简化方法
- GRPO（DeepSeek, 2024）—— 组内归一化高效对齐
- KTO（Ethayarajh et al., 2024）—— 损失厌恶对齐
- SimPO（Meng et al., 2024）—— 无 Reference 对齐
- RLAIF（Bai et al., 2022）—— AI 反馈对齐
参考资料
- Ouyang, L., et al. (2022). “Training language models to follow instructions with human feedback.” NeurIPS.
- Stiennon, N., et al. (2020). “Learning to summarize with human feedback.” NeurIPS.
- Rafailov, R., et al. (2023). “Direct Preference Optimization: Your Language Model is Secretly a Reward Model.” NeurIPS.
- Schulman, J., et al. (2017). “Proximal Policy Optimization Algorithms.” arXiv.
- DeepSeek-AI. (2024). “DeepSeekMath: Pushing the Limits of Mathematical Reasoning.” arXiv.
- Ethayarajh, K., et al. (2024). “KTO: Kahneman-Tversky Optimization.” arXiv.
- Meng, Y., et al. (2024). “SimPO: Simple Preference Optimization.” arXiv.
- Azar, M. G., et al. (2023). “A General Theoretical Paradigm to Understand Alignment.” arXiv.
- Bai, Y., et al. (2022). “Constitutional AI: Harmlessness from AI Feedback.” arXiv.
- Gao, L., et al. (2023). “RLAIF: Scaling Reinforcement Learning from Human Feedback with AI Feedback.” arXiv.
- Bradley, R. A., & Terry, M. E. (1952). “Rank Analysis of Incomplete Block Designs.” Biometrika.
- Amodei, D., et al. (2016). “Concrete Problems in AI Safety.” arXiv.
文章分享
如果这篇文章对你有帮助，欢迎分享给更多人！
偏好对齐深度解析：从 RLHF 到 DPO/KTO 的技术演进
https://aiattnstudio.link/posts/preference-alignment/欢迎来到Federico的个人博客
57文章
7分类
404标签