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 年已经是偏好对齐的主流方法，关键建议：
...
