手写 DPO：从零实现大模型直接偏好优化训练

手写系列第 N+1 弹：继手写 RLHF 之后，这次带来更简洁、更强大的 DPO（Direct Preference Optimization）。不用繁琐的奖励模型，不用复杂的 PPO 稳定性调优，一个简单的二元交叉熵损失就能让大模型学会对齐人类偏好。DeepSeek R1 的推理能力背后，DPO 正是关键技术之一。

一、从 RLHF 到 DPO：为什么我们需要更简洁的对齐方法？

1.1 RLHF 的困境

大模型对齐（Alignment）是让模型输出符合人类期望的关键步骤。RLHF（Reinforcement Learning from Human Feedback）自 InstructGPT 以来一直是主流范式，但它在工程落地中面临三个核心难题：

奖励模型难以训练。RLHF 需要单独训练一个 Reward Model，对模型输出进行打分。这个奖励模型本身就需要大量人工标注数据，且很容易出现 reward hacking（奖励欺骗）——模型学会了利用奖励函数的漏洞，而非真正理解人类偏好。

PPO 训练不稳定。PPO（Proximal Policy Optimization）是个出了名的"调参噩梦"。学习率、clip 范围、KL 散度系数、GAE 参数……任何一项没调好，训练直接崩溃。即使调好了，一个 batch 内可能还是会出现策略坍塌。

计算开销巨大。整个流程需要同时加载：参考模型（Reference Model）、策略模型（Policy Model）、奖励模型、价值模型（Critic）。四份模型参数，对显存的要求极不友好。

1.2 DPO 的核心洞察

2023 年，斯坦福大学的研究者在论文《Direct Preference Optimization: Your Language Model is Secretly a Reward Model》中提出了一个关键洞察：

语言模型自身的概率比，就已经隐含了奖励信息。

换句话说，我们根本不需要额外的奖励模型。给定偏好对 $(y_w, y_l)$（win vs lose），DPO 直接基于策略模型和参考模型的对数概率比来构造损失函数，将偏好学习转化为一个简单的二元分类问题：

$$L_{DPO}(\\pi_\\theta; \\pi_{ref}) = -\\mathbb{E}{(x,y_w,y_l) \\sim D} \\left[ \\log \\sigma \\left( \\beta \\log \\frac{\\pi\\theta(y_w|x)}{\\pi_{ref}(y_w|x)} – \\beta \\log \\frac{\\pi_\\theta(y_l|x)}{\\pi_{ref}(y_l|x)} \\right) \\right]$$

这个公式看起来有点复杂，但本质思想很简单：我们希望模型更倾向于生成偏好回答 $y_w$ 而非非偏好回答 $y_l$，同时用参考模型 $\\pi_{ref}$ 做正则化，防止模型偏离初始化太远。

这就是 DPO 的全部核心——一个损失函数，取代了 RLHF 整个复杂的三阶段流水线。

二、DPO 的数学原理详解

2.1 Bradley-Terry 偏好模型

在理解 DPO 之前，我们需要先了解它背后的偏好建模理论。Bradley-Terry 模型是经典的成对比较模型，它假设每个回答 $y$ 有一个隐式的"强度"分值 $r(x,y)$，用户偏好 $y_w$ 胜过 $y_l$ 的概率为：

$$p^*(y_w \\succ y_l | x) = \\frac{\\exp(r(x,y_w))}{\\exp(r(x,y_w)) + \\exp(r(x,y_l))} = \\sigma(r(x,y_w) – r(x,y_l))$$

其中 $\\sigma$ 是 sigmoid 函数。RLHF 的做法是训练一个神经网络 $r_\\phi(x,y)$ 来近似这个隐含奖励，然后用 PPO 基于该奖励优化策略。

2.2 DPO 的巧妙推导

DPO 的关键一步是：如果我们把策略模型 $\\pi_\\theta$ 看作一个隐式奖励函数，那么最优策略 $\\pi^*$ 和其对应的奖励函数之间存在一一对应的解析关系。

根据 RLHF 中带 KL 散度约束的最优策略解：

$$\\pi^*(y|x) = \\frac{1}{Z(x)} \\pi_{ref}(y|x) \\exp\\left(\\frac{1}{\\beta} r(x,y)\\right)$$

解出奖励函数：

$$r(x,y) = \\beta \\log \\frac{\\pi^*(y|x)}{\\pi_{ref}(y|x)} + \\beta \\log Z(x)$$

将 $r(x,y)$ 代入 Bradley-Terry 模型，配分函数 $Z(x)$ 被消去，得到：

$$p^(y_w \\succ y_l | x) = \\sigma \\left( \\beta \\log \\frac{\\pi^(y_w|x)}{\\pi_{ref}(y_w|x)} – \\beta \\log \\frac{\\pi^*(y_l|x)}{\\pi_{ref}(y_l|x)} \\right)$$

于是我们有了一个由策略模型 $\\pi_\\theta$ 参数化的偏好概率。最大化对数似然就得到了 DPO 损失函数。

这个推导的精妙之处在于：它将奖励建模和策略优化合并为一步。DPO 不使用显式的奖励函数，而是通过策略模型的概率比隐式地编码了偏好信息。

2.3 DPO 损失函数的梯度分析

为了更好地理解 DPO 的训练动态，我们来看看损失函数的梯度形式：

$$\\nabla_\\theta L_{DPO} = -\\beta \\mathbb{E}{(x,y_w,y_l) \\sim D} \\Big[ \\sigma(\\hat{r}\\theta(x,y_l) – \\hat{r}\\theta(x,y_w)) \\Big( \\nabla\\theta \\log \\pi_\\theta(y_w|x) – \\nabla_\\theta \\log \\pi_\\theta(y_l|x) \\Big) \\Big]$$

其中 $\\hat{r}\\theta(x,y) = \\beta \\log \\frac{\\pi\\theta(y|x)}{\\pi_{ref}(y|x)}$ 是隐式奖励。

这个梯度形式告诉我们三件事：

三、从零实现 DPO：完整代码解析

现在进入实战环节。我们将用 PyTorch 从零实现一个完整的 DPO 训练框架，包括数据处理、损失计算、训练循环和评估。

3.1 项目结构

dpo-from-scratch/
├── data/ # 偏好数据集
├── model/ # 模型定义
│ ├── base_model.py # 基础语言模型
│ └── reference.py # 参考模型封装
├── trainer/
│ ├── dpo_loss.py # DPO 核心损失函数
│ └── dpo_trainer.py # DPO 训练器
├── utils/
│ ├── data_utils.py # 数据处理工具
│ └── eval_utils.py # 评估工具
└── train_dpo.py # 训练入口

3.2 基础语言模型

首先，我们需要一个可训练的语言模型。为了降低硬件门槛，我们基于一个简化的 GPT-2 风格 Transformer 实现：

# model/base_model.py
import torch
import torch.nn as nn
import torch.nn.functional as F
from transformers import AutoModelForCausalLM, AutoTokenizer
from typing import Optional, Tuple, List

class GPT2LikeModel(nn.Module):
"""简化的 GPT-2 风格语言模型，支持因果语言建模"""

def __init__(self, vocab_size: int = 50257, hidden_size: int = 768,
num_layers: int = 12, num_heads: int = 12, max_length: int = 1024):
super().__init__()

self.token_embedding = nn.Embedding(vocab_size, hidden_size)
self.position_embedding = nn.Embedding(max_length, hidden_size)

self.layers = nn.ModuleList([
TransformerBlock(hidden_size, num_heads)
for _ in range(num_layers)
])

self.ln_final = nn.LayerNorm(hidden_size)
self.lm_head = nn.Linear(hidden_size, vocab_size, bias=False)

# 权重绑定：输入嵌入和输出头共享权重
self.token_embedding.weight = self.lm_head.weight

self.hidden_size = hidden_size
self.max_length = max_length

def forward(self, input_ids: torch.Tensor,
attention_mask: Optional[torch.Tensor] = None,
labels: Optional[torch.Tensor] = None) -> Tuple[torch.Tensor, torch.Tensor]:
"""
Args:
input_ids: (batch_size, seq_len)
attention_mask: (batch_size, seq_len) – 0 表示 padding
labels: (batch_size, seq_len) – -100 表示忽略的 token
Returns:
logits: (batch_size, seq_len, vocab_size)
loss: 标量（训练时返回）
"""
batch_size, seq_len = input_ids.shape

# 位置编码
positions = torch.arange(seq_len, device=input_ids.device).unsqueeze(0)

# 词嵌入 + 位置嵌入
x = self.token_embedding(input_ids) + self.position_embedding(positions)

# 构建因果注意力掩码
if attention_mask is not None:
# 将 padding mask (0/1) 扩展为 2D 注意力掩码
# 0 表示 padding，1 表示有效 token
extended_mask = attention_mask[:, None, None, :] # (B, 1, 1, S)
extended_mask = extended_mask.to(dtype=x.dtype)
extended_mask = (1.0 – extended_mask) * torch.finfo(x.dtype).min
else:
extended_mask = None

# 逐层传递
for layer in self.layers:
x = layer(x, attention_mask=extended_mask)

x = self.ln_final(x)
logits = self.lm_head(x) # (B, S, V)

loss = None
if labels is not None:
shift_logits = logits[…, :-1, :].contiguous()
shift_labels = labels[…, 1:].contiguous()
loss = F.cross_entropy(
shift_logits.view(-1, shift_logits.size(-1)),
shift_labels.view(-1),
ignore_index=-100
)

return logits, loss

class TransformerBlock(nn.Module):
"""标准 Transformer 块：多头注意力 + FFN"""

def __init__(self, hidden_size: int, num_heads: int):
super().__init__()
self.ln_1 = nn.LayerNorm(hidden_size)
self.attn = CausalSelfAttention(hidden_size, num_heads)
self.ln_2 = nn.LayerNorm(hidden_size)
self.mlp = FeedForward(hidden_size)

def forward(self, x: torch.Tensor, attention_mask: Optional[torch.Tensor] = None):
x = x + self.attn(self.ln_1(x), attention_mask)
x = x + self.mlp(self.ln_2(x))
return x

class CausalSelfAttention(nn.Module):
"""因果自注意力：确保当前位置只能看到之前的位置"""

def __init__(self, hidden_size: int, num_heads: int):
super().__init__()
self.num_heads = num_heads
self.head_dim = hidden_size // num_heads
assert hidden_size % num_heads == 0, "hidden_size must be divisible by num_heads"

self.c_attn = nn.Linear(hidden_size, 3 * hidden_size)
self.c_proj = nn.Linear(hidden_size, hidden_size)

# 下三角因果掩码，注册为 buffer
self.register_buffer(
"bias",
torch.tril(torch.ones(1, 1, 1024, 1024)).view(1, 1, 1024, 1024)
)

def forward(self, x: torch.Tensor, attention_mask: Optional[torch.Tensor] = None):
B, T, C = x.shape

qkv = self.c_attn(x)
q, k, v = qkv.chunk(3, dim=-1)

# 多头拆分
q = q.view(B, T, self.num_heads, self.head_dim).transpose(1, 2)
k = k.view(B, T, self.num_heads, self.head_dim).transpose(1, 2)
v = v.view(B, T, self.num_heads, self.head_dim).transpose(1, 2)

# 计算注意力分数
att = (q @ k.transpose(-2, -1)) * (1.0 / self.head_dim ** 0.5)
att = att.masked_fill(self.bias[:, :, :T, :T] == 0, float('-inf'))

# 叠加 padding mask
if attention_mask is not None:
att = att + attention_mask[:, :, :T, :]

att = F.softmax(att, dim=-1)
y = att @ v

# 合并多头
y = y.transpose(1, 2).contiguous().view(B, T, C)
y = self.c_proj(y)

return y

class FeedForward(nn.Module):
"""前馈网络：GELU 激活 + 投影"""

def __init__(self, hidden_size: int):
super().__init__()
self.c_fc = nn.Linear(hidden_size, 4 * hidden_size)
self.c_proj = nn.Linear(4 * hidden_size, hidden_size)

def forward(self, x: torch.Tensor):
x = self.c_fc(x)
x = F.gelu(x)
x = self.c_proj(x)
return x

3.3 DPO 核心损失函数

这是整个框架的核心——DPO 损失的一行即可重现的代：

# trainer/dpo_loss.py
import torch
import torch.nn.functional as F
from typing import Optional, Tuple

def dpo_loss(
policy_chosen_logps: torch.Tensor, # 策略模型在偏好回答上的对数概率
policy_rejected_logps: torch.Tensor, # 策略模型在非偏好回答上的对数概率
ref_chosen_logps: torch.Tensor, # 参考模型在偏好回答上的对数概率
ref_rejected_logps: torch.Tensor, # 参考模型在非偏好回答上的对数概率
beta: float = 0.1, # KL 散度系数
label_smoothing: float = 0.0, # 标签平滑
ipo: bool = False # 是否使用 IPO 变体
) -> Tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
"""
计算 DPO 损失。

Args:
policy_chosen_logps: (batch_size,) 策略模型对 chosen 序列的对数概率
policy_rejected_logps: (batch_size,) 策略模型对 rejected 序列的对数概率
ref_chosen_logps: (batch_size,) 参考模型对 chosen 序列的对数概率
ref_rejected_logps: (batch_size,) 参考模型对 rejected 序列的对数概率
beta: 控制 KL 约束的强度，越大对参考模型的偏离惩罚越重
label_smoothing: 标签平滑系数，防止过拟合
ipo: 如果 True，使用 Identity Preference Optimization 变体

Returns:
loss: 标量损失
chosen_rewards: 偏好回答的隐式奖励
rejected_rewards: 非偏好回答的隐式奖励
"""
# 计算对数概率比值
# log π_θ(y_w|x) – log π_ref(y_w|x) = π_θ 相对于 π_ref 的 log 比率
chosen_log_ratio = policy_chosen_logps – ref_chosen_logps
rejected_log_ratio = policy_rejected_logps – ref_rejected_logps

# 隐式奖励 (乘以 beta)
chosen_rewards = beta * chosen_log_ratio.detach()
rejected_rewards = beta * rejected_log_ratio.detach()

# DPO 核心：偏好差异的对数优势
logits = chosen_log_ratio – rejected_log_ratio # (batch_size,)

if ipo:
# IPO (Identity Preference Optimization) 变体
# 使用 MSE 损失替代二元交叉熵，对偏好噪声更鲁棒
loss = (logits – 1.0 / (2 * beta)) ** 2
else:
# 标准 DPO：对 logit 应用 sigmoid 和交叉熵
# -log σ(β * (log_ratio_chosen – log_ratio_rejected))
# 等价于：-log σ(β * π_θ 偏好差距)
losses = -F.logsigmoid(beta * logits)

if label_smoothing > 0:
# 标签平滑：软话标签，防止对极端偏好过拟合
losses = (1 – label_smoothing) * losses + label_smoothing * (-F.logsigmoid(-beta * logits))

loss = losses.mean()

return loss, chosen_rewards, rejected_rewards

def compute_log_probs(
model: torch.nn.Module,
input_ids: torch.Tensor,
attention_mask: torch.Tensor,
labels: torch.Tensor
) -> torch.Tensor:
"""
计算序列的对数概率。

Args:
model: 语言模型
input_ids: (batch_size, seq_len)
attention_mask: (batch_size, seq_len)
labels: (batch_size, seq_len) – 只有需要计算的位置有值，别的为 -100

Returns:
log_probs: (batch_size,) 每个样本的对数概率和
"""
logits, _ = model(input_ids, attention_mask=attention_mask)

# 计算每个 token 的 log softmax
log_probs = F.log_softmax(logits, dim=-1) # (B, S, V)

# 获取真实 token 位置的对数概率
# labels 中 -100 的位置被忽略
token_log_probs = torch.gather(
log_probs[:, :-1, :],
dim=-1,
index=labels[:, 1:].unsqueeze(-1)
).squeeze(-1) # (B, S-1)

# 仅在标签有效的位置求和
mask = (labels[:, 1:] != -100).float()
log_probs_sum = (token_log_probs * mask).sum(dim=-1)

return log_probs_sum

3.4 DPO 训练器

训练器负责管理策略模型和参考模型，执行每个 batch 的 DPO 训练步骤：

# trainer/dpo_trainer.py
import torch
import torch.nn as nn
from torch.utils.data import DataLoader
from typing import Optional, Dict
from tqdm import tqdm
from trainer.dpo_loss import dpo_loss, compute_log_probs

class DPOTrainer:
"""
DPO 训练器：管理策略模型、参考模型和优化过程。

DPO 的核心思想是直接优化偏好，不需要额外的奖励模型：
1. 加载策略模型 π_θ 和参考模型 π_ref（冻结）
2. 对每个 batch，计算 π_θ 和 π_ref 在偏好/非偏好回答上的对数概率
3. 计算 DPO 损失并更新 π_θ
"""

def __init__(
self,
policy_model: nn.Module,
ref_model: Optional[nn.Module] = None,
beta: float = 0.1,
learning_rate: float = 1e-5,
max_grad_norm: float = 1.0,
use_wandb: bool = False,
):
self.policy_model = policy_model
self.ref_model = ref_model
self.beta = beta
self.max_grad_norm = max_grad_norm

# 冻结参考模型
if self.ref_model is not None:
for param in self.ref_model.parameters():
param.requires_grad = False
self.ref_model.eval()

# 优化器：AdamW 是训练 LLM 的标准选择
self.optimizer = torch.optim.AdamW(
self.policy_model.parameters(),
lr=learning_rate,
weight_decay=0.01,
)

# 学习率调度器：余弦退火
self.scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(
self.optimizer, T_max=1000, eta_min=learning_rate * 0.1
)

self.use_wandb = use_wandb

def train_step(
self,
batch: Dict[str, torch.Tensor]
) -> Dict[str, float]:
"""
执行一个 DPO 训练步骤。

Batch 结构:
chosen_input_ids: (B, S_c) 偏好回答
chosen_attention_mask: (B, S_c)
chosen_labels: (B, S_c)
rejected_input_ids: (B, S_r) 非偏好回答
rejected_attention_mask: (B, S_r)
rejected_labels: (B, S_r)
"""
self.policy_model.train()

# 计算策略模型的对数概率
policy_chosen_logps = compute_log_probs(
self.policy_model,
batch['chosen_input_ids'],
batch['chosen_attention_mask'],
batch['chosen_labels']
)

policy_rejected_logps = compute_log_probs(
self.policy_model,
batch['rejected_input_ids'],
batch['rejected_attention_mask'],
batch['rejected_labels']
)

# 计算参考模型的对数概率（不需要梯度）
with torch.no_grad():
ref_chosen_logps = compute_log_probs(
self.ref_model,
batch['chosen_input_ids'],
batch['chosen_attention_mask'],
batch['chosen_labels']
)

ref_rejected_logps = compute_log_probs(
self.ref_model,
batch['rejected_input_ids'],
batch['rejected_attention_mask'],
batch['rejected_labels']
)

# DPO 损失
loss, chosen_rewards, rejected_rewards = dpo_loss(
policy_chosen_logps,
policy_rejected_logps,
ref_chosen_logps,
ref_rejected_logps,
beta=self.beta,
)

# 反向传播
self.optimizer.zero_grad()
loss.backward()

# 梯度裁剪：防止梯度爆炸
nn.utils.clip_grad_norm_(
self.policy_model.parameters(),
self.max_grad_norm
)

self.optimizer.step()

# 计算准确率：模型是否更偏好 chosen
with torch.no_grad():
accuracy = (policy_chosen_logps > policy_rejected_logps).float().mean()

metrics = {
'loss': loss.item(),
'accuracy': accuracy.item(),
'chosen_reward': chosen_rewards.mean().item(),
'rejected_reward': rejected_rewards.mean().item(),
'reward_margin': (chosen_rewards – rejected_rewards).mean().item(),
'learning_rate': self.optimizer.param_groups[0]['lr'],
}

return metrics

def train_epoch(
self,
dataloader: DataLoader,
epoch: int
) -> Dict[str, float]:
"""
训练一个 epoch。
"""
total_metrics = {
'loss': 0.0, 'accuracy': 0.0,
'reward_margin': 0.0
}
num_batches = 0

progress_bar = tqdm(dataloader, desc=f'Epoch {epoch}')

for batch in progress_bar:
metrics = self.train_step(batch)

for k in total_metrics:
total_metrics[k] += metrics[k]
num_batches += 1

# 更新进度条
progress_bar.set_postfix({
'loss': metrics['loss'],
'acc': metrics['accuracy'],
'margin': metrics['reward_margin'],
})

# 平均指标
for k in total_metrics:
total_metrics[k] /= num_batches

self.scheduler.step()

return total_metrics

3.5 偏好数据集处理

DPO 需要结构化的偏好数据——每个样本包含一个 prompt 和两个回答（chosen 和 rejected）：

# utils/data_utils.py
import torch
from torch.utils.data import Dataset
from typing import Optional, List, Dict, Tuple
from transformers import PreTrainedTokenizer
import json

class PreferenceDataset(Dataset):
"""
偏好数据集：每个样本包含一个 prompt 以及 chosen/rejected 两个回答。

数据格式:
{
"prompt": "什么是机器学习？",
"chosen": "机器学习是一种人工智能技术…",
"rejected": "机器学习就是让电脑学东西…"
}
"""

def __init__(
self,
data_path: str,
tokenizer: PreTrainedTokenizer,
max_prompt_length: int = 512,
max_response_length: int = 256,
):
self.tokenizer = tokenizer
self.max_prompt_length = max_prompt_length
self.max_response_length = max_response_length

# 加载数据
with open(data_path, 'r', encoding='utf-8') as f:
self.data = [json.loads(line) for line in f]

def __len__(self):
return len(self.data)

def __getitem__(self, idx: int) -> Dict[str, torch.Tensor]:
item = self.data[idx]
prompt = item['prompt']
chosen = item['chosen']
rejected = item['rejected']

return self._tokenize_pair(prompt, chosen, rejected)

def _tokenize_pair(
self,
prompt: str,
chosen: str,
rejected: str
) -> Dict[str, torch.Tensor]:
"""
将 (prompt, chosen, rejected) 三元组 tokenize。

关键策略：
1. prompt 和 response 分别 tokenize 后拼接
2. labels 中只有 response 部分参与损失计算
3. prompt 部分的 labels 设为 -100（忽略）
"""
# Tokenize prompt
prompt_tokens = self.tokenizer(
prompt,
truncation=True,
max_length=self.max_prompt_length,
add_special_tokens=False,
)

# Tokenize chosen response
chosen_tokens = self.tokenizer(
chosen,
truncation=True,
max_length=self.max_response_length,
add_special_tokens=False,
)

# Tokenize rejected response
rejected_tokens = self.tokenizer(
rejected,
truncation=True,
max_length=self.max_response_length,
add_special_tokens=False,
)

# 拼接 prompt + chosen
chosen_input_ids = prompt_tokens['input_ids'] + chosen_tokens['input_ids'] + [self.tokenizer.eos_token_id]
chosen_attention_mask = [1] * len(chosen_input_ids)
# Labels: -100 表示 prompt 部分不参与损失计算
chosen_labels = [-100] * len(prompt_tokens['input_ids']) + chosen_tokens['input_ids'] + [self.tokenizer.eos_token_id]

# 拼接 prompt + rejected
rejected_input_ids = prompt_tokens['input_ids'] + rejected_tokens['input_ids'] + [self.tokenizer.eos_token_id]
rejected_attention_mask = [1] * len(rejected_input_ids)
rejected_labels = [-100] * len(prompt_tokens['input_ids']) + rejected_tokens['input_ids'] + [self.tokenizer.eos_token_id]

return {
'chosen_input_ids': torch.tensor(chosen_input_ids),
'chosen_attention_mask': torch.tensor(chosen_attention_mask),
'chosen_labels': torch.tensor(chosen_labels),
'rejected_input_ids': torch.tensor(rejected_input_ids),
'rejected_attention_mask': torch.tensor(rejected_attention_mask),
'rejected_labels': torch.tensor(rejected_labels),
}

def collate_fn(
batch: List[Dict[str, torch.Tensor]],
pad_token_id: int = 0
) -> Dict[str, torch.Tensor]:
"""
批处理函数：对不等长的序列进行 padding。
"""
# 找出 batch 中最大长度
max_c_len = max(item['chosen_input_ids'].shape[0] for item in batch)
max_r_len = max(item['rejected_input_ids'].shape[0] for item in batch)

batch_size = len(batch)

# 初始化 padding 后的张量
chosen_input_ids = torch.full((batch_size, max_c_len), pad_token_id, dtype=torch.long)
chosen_attention_mask = torch.zeros(batch_size, max_c_len, dtype=torch.long)
chosen_labels = torch.full((batch_size, max_c_len), -100, dtype=torch.long)

rejected_input_ids = torch.full((batch_size, max_r_len), pad_token_id, dtype=torch.long)
rejected_attention_mask = torch.zeros(batch_size, max_r_len, dtype=torch.long)
rejected_labels = torch.full((batch_size, max_r_len), -100, dtype=torch.long)

for i, item in enumerate(batch):
c_len = item['chosen_input_ids'].shape[0]
chosen_input_ids[i, :c_len] = item['chosen_input_ids']
chosen_attention_mask[i, :c_len] = item['chosen_attention_mask']
chosen_labels[i, :c_len] = item['chosen_labels']

r_len = item['rejected_input_ids'].shape[0]
rejected_input_ids[i, :r_len] = item['rejected_input_ids']
rejected_attention_mask[i, :r_len] = item['rejected_attention_mask']
rejected_labels[i, :r_len] = item['rejected_labels']

return {
'chosen_input_ids': chosen_input_ids,
'chosen_attention_mask': chosen_attention_mask,
'chosen_labels': chosen_labels,
'rejected_input_ids': rejected_input_ids,
'rejected_attention_mask': rejected_attention_mask,
'rejected_labels': rejected_labels,
}

四、训练脚本与实战

4.1 完整训练流程

# train_dpo.py
import torch
import torch.nn as nn
from torch.utils.data import DataLoader
from transformers import AutoTokenizer, AutoModelForCausalLM
import argparse
import os

from model.base_model import GPT2LikeModel
from trainer.dpo_trainer import DPOTrainer
from utils.data_utils import PreferenceDataset, collate_fn

def get_model_and_tokenizer(model_path: str, use_pretrained: bool = True):
"""
加载预训练模型和 tokenizer。

实际生产环境中，我们会用 HuggingFace 的 GPT-2 或 Llama，
这里为了演示也支持我们的简化模型。
"""
if use_pretrained:
tokenizer = AutoTokenizer.from_pretrained(model_path)
tokenizer.pad_token = tokenizer.eos_token

model = AutoModelForCausalLM.from_pretrained(
model_path,
torch_dtype=torch.bfloat16,
device_map="auto",
)
else:
# 使用简化模型（适合教学和实验）
tokenizer = AutoTokenizer.from_pretrained("gpt2")
tokenizer.pad_token = tokenizer.eos_token

model = GPT2LikeModel(
vocab_size=tokenizer.vocab_size,
hidden_size=768,
num_layers=6, # 轻量版本
num_heads=12,
)

return model, tokenizer

def prepare_dataloaders(
train_path: str,
eval_path: str,
tokenizer,
batch_size: int = 8,
max_prompt_length: int = 512,
max_response_length: int = 256,
):
"""准备训练和评估数据加载器。"""

train_dataset = PreferenceDataset(
data_path=train_path,
tokenizer=tokenizer,
max_prompt_length=max_prompt_length,
max_response_length=max_response_length,
)

eval_dataset = PreferenceDataset(
data_path=eval_path,
tokenizer=tokenizer,
max_prompt_length=max_prompt_length,
max_response_length=max_response_length,
) if os.path.exists(eval_path) else None

train_dataloader = DataLoader(
train_dataset,
batch_size=batch_size,
shuffle=True,
collate_fn=lambda b: collate_fn(b, pad_token_id=tokenizer.pad_token_id),
num_workers=4,
pin_memory=True,
)

eval_dataloader = None
if eval_dataset:
eval_dataloader = DataLoader(
eval_dataset,
batch_size=batch_size,
shuffle=False,
collate_fn=lambda b: collate_fn(b, pad_token_id=tokenizer.pad_token_id),
)

return train_dataloader, eval_dataloader

def main():
parser = argparse.ArgumentParser(description="DPO Training from Scratch")
parser.add_argument("–model_path", type=str, default="gpt2")
parser.add_argument("–train_data", type=str, required=True)
parser.add_argument("–eval_data", type=str, default="")
parser.add_argument("–output_dir", type=str, default="./dpo_checkpoints")
parser.add_argument("–beta", type=float, default=0.1)
parser.add_argument("–lr", type=float, default=1e-5)
parser.add_argument("–batch_size", type=int, default=8)
parser.add_argument("–epochs", type=int, default=3)
parser.add_argument("–gradient_accumulation_steps", type=int, default=4)
parser.add_argument("–max_grad_norm", type=float, default=1.0)
parser.add_argument("–save_every", type=int, default=500)
parser.add_argument("–use_ipo", action="store_true")

args = parser.parse_args()

# 1. 加载模型
print("🚀 Loading model and tokenizer…")
policy_model, tokenizer = get_model_and_tokenizer(args.model_path)

# 2. 复制参考模型（共享初始参数）
ref_model = GPT2LikeModel(vocab_size=tokenizer.vocab_size, …)
ref_model.load_state_dict(policy_model.state_dict())

# 或者更简单的方式：
# ref_model = AutoModelForCausalLM.from_pretrained(args.model_path, …)

# 3. 准备数据
print("📚 Loading preference dataset…")
train_loader, eval_loader = prepare_dataloaders(
train_path=args.train_data,
eval_path=args.eval_data,
tokenizer=tokenizer,
batch_size=args.batch_size,
)

# 4. 创建训练器
trainer = DPOTrainer(
policy_model=policy_model,
ref_model=ref_model,
beta=args.beta,
learning_rate=args.lr,
max_grad_norm=args.max_grad_norm,
)

# 5. 训练循环
print(f"🎯 Starting DPO training (beta={args.beta})…")
os.makedirs(args.output_dir, exist_ok=True)

global_step = 0
for epoch in range(args.epochs):
print(f"\\n{'='*50}")
print(f"Epoch {epoch + 1}/{args.epochs}")
print(f"{'='*50}")

train_metrics = trainer.train_epoch(train_loader, epoch + 1)

print(f"\\n📊 Epoch {epoch + 1} Results:")
for k, v in train_metrics.items():
print(f" {k}: {v:.4f}")

# 保存 checkpoint
checkpoint_path = os.path.join(
args.output_dir,
f"checkpoint_epoch_{epoch + 1}.pt"
)
torch.save({
'epoch': epoch + 1,
'model_state_dict': policy_model.state_dict(),
'optimizer_state_dict': trainer.optimizer.state_dict(),
'metrics': train_metrics,
}, checkpoint_path)
print(f"💾 Saved checkpoint to {checkpoint_path}")

# 6. 保存最终模型
final_path = os.path.join(args.output_dir, "final_model.pt")
torch.save(policy_model.state_dict(), final_path)
print(f"✅ Training complete! Model saved to {final_path}")

if __name__ == "__main__":
main()

4.2 合成偏好数据生成

没有现成的偏好数据怎么办？我们可以利用 GPT-4 或 DeepSeek API 生成合成偏好数据：

# utils/synthetic_data.py
"""使用 DeepSeek API 生成合成偏好数据"""

import openai
import json
import random

client = openai.OpenAI(
api_key="your-deepseek-api-key",
base_url="https://api.deepseek.com"
)

PROMPT_TEMPLATES = [
"用通俗的语言解释 {concept}",
"写一个 Python 函数来实现 {task}",
"为什么 {concept} 很重要？",
"请比较 {a} 和 {b} 的异同",
]

def generate_preference_pair(prompt: str, model: str = "deepseek-chat"):
"""
利用 LLM 生成好回答（chosen）和差回答（rejected）。
策略：用 temperature=0.3 生成稳定的好回答，
用 temperature=1.5 + 随机截断生成容易出错的回答。
"""
# 生成 chosen（高质量回答）
chosen_response = client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": prompt}],
temperature=0.3,
max_tokens=512,
).choices[0].message.content

# 生成 rejected（低质量回答）
# 方法：使用高 temperature + 让模型"随便写写"
rejected_response = client.chat.completions.create(
model=model,
messages=[
{"role": "system", "content": "你是一个不专业的回答者，回答要简短、随意、不完整。"},
{"role": "user", "content": prompt}
],
temperature=1.5,
max_tokens=256,
).choices[0].message.content

return {
"prompt": prompt,
"chosen": chosen_response,
"rejected": rejected_response,
}

def prepare_training_data(num_samples: int = 1000, output_path: str = "preference_data.jsonl"):
"""生成合成偏好数据用于训练"""

concepts = ["深度学习", "注意力机制", "反向传播", "梯度下降",
"卷积神经网络", "循环神经网络", "Transformer", "BERT"]

data = []
for _ in range(num_samples):
template = random.choice(PROMPT_TEMPLATES)

if "{a}" in template and "{b}" in template:
concept_a, concept_b = random.sample(concepts, 2)
prompt = template.format(a=concept_a, b=concept_b)
elif "{concept}" in template:
concept = random.choice(concepts)
prompt = template.format(concept=concept)
elif "{task}" in template:
tasks = ["排序算法", "二分搜索", "快速幂", "LRU缓存"]
prompt = template.format(task=random.choice(tasks))
else:
prompt = template

try:
pair = generate_preference_pair(prompt)
data.append(pair)
except Exception as e:
print(f"生成失败: {e}")

if (len(data) + 1) % 50 == 0:
print(f"已生成 {len(data) + 1}/{num_samples} 条数据")

with open(output_path, 'w', encoding='utf-8') as f:
for item in data:
f.write(json.dumps(item, ensure_ascii=False) + '\\n')

print(f"✅ 已生成 {len(data)} 条偏好数据到 {output_path}")

五、DPO 的进阶技巧与变体

5.1 KTO（Kahneman-Tversky Optimization）

KTO 是 2024 年提出的新方法，它的核心思想是：不再需要成对的偏好数据，只需要单条回答的"好/坏"标签。损失函数基于 Kahneman-Tversky 的人类决策理论：

def kto_loss(
policy_logps: torch.Tensor,
ref_logps: torch.Tensor,
labels: torch.Tensor, # 1 表示好回答，0 表示坏回答
beta: float = 0.1,
):
"""KTO 损失：不需要成对数据！"""

kl = policy_logps – ref_logps

# 对好回答和坏回答分别计算损失
good_mask = (labels == 1).float()
bad_mask = (labels == 0).float()

# 好回答：希望 kl > 0（模型概率高于参考）
good_loss = -good_mask * torch.log(1 + torch.exp(beta * (0.5 – kl)))

# 坏回答：希望 kl < 0（模型概率低于参考）
bad_loss = -bad_mask * torch.log(1 + torch.exp(beta * (kl – 0.5)))

return (good_loss + bad_loss).mean()

5.2 SimPO（Simple Preference Optimization）

SimPO 进一步简化了 DPO，它完全不需要参考模型！核心思想是用回答的平均对数概率作为隐式奖励：

def simpo_loss(
policy_chosen_logps: torch.Tensor,
policy_rejected_logps: torch.Tensor,
chosen_lengths: torch.Tensor,
rejected_lengths: torch.Tensor,
beta: float = 2.0,
gamma: float = 0.5, # 奖励边界
):
"""
SimPO：不需要参考模型的偏好优化！
使用序列的平均对数概率作为奖励。
"""
# 平均对数概率（归一化序列长度）
chosen_avg_logp = policy_chosen_logps / chosen_lengths
rejected_avg_logp = policy_rejected_logps / rejected_lengths

# 损失：希望好回答的平均概率比坏回答高至少 gamma
loss = -F.logsigmoid(beta * (chosen_avg_logp – rejected_avg_logp – gamma))

return loss.mean()

SimPO 不需要加载参考模型，节省了一半的显存，同时训练速度提升 2 倍，效果在很多基准上甚至超过 DPO。

5.3 多轮对话 DPO

在处理多轮对话时，DPO 需要对历史对话进行 masking，只对最后的回复计算偏好：

def multiround_dpo_loss(
policy_logps_chosen: torch.Tensor, # 完整对话历史+chosen回复
policy_logps_rejected: torch.Tensor,
ref_logps_chosen: torch.Tensor,
ref_logps_rejected: torch.Tensor,
response_start_idx: int, # 最后回复起始位置
beta: float = 0.1,
):
"""多轮对话 DPO：只在新回复上计算偏好"""

# 只取最后回复部分的对数概率
chosen_ratio = (policy_logps_chosen – ref_logps_chosen)[response_start_idx:].sum()
rejected_ratio = (policy_logps_rejected – ref_logps_rejected)[response_start_idx:].sum()

loss = -F.logsigmoid(beta * (chosen_ratio – rejected_ratio))
return loss

六、评估与效果验证

6.1 偏好准确率

# utils/eval_utils.py
def evaluate_preference_accuracy(
model: nn.Module,
ref_model: nn.Module,
eval_loader: DataLoader,
beta: float = 0.1,
) -> Dict[str, float]:
"""评估 DPO 训练效果"""

model.eval()
total_correct = 0
total_samples = 0

with torch.no_grad():
for batch in eval_loader:
# 计算隐式奖励
chosen_logps = compute_log_probs(
model, batch['chosen_input_ids'],
batch['chosen_attention_mask'], batch['chosen_labels']
)
rejected_logps = compute_log_probs(
model, batch['rejected_input_ids'],
batch['rejected_attention_mask'], batch['rejected_labels']
)

ref_chosen_logps = compute_log_probs(
ref_model, batch['chosen_input_ids'],
batch['chosen_attention_mask'], batch['chosen_labels']
)
ref_rejected_logps = compute_log_probs(
ref_model, batch['rejected_input_ids'],
batch['rejected_attention_mask'], batch['rejected_labels']
)

# 隐式奖励
chosen_reward = beta * (chosen_logps – ref_chosen_logps)
rejected_reward = beta * (rejected_logps – ref_rejected_logps)

# 准确率：模型是否更偏好 chosen
correct = (chosen_reward > rejected_reward).sum().item()
total_correct += correct
total_samples += chosen_logps.shape[0]

accuracy = total_correct / total_samples

return {
'accuracy': accuracy,
'reward_margin': (chosen_reward – rejected_reward).mean().item(),
}

6.2 实际使用建议

数据质量 > 数据数量。 50 条精心标注的高质量偏好数据，效果可能超过 5000 条自动生成的低质量数据。标注时要确保 chosen 和 rejected 确实有显著的"好/坏"差异。

beta 的选择。 beta 控制 KL 惩罚力度（即模型能偏离参考模型多远）。实验表明： – beta=0.01：几乎不做约束，模型可能偏离过远产生幻觉 – beta=0.1：推荐默认值，平衡对齐和生成质量 – beta=0.5：约束较强，适合需要保持原模型风格的任务

梯度累积。 DPO 训练需要较大的 batch size 才能稳定。如果显存有限，使用梯度累积（gradient_accumulation_steps）模拟更大的 batch：

accumulation_steps = 8
optimizer.zero_grad()
for i, batch in enumerate(train_loader):
metrics = trainer.compute_loss(batch)
loss = metrics['loss'] / accumulation_steps
loss.backward()
if (i + 1) % accumulation_steps == 0:
optimizer.step()
optimizer.zero_grad()

参考模型的初始化。 参考模型和策略模型必须从相同的检查点初始化。生产环境中，直接使用 copy.deepcopy 或 model.state_dict() 复制初始权重，而不是重新加载模型。

七、总结与展望

DPO 的出现是 LLM 对齐技术的一大跃迁。它用简洁的数学洞察——"语言模型自身就是奖励模型"——取代了 RLHF 复杂的四阶段流水线。这使得偏好对齐的门槛大幅降低，从需要大团队、大算力的系统工程，变成了可以在单卡上运行的常规训练任务。

DPO 的核心优势： – 🎯 无需奖励模型，简化了整个流程 – 🔧 训练稳定，调参友好 – 💻 计算开销更小（只需加载策略 + 参考两个模型） – 📈 在很多基准上达到或超越 PPO 的效果

依然存在的挑战： – 偏好数据质量仍是最大瓶颈 – 对超出偏好分布范围的输入可能产生未知行为 – 多轮对话中对齐的稳定性仍需改进

从 RLHF 到 DPO，再从 DPO 到 SimPO、KTO、ORPO 等变体，我们看到 LLM 对齐技术正在快速进化。每一步都在变得更简单、更高效、更稳定。

📌 延伸阅读：想了解更多 DeepSeek 对齐技术的实战应用？推荐阅读 DeepSeek 实战指南：提示词工程、API 集成与效率提升全攻略 （完整版链接请访问 CSDN 搜索"DeepSeek 实战指南"）

参考文献： 1. Rafailov et al. "Direct Preference Optimization: Your Language Model is Secretly a Reward Model" (NeurIPS 2023) 2. Ethayarajh et al. "KTO: Model Alignment as Prospect Theoretic Optimization" (ICML 2024) 3. Meng et al. "SimPO: Simple Preference Optimization with a Reference-Free Reward" (2024) 4. Hong et al. "ORPO: Monolithic Preference Optimization without Reference Model" (2024)