偏好对齐

偏好对齐让模型的输出符合人类期望，是 LLM 安全和有用的关键

在大模型体系中的位置

预训练 (Pre-training)          → 学习语言知识和世界知识
    ↓
监督微调 (SFT)                 → 学习指令跟随能力
    ↓
偏好对齐  ← 你在这里            → 学习人类偏好，安全有用
  ├── RLHF (PPO)               → 经典方案：奖励模型 + 强化学习
  ├── DPO                      → 无需奖励模型，直接偏好优化
  ├── GRPO                     → DeepSeek 方案：组内相对排名
  └── KTO                      → 无需成对数据，前景理论启发

SFT 后的模型能遵循指令，但可能生成有害、不准确或低质量的回答。偏好对齐通过人类反馈引导模型学习"什么样的回答更好"，使模型既 有用 (helpful) 又 安全 (harmless)。

RLHF 概述

RLHF (Reinforcement Learning from Human Feedback) 是 ChatGPT 成功的关键技术，完整 pipeline 包括三个阶段：

阶段一：SFT 训练       → 得到 SFT 模型 (π_sft)
    ↓
阶段二：训练 Reward Model → 学习人类偏好打分
    ↓
阶段三：PPO 优化        → 用 RL 最大化奖励，同时不偏离 SFT 模型太远

Bradley-Terry 偏好模型

RLHF 的数学基础是 Bradley-Terry 模型——给定提示 $x$，人类偏好回答 $y_w$（chosen）胜过 $y_l$（rejected）的概率为：

$$P(y_w\succ y_l|x)=\sigma (r_{\theta }(x,y_w)-r_{\theta }(x,y_l))$$

其中 $\sigma$ 是 sigmoid 函数，$r_{\theta }$ 是奖励模型。训练目标是最大化这个概率：

$${\mathcal{L}}_{\text{RM}}=-\log \sigma (r_{\theta }(x,y_w)-r_{\theta }(x,y_l))$$

Reward Model 训练

模型架构

奖励模型的架构与预训练模型相同，但将 next-token prediction 的分类头替换为输出标量奖励的回归头：

from transformers import LlamaForCausalLM, LlamaForSequenceClassification, LlamaConfig

config = LlamaConfig(
    vocab_size=100, hidden_size=256,
    intermediate_size=512, num_hidden_layers=2,
    num_attention_heads=4, num_key_value_heads=4,
)

# 从预训练模型初始化奖励模型
model = LlamaForCausalLM(config)
model.save_pretrained('./lm_pretrained')

# 关键：num_labels=1，输出标量奖励
rm_model = LlamaForSequenceClassification.from_pretrained(
    './lm_pretrained', num_labels=1
)
# 原始的 lm_head 被替换为: score = Linear(hidden_size, 1)

训练（含 Margin Loss）

LLaMA 2 使用了带 margin 的损失函数，margin 由偏好评级决定：

$$\mathcal{L}=-\log \sigma (r_{\theta }(x,y_c)-r_{\theta }(x,y_r)-m(r))$$

# 奖励模型训练：让 chosen 的得分高于 rejected
X_chosen = torch.randint(0, 100, (1, 10))
X_rejected = torch.randint(0, 100, (1, 10))
margin = 3.0  # Margin 越大，表示 chosen 明显优于 rejected

rm_chosen = rm_model(input_ids=X_chosen).logits     # chosen 的奖励分数
rm_rejected = rm_model(input_ids=X_rejected).logits  # rejected 的奖励分数

# 标准 loss
loss = -torch.sigmoid(rm_chosen - rm_rejected).log()
# 带 margin 的 loss（LLaMA 2 方案）
loss_with_margin = -torch.sigmoid(rm_chosen - rm_rejected - margin).log()

LLaMA 2 的双奖励选择

LLaMA 2 使用两个奖励模型——安全性和有用性——并根据安全分数选择：

$$R_c(g|p)=\{\begin{array}{ll}{R}_s(g|p)& \text{if is\_safety}(p)\text{ or }{R}_s(g|p)<0.15\\ R_h(g|p)& \text{otherwise}\end{array}$$

def llama2_reward_select(reward_safety, reward_helpfulness):
    """安全分数低时优先使用安全奖励，否则用有用性奖励"""
    return reward_safety if reward_safety < 0.15 else reward_helpfulness

奖励后处理：Whiten + KL 惩罚

# 1. 逆 Sigmoid（Logit 函数）：稳定数值
def inverse_sigmoid(x):
    return torch.log(x / (1 - x))

# 2. Whiten：标准化奖励分布
def whiten(values, shift_mean=True):
    mean, var = torch.mean(values), torch.var(values)
    whitened = (values - mean) * torch.rsqrt(var + 1e-8)
    return whitened

# 3. KL 惩罚：防止模型偏离 SFT 策略太远
# R(g|p) = R_hat(g|p) - beta * D_KL(pi_theta || pi_0)
beta = 0.01
kl = logprobs_new - logprobs_ref
reward_final = rm_score - beta * kl

PPO (Proximal Policy Optimization)

PPO 是 RLHF 中最常用的强化学习算法。它通过 clip 机制 限制策略更新幅度，确保训练稳定。

RLHF-PPO 需要四个模型

class RLHFModelBundle():
    def __init__(self, policy, ref_policy, reward_model, value_net):
        self.policy = policy            # 策略模型（要优化的 LLM）
        self.ref_policy = ref_policy    # 参考模型（冻结的 SFT 模型）
        self.reward_model = reward_model  # 奖励模型（冻结）
        self.value_net = value_net      # 价值网络（估计状态价值）

# 价值网络：在 LLM 基础上加一个线性头输出标量
class ModelValueHead(torch.nn.Module):
    def __init__(self, model):
        super().__init__()
        self.model = model
        self.dropout = torch.nn.Dropout(0.05)
        self.summary = torch.nn.Linear(model.config.hidden_size, 1)

    def forward(self, xy):
        hidden_states = self.model(**xy, output_hidden_states=True).hidden_states
        last_hidden = hidden_states[-1]
        output = self.dropout(last_hidden)
        return self.summary(output)[:, :, 0]  # 输出标量

PPO 训练流程

class TrainingArgs():
    def __init__(self):
        self.ppo_epochs = 3         # 每批数据训练轮数
        self.mini_batch_size = 1
        self.epochs = 2             # 总轮数
        self.kl_ctl = 0.01          # KL 惩罚系数
        self.vl_coef = 0.01         # 价值损失系数
        self.lam = 0.9              # GAE lambda
        self.gamma = 0.9            # 折扣因子
        self.cliprange = 0.2        # PPO clip 范围
        self.cliprange_value = 0.2  # 价值函数 clip 范围

def ppo_train(bundle, args, x):
    """PPO 训练主循环"""
    for i in range(args.epochs):
        # 1. 策略模型生成回答
        xy = get_generate(bundle.policy, x, max_new_tokens)
        # 2. Reward Model 打分
        rewards = get_reward(bundle.reward_model, xy)
        # 3. PPO 更新
        loss, reward = ppo_step(bundle, args, x, xy, rewards)
    return loss, reward

KL 惩罚奖励

def compute_rewards_kl(reward, ref_logprobs, old_logprobs, kl_ctl):
    """将 KL 惩罚融入奖励：每个 token 位置减去 KL 散度，最后一个 token 加上 RM 分数"""
    kl = old_logprobs - ref_logprobs
    kl_reward = -kl_ctl * kl
    kl_reward[:, -1] += reward[:, 0]  # RM 奖励只加在最后一个 token
    return kl_reward

GAE (Generalized Advantage Estimation)

def get_GAE(rewards, mask, values, gamma, lam):
    """计算广义优势估计：结合 TD 误差和折扣因子"""
    lastgaelam = 0
    advantages_reversed = []
    gen_len = rewards.shape[-1]
    values = values * mask
    rewards = rewards * mask

    for t in reversed(range(gen_len)):
        nextvalues = values[:, t + 1] if t < gen_len - 1 else 0.0
        delta = rewards[:, t] + gamma * nextvalues - values[:, t]
        lastgaelam = delta + gamma * lam * lastgaelam
        advantages_reversed.append(lastgaelam)

    advantages = torch.stack(advantages_reversed[::-1]).transpose(0, 1)
    return advantages

PPO 的核心损失函数

PPO 的总损失由 策略损失 和 价值损失 组成：

def get_policy_loss(logprobs, logprobs_old, advantages, mask, cliprange):
    """PPO 策略损失：clip 机制限制更新幅度"""
    ratio = torch.exp(logprobs - logprobs_old)                # 新旧策略比
    pg_losses = -advantages * ratio                           # 无 clip 损失
    pg_losses2 = -advantages * torch.clamp(                   # 有 clip 损失
        ratio, 1.0 - cliprange, 1.0 + cliprange
    )
    pg_loss = masked_mean(torch.max(pg_losses, pg_losses2), mask)
    return pg_loss

def get_value_loss(advantages, values, values_old, mask, cliprange_value):
    """价值网络损失：同样使用 clip"""
    returns = advantages + values_old
    vpredclipped = clip_by_value(
        values, values_old - cliprange_value, values_old + cliprange_value
    )
    vf_losses1 = (values - returns) ** 2
    vf_losses2 = (vpredclipped - returns) ** 2
    return 0.5 * masked_mean(torch.max(vf_losses1, vf_losses2), mask)

def compute_loss(mini_batchs, ppo_config):
    """PPO 总损失 = 策略损失 + 价值损失系数 * 价值损失"""
    GAE = get_GAE(mini_batchs['rewards_kl'], mini_batchs['mask'],
                  mini_batchs['values'], ppo_config.gamma, ppo_config.lam)
    pg_loss = get_policy_loss(mini_batchs['logprobs'], mini_batchs['logprobs_old'],
                              GAE, mini_batchs['mask'], ppo_config.cliprange)
    vl_loss = get_value_loss(GAE, mini_batchs['values'], mini_batchs['values_old'],
                             mini_batchs['mask'], ppo_config.cliprange_value)
    loss = pg_loss + ppo_config.vl_coef * vl_loss
    return loss, pg_loss, vl_loss

DPO (Direct Preference Optimization)

DPO 的核心突破：消除了 Reward Model 和 RL 训练，将偏好对齐简化为一个分类损失。

从 RLHF 到 DPO 的数学推导

RLHF 的优化目标是：

$$\underset{{\pi }_{\theta }}{max}{\mathbb{E}}_{x,y\sim {\pi }_{\theta }}[r(x,y)]-\beta D_{KL}[{\pi }_{\theta }(y|x)||{\pi }_{\text{ref}}(y|x)]$$

这个 KL 约束的优化问题有闭合解：

$${\pi }^{\ast }(y|x)=\frac{1}{Z(x)}{\pi }_{\text{ref}}(y|x)\exp (\frac{1}{\beta }r(x,y))$$

将奖励函数反解出来：

$$r(x,y)=\beta \log \frac{{\pi }^{\ast }(y|x)}{{\pi }_{\text{ref}}(y|x)}+\beta \log Z(x)$$

代入 Bradley-Terry 模型（$Z(x)$ 被消掉），得到 DPO 损失：

$${\mathcal{L}}_{\text{DPO}}({\pi }_{\theta };{\pi }_{\text{ref}})=-{\mathbb{E}}_{(x,y_w,y_l)\sim \mathcal{D}}[\log \sigma (\beta \log \frac{{\pi }_{\theta }(y_w|x)}{{\pi }_{\text{ref}}(y_w|x)}-\beta \log \frac{{\pi }_{\theta }(y_l|x)}{{\pi }_{\text{ref}}(y_l|x)})]$$

DPO 损失的直觉

• 增大 ${\pi }_{\theta }(y_w|x)$：让模型更可能生成 chosen 回答
• 减小 ${\pi }_{\theta }(y_l|x)$：让模型更不可能生成 rejected 回答
• $\beta$ 控制偏离参考策略的程度：$\beta$ 越大，优化越保守

PyTorch 实现 DPO

import torch.nn.functional as F

def compute_log_probs(model_logits, token_ids):
    """从模型输出的 logits 中提取每个位置对应 token 的 log 概率
    
    Args:
        model_logits: 模型输出  [batch, seq_len, vocab_size]
        token_ids:    目标 token [batch, seq_len]
    Returns:
        per_token_logps: 每个 token 的 log 概率 [batch, seq_len]
    """
    log_dist = F.log_softmax(model_logits, dim=-1)            # 归一化为 log 概率分布
    per_token_logps = log_dist.gather(2, token_ids.unsqueeze(2)).squeeze(2)
    return per_token_logps

# DPO 前向：需要当前模型和参考模型分别对 chosen/rejected 计算 log prob
with torch.no_grad():
    ref_chosen_logits = ref_model(input_ids=x_chosen).logits
    ref_rejected_logits = ref_model(input_ids=x_rejected).logits
policy_chosen_logits = model(input_ids=x_chosen).logits
policy_rejected_logits = model(input_ids=x_rejected).logits

# 计算 log 概率
logp_chosen_ref = compute_log_probs(ref_chosen_logits, x_chosen)
logp_chosen = compute_log_probs(policy_chosen_logits, x_chosen)
logp_rejected_ref = compute_log_probs(ref_rejected_logits, x_rejected)
logp_rejected = compute_log_probs(policy_rejected_logits, x_rejected)

# DPO Loss 计算
beta = 0.1
policy_diff = logp_chosen - logp_rejected             # log(pi(yw)/pi(yl))
baseline_diff = logp_chosen_ref - logp_rejected_ref    # log(pi_ref(yw)/pi_ref(yl))
dpo_logits = policy_diff - baseline_diff               # DPO logits
losses = -F.logsigmoid(beta * dpo_logits)              # DPO loss
loss = losses.mean()

DPO 完整训练循环

optimizer = optim.SGD(model.parameters(), lr=0.1)
num_epochs = 100

for epoch in range(num_epochs):
    optimizer.zero_grad()

    # 参考模型不更新梯度
    with torch.no_grad():
        ref_chosen_out = ref_model(**x_chosen).logits
        ref_rejected_out = ref_model(**x_rejected).logits
    policy_chosen_out = model(**x_chosen).logits
    policy_rejected_out = model(**x_rejected).logits

    # 计算 log prob
    logp_chosen = compute_log_probs(policy_chosen_out, prompt_chosen)
    logp_rejected = compute_log_probs(policy_rejected_out, prompt_rejected)
    logp_chosen_ref = compute_log_probs(ref_chosen_out, prompt_chosen)
    logp_rejected_ref = compute_log_probs(ref_rejected_out, prompt_rejected)

    # DPO loss
    beta = 0.1
    policy_diff = logp_chosen - logp_rejected
    baseline_diff = logp_chosen_ref - logp_rejected_ref
    dpo_logits = policy_diff - baseline_diff
    losses = -F.logsigmoid(beta * dpo_logits) * label
    loss = losses.sum(-1) / attention_mask.sum()

    loss.backward()
    optimizer.step()

DPO 的过拟合问题与 IPO

DPO 存在过拟合风险：${\pi }_{\theta }(y_l)$ 可能被压到 0，导致 BT 模型得分趋向 $+\mathrm{\infty }$。IPO (Identity Preference Optimization) 通过将损失改为二次形式来避免：

$${\mathcal{L}}_{\text{IPO}}=\mathbb{E}{(h_{\pi }(y_w,y_l,x)-\frac{{\tau }^{-1}}{2})}^2$$

# IPO loss：回归损失而非分类损失
constant = 1.0 / (beta * 2.0)
losses_ipo = torch.square(dpo_logits - constant) * label

GRPO (Group Relative Policy Optimization)

GRPO 是 DeepSeek 提出的方法，创新在于：不需要 Reward Model 和 Critic 网络，通过组内相对排名计算优势函数。

GRPO 损失函数

$${\mathcal{L}}_{\text{GRPO}}(\theta )=-\frac{1}{G}\sum _{i=1}^G\frac{1}{|o_i|}\sum _{t=1}^{|o_i|}[\frac{{\pi }_{\theta }(o_{i,t}|q,o_{i,<t})}{{[{\pi }_{\theta }(o_{i,t}|q,o_{i,<t})]}_{\text{no\_grad}}}{\hat{A}}_{i,t}-\beta D_{KL}[{\pi }_{\theta }||{\pi }_{\text{ref}}]]$$

步骤 1：Group Sampling（组采样）

对同一个问题，模型生成 $G$ 个不同的回答：

def grpo_rejection_sampling(model, x, max_new_tokens=10):
    """对同一输入采样多个回答"""
    y = model.generate(input_ids=x, max_new_tokens=max_new_tokens, do_sample=True)
    return y

# 将同一个问题复制 G 次，一次性采样 G 个回答
grpo_samples_nums = 3
input_x = [format_prompt(X[0])] * grpo_samples_nums
input_x_tensor = torch.tensor(input_x, dtype=torch.long)
grpo_xy = grpo_rejection_sampling(model, input_x_tensor, max_new_tokens)

步骤 2：Rule-based Reward（规则奖励）

GRPO 的奖励通常是规则定义的（特别适合数学/代码任务）：

def rule_reward(response, label_id):
    """规则奖励：检查回答是否包含正确答案"""
    for i in range(len(response) - 2):
        if (response[i] == DEFINE_ANSWER_START and
            response[i + 1] == label_id and
            response[i + 2] == DEFINE_ANSWER_END):
            return True
    return False

def think_reward(response):
    """格式奖励：检查回答是否包含 <think>...</think> 标签"""
    found_start = False
    for num in response:
        if num == DEFINE_THINK_START:
            found_start = True
        elif num == DEFINE_THINK_END and found_start:
            return True
    return False

步骤 3：Group Relative Advantage（组内相对优势）

这是 GRPO 的核心创新——不需要 Critic 网络，直接用组内统计计算优势：

$${\hat{A}}_i=\frac{r_i-\text{mean}(\mathbf{r})}{\text{std}(\mathbf{r})+ϵ}$$

def grpo_advantage(rewards):
    """组内相对优势：标准化组内奖励"""
    epsilon = 0.0001
    rewards = torch.tensor(rewards, dtype=torch.float)
    A = (rewards - rewards.mean()) / (rewards.std() + epsilon)
    return A

# 不同采样结果的优势值
print(grpo_advantage([1,0,0,0,0,0]))  # 唯一正例，优势最大
# tensor([ 2.2361, -0.4472, -0.4472, -0.4472, -0.4472, -0.4472])

print(grpo_advantage([1,1,1,1,1,1]))  # 全部正确，优势为 0（无法学习）
# tensor([0., 0., 0., 0., 0., 0.])

print(grpo_advantage([0,0,0,0,0,0]))  # 全部错误，优势为 0（无法学习）
# tensor([0., 0., 0., 0., 0., 0.])

关键洞察：全对或全错时优势为 0，模型无法学习。越少正例，正确回答的优势越大。采样数 $G$ 越大，优势估计越准确。

步骤 4：GRPO Loss

def approx_kl(pi_logprob, pi_ref_logprob):
    """近似 KL 散度（K3 估计）"""
    return pi_ref_logprob.exp() / pi_logprob.exp() - (pi_ref_logprob - pi_logprob) - 1

def grpo_loss(pi_logprob, pi_old_logprob, pi_ref_logprob, advantage, input_len):
    """GRPO 损失函数"""
    epsilon = 0.2
    beta = 0.01
    bs, seq_len = pi_logprob.shape

    advantage = advantage.unsqueeze(dim=1)  # 广播到每个 token

    # PPO 风格的 ratio 和 clip
    ratio = torch.exp(pi_logprob - pi_old_logprob)
    ratio_clip = torch.clamp(ratio, 1 - epsilon, 1 + epsilon)
    policy_gradient = torch.minimum(ratio * advantage, ratio_clip * advantage)

    # KL 正则
    kl = approx_kl(pi_logprob, pi_ref_logprob)

    # 只对 response 部分计算 loss
    group_num, len_oi = pi_logprob.shape
    len_oi = len_oi - input_len
    len_oi = torch.tensor([len_oi] * group_num, dtype=torch.long)

    mask = torch.zeros(bs, seq_len)
    mask[:, input_len:] = 1

    loss = (policy_gradient - beta * kl) * mask
    loss = (-1 / group_num) * loss / len_oi.unsqueeze(dim=1)
    return loss.sum()

GRPO 中的 KL 散度分析

GRPO 使用 KL 散度约束策略不偏离参考模型太远。不同的 KL 估计方式在偏差和方差上有显著差异，直接影响训练稳定性。

三种 KL 估计方式

设 $r=\frac{p(x)}{q(x)}$，从 $q$ 分布采样时，有三种常用的 KL 估计：

估计方式	公式	特点
K1 (log-ratio)	$-\log r$	无偏但有负值，方差最大
K2 (squared log)	$\frac{(\log r{)}^2}{2}$	恒正，方差较小，但有偏（偏大）
K3 (Schulman 近似)	$r-1-\log r$	恒正，方差较小，近似无偏

参考：John Schulman - Approximating KL Divergence

为什么 K3 恒为正？

K3 的形式为 $f(r)=r-1-\log r$。由于 $\log r\le r-1$（对数不等式），所以 $r-1-\log r\ge 0$，等号当且仅当 $r=1$（即 $p=q$）时成立。

import torch

# 验证三种 KL 估计
p = torch.distributions.Normal(0, 1)
q = torch.distributions.Normal(0.5, 1)
x = q.sample((100_000,))

true_kl = torch.distributions.kl_divergence(p, q)  # 解析解 = 0.125
logr = p.log_prob(x) - q.log_prob(x)

k1 = -logr                          # 无偏，但方差 ~4x true_kl
k2 = logr ** 2 / 2                  # 有偏（偏大 ~5%），方差 ~1.5x
k3 = logr.exp() - 1 - logr          # 近似无偏，方差 ~1.5x

print(f"True KL: {true_kl:.4f}")
print(f"K1: mean={k1.mean():.4f}, std={k1.std():.4f}")
print(f"K2: mean={k2.mean():.4f}, std={k2.std():.4f}")
print(f"K3: mean={k3.mean():.4f}, std={k3.std():.4f}")
# K3 均值最接近真实 KL，且方差远小于 K1

GRPO 使用 K3 的原因

GRPO 选择 K3（即 exp(logr) - logr - 1）作为 KL penalty：

def approx_kl(pi_logprob, pi_ref_logprob):
    """KL penalty（K3 形式）"""
    # r = pi_ref / pi_theta = exp(log_ref - log_pi)
    logr = pi_ref_logprob - pi_logprob
    return logr.exp() - logr - 1      # 恒 >= 0，方差小

相比 K1（log_pi - log_ref），K3 不会产生负值，训练更稳定。相比 K2，K3 的偏差更小。这是 GRPO 中 KL 约束项的默认选择。

KL 估计方式的选择原则

• K1：数学上最优（无偏），但方差太大，实际训练中波动剧烈
• K2：简单高效，但系统性偏大，会过度惩罚策略偏移
• K3：兼顾低偏差和低方差，是 GRPO / PPO 实践中的首选

KTO (Kahneman-Tversky Optimization)

KTO 基于行为经济学的 前景理论 (Prospect Theory)，核心优势：不需要成对的偏好数据，只需要独立标注"好的回答"和"差的回答"。

动机：为什么不需要成对数据？

DPO 要求每条训练样本包含 (prompt, chosen, rejected) 三元组——同一个 prompt 下的一对好/坏回答。但在实际标注中：

• 成对标注成本高（标注员需要同时阅读两个回答并比较）
• 很多场景下只有独立的质量判断（"这个回答好" / "这个回答差"）
• 不同 prompt 下的好/坏回答也包含偏好信息

KTO 的突破在于：每条样本只需要一个 binary 标签（好/坏），不需要成对比较。

前景理论 (Prospect Theory) 的启发

KTO 的损失函数设计源自 Kahneman 和 Tversky 的 前景理论（2002 年诺贝尔经济学奖）：

1. 损失厌恶 (Loss Aversion)：人对损失的敏感度高于对同等收益的敏感度。丢 100 元的痛苦 > 捡到 100 元的快乐
2. 参考点依赖 (Reference Dependence)：人评估收益/损失时依赖一个参考点，而非绝对值

映射到 LLM 对齐：

• 参考点 = 参考模型 ${\pi }_{\text{ref}}$ 的 KL 散度
• 收益 = chosen 回答相对参考点的提升
• 损失 = rejected 回答相对参考点的下降
• 损失厌恶 = 对差回答施加更大的惩罚权重

数学公式

KTO 的损失函数：

$${\mathcal{L}}_{\text{KTO}}({\pi }_{\theta },{\pi }_{\text{ref}})={\mathbb{E}}_{x,y\sim \mathcal{D}}[w(y)(1-v_{\text{KTO}}(x,y;\beta ))]$$

其中：

$$r_{\text{KTO}}(x,y)=\beta \log \frac{{\pi }_{\theta }(y|x)}{{\pi }_{\text{ref}}(y|x)},z_{\text{ref}}={\mathbb{E}}_{x^{\prime }}[\beta \text{KL}({\pi }_{\theta }\parallel {\pi }_{\text{ref}}|x^{\prime })]$$$$v_{\text{KTO}}(x,y;\beta )=\{\begin{array}{ll}\sigma (r_{\text{KTO}}-z_{\text{ref}})& \text{if }y\sim y_{\text{desirable}}\\ \sigma (z_{\text{ref}}-r_{\text{KTO}})& \text{if }y\sim y_{\text{undesirable}}\end{array}$$$$w(y)=\{\begin{array}{ll}{\lambda }_D& \text{if }y\sim y_{\text{desirable}}\\ {\lambda }_U& \text{if }y\sim y_{\text{undesirable}}\end{array}$$

• ${\lambda }_D/{\lambda }_U$ 是非对称权重，体现损失厌恶（通常 ${\lambda }_D>{\lambda }_U$）
• $z_{\text{ref}}$ 是参考点（KL 散度的期望值）

数据格式

# KTO 数据不需要成对，每条独立标注
kto_dataset = {
    "prompt": ["Hey, hello", "How are you", "What is your name?", "What is your name?"],
    "completion": ["Hi!", "Fine", "I'm a bot.", "I'm Bob."],
    "label": [True, False, True, False],  # True=好回答, False=差回答
}

KTO 损失函数实现

KTO 实现参考了 Ethayarajh et al. (2024) 原始论文的损失函数定义。

import torch
import torch.nn.functional as F

def kto_loss(logits, ref_logits, input_ids, label_mask, labels, beta=0.1):
    """
    KTO 损失函数：基于前景理论的非对称优化

    Args:
        logits:     当前策略的 logits       [batch, seq_len, vocab]
        ref_logits: 参考策略的 logits       [batch, seq_len, vocab]
        input_ids:  输入 token ids          [batch, seq_len]
        label_mask: completion 部分的 mask   [batch, seq_len]
        labels:     binary 标签 (True/False) [batch]
        beta:       温度参数
    """
    # 1. 计算 log probabilities
    def get_logps(lgt, ids, mask):
        per_token = torch.gather(
            lgt.log_softmax(-1), dim=2, index=ids.unsqueeze(2)
        ).squeeze(2)
        return (per_token * mask / mask.sum()).sum(-1)

    logps = get_logps(logits, input_ids, label_mask)
    ref_logps = get_logps(ref_logits, input_ids, label_mask)

    # 2. 计算 log ratio（隐式奖励）
    ratio = logps - ref_logps                    # r_KTO = beta * log(pi/pi_ref)
    kl = (logps - ref_logps).mean().detach()     # 参考点 z_ref

    # 3. 分离 chosen / rejected
    chosen_mask = labels == True
    rejected_mask = labels == False

    # 4. 非对称损失（前景理论的核心）
    desirable_weight = 1.33    # lambda_D：好回答权重更大
    undesirable_weight = 1.0   # lambda_U：差回答权重

    chosen_losses = 1 - F.sigmoid(beta * (ratio[chosen_mask] - kl))
    rejected_losses = 1 - F.sigmoid(beta * (kl - ratio[rejected_mask]))

    loss = torch.cat([
        desirable_weight * chosen_losses,
        undesirable_weight * rejected_losses
    ]).mean()
    return loss

KTO vs DPO 对比

对比项	DPO	KTO
数据格式	成对偏好 (chosen, rejected)	独立 binary (好/坏)
标注成本	高（需要比较两个回答）	低（只需判断好坏）
理论基础	Bradley-Terry 偏好模型	Kahneman-Tversky 前景理论
参考点	隐含在 log-ratio 中	显式的 KL 参考点 $z_{\text{ref}}$
损失对称性	对称（chosen 和 rejected 等权）	非对称（损失厌恶）
效果	成对数据充足时更优	数据稀缺或标注预算有限时更实用

什么时候选 KTO？

1. 数据只有 thumbs up/down：用户反馈通常是"赞/踩"，天然适合 KTO
2. 标注预算有限：成对标注成本约为独立标注的 2-3 倍
3. 数据来源异构：不同标注员标注的样本难以配对，KTO 直接使用

论文：KTO: Model Alignment as Prospect Theoretic Optimization

各方法对比

方法	需要的模型	数据要求	计算成本	适用场景
PPO	Actor + Ref + RM + Critic (4个)	成对偏好数据 + RM	最高	通用对齐，效果最稳定
DPO	Policy + Ref (2个)	成对偏好数据	低	资源受限，快速对齐
GRPO	Policy + Ref (2个)	规则可判定的任务	中等	数学/代码/推理任务
KTO	Policy + Ref (2个)	非成对标注数据	低	数据收集成本高的场景

选型建议：

• 有充足算力和高质量成对数据 → PPO
• 有成对数据但算力有限 → DPO
• 推理/数学/代码任务 → GRPO（DeepSeek-R1 的成功验证）
• 只有好/差的独立标注 → KTO

更多对齐算法

上面介绍的 PPO、DPO、GRPO、KTO 是最主流的四种方法。随着研究推进，社区涌现了更多变体，各有侧重。

ORPO (Odds Ratio Preference Optimization)

ORPO 的核心创新：把 SFT 和对齐合并成一个阶段，通过在语言建模损失上叠加一个 odds ratio 惩罚项来实现偏好对齐。

$$L_{\text{ORPO}}=L_{\text{SFT}}+\lambda \cdot \log \frac{\text{odds}(y_w|x)}{\text{odds}(y_l|x)}$$

其中 $\text{odds}(y|x)=\frac{P(y|x)}{1-P(y|x)}$，$y_w$ 为 chosen，$y_l$ 为 rejected。

优势

• 不需要参考模型：比 DPO 更简单，省掉 reference model 的显存开销
• 一步到位：SFT + 对齐同时完成，省去多阶段训练的复杂度

论文：ORPO: Monolithic Preference Optimization without Reference Model

CPO (Contrastive Preference Optimization)

CPO 同样移除了对参考模型的依赖，使用 对比损失 (contrastive loss) 直接在偏好对上优化。

SimPO — CPO 的实用变体

SimPO (Simple Preference Optimization) 在 CPO 基础上增加了两个改进：

1. 长度归一化奖励：用序列平均 log-probability 作为隐式奖励，避免模型偏好长回答
2. 目标奖励间距 (target reward margin)：在 chosen 和 rejected 之间强制一个最小间距 $\gamma$

$$L_{\text{SimPO}}=-\log \sigma (\frac{\beta }{|y_w|}\log \pi (y_w|x)-\frac{\beta }{|y_l|}\log \pi (y_l|x)-\gamma )$$

RLOO (REINFORCE Leave-One-Out)

RLOO 是一种 无需 Critic 网络 的在线 RL 方法，使用 leave-one-out baseline 来估计优势。

核心思路：对每个 prompt 采样 $K$ 个 response，每个 response 的 baseline = 其他 $K-1$ 个 response 的平均奖励。

$$A_i=r_i-\frac{1}{K-1}\sum _{j\ne i}{r}_j$$

与 GRPO 的关系

RLOO 和 GRPO 思路相似——都用组内其他样本来估计 baseline，不需要额外的 value network。区别在于 GRPO 用组内均值和标准差做归一化（advantage normalization），而 RLOO 直接用 leave-one-out 均值。RLOO 的方差更低，理论性质更好。

Online DPO

标准 DPO 是 离线 (offline) 的——在固定的偏好数据集上训练。Online DPO 在训练过程中 实时生成 response 并排序：

1. 当前策略对 prompt 生成多个 response
2. 用 judge 模型 / reward model 对生成的 response 打分排序
3. 构造新的偏好对 (chosen, rejected) 进行 DPO 更新

为什么需要 Online？

离线 DPO 的偏好数据来自其他模型（如 GPT-4），与当前策略的分布存在 分布偏移 (distribution shift)。Online DPO 用自身生成的数据训练，弥合了这一差距，效果通常优于离线版本。

PRM (Process Reward Model)

传统的 Reward Model 是 结果奖励模型 (ORM)：只对最终回答打分。PRM 提供 步级监督 (step-level supervision)：对推理过程中的每一步都给出奖励。

	ORM (结果奖励)	PRM (过程奖励)
监督粒度	整个回答	每个推理步骤
训练数据	好/坏回答对	每步标注 correct/incorrect
适用场景	通用对话	数学、代码、逻辑推理
优势	标注成本低	定位错误更精确，信用分配更清晰

prompt: "求解 2x + 3 = 7"

Step 1: 2x + 3 = 7        ✅ (PRM: +1)
Step 2: 2x = 4            ✅ (PRM: +1)  
Step 3: x = 2             ✅ (PRM: +1)

vs.

Step 1: 2x + 3 = 7        ✅ (PRM: +1)
Step 2: 2x = 10           ❌ (PRM: -1)  ← PRM 能精确定位这一步出错
Step 3: x = 5             ❌ (PRM: -1)

PRM 与 RL 的结合

PRM 可以作为 PPO / GRPO 等 RL 算法的奖励信号源，用步级奖励替代结果级奖励，让策略梯度的信用分配更准确。OpenAI 的 Let's Verify Step by Step 是这一方向的代表工作。

TRL 实战：对齐训练实现

TRL (Transformer Reinforcement Learning) 是 Hugging Face 提供的对齐训练框架，支持 DPO、GRPO、PPO、KTO 等主流算法。下面通过代码展示核心用法。

DPO 训练

from trl import DPOConfig, DPOTrainer

training_args = DPOConfig(
    output_dir="dpo-model",
    beta=0.1,            # KL 惩罚强度，越大越保守
    loss_type="sigmoid",  # 可选 "sigmoid", "hinge", "ipo"
)

trainer = DPOTrainer(
    model=model,
    args=training_args,
    train_dataset=dataset,  # 需要 "prompt", "chosen", "rejected" 三列
)
trainer.train()

DPO 数据格式示例

dataset = {
    "prompt": ["请解释量子计算", "写一首关于秋天的诗"],
    "chosen": ["量子计算利用量子力学原理...", "秋风起，落叶知..."],
    "rejected": ["不知道", "春天来了..."],
}

GRPO 训练

from trl import GRPOConfig, GRPOTrainer

training_args = GRPOConfig(
    output_dir="grpo-model",
    num_generations=8,      # 每个 prompt 生成 G 个 completion
    beta=0.0,               # KL 惩罚（0 = 不使用，当前主流实践）
    scale_rewards=True,     # 按 std(rewards) 归一化
)

trainer = GRPOTrainer(
    model="Qwen/Qwen2-0.5B-Instruct",
    reward_funcs=accuracy_reward,  # 可以是函数或奖励模型
    train_dataset=dataset,
)
trainer.train()

GRPO 关键参数解读：

• num_generations：每个 prompt 生成 $G$ 个 completion，组内相对排名计算 advantage。$G$ 越大估计越准但显存开销越高，通常取 4~16
• reward_funcs：可以是 基于规则的函数（如数学题答案对错判断），也可以是 奖励模型
• scale_rewards=True：将奖励除以组内标准差进行归一化。设为 False 可避免难度偏差——简单题组内方差小，归一化后梯度被放大（来自 Understanding R1-Zero-Like Training）
• beta=0.0：不使用 KL 约束，这是 DeepSeek-R1 等工作验证的主流实践

beta=0 的风险

不使用 KL 约束时，模型可能产生 格式退化（输出变得不可读）或 reward hacking。实践中需要配合格式奖励（format reward）来约束输出格式。

PPO 训练

PPO 是最复杂的对齐方法，需要 4 个模型同时在显存中：

from trl import PPOConfig, PPOTrainer

# PPO 需要的 4 个模型：
# 1. policy model      — 被优化的策略
# 2. reference model   — 冻结的 SFT 模型（用于 KL 约束）
# 3. reward model      — 打分器
# 4. value model       — Critic，估计状态价值

config = PPOConfig(
    output_dir="ppo-model",
    kl_coef=0.05,           # KL 惩罚系数
    cliprange=0.2,          # PPO clip 范围
)

trainer = PPOTrainer(
    model=model,              # policy model
    reward_model=reward_model,
    value_model=value_model,
    args=config,
    train_dataset=dataset,
)
trainer.train()

显存优化

PPO 的 4 个模型非常吃显存。常见做法：

• LoRA：policy 和 value model 只训练 adapter
• 模型共享：policy 和 value model 共享 backbone，分别加不同的 head
• DeepSpeed ZeRO：多卡分片

奖励函数设计模式

# 模式 1：基于规则的奖励（适合数学/代码等有明确答案的任务）
def accuracy_reward(completions, ground_truth):
    """检查回答是否包含正确答案"""
    rewards = []
    for completion, answer in zip(completions, ground_truth):
        if answer in completion:
            rewards.append(1.0)
        else:
            rewards.append(0.0)
    return rewards

# 模式 2：格式奖励（约束输出格式）
def format_reward(completions):
    """检查输出是否符合指定格式"""
    import re
    rewards = []
    for completion in completions:
        # 例：要求输出包含 <think>...</think><answer>...</answer>
        if re.search(r"<think>.*?</think>.*<answer>.*?</answer>", completion, re.DOTALL):
            rewards.append(1.0)
        else:
            rewards.append(0.0)
    return rewards

# 模式 3：多奖励组合
def combined_reward(completions, ground_truth):
    """组合多个奖励信号"""
    acc = accuracy_reward(completions, ground_truth)
    fmt = format_reward(completions)
    return [0.8 * a + 0.2 * f for a, f in zip(acc, fmt)]

对齐算法选择指南

场景	推荐算法	原因
有成对偏好数据，追求简单	DPO	无需 RM，训练稳定
有成对数据，想一步完成 SFT + 对齐	ORPO	省掉 SFT 阶段，流程最简
数学/推理任务，可定义规则奖励	GRPO	DeepSeek 验证，支持规则奖励
只有好/坏标签（非成对）	KTO	不需要成对数据
追求最高质量，有充足资源	PPO	最灵活但最复杂
需要过程监督（数学/推理）	PRM + RL	步级奖励更精确

选择决策流程：

有成对偏好数据？
├── 是 → 需要同时做 SFT？
│       ├── 是 → ORPO
│       └── 否 → 有在线生成能力？
│               ├── 是 → Online DPO / PPO
│               └── 否 → DPO
└── 否 → 有规则可定义的奖励？
        ├── 是 → GRPO
        └── 否 → 有好/坏标签？
                ├── 是 → KTO
                └── 否 → 先收集数据 😅

实践建议

1. 起步用 DPO：最简单，效果不差，适合快速迭代
2. 数学/代码用 GRPO：规则奖励天然适合，DeepSeek-R1 已充分验证
3. PPO 是上限最高的选择，但工程复杂度也最高，建议有经验后再尝试
4. 关注 Online 方法：Online DPO / GRPO 等在线方法通常优于离线版本

苏格拉底时刻 💡

1. DPO 真的不需要 Reward Model 吗？ DPO 隐含了一个奖励模型 $r(x,y)=\beta \log \frac{{\pi }_{\theta }(y|x)}{{\pi }_{\text{ref}}(y|x)}$，只是不需要显式训练。当偏好数据分布与模型能力差距较大时，DPO 可能不如 RLHF。
2. GRPO 为什么在推理任务上特别好用？ 因为推理任务有明确的正确答案，可以用规则奖励（答案对错）取代人工标注，而开放式对话难以定义规则。
3. KL 惩罚为什么必不可少？ 没有 KL 约束，模型会 reward hacking——找到奖励模型的漏洞，生成得分高但质量低的输出。
4. 全对/全错时 GRPO 的优势为 0，怎么办？ 需要调整采样温度和采样数量，确保组内有足够的多样性。如果某个难度的题目模型总是全对或全错，说明这个难度不适合当前阶段的 RL 训练。
5. 偏好对齐是否限制模型能力？ "对齐税"确实存在，但通过精心设计训练策略（如较小的 $\beta$、高质量数据）可以将其最小化。

常见问题 & 面试考点

• Q: PPO 和 DPO 哪个效果更好？ 取决于场景。PPO 在 online 设定下更强（可以探索），DPO 在 offline 设定下更稳定。
• Q: DPO 的 $\beta$ 如何选择？ $\beta$ 越大越保守（更接近 SFT 模型），$\beta$ 越小优化越激进。通常在 0.1~0.5 之间搜索。
• Q: GRPO 和 PPO 的本质区别是什么？ GRPO 用组内相对排名代替了 Critic 网络来估计优势，省去了一半的模型。
• Q: 为什么 KTO 不需要成对数据？ KTO 分别处理好/差回答，用 KL 散度作为"锚点"来衡量偏离程度，不需要直接比较两个回答。

推荐资源

• Training language models to follow instructions with human feedback - InstructGPT / RLHF 论文
• Direct Preference Optimization - DPO 论文
• A General Theoretical Paradigm to Understand Learning from Human Feedback - IPO 论文
• DeepSeek-R1: Incentivizing Reasoning Capability in LLMs via RL - GRPO / DeepSeek-R1
• Proximal Policy Optimization Algorithms - PPO 论文
• KTO: Model Alignment as Prospect Theoretic Optimization - KTO 论文
• Hugging Face TRL - 实现 DPO/PPO/GRPO/KTO 的框架