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

[2026-08-19] mod:2026-11-05 wc:1.9k rt:8

DPO：直接偏好优化

无需强化学习的简化对齐

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

学习目标

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

DPO 概述

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

DPO vs RLHF

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

理论基础

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

DPO 损失函数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
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

数据集实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
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(),
        }

模型实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
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

训练循环

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
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

评估指标

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
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

生成和评估

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
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]

超参数调优

1
2
3
4
5
6
7
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
2
3
4
5
6
7
8
9
10
11
12
13
14
# 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 风险。