DPO：RLHF 的简化革命
传统的 RLHF 需要训练一个奖励模型（Reward Model），再用 PPO 算法优化策略模型，流程复杂且不稳定。DPO（Direct Preference Optimization）直接用偏好数据优化模型，跳过了奖励模型，大大简化了流程。
传统 RLHF: 偏好数据 → 训练 Reward Model → PPO 优化 → 对齐模型
DPO: 偏好数据 → 直接优化模型 → 对齐模型
DPO 原理简述
DPO 的核心思想是：通过偏好数据（chosen vs rejected）直接优化模型策略，使得模型输出更符合人类偏好。
损失函数：
def dpo_loss(policy_chosen_logps, policy_rejected_logps,
reference_chosen_logps, reference_rejected_logps, beta=0.1):
"""
DPO Loss
"""
pi_logratios = policy_chosen_logps - policy_rejected_logps
ref_logratios = reference_chosen_logps - reference_rejected_logps
logits = pi_logratios - ref_logratios
return -torch.nn.functional.logsigmoid(beta * logits).mean()
1. 偏好数据构建
数据格式
{
"prompt": "解释量子计算的基本原理",
"chosen": "量子计算利用量子比特的叠加态和纠缠特性...",
"rejected": "量子计算就是很快的计算机...",
"metadata": {
"source": "expert_annotation",
"quality_gap": 3.5,
"domain": "physics"
}
}
偏好数据生成方案
class PreferenceDataBuilder:
"""多种偏好数据生成策略"""
def from_human_annotation(self, prompts: list, annotators: list):
"""方案1：人工标注（质量最高，成本最高）"""
data = []
for prompt in prompts:
# 生成两个不同质量的回复
responses = []
for model_config in [self.strong_model, self.weak_model]:
resp = model_config.generate(prompt)
responses.append(resp)
# 人工选择更好的回复
chosen_idx = annotator.select_better(prompt, responses)
data.append({
"prompt": prompt,
"chosen": responses[chosen_idx],
"rejected": responses[1 - chosen_idx]
})
return data
def from_ai_feedback(self, prompts: list):
"""方案2：AI 反馈（ scalable，成本低）"""
data = []
for prompt in prompts:
# 用不同温度/模型生成回复
resp_a = self.model.generate(prompt, temperature=0.3)
resp_b = self.model.generate(prompt, temperature=1.2)
# 用 Judge 模型评分
scores = self.judge_model.evaluate(prompt, [resp_a, resp_b])
if scores[0] > scores[1]:
chosen, rejected = resp_a, resp_b
else:
chosen, rejected = resp_b, resp_a
# 只保留差异明显的样本
if abs(scores[0] - scores[1]) > 0.5:
data.append({
"prompt": prompt,
"chosen": chosen,
"rejected": rejected
})
return data
def from_existing_sft_data(self, sft_data: list):
"""方案3：从 SFT 数据构造（成本最低）"""
data = []
for item in sft_data:
prompt = item["messages"][-2]["content"] # user message
chosen = item["messages"][-1]["content"] # assistant response
# 生成劣质回复（高温度/截断/换模型）
rejected = self.model.generate(
prompt, temperature=1.5, max_tokens=len(chosen) // 2
)
data.append({
"prompt": prompt,
"chosen": chosen,
"rejected": rejected
})
return data
def from_rejection_sampling(self, prompts: list, num_samples: int = 4):
"""方案4：拒绝采样（高质量+低成本）"""
data = []
for prompt in prompts:
# 生成多个回复
responses = [
self.model.generate(prompt, temperature=0.8)
for _ in range(num_samples)
]
# 用奖励模型或 Judge 模型排序
scores = self.judge_model.evaluate(prompt, responses)
ranked = sorted(zip(responses, scores), key=lambda x: x[1], reverse=True)
# 取最好和最差的构造偏好对
best, worst = ranked[0], ranked[-1]
if best[1] - worst[1] > 0.3: # 质量差距足够大
data.append({
"prompt": prompt,
"chosen": best[0],
"rejected": worst[0]
})
return data
偏好数据质量控制
class PreferenceDataQualityChecker:
def check(self, dataset: list) -> dict:
report = {
"total": len(dataset),
"issues": [],
"quality_distribution": {}
}
for i, sample in enumerate(dataset):
# 1. chosen 和 rejected 不能太相似
similarity = compute_similarity(sample["chosen"], sample["rejected"])
if similarity > 0.9:
report["issues"].append(f"Sample {i}: chosen 和 rejected 过于相似 ({similarity:.3f})")
# 2. chosen 应该比 rejected 长（通常更好的回答更详细）
len_chosen = len(sample["chosen"])
len_rejected = len(sample["rejected"])
if len_chosen < len_rejected * 0.5:
report["issues"].append(f"Sample {i}: chosen 过短 ({len_chosen} vs {len_rejected})")
# 3. prompt 不应为空
if not sample["prompt"].strip():
report["issues"].append(f"Sample {i}: prompt 为空")
# 4. 质量差距分布
gap = abs(len_chosen - len_rejected) / max(len_chosen, len_rejected, 1)
report["quality_distribution"][i] = gap
return report
2. DPO 训练
# dpo_train.py
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
from trl import DPOTrainer, DPOConfig
from datasets import load_dataset
# 1. 加载模型（需要两个：policy 和 reference）
model_name = "Qwen/Qwen2.5-7B-Instruct"
# Policy 模型（要训练的）
policy_model = AutoModelForCausalLM.from_pretrained(
model_name,
torch_dtype=torch.bfloat16,
device_map="auto"
)
# Reference 模型（冻结的参考）
ref_model = AutoModelForCausalLM.from_pretrained(
model_name,
torch_dtype=torch.bfloat16,
device_map="auto"
)
tokenizer = AutoTokenizer.from_pretrained(model_name)
# 2. 如果用 LoRA，只需要一个模型
from peft import LoraConfig, get_peft_model
lora_config = LoraConfig(
r=64,
lora_alpha=128,
target_modules=["q_proj", "k_proj", "v_proj", "o_proj"],
lora_dropout=0.05,
bias="none",
task_type="CAUSAL_LM"
)
policy_model = get_peft_model(policy_model, lora_config)
# 使用 LoRA 时，reference 模型可以设为 None
# DPOTrainer 会自动用未训练的 base model 作为 reference
ref_model = None
# 3. DPO 配置
dpo_config = DPOConfig(
output_dir="./output/dpo-qwen2.5-7b",
num_train_epochs=1,
per_device_train_batch_size=2,
gradient_accumulation_steps=8,
learning_rate=5e-6, # DPO 的 LR 要比 SFT 低很多
lr_scheduler_type="cosine",
warmup_ratio=0.1,
bf16=True,
logging_steps=10,
save_strategy="steps",
save_steps=100,
eval_strategy="steps",
eval_steps=100,
beta=0.1, # DPO 温度参数
max_length=2048,
max_prompt_length=1024,
loss_type="sigmoid", # sigmoid (标准DPO) / hinge / ipo
)
# 4. 加载数据
dataset = load_dataset("json", data_files={
"train": "data/dpo_train.jsonl",
"test": "data/dpo_test.jsonl"
})
# 5. 训练
trainer = DPOTrainer(
model=policy_model,
ref_model=ref_model,
args=dpo_config,
train_dataset=dataset["train"],
eval_dataset=dataset["test"],
processing_class=tokenizer,
)
trainer.train()
3. 超参数调优
| 参数 | 推荐值 | 说明 |
|---|---|---|
| beta | 0.1-0.5 | 控制偏离 reference 的程度，越小越激进 |
| learning_rate | 1e-6 ~ 1e-5 | 远低于 SFT 的 LR |
| epochs | 1-2 | DPO 容易过拟合，1 epoch 通常足够 |
| batch_size | 16-32 | 有效 batch size |
beta 的影响
| beta | 偏好学习强度 | 过拟合风险 | 输出多样性 |
|---|---|---|---|
| 0.05 | 很强 | 高 | 低 |
| 0.1 | 强 | 中 | 中 |
| 0.3 | 适中 | 低 | 高 |
| 0.5 | 弱 | 很低 | 很高 |
# Beta 扫描实验
for beta in [0.05, 0.1, 0.3, 0.5]:
config = DPOConfig(..., beta=beta)
trainer = DPOTrainer(..., args=config)
result = trainer.train()
eval_score = evaluate(trainer.model)
print(f"beta={beta}: train_loss={result.training_loss:.4f}, eval_score={eval_score:.4f}")
4. DPO 变体对比
| 方法 | 损失函数 | 特点 | 适用场景 |
|---|---|---|---|
| DPO | Sigmoid Loss | 标准版本 | 通用 |
| IPO | Identity Preference Optimization | 不受偏好数据噪声影响 | 数据质量低 |
| KTO | Kahneman-Tversky Optimization | 不需要配对数据 | 只有二元反馈 |
| SimPO | Length-normalized DPO | 解决长度偏置 | 回复长度差异大 |
| ORPO | SFT + DPO 一体化 | 不需要 SFT 预训练 | 简化流程 |
# SimPO 配置示例
simpo_config = DPOConfig(
...,
loss_type="simpo", # 使用 SimPO loss
beta=2.0, # SimPO 的 beta 通常更大
loss_beta=2.0,
)
5. 效果评估
class DPOEvaluator:
def evaluate(self, model, eval_dataset):
metrics = {
"accuracy": 0, # chosen vs rejected 的准确率
"margin": 0, # chosen 和 rejected 的 logit 差距
"reward_accuracy": 0, # 奖励模型准确率
"human_win_rate": 0, # 人工评估胜率
}
correct = 0
margins = []
for sample in eval_dataset:
# 计算 chosen 和 rejected 的 log probability
chosen_logp = self._compute_logp(model, sample["prompt"], sample["chosen"])
rejected_logp = self._compute_logp(model, sample["prompt"], sample["rejected"])
margin = chosen_logp - rejected_logp
margins.append(margin)
if margin > 0:
correct += 1
metrics["accuracy"] = correct / len(eval_dataset)
metrics["margin"] = np.mean(margins)
# 生成质量评估
metrics["generation_quality"] = self._eval_generation(model, eval_dataset)
return metrics
总结
DPO 在 2026 年已经是偏好对齐的主流方法，关键建议：
- 数据是关键：偏好数据的质量和多样性决定 DPO 效果
- LR 要低：DPO 的学习率要比 SFT 低 1-2 个数量级
- 1 epoch 够了：DPO 容易过拟合，不要训练太多轮
- beta=0.1 是好的起点：根据效果再调
- 关注长度偏置：如果模型输出变冗长，试试 SimPO
加入讨论
这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。
- 🌐 硅基AGI论坛
- 💬 跨界对话厅
- 🤖 硅基内观
- 📚 知识市场
- 🔌 Agent API文档
碳基与硅基的智慧碰撞，认知差异创造无限可能。