DPO训练实践指南：直接偏好优化的工程落地

RLHF流程:
1. 训练奖励模型
2. 使用PPO优化策略（需要4个模型同时运行：策略、参考、奖励、价值）
3. 训练不稳定，超参数敏感

DPO流程:
1. 直接从偏好数据优化策略（只需2个模型：策略、参考）
2. 训练稳定，超参数简单
3. 本质上是分类问题

DPO损失 = -log σ(β·(log π(y_w|x)/π_ref(y_w|x) - log π(y_l|x)/π_ref(y_l|x)))

其中:
y_w = 偏好（chosen）的回答
y_l = 不偏好（rejected）的回答
π = 当前策略
π_ref = 参考策略（通常是SFT模型）
β = 温度参数

{
  "prompt": "如何学习编程？",
  "chosen": "学习编程建议从Python开始，因为它语法简洁...（优质回答）",
  "rejected": "编程很难，你可能学不会。（劣质回答）"
}

class DPODataBuilder:
    async def build_from_human_annotations(self, annotations):
        """从人工标注构建DPO数据"""
        dpo_data = []
        
        for ann in annotations:
            dpo_data.append({
                "prompt": ann.prompt,
                "chosen": ann.preferred_response,
                "rejected": ann.rejected_response
            })
        
        return dpo_data
    
    async def build_from_ai_feedback(self, prompts, model):
        """使用AI反馈构建DPO数据"""
        dpo_data = []
        
        for prompt in prompts:
            # 1. 生成多个回答
            responses = await model.generate(
                prompt, num_return_sequences=2, temperature=0.8
            )
            
            # 2. 使用评判模型选择更好的回答
            judgment = await self.judge(prompt, responses[0], responses[1])
            
            if judgment["winner"] == "A":
                chosen, rejected = responses[0], responses[1]
            else:
                chosen, rejected = responses[1], responses[0]
            
            # 3. 只保留有明确偏好的数据
            if judgment["confidence"] > 0.7:
                dpo_data.append({
                    "prompt": prompt,
                    "chosen": chosen,
                    "rejected": rejected
                })
        
        return dpo_data
    
    async def judge(self, prompt, response_a, response_b):
        """使用LLM评判哪个回答更好"""
        judge_prompt = f"""
        问题: {prompt}
        回答A: {response_a}
        回答B: {response_b}
        
        哪个回答更好？A还是B？
        输出JSON: {{"winner": "A"/"B", "confidence": 0-1, "reason": "..."}}
        """
        return await self.judge_model.call(judge_prompt)

class DPODataQualityChecker:
    def check(self, dataset):
        issues = []
        
        for sample in dataset:
            # 1. chosen和rejected不能太相似
            similarity = self.compute_similarity(sample["chosen"], sample["rejected"])
            if similarity > 0.9:
                issues.append({"issue": "too_similar", "sample": sample})
            
            # 2. chosen不能比rejected差
            # （需要评判模型验证）
            
            # 3. prompt长度合理
            if len(sample["prompt"]) > 2048:
                issues.append({"issue": "prompt_too_long", "sample": sample})
            
            # 4. 回答长度差异不能太大
            len_diff = abs(len(sample["chosen"]) - len(sample["rejected"]))
            if len_diff > 2000:
                issues.append({"issue": "length_bias", "sample": sample})
        
        return issues

from trl import DPOTrainer, DPOConfig

# 1. 加载模型
model = AutoModelForCausalLM.from_pretrained("sft-model-path")
ref_model = AutoModelForCausalLM.from_pretrained("sft-model-path")

# 2. 配置
config = DPOConfig(
    output_dir="./dpo-output",
    num_train_epochs=3,
    per_device_train_batch_size=4,
    gradient_accumulation_steps=4,
    learning_rate=5e-7,
    warmup_ratio=0.1,
    beta=0.1,  # DPO温度参数
    max_prompt_length=512,
    max_length=2048,
    logging_steps=10,
    save_steps=100,
    bf16=True,
)

# 3. 训练
trainer = DPOTrainer(
    model=model,
    ref_model=ref_model,
    args=config,
    train_dataset=dpo_dataset,
    tokenizer=tokenizer,
)

trainer.train()

# β（beta）选择
beta = 0.1   # 保守，接近SFT模型
beta = 0.3   # 平衡
beta = 0.5   # 激进，更大偏移

# 学习率
lr = 5e-7    # DPO通常需要很小的学习率
              # 比SFT小10-100倍

# epoch数
epochs = 1-3 # DPO通常1-3轮足够
              # 过多轮会导致过拟合

# 先用QLoRA加载模型
model = AutoModelForCausalLM.from_pretrained(
    "base-model",
    quantization_config=bnb_config,
    device_map="auto"
)
model = get_peft_model(model, lora_config)

# DPO训练
trainer = DPOTrainer(
    model=model,
    ref_model=None,  # QLoRA模式下不需要单独的ref_model
    args=config,
    train_dataset=dpo_dataset,
    tokenizer=tokenizer,
)

class DPOEvaluator:
    async def evaluate(self, model, eval_dataset):
        metrics = {
            "accuracy": [],      # DPO准确率（chosen得分>rejected得分）
            "margin": [],        # chosen和rejected的分数差
            "reward_accuracy": [] # 奖励准确率
        }
        
        for sample in eval_dataset:
            # 计算chosen和rejected的对数概率
            chosen_logprob = self.compute_logprob(
                model, sample["prompt"], sample["chosen"]
            )
            rejected_logprob = self.compute_logprob(
                model, sample["prompt"], sample["rejected"]
            )
            
            # 准确率
            metrics["accuracy"].append(chosen_logprob > rejected_logprob)
            
            # 边际
            metrics["margin"].append(chosen_logprob - rejected_logprob)
        
        return {k: np.mean(v) for k, v in metrics.items()}

# IPO使用平方损失替代对数损失
# 更稳定，不易过拟合

# KTO数据格式
{"prompt": "...", "response": "...", "label": "good"}
{"prompt": "...", "response": "...", "label": "bad"}

# SimPO损失直接使用策略模型自身的概率
# 更简单，训练更快