DPO偏好对齐训练：数据工程与超参调优

RLHF Pipeline (复杂):
  SFT → Reward Model → PPO → 对齐模型
  (3个模型，4个训练阶段)

DPO Pipeline (简洁):
  SFT → DPO → 对齐模型
  (1个模型，2个训练阶段)

# DPO Loss公式
# L_DPO = -log σ(β * (log π(y_w|x)/π_ref(y_w|x) - log π(y_l|x)/π_ref(y_l|x)))

# 其中：
# π: 当前策略模型
# π_ref: 参考模型（SFT后的模型，冻结）
# y_w: 偏好的回答 (winner)
# y_l: 不偏好的回答 (loser)
# β: 温度参数，控制偏离参考模型的程度
# σ: sigmoid函数

{
  "prompt": "解释什么是RAG？",
  "chosen": "RAG（检索增强生成）是一种结合外部知识库和大语言模型的技术。它通过在生成回答前先从知识库中检索相关文档，然后将这些文档作为上下文提供给模型，从而提高回答的准确性和时效性。\n\nRAG的主要优势包括：\n1. 知识更新无需重新训练\n2. 减少模型幻觉\n3. 可追溯信息来源",
  "rejected": "RAG就是让AI去搜索资料然后再回答问题的一种方法。"
}

# 标注界面数据格式
annotation_schema = {
    "prompt": "用户问题",
    "response_a": "模型A的回答",
    "response_b": "模型B的回答",
    "preference": "A" or "B" or "tie",  # 偏好选择
    "annotator": "标注者ID",
    "confidence": 1-5,  # 标注信心
    "reason": "偏好理由",
}

# 过滤低质量标注
def filter_annotations(data):
    filtered = []
    for item in data:
        # 排除低信心标注
        if item["confidence"] < 3:
            continue
        # 排除平局
        if item["preference"] == "tie":
            continue
        # 排除标注时间过短的（可能敷衍）
        if item.get("time_spent", 0) < 10:
            continue
        filtered.append(item)
    return filtered

def generate_preference_data(prompt, model_a_response, model_b_response, judge_model):
    """用强模型作为裁判生成偏好数据"""
    
    judge_prompt = f"""你是一个回答质量评估专家。请比较以下两个回答的优劣。

问题: {prompt}

回答A: {model_a_response}

回答B: {model_b_response}

评估维度:
1. 准确性: 信息是否正确
2. 完整性: 是否充分回答了问题
3. 清晰度: 表达是否清晰有条理
4. 有用性: 对用户是否有帮助

请返回JSON:
{{
  "preference": "A" or "B" or "tie",
  "scores": {{"accuracy": {{"A": 1-10, "B": 1-10}}, ...}},
  "reason": "偏好理由"
}}"""
    
    result = judge_model.invoke(judge_prompt)
    return json.loads(result)

def self_generate_preferences(model, prompts):
    """用模型自身生成偏好数据"""
    
    preferences = []
    for prompt in prompts:
        # 生成两个回答（不同温度/采样）
        response_good = model.generate(
            prompt, temperature=0.3, top_p=0.9  # 低温度=高质量
        )
        response_bad = model.generate(
            prompt, temperature=1.2, top_p=0.95  # 高温度=低质量
        )
        
        # 自动评估：用规则或小模型判断
        if score(response_good) > score(response_bad):
            preferences.append({
                "prompt": prompt,
                "chosen": response_good,
                "rejected": response_bad,
            })
    
    return preferences

def analyze_preference_data(dataset):
    """偏好数据集质量分析"""
    
    stats = {
        "total": len(dataset),
        "avg_chosen_length": np.mean([len(d["chosen"]) for d in dataset]),
        "avg_rejected_length": np.mean([len(d["rejected"]) for d in dataset]),
        "length_bias": 0,  # 是否存在"长=好"的偏差
        "overlap_rate": 0,  # chosen和rejected的相似度
    }
    
    # 长度偏差检查
    longer_chosen = sum(
        1 for d in dataset if len(d["chosen"]) > len(d["rejected"])
    )
    stats["length_bias"] = longer_chosen / len(dataset)
    # 如果>80%，说明存在严重的"长=好"偏差
    
    # 重叠度检查
    overlaps = []
    for d in dataset:
        # 计算chosen和rejected的编辑距离比例
        overlap = 1 - edit_distance(d["chosen"], d["rejected"]) / max(
            len(d["chosen"]), len(d["rejected"])
        )
        overlaps.append(overlap)
    stats["avg_overlap"] = np.mean(overlaps)
    # 如果重叠度>80%，chosen和rejected太相似，训练信号弱
    
    return stats

from datasets import Dataset
from transformers import AutoModelForCausalLM, AutoTokenizer
from trl import DPOTrainer, DPOConfig
import torch

# 加载SFT模型（作为初始策略和参考模型）
model = AutoModelForCausalLM.from_pretrained(
    "./sft-model",
    torch_dtype=torch.bfloat16,
    device_map="auto",
)

# 参考模型（冻结，不训练）
ref_model = AutoModelForCausalLM.from_pretrained(
    "./sft-model",
    torch_dtype=torch.bfloat16,
    device_map="auto",
)

tokenizer = AutoTokenizer.from_pretrained("./sft-model")
tokenizer.pad_token = tokenizer.eos_token

# 准备偏好数据
train_dataset = Dataset.from_list([
    {
        "prompt": "什么是GraphRAG？",
        "chosen": "GraphRAG是一种结合知识图谱的检索增强生成技术...",
        "rejected": "GraphRAG是一个AI工具。"
    },
    # ... 更多数据
])

# DPO配置
dpo_config = DPOConfig(
    output_dir="./dpo-output",
    
    # 核心超参
    beta=0.1,              # 温度参数，控制偏离参考模型的程度
    loss_type="sigmoid",   # 损失函数类型
    
    # 训练参数
    num_train_epochs=1,     # DPO通常1-2个epoch即可
    per_device_train_batch_size=4,
    gradient_accumulation_steps=4,
    
    # 学习率（DPO用比SFT更小的学习率）
    learning_rate=5e-7,     # 关键！DPO学习率要比SFT小100倍
    lr_scheduler_type="cosine",
    warmup_ratio=0.1,
    
    # 精度
    bf16=True,
    
    # 序列长度
    max_length=2048,
    max_prompt_length=1024,
    
    # 其他
    gradient_checkpointing=True,
    optim="adamw_torch",
    logging_steps=10,
    save_steps=500,
    save_total_limit=3,
    remove_unused_columns=False,
)

# 创建Trainer
dpo_trainer = DPOTrainer(
    model=model,
    ref_model=ref_model,
    args=dpo_config,
    train_dataset=train_dataset,
    tokenizer=tokenizer,
)

# 开始训练
dpo_trainer.train()

from peft import LoraConfig, get_peft_model
from transformers import BitsAndBytesConfig

# 4-bit量化加载
bnb_config = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_quant_type="nf4",
    bnb_4bit_compute_dtype=torch.bfloat16,
)

model = AutoModelForCausalLM.from_pretrained(
    "./sft-model",
    quantization_config=bnb_config,
    device_map="auto",
)

# 应用LoRA
lora_config = LoraConfig(
    r=64,
    lora_alpha=16,
    target_modules=["q_proj", "v_proj", "k_proj", "o_proj"],
    lora_dropout=0.05,
    bias="none",
    task_type="CAUSAL_LM",
)
model = get_peft_model(model, lora_config)

# QLoRA DPO训练
dpo_trainer = DPOTrainer(
    model=model,
    ref_model=None,  # QLoRA模式可以不传ref_model，内部自动处理
    args=dpo_config,
    train_dataset=train_dataset,
    tokenizer=tokenizer,
    peft_config=lora_config,
)

# Beta调参实验
for beta in [0.01, 0.05, 0.1, 0.2, 0.5]:
    config = DPOConfig(beta=beta, ...)
    trainer = DPOTrainer(model=model, ...)
    trainer.train()
    
    # 评估
    score = evaluate(model)
    print(f"beta={beta}: reward_accuracy={score}")

# DPO学习率推荐范围：1e-7 到 5e-6
# 比SFT小100倍，比LoRA小10倍

# SFT学习率: 2e-4
# LoRA学习率: 2e-4
# DPO学习率: 5e-7 (直接微调)
# DPO+LoRA学习率: 5e-6 (LoRA参数更少，可以稍大)

# DPO训练日志解读
"""
Step 10:
  rewards/accuracies: 0.62      # 偏好准确率，应持续上升
  rewards/margins: 0.15         # chosen和rejected的reward差，应增大
  rewards/chosen: 0.08          # chosen的reward，应上升
  rewards/rejected: -0.07       # rejected的reward，应下降
  loss: 0.68                    # DPO loss，应下降
  logps/chosen: -145.3          # chosen的对数概率
  logps/rejected: -152.1        # rejected的对数概率
"""

def diagnose_dpo_training(trainer_logs):
    """诊断DPO训练问题"""
    
    final_accuracy = trainer_logs[-1]["rewards/accuracies"]
    final_margin = trainer_logs[-1]["rewards/margins"]
    
    if final_accuracy < 0.6:
        return "偏好准确率低：数据质量可能有问题，检查chosen/rejected差异"
    
    if final_margin < 0.05:
        return "Reward margin小：β可能过小或学习率过低"
    
    chosen_logp_trend = [l["logps/chosen"] for l in trainer_logs]
    if chosen_logp_trend[-1] < chosen_logp_trend[0] - 50:
        return "Chosen概率大幅下降：模型在退化，降低学习率或减少epoch"
    
    return "训练状态正常"