【大模型】- DPO：直接偏好优化

DPO：直接偏好优化

无需强化学习的简化对齐

类型： 学习　｜　语言： Python　｜　🏷 前置：《DPO》（本系列第 7 篇）

学习目标

• 理解 DPO 与 RLHF 的区别
• 实现 DPO 损失函数
• 训练 DPO 模型
• 比较 DPO 和 RLHF 的性能
• 处理 DPO 训练的常见问题

DPO 概述

DPO（直接偏好优化）是一种简化的对齐方法，它直接从偏好数据优化策略，无需训练单独的奖励模型或使用强化学习。

DPO vs RLHF

方面	RLHF	DPO
奖励模型	需要单独训练	隐式包含在策略中
训练稳定性	较低（RL 不稳定）	较高
计算成本	较高（PPO 复杂）	较低
超参数	较多	较少
理论基础	策略梯度	对比学习

理论基础

DPO 的核心思想是直接优化策略，使其最大化与人类偏好对齐的奖励，同时最小化与参考策略的 KL 散度。

DPO 损失函数

import torch
import torch.nn.functional as F

def dpo_loss(policy_chosen_logps, policy_rejected_logps,
             reference_chosen_logps, reference_rejected_logps,
             beta=0.1):
    """
    计算 DPO 损失
    
    Args:
        policy_chosen_logps: 策略模型对 chosen 回答的对数概率
        policy_rejected_logps: 策略模型对 rejected 回答的对数概率
        reference_chosen_logps: 参考模型对 chosen 回答的对数概率
        reference_rejected_logps: 参考模型对 rejected 回答的对数概率
        beta: 温度参数
    """
    # 计算对数概率差异
    chosen_logratios = policy_chosen_logps - reference_chosen_logps
    rejected_logratios = policy_rejected_logps - reference_rejected_logps
    
    # DPO 损失
    logits = beta * (chosen_logratios - rejected_logratios)
    loss = -F.logsigmoid(logits).mean()
    
    return loss

数据集实现

from torch.utils.data import Dataset

class DPODataset(Dataset):
    def __init__(self, data, tokenizer, max_length=512):
        self.data = data
        self.tokenizer = tokenizer
        self.max_length = max_length
    
    def __len__(self):
        return len(self.data)
    
    def __getitem__(self, idx):
        item = self.data[idx]
        
        # 分词 chosen 和 rejected
        chosen_encodings = self.tokenizer(
            item["prompt"] + item["chosen"],
            max_length=self.max_length,
            padding="max_length",
            truncation=True,
            return_tensors="pt"
        )
        
        rejected_encodings = self.tokenizer(
            item["prompt"] + item["rejected"],
            max_length=self.max_length,
            padding="max_length",
            truncation=True,
            return_tensors="pt"
        )
        
        return {
            "chosen_input_ids": chosen_encodings["input_ids"].squeeze(),
            "chosen_attention_mask": chosen_encodings["attention_mask"].squeeze(),
            "rejected_input_ids": rejected_encodings["input_ids"].squeeze(),
            "rejected_attention_mask": rejected_encodings["attention_mask"].squeeze(),
        }

模型实现

class DPOModel(nn.Module):
    def __init__(self, model):
        super().__init__()
        self.model = model
    
    def forward(self, input_ids, attention_mask):
        outputs = self.model(input_ids, attention_mask)
        return outputs.logits
    
    def get_logprobs(self, input_ids, attention_mask):
        """计算序列的对数概率"""
        outputs = self.model(input_ids, attention_mask)
        logits = outputs.logits
        
        # 对数概率
        log_probs = F.log_softmax(logits, dim=-1)
        
        # 收集每个 token 的对数概率
        sequence_log_probs = torch.gather(
            log_probs[:, :-1],
            dim=-1,
            index=input_ids[:, 1:].unsqueeze(-1)
        ).squeeze(-1)
        
        # 应用注意力掩码
        mask = attention_mask[:, 1:].float()
        sequence_log_probs = (sequence_log_probs * mask).sum(dim=-1)
        
        return sequence_log_probs

训练循环

def train_dpo(model, ref_model, train_dataloader, config):
    optimizer = torch.optim.AdamW(model.parameters(), lr=config["lr"])
    
    for epoch in range(config["epochs"]):
        model.train()
        total_loss = 0
        
        for batch in train_dataloader:
            # 移到设备
            chosen_input = batch["chosen_input_ids"].to(config["device"])
            chosen_mask = batch["chosen_attention_mask"].to(config["device"])
            rejected_input = batch["rejected_input_ids"].to(config["device"])
            rejected_mask = batch["rejected_attention_mask"].to(config["device"])
            
            # 计算策略模型的对数概率
            policy_chosen_logps = model.get_logprobs(chosen_input, chosen_mask)
            policy_rejected_logps = model.get_logprobs(rejected_input, rejected_mask)
            
            # 计算参考模型的对数概率
            with torch.no_grad():
                reference_chosen_logps = ref_model.get_logprobs(chosen_input, chosen_mask)
                reference_rejected_logps = ref_model.get_logprobs(rejected_input, rejected_mask)
            
            # 计算 DPO 损失
            loss = dpo_loss(
                policy_chosen_logps,
                policy_rejected_logps,
                reference_chosen_logps,
                reference_rejected_logps,
                beta=config["beta"]
            )
            
            # 反向传播
            loss.backward()
            torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0)
            optimizer.step()
            optimizer.zero_grad()
            
            total_loss += loss.item()
        
        # 计算准确率
        accuracy = compute_accuracy(model, ref_model, val_dataloader, config)
        
        print(f"Epoch {epoch + 1}:")
        print(f"  Loss: {total_loss / len(train_dataloader):.4f}")
        print(f"  Accuracy: {accuracy:.4f}")
    
    return model

评估指标

def compute_accuracy(model, ref_model, dataloader, config):
    """计算 DPO 准确率：模型是否偏好 chosen 回答"""
    correct = 0
    total = 0
    
    model.eval()
    with torch.no_grad():
        for batch in dataloader:
            chosen_input = batch["chosen_input_ids"].to(config["device"])
            chosen_mask = batch["chosen_attention_mask"].to(config["device"])
            rejected_input = batch["rejected_input_ids"].to(config["device"])
            rejected_mask = batch["rejected_attention_mask"].to(config["device"])
            
            # 计算对数概率差异
            policy_chosen_logps = model.get_logprobs(chosen_input, chosen_mask)
            policy_rejected_logps = model.get_logprobs(rejected_input, rejected_mask)
            
            reference_chosen_logps = ref_model.get_logprobs(chosen_input, chosen_mask)
            reference_rejected_logps = ref_model.get_logprobs(rejected_input, rejected_mask)
            
            # DPO 分数
            chosen_scores = config["beta"] * (policy_chosen_logps - reference_chosen_logps)
            rejected_scores = config["beta"] * (policy_rejected_logps - reference_rejected_logps)
            
            # 如果 chosen 得分更高，则正确
            correct += (chosen_scores > rejected_scores).sum().item()
            total += len(chosen_scores)
    
    return correct / total

def compute_rewards(model, ref_model, input_ids, attention_mask, beta=0.1):
    """计算隐式奖励"""
    with torch.no_grad():
        policy_logps = model.get_logprobs(input_ids, attention_mask)
        reference_logps = ref_model.get_logprobs(input_ids, attention_mask)
    
    # 隐式奖励
    rewards = beta * (policy_logps - reference_logps)
    return rewards

生成和评估

def generate_dpo(model, tokenizer, prompt, max_new_tokens=512):
    model.eval()
    
    inputs = tokenizer(prompt, return_tensors="pt").to(model.device)
    
    with torch.no_grad():
        outputs = model.generate(
            **inputs,
            max_new_tokens=max_new_tokens,
            temperature=0.7,
            do_sample=True,
            top_p=0.9
        )
    
    response = tokenizer.decode(outputs[0], skip_special_tokens=True)
    return response.split(prompt)[-1]

超参数调优

config = {
    "lr": 5e-7,           # 比 RLHF 更小的学习率
    "beta": 0.1,          # KL 惩罚系数
    "epochs": 1,          # 通常 1-3 个 epoch
    "batch_size": 4,
    "max_length": 512,
}

常见问题

1. 过拟合：DPO 容易在小数据集上过拟合

   • 解决方案：使用数据增强，减少训练轮数

2. 模式崩溃：模型只生成一种类型的回答

   • 解决方案：调整 beta，增加数据多样性

3. 奖励黑客：模型学会利用偏好数据的模式

   • 解决方案：使用更多样化的数据，定期评估

完整训练流程

# 1. 加载模型
model = DPOModel(AutoModelForCausalLM.from_pretrained("sft_model"))
ref_model = DPOModel(AutoModelForCausalLM.from_pretrained("sft_model"))
ref_model.eval()

# 2. 准备数据
dataset = DPODataset(train_data, tokenizer)
dataloader = DataLoader(dataset, batch_size=config["batch_size"])

# 3. 训练
trained_model = train_dpo(model, ref_model, dataloader, config)

# 4. 保存
trained_model.model.save_pretrained("dpo_model")

总结

DPO 是 RLHF 的简化替代方案，直接从偏好数据优化策略。它更稳定、计算成本更低，但需要仔细调整超参数以避免过拟合和模式崩溃。

下一步

下一课将介绍 Constitutional AI，一种自我改进的对齐方法。

📚 本文改编自 AI Engineering from Scratch（MIT License · 作者 Rohit Ghumare），中文内容来自官方中文镜像。原课程共 503 课 · 20 阶段 · 免费开源，教程网站见 aiengineeringfromscratch.com。

📝 自我检查（课程配套测验）

• 第1题 · 学前
• 第2题 · 学前
• 第3题 · 学后
• 第4题 · 学后
• 第5题 · 学后

Q1（学前） DPO 相比 RLHF 的主要优势是什么？

A. 需要更多人类标注
B. 无需单独奖励模型和 PPO 训练——直接在偏好对上优化，更简单稳定
C. 产生更大模型
D. 只适用于分类任务

答案： B　解析： DPO（直接偏好优化）将 RLHF 的奖励建模和 RL 阶段合并为单一监督学习目标。直接在（chosen, rejected）回答对上训练，无需 PPO 或独立奖励模型。

Q2（学前） DPO 损失函数鼓励什么？

A. 模型对所有输入给出相同回答
B. 提高 chosen 回答相对 rejected 回答的对数概率，同时保持接近参考模型
C. 最大化训练 loss
D. 最小化模型大小

答案： B　解析： DPO 损失增加模型对 preferred（chosen）回答的概率，降低对 rejected 回答的概率，用参考模型（通常是 SFT 模型）的隐式 KL 约束防止偏离过远。

Q3（学后） DPO 中的 β（beta）参数控制什么？

A. 学习率
B. 对参考模型的约束强度：β 越大，优化后的模型越接近参考模型
C. batch 大小
D. 训练 epoch 数

答案： B　解析： β 是 DPO 损失中的温度参数，控制偏离参考（SFT）策略的惩罚。高 β 使模型保守（贴近 SFT）；低 β 允许更大变化以更好匹配偏好，但有遗忘风险。

Q4（学后） 为什么 DPO 在 2024–2025 年广泛采用？

A. 它比 RLHF 需要更多 GPU
B. 实现更简单、训练更稳定、算力需求更低，同时达到与 RLHF 相当的对齐效果
C. 它完全取代预训练
D. 它只适用于开源模型

答案： B　解析： DPO 消除了 RLHF 的奖励模型和 PPO 阶段，将偏好对齐简化为类似 SFT 的训练。更稳定、更易复现、资源需求更低，被 Llama、Mistral 等广泛采用。

Q5（学后） DPO 仍可能有什么问题？

A. 无法使用偏好数据
B. 偏好数据质量至关重要——有噪声或有偏的偏好会导致错误对齐；且 DPO 仍可能被 exploit
C. 不能用于对话模型
D. 需要 RL 基础设施

答案： B　解析： DPO 直接从偏好数据学习——垃圾输入产生垃圾对齐。标注者分歧、文化偏见或标注错误会直接变成模型行为。DPO 也面临与 RLHF 类似的 reward hacking 风险。