DPO微调算法

基本概念

DPO算法全称Direct Preference Optimization，即直接偏好优化，是直接利用偏好数据将模型与人类偏好进行对齐的一种强化学习算法。DPO绕开了奖励模型的复杂机制，直接利用偏好数据通过损失函数调整模型参数，使得模型输出结果更符合人类偏好。

DPO的核心是一个三元组（prompt(问题), chosen(偏好), rejected(非偏好)）和损失函数 $$\begin{equation}
\mathcal{L}{\text{DPO}} = - \mathbb{E}{(x, y_w, y_l) \sim \mathcal{D}} \left[
\log \sigma \left(
\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)}
\right)
\right]
\end{equation}$$, 其中：

  • ywy_w 和 yly_l 分别表示偏好回答和非偏好回答。
  • πθ\pi_\theta 是待优化的策略模型，πref\pi_{ref} 是参考模型（通常为初始预训练模型）。
  • β\beta 是温度参数，控制偏好强度。

DPO的优势

  • 效率高：省去奖励模型训练和PPO微调步骤，计算成本更低。
  • 稳定性强：避免强化学习中的超参数敏感性和训练不稳定性。
  • 直接优化偏好：通过概率模型显式最大化偏好数据的似然。

一个基于gpt2微调的简单DPO算法示例

# 尝试使用gpt-2进行DPO算法复现
from datasets import Dataset

# 一个简单的示例
# DPO需要有三部分组成，偏好问题（prompt）偏好答案（chosen）不偏好答案（rejected），直接利用这个三元组调整损失函数
# 而不需要单独训练奖励模型
data = {
    "prompt": ["解释相对论", "如何学习Python"],
    "chosen": ["爱因斯坦提出的时空理论...", "从基础语法开始，多写代码..."],
    "rejected": ["相对论是关于速度的", "看视频就够了"]
}

dataset = Dataset.from_dict(data)

# 准备数据
from torch.utils.data import DataLoader

def collate_fn(batch):
    return {
        "prompt": [x["prompt"] for x in batch],
        "chosen": [x["chosen"] for x in batch],
        "rejected": [x["rejected"] for x in batch]
    }

data_loader = DataLoader(dataset, batch_size=2, collate_fn=collate_fn)

# 加载模型
from transformers import AutoModelForCausalLM, AutoTokenizer
# 暂时使用gpt2，模型小便于部署，并且小模型方便观察DPO带来的效果？
model_name = "gpt2"
tokenizer = AutoTokenizer.from_pretrained(model_name)
tokenizer.pad_token = tokenizer.eos_token  # 设置填充token
# 策略模型
policy_model = AutoModelForCausalLM.from_pretrained(model_name)
# 参考模型（固定参数）
ref_model = AutoModelForCausalLM.from_pretrained(model_name)
for param in ref_model.parameters():
    param.requires_grad = False

# 核心DPO损失函数实现
import torch
def get_log_probs(model, tokenizer, prompts, responses):
    # 拼接prompt和response
    texts = [p + r for p, r in zip(prompts, responses)]
    inputs = tokenizer(texts, return_tensors="pt", padding=True, truncation=True)
    # 计算对数概率
    # with torch.no_grad():
    outputs = model(**inputs, labels=inputs["input_ids"])
    # 对数概率 = 负损失 * 序列长度
    log_probs = -outputs.loss * inputs["input_ids"].shape[1]
    return log_probs
# DPO损失函数
import torch.nn.functional as F
def dpo_loss(policy_chosen_logps, policy_reject_logps, ref_chosen_logps, ref_reject_logps, beta=0.1):
    # 计算策略模型和参考模型的对数概率差
    policy_logratios = policy_chosen_logps - policy_reject_logps
    ref_logratios = ref_chosen_logps - ref_reject_logps
    losses = -F.logsigmoid(beta * (policy_logratios - ref_logratios))
    return losses.mean()

from torch.optim import AdamW

optimizer = AdamW(policy_model.parameters(), lr=5e-5)
beta = 0.1

# 开始训练
from tqdm import tqdm
for epoch in range(3):
    for batch in tqdm(data_loader):
        # 依次计算对数概率
        policy_chosen_logps = get_log_probs(policy_model, tokenizer, batch["prompt"], batch["chosen"])
        policy_reject_logps = get_log_probs(policy_model, tokenizer, batch["prompt"], batch["rejected"])
        ref_chosen_logps = get_log_probs(ref_model, tokenizer, batch["prompt"], batch["chosen"])
        ref_reject_logps = get_log_probs(ref_model, tokenizer, batch["prompt"], batch["rejected"])

        losses = dpo_loss(policy_chosen_logps, policy_reject_logps, ref_chosen_logps, ref_reject_logps, beta)

        # 反向传播
        optimizer.zero_grad()
        losses.backward()
        optimizer.step()
    print(f"Epoch: {epoch}, Loss: {losses.item(): .4f}")

def generate_response(model, tokenizer, prompt):
    inputs = tokenizer(prompt, return_tensors="pt")
    outputs = model.generate(**inputs, max_length=100)
    return tokenizer.decode(outputs[0], skip_special_tokens=True)

test_prompt = "怎么学习Python"
print("优化后：", generate_response(policy_model, tokenizer, test_prompt))
print("参考模型输出：", generate_response(ref_model, tokenizer, test_prompt))

# 保存模型
policy_model.save_pretrained("dpo_finetuned_model")
tokenizer.save_pretrained("dpo_finetuned_model")