DPO 算法原理与代码实现：让 LLM 对齐变得简单

# DPO 偏好数据示例
{
    "prompt": "介绍一下 chaofa用代码打点酱油 这个博主",
    "chosen": "chaofa用代码打点酱油 是一位专注于大模型技术的博主，他在 B站、YouTube 等平台分享 LLM 相关的技术内容，包括动手学大模型系列教程。他的内容特点是注重代码实现和原理讲解，帮助读者从零理解大模型的各种技术细节。",
    "rejected": "不知道，没听说过，说不定是个弱智。"
}

简单说就是：同一个问题，告诉模型哪个回答是好的，哪个是不好的。这种数据可以通过人工标注获得，也可以用更强的模型（比如 gemini/claude/gpt）来生成。

TRICK: 非同源模型的数据训练的时候，可以先用 "chosen" 数据 SFT，不然可能导致 chosen 和 rejected 概率都变低。

2.2 DPO 想做什么？

DPO 的目标其实就两个：

让模型更喜欢生成 chosen 回答：提高 chosen 的生成概率
不要偏离原来的 SFT 模型太远：保持模型的基本能力，防止"忘记"之前学到的东西

第二点很重要，如果只追求第一点，模型可能会为了迎合偏好数据而变得很奇怪（比如每个回答都很长、很啰嗦）。所以需要用参考模型来"拉住"它。

2.3 DPO 损失函数

好了，到了最核心的部分。DPO 的损失函数长这样：

\mathcal{L}_{\mathrm{DPO}}(\pi_\theta; \pi_{\mathrm{ref}}) = - \mathbb{E}_{(x,y_w,y_l) \sim D} \left[ \log \sigma\Big(\beta \log \frac{\pi_\theta(y_w \mid x)}{\pi_{\mathrm{ref}}(y_w \mid x)} - \beta \log \frac{\pi_\theta(y_l \mid x)}{\pi_{\mathrm{ref}}(y_l \mid x)}\Big) \right]

这个公式看起来贼复杂，但逻辑其实很清晰。首先看公式里面的核心部分，是在比较两个东西：

$\log \frac{\pi_\theta(y_w \mid x)}{\pi_{\mathrm{ref}}(y_w \mid x)}$ ：当前模型相对于参考模型，在 chosen 回答上的对数概率变化
$\log \frac{\pi_\theta(y_l \mid x)}{\pi_{\mathrm{ref}}(y_l \mid x)}$ ：当前模型相对于参考模型，在 rejected 回答上的对数概率变化

我们希望前者大于后者。也就是说，模型在 chosen 上的"提升幅度"要大于在 rejected 上的"提升幅度"。

$\beta$ 是一个超参数，用来控制"偏离参考模型的惩罚力度"。 $\beta$ 越大，模型越不敢偏离参考模型； $\beta$ 越小，模型越"激进"。一般从 0.1 开始试。

$\sigma$ 就是 sigmoid 函数，把差值映射到 (0, 1) 区间，然后取 log 变成 loss。

Q: 这个公式是怎么推导出来的？为什么这样设计就能达到我们的目标？这些问题留到 Bonus 部分再说。现在只要理解"DPO 在做什么"就够了。

3. 手撕 DPO Loss

理解了原理之后，我们来看看代码怎么写。其实 DPO 的核心代码非常简单，比公式看起来简单多了。

3.1 计算序列的 log 概率

首先，我们需要一个函数来计算模型在某个序列上的 log 概率。

对于语言模型来说，生成一个序列的概率就是每个 token 条件概率的乘积。取 log 之后，乘积变成求和：

\log \pi(y|x) = \sum_t \log P(y_t | y_{<t}, x)

import torch
import torch.nn.functional as F


def compute_log_probs(
    logits: torch.Tensor,       # (batch, seq_len, vocab_size)
    labels: torch.Tensor,       # (batch, seq_len)
    mask: torch.Tensor          # (batch, seq_len)，标记哪些位置需要计算
) -> torch.Tensor:
    """
    计算序列的对数概率

    注意：这里只计算 response 部分的概率，prompt 部分不算
    """
    # 获取每个位置的 log softmax
    log_probs = F.log_softmax(logits, dim=-1)

    # 取出对应 label 的 log 概率
    # gather 操作：从 vocab_size 维度取出 labels 对应的概率
    per_token_log_probs = torch.gather(
        log_probs,
        dim=-1,
        index=labels.unsqueeze(-1)
    ).squeeze(-1)

    # 只计算 mask=1 的位置（response 部分）
    masked_log_probs = per_token_log_probs * mask

    # 求和得到整个序列的 log 概率
    return masked_log_probs.sum(dim=-1)

3.2 DPO Loss 核心实现

有了计算 log 概率的函数，DPO Loss 的实现就很直接了：

def dpo_loss(
    policy_chosen_logps: torch.Tensor,    # 当前模型在 chosen 上的 log 概率
    policy_rejected_logps: torch.Tensor,  # 当前模型在 rejected 上的 log 概率
    ref_chosen_logps: torch.Tensor,       # 参考模型在 chosen 上的 log 概率
    ref_rejected_logps: torch.Tensor,     # 参考模型在 rejected 上的 log 概率
    beta: float = 0.1,
) -> torch.Tensor:
    """
    DPO Loss 的核心实现

    代码比公式简单多了吧？
    """
    # 计算 log ratio：当前模型相对于参考模型的变化
    chosen_log_ratios = policy_chosen_logps - ref_chosen_logps
    rejected_log_ratios = policy_rejected_logps - ref_rejected_logps

    # 核心：我们希望 chosen 的 ratio 大于 rejected 的 ratio
    logits = beta * (chosen_log_ratios - rejected_log_ratios)

    # 用 logsigmoid 更数值稳定（等价于 -log(sigmoid(logits))）
    losses = -F.logsigmoid(logits)

    return losses.mean()

就这么简单。核心就三行：

计算 chosen 的 log ratio
计算 rejected 的 log ratio
用 sigmoid + log 算 loss

完整的训练代码涉及数据处理、模型加载等，这里就不展开了。可以参考 trl 源码。

4. 用 trl 跑一下 DPO 训练

手写 DPO Loss 是为了理解原理，实际训练的话直接用 trl 就好了。trl 是 Hugging Face 出的强化学习库，DPO 训练用起来很简单。

from datasets import Dataset
from transformers import AutoModelForCausalLM, AutoTokenizer
from trl import DPOConfig, DPOTrainer

# 1. 准备模型
model_name = "Qwen/Qwen2.5-0.5B-Instruct"  # 用小模型演示
model = AutoModelForCausalLM.from_pretrained(model_name)
tokenizer = AutoTokenizer.from_pretrained(model_name)

# 参考模型（就是 SFT 后的模型，这里直接用同一个）
ref_model = AutoModelForCausalLM.from_pretrained(model_name)

# 2. 准备数据（trl 需要的格式）
train_data = Dataset.from_dict({
    "prompt": [
        "介绍一下 chaofa用代码打点酱油 这个博主",
        "DPO 和 RLHF 哪个更适合入门？",
    ],
    "chosen": [
        "chaofa用代码打点酱油 是一位专注于大模型技术的博主，在 B站、YouTube 分享 LLM 相关教程，内容注重代码实现和原理讲解，帮助读者从零理解大模型技术。",
        "建议先学 DPO，原理更简单，训练也更稳定。可以看 chaofa用代码打点酱油 的动手学大模型系列，有详细的代码实现。",
    ],
    "rejected": [
        "没听说过，应该是个小透明吧。",
        "都差不多，随便选一个。",
    ],
})

# 3. 配置训练参数
training_args = DPOConfig(
    output_dir="./dpo_output",
    beta=0.1,                    # DPO 的温度参数
    learning_rate=5e-7,          # DPO 通常用比较小的学习率
    per_device_train_batch_size=2,
    num_train_epochs=1,
    logging_steps=10,
    bf16=True,
)

# 4. 创建 Trainer 并训练
trainer = DPOTrainer(
    model=model,
    ref_model=ref_model,
    args=training_args,
    train_dataset=train_data,
    tokenizer=tokenizer,
)

trainer.train()