DPO 直接偏好优化 #
DPO 概述 #
DPO（Direct Preference Optimization，直接偏好优化）是 2023 年提出的一种简化 RLHF 流程的方法。它跳过了奖励模型训练和 PPO 优化，直接使用偏好数据优化语言模型。
DPO vs RLHF #
text
┌─────────────────────────────────────────────────────────────┐
│ RLHF vs DPO 流程对比 │
├─────────────────────────────────────────────────────────────┤
│ │
│ RLHF 流程： │
│ ┌───────────┐ ┌───────────┐ ┌───────────┐ │
│ │ SFT 模型 │ → │ 奖励模型 │ → │ PPO 训练 │ │
│ └───────────┘ └───────────┘ └───────────┘ │
│ ↑ ↑ ↑ │
│ SFT 数据 偏好数据 偏好数据 │
│ │
│ DPO 流程： │
│ ┌───────────┐ ┌───────────┐ │
│ │ SFT 模型 │ → │ DPO 训练 │ │
│ └───────────┘ └───────────┘ │
│ ↑ ↑ │
│ SFT 数据 偏好数据 │
│ │
│ DPO 优势： │
│ ├── 无需训练奖励模型 │
│ ├── 无需复杂的 PPO 训练 │
│ ├── 训练流程更简单 │
│ ├── 计算资源需求更低 │
│ └── 更容易实现和调试 │
│ │
└─────────────────────────────────────────────────────────────┘
DPO 核心原理 #
从 RLHF 到 DPO #
text
RLHF 目标：
────────────────────────
max E[r(x, y)] - β * KL(π_θ || π_ref)
其中：
├── r(x, y) 是奖励函数
├── π_θ 是待优化策略
├── π_ref 是参考策略（SFT 模型）
└── β 是 KL 约束系数
关键洞察：
────────────────────────
最优策略可以表示为：
π*(y|x) = (1/Z(x)) * π_ref(y|x) * exp(r(x,y)/β)
其中 Z(x) 是配分函数。
由此可得奖励函数：
r(x,y) = β * log(π*(y|x) / π_ref(y|x)) + β * log Z(x)
DPO 损失函数 #
text
将奖励函数代入 Bradley-Terry 模型：
P(y_w > y_l | x) = σ(r(x, y_w) - r(x, y_l))
= σ(β * log(π_θ(y_w|x) / π_ref(y_w|x))
- β * log(π_θ(y_l|x) / π_ref(y_l|x)))
DPO 损失函数：
────────────────────────
L_DPO = -E[log σ(β * (log(π_θ(y_w|x) / π_ref(y_w|x))
- log(π_θ(y_l|x) / π_ref(y_l|x))))]
直观理解：
────────────────────────
├── 增加被选择回复的概率（相对于参考模型）
├── 降低被拒绝回复的概率（相对于参考模型）
├── β 控制优化强度
└── 无需显式的奖励模型
DPO 目标函数图解 #
text
┌─────────────────────────────────────────────────────────────┐
│ DPO 优化方向 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 对被选择的回复 y_w： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 目标：增加 π_θ(y_w|x) / π_ref(y_w|x) │ │
│ │ │ │
│ │ π_θ(y_w|x) │ │
│ │ ↑ │ │
│ │ │ ──────── │ │
│ │ │ / │ │
│ │ │ / 优化方向 │ │
│ │ │ / │ │
│ │ └──┼───────────────→ 训练步数 │ │
│ │ π_ref(y_w|x) │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 对被拒绝的回复 y_l： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 目标：降低 π_θ(y_l|x) / π_ref(y_l|x) │ │
│ │ │ │
│ │ π_θ(y_l|x) │ │
│ │ ↑ │ │
│ │ │ \ │ │
│ │ │ \ 优化方向 │ │
│ │ │ \ │ │
│ │ │ ──────── │ │
│ │ └──┼───────────────→ 训练步数 │ │
│ │ π_ref(y_l|x) │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘
DPO 实现 #
基础实现 #
python
import torch
import torch.nn.functional as F
from transformers import AutoModelForCausalLM, AutoTokenizer
class DPOTrainer:
def __init__(
self,
model,
ref_model,
tokenizer,
beta=0.1,
learning_rate=1e-6,
):
self.model = model
self.ref_model = ref_model
self.tokenizer = tokenizer
self.beta = beta
self.optimizer = torch.optim.AdamW(
model.parameters(),
lr=learning_rate
)
self.ref_model.eval()
for param in self.ref_model.parameters():
param.requires_grad = False
def get_log_probs(self, model, input_ids, attention_mask):
outputs = model(
input_ids=input_ids,
attention_mask=attention_mask,
labels=input_ids,
)
logits = outputs.logits[:, :-1, :]
labels = input_ids[:, 1:]
log_probs = F.log_softmax(logits, dim=-1)
per_token_log_probs = log_probs.gather(
2, labels.unsqueeze(-1)
).squeeze(-1)
attention_mask_shifted = attention_mask[:, 1:]
per_token_log_probs = per_token_log_probs * attention_mask_shifted
return per_token_log_probs.sum(dim=-1)
def compute_dpo_loss(
self,
input_ids_chosen,
attention_mask_chosen,
input_ids_rejected,
attention_mask_rejected,
):
policy_log_probs_chosen = self.get_log_probs(
self.model, input_ids_chosen, attention_mask_chosen
)
policy_log_probs_rejected = self.get_log_probs(
self.model, input_ids_rejected, attention_mask_rejected
)
with torch.no_grad():
ref_log_probs_chosen = self.get_log_probs(
self.ref_model, input_ids_chosen, attention_mask_chosen
)
ref_log_probs_rejected = self.get_log_probs(
self.ref_model, input_ids_rejected, attention_mask_rejected
)
policy_log_ratio_chosen = policy_log_probs_chosen - ref_log_probs_chosen
policy_log_ratio_rejected = policy_log_probs_rejected - ref_log_probs_rejected
logits = self.beta * (
policy_log_ratio_chosen - policy_log_ratio_rejected
)
loss = -F.logsigmoid(logits).mean()
chosen_rewards = self.beta * policy_log_ratio_chosen
rejected_rewards = self.beta * policy_log_ratio_rejected
accuracy = (chosen_rewards > rejected_rewards).float().mean()
return loss, {
"loss": loss.item(),
"chosen_rewards": chosen_rewards.mean().item(),
"rejected_rewards": rejected_rewards.mean().item(),
"accuracy": accuracy.item(),
}
def train_step(self, batch):
loss, metrics = self.compute_dpo_loss(
batch["input_ids_chosen"],
batch["attention_mask_chosen"],
batch["input_ids_rejected"],
batch["attention_mask_rejected"],
)
self.optimizer.zero_grad()
loss.backward()
torch.nn.utils.clip_grad_norm_(self.model.parameters(), 1.0)
self.optimizer.step()
return metrics
数据集实现 #
python
from torch.utils.data import Dataset
class DPODataset(Dataset):
def __init__(self, data_path, tokenizer, max_length=512):
self.tokenizer = tokenizer
self.max_length = max_length
self.data = self._load_data(data_path)
def _load_data(self, data_path):
import json
data = []
with open(data_path, 'r') as f:
for line in f:
data.append(json.loads(line))
return data
def __len__(self):
return len(self.data)
def __getitem__(self, idx):
item = self.data[idx]
prompt = item["prompt"]
chosen = item["chosen"]
rejected = item["rejected"]
chosen_text = prompt + chosen
rejected_text = prompt + rejected
chosen_enc = self.tokenizer(
chosen_text,
max_length=self.max_length,
padding="max_length",
truncation=True,
return_tensors="pt"
)
rejected_enc = self.tokenizer(
rejected_text,
max_length=self.max_length,
padding="max_length",
truncation=True,
return_tensors="pt"
)
return {
"input_ids_chosen": chosen_enc["input_ids"].squeeze(0),
"attention_mask_chosen": chosen_enc["attention_mask"].squeeze(0),
"input_ids_rejected": rejected_enc["input_ids"].squeeze(0),
"attention_mask_rejected": rejected_enc["attention_mask"].squeeze(0),
}
完整训练脚本 #
python
from torch.utils.data import DataLoader
from transformers import get_linear_schedule_with_warmup
from tqdm import tqdm
def train_dpo(
model_name,
train_data_path,
output_dir,
beta=0.1,
learning_rate=1e-6,
num_epochs=3,
batch_size=4,
gradient_accumulation_steps=4,
):
tokenizer = AutoTokenizer.from_pretrained(model_name)
tokenizer.pad_token = tokenizer.eos_token
model = AutoModelForCausalLM.from_pretrained(
model_name,
torch_dtype=torch.bfloat16,
device_map="auto",
)
ref_model = AutoModelForCausalLM.from_pretrained(
model_name,
torch_dtype=torch.bfloat16,
device_map="auto",
)
train_dataset = DPODataset(train_data_path, tokenizer)
train_dataloader = DataLoader(
train_dataset,
batch_size=batch_size,
shuffle=True,
)
trainer = DPOTrainer(
model=model,
ref_model=ref_model,
tokenizer=tokenizer,
beta=beta,
learning_rate=learning_rate,
)
num_training_steps = len(train_dataloader) * num_epochs // gradient_accumulation_steps
scheduler = get_linear_schedule_with_warmup(
trainer.optimizer,
num_warmup_steps=int(0.03 * num_training_steps),
num_training_steps=num_training_steps,
)
for epoch in range(num_epochs):
print(f"Epoch {epoch + 1}/{num_epochs}")
for step, batch in enumerate(tqdm(train_dataloader)):
batch = {k: v.to(model.device) for k, v in batch.items()}
metrics = trainer.train_step(batch)
if (step + 1) % gradient_accumulation_steps == 0:
scheduler.step()
if step % 10 == 0:
print(f"Step {step}: loss={metrics['loss']:.4f}, "
f"accuracy={metrics['accuracy']:.4f}")
model.save_pretrained(f"{output_dir}/epoch-{epoch + 1}")
tokenizer.save_pretrained(f"{output_dir}/epoch-{epoch + 1}")
model.save_pretrained(output_dir)
tokenizer.save_pretrained(output_dir)
DPO 变体 #
IPO（Identity Preference Optimization） #
python
def compute_ipo_loss(
policy_log_ratio_chosen,
policy_log_ratio_rejected,
beta=0.1,
):
diff = policy_log_ratio_chosen - policy_log_ratio_rejected
loss = (diff - 1 / (2 * beta)) ** 2
return loss.mean()
KTO（Kahneman-Tversky Optimization） #
python
def compute_kto_loss(
policy_log_probs,
ref_log_probs,
is_preferred,
beta=0.1,
desirable_weight=1.0,
undesirable_weight=0.5,
):
log_ratio = policy_log_probs - ref_log_probs
reward = beta * log_ratio
if is_preferred:
loss = desirable_weight * (1 - torch.sigmoid(reward))
else:
loss = undesirable_weight * torch.sigmoid(reward)
return loss.mean()
ORPO（Odds Ratio Preference Optimization） #
python
def compute_orpo_loss(
policy_log_probs_chosen,
policy_log_probs_rejected,
ref_log_probs_chosen,
ref_log_probs_rejected,
beta=0.1,
):
odds_ratio = torch.exp(
policy_log_probs_chosen - policy_log_probs_rejected
)
log_odds_ratio = torch.log(odds_ratio)
sig_ratio = torch.sigmoid(log_odds_ratio)
preference_loss = -torch.log(sig_ratio)
sft_loss = -policy_log_probs_chosen.mean()
total_loss = sft_loss + beta * preference_loss.mean()
return total_loss
DPO vs RLHF 详细对比 #
性能对比 #
text
┌─────────────────────────────────────────────────────────────┐
│ DPO vs RLHF 性能对比 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 训练效率： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 指标 │ RLHF │ DPO │ │
│ ├─────────────────────────────────────────────────────┤ │
│ │ 训练时间 │ 长 │ 短 │ │
│ │ 计算资源 │ 高 │ 低 │ │
│ │ 实现复杂度 │ 高 │ 低 │ │
│ │ 调参难度 │ 高 │ 低 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 模型效果： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 指标 │ RLHF │ DPO │ │
│ ├─────────────────────────────────────────────────────┤ │
│ │ 对齐效果 │ 优秀 │ 良好 │ │
│ │ 灵活性 │ 高 │ 中 │ │
│ │ 迭代优化 │ 支持 │ 支持 │ │
│ │ 大规模模型 │ 成熟 │ 可行 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘
适用场景 #
text
选择 DPO 的场景：
────────────────────────
├── 计算资源有限
├── 快速原型开发
├── 中小规模模型
├── 简单对齐需求
└── 团队经验有限
选择 RLHF 的场景：
────────────────────────
├── 追求最佳效果
├── 大规模模型
├── 复杂偏好学习
├── 需要精细控制
└── 有充足资源
DPO 最佳实践 #
超参数选择 #
text
关键超参数：
────────────────────────
β（KL 约束系数）：
├── 典型值：0.1 ~ 0.5
├── 较小：优化更激进，可能过拟合
├── 较大：优化更保守，更稳定
└── 建议：从 0.1 开始，根据效果调整
学习率：
├── 典型值：1e-7 ~ 1e-5
├── 较大：训练快，可能不稳定
├── 较小：稳定，但收敛慢
└── 建议：1e-6 是常用起点
批次大小：
├── 典型值：16 ~ 128
├── 较大：更稳定，需要更多显存
├── 较小：梯度噪声大
└── 建议：根据显存选择最大可能值
数据质量 #
text
高质量偏好数据要求：
────────────────────────
├── 明显的偏好差异
│ └── chosen 和 rejected 应该有清晰的质量差距
│
├── 多样化的提示
│ └── 覆盖不同任务类型和难度
│
├── 一致的标注标准
│ └── 标注者对偏好标准有共识
│
└── 足够的数据量
└── 通常需要数万条偏好数据
训练技巧 #
python
class ImprovedDPOTrainer(DPOTrainer):
def __init__(self, *args, label_smoothing=0.0, **kwargs):
super().__init__(*args, **kwargs)
self.label_smoothing = label_smoothing
def compute_dpo_loss(self, *args, **kwargs):
loss, metrics = super().compute_dpo_loss(*args, **kwargs)
if self.label_smoothing > 0:
loss = loss * (1 - self.label_smoothing) + 0.5 * self.label_smoothing
return loss, metrics
def train_step(self, batch):
metrics = super().train_step(batch)
if metrics["accuracy"] < 0.5:
print("Warning: Low accuracy, consider adjusting beta or data quality")
return metrics
下一步 #
现在你已经掌握了 DPO 的原理和实现，接下来学习 高级技术，了解更多 RLHF 变体和前沿方法！
最后更新：2026-04-05