DPO 深度解析：绕过强化学习的直接偏好优化

1
from dataclasses import dataclass
2
from typing import List
3

4
@dataclass
5
class PreferenceSample:
6
    """偏好数据的一个样本"""
7
    prompt: str              # 用户指令/问题
8
    chosen: str               # 人类标注为"更好"的回答
9
    rejected: str             # 人类标注为"更差"的回答
10

11
    # 可选元数据
12
    chosen_score: float = 0.0   # 打分（如果有）
13
    source: str = ""             # 数据来源（人标/AI标）
14
    difficulty: str = ""         # 简单/中等/困难
15

16
# 典型的偏好数据集格式
17
# Anthropic HH-RLHF, Stanford SHP, OpenAssistant, LMSYS-Chat
18
preference_dataset: List[PreferenceSample] = load_preference_data()

1
def tokenize_preference_batch(
2
    samples: List[PreferenceSample],
3
    tokenizer,
4
    max_prompt_len: int = 512,
5
    max_response_len: int = 512,
6
):
7
    """
8
    将偏好数据 tokenize，返回：
9
    - chosen_input_ids: [B, prompt + chosen]
10
    - rejected_input_ids: [B, prompt + rejected]
11
    - chosen_response_mask: 标记 response 部分的 mask（用于计算 log prob）
12
    - rejected_response_mask: 同上
13
    """
14
    chosen_enc = tokenizer(
15
        [s.prompt + s.chosen for s in samples],
16
        max_length=max_prompt_len + max_response_len,
17
        truncation=True,
18
        padding=True,
19
        return_tensors="pt",
20
    )
21
    rejected_enc = tokenizer(
22
        [s.prompt + s.rejected for s in samples],
23
        max_length=max_prompt_len + max_response_len,
24
        truncation=True,
25
        padding=True,
26
        return_tensors="pt",
27
    )
28

29
    # 构建 response mask（tokenize 后的 answer 部分）
30
    chosen_response_mask = build_response_mask(
31
        tokenizer, samples, chosen_enc, "chosen"
32
    )
33
    rejected_response_mask = build_response_mask(
34
        tokenizer, samples, rejected_enc, "rejected"
35
    )
36

37
    return {
38
        "chosen_input_ids": chosen_enc["input_ids"],
39
        "chosen_attention_mask": chosen_enc["attention_mask"],
40
        "chosen_response_mask": chosen_response_mask,
41
        "rejected_input_ids": rejected_enc["input_ids"],
42
        "rejected_attention_mask": rejected_enc["attention_mask"],
43
        "rejected_response_mask": rejected_response_mask,
44
    }
45

46

47
def build_response_mask(tokenizer, samples, enc, field):
48
    """构建只覆盖 response 部分的 mask"""
49
    mask = torch.zeros_like(enc["input_ids"], dtype=torch.bool)
50
    for i, sample in enumerate(samples):
51
        prompt_len = len(tokenizer.encode(sample.prompt, add_special_tokens=True))
52
        mask[i, prompt_len:] = True
53
    return mask

1
import torch
2
import torch.nn.functional as F
3

4
def compute_token_log_probs(
5
    model,
6
    input_ids: torch.Tensor,
7
    attention_mask: torch.Tensor,
8
) -> torch.Tensor:
9
    """
10
    计算每个 token 的 log π(y_i | x, y_{<i})
11
    输入: (B, L)
12
    输出: (B, L-1)  — 最后一个 token 无法预测下一个
13
    """
14
    outputs = model(input_ids=input_ids, attention_mask=attention_mask)
15
    logits = outputs.logits[:, :-1, :]  # (B, L-1, V)
16
    labels = input_ids[:, 1:]          # (B, L-1)
17
    log_probs = F.log_softmax(logits, dim=-1)
18
    token_log_probs = log_probs.gather(2, labels.unsqueeze(-1)).squeeze(-1)  # (B, L-1)
19
    return token_log_probs
20

21

22
def compute_sequence_log_prob(
23
    model, input_ids, attention_mask, response_mask
24
) -> torch.Tensor:
25
    """
26
    计算 response 部分的 log π(response | prompt) 总和。
27
    只对 response 部分 token 求和，prompt 部分 mask 掉。
28

29
    公式: Σ_{t∈response} log π(y_t | prompt, y_{<t})
30
    """
31
    token_log_probs = compute_token_log_probs(model, input_ids, attention_mask)
32
    # 对应位置向前移动一位（log prob 与 labels 对齐）
33
    response_log_probs = token_log_probs * response_mask[:, 1:]
34
    return response_log_probs.sum(dim=-1)  # (B,)
35

36

37
def compute_log_ratios(
38
    policy_chosen_lps: torch.Tensor,   # (B,)
39
    policy_rejected_lps: torch.Tensor, # (B,)
40
    ref_chosen_lps: torch.Tensor,      # (B,)
41
    ref_rejected_lps: torch.Tensor,   # (B,)
42
) -> torch.Tensor:
43
    """
44
    计算 Δ_θ = log(π_θ(y_c)/π_ref(y_c)) - log(π_θ(y_r)/π_ref(y_r))
45
    这是 DPO 损失的核心量
46
    """
47
    return (
48
        (policy_chosen_lps - ref_chosen_lps)
49
        - (policy_rejected_lps - ref_rejected_lps)
50
    )

1
def dpo_loss(
2
    policy_chosen_lps: torch.Tensor,
3
    policy_rejected_lps: torch.Tensor,
4
    ref_chosen_lps: torch.Tensor,
5
    ref_rejected_lps: torch.Tensor,
6
    beta: float = 0.1,
7
):
8
    """
9
    DPO 损失。
10
    所有输入: (B,)
11

12
    损失: -E[log σ(β * Δ_θ)]
13
    其中 Δ_θ = log(π_θ(y_c)/π_ref(y_c)) - log(π_θ(y_r)/π_ref(y_r))
14
    """
15
    policy_logratios = policy_chosen_lps - policy_rejected_lps
16
    ref_logratios = ref_chosen_lps - ref_rejected_lps
17

18
    # β * (Δ_π - Δ_ref) = β * Δ_θ
19
    logits = beta * (policy_logratios - ref_logratios)
20

21
    loss = -F.logsigmoid(logits).mean()
22

23
    # 辅助指标
24
    with torch.no_grad():
25
        chosen_rewards = beta * (policy_chosen_lps - ref_chosen_lps)
26
        rejected_rewards = beta * (policy_rejected_lps - ref_rejected_lps)
27
        reward_margin = (chosen_rewards - rejected_rewards).mean()
28
        accuracy = (chosen_rewards > rejected_rewards).float().mean()
29

30
    return loss, {
31
        "loss": loss.item(),
32
        "reward_margin": reward_margin.item(),
33
        "accuracy": accuracy.item(),
34
        "chosen_reward_mean": chosen_rewards.mean().item(),
35
        "rejected_reward_mean": rejected_rewards.mean().item(),
36
    }
37

38

39
def train_dpo(
40
    policy_model,
41
    ref_model,
42
    train_loader,
43
    beta: float = 0.1,
44
    lr: float = 1e-6,
45
    num_epochs: int = 1,
46
    max_grad_norm: float = 1.0,
47
    log_interval: int = 10,
48
):
49
    """
50
    完整的 DPO 训练循环。
51
    """
52
    optimizer = torch.optim.AdamW(policy_model.parameters(), lr=lr)
53

54
    ref_model.eval()  # 冻结参考模型
55

56
    global_step = 0
57
    for epoch in range(num_epochs):
58
        for step, batch in enumerate(train_loader):
59

60
            # === 1. Policy forward pass ===
61
            policy_chosen_lps = compute_sequence_log_prob(
62
                policy_model,
63
                batch["chosen_input_ids"],
64
                batch["chosen_attention_mask"],
65
                batch["chosen_response_mask"],
66
            )
67
            policy_rejected_lps = compute_sequence_log_prob(
68
                policy_model,
69
                batch["rejected_input_ids"],
70
                batch["rejected_attention_mask"],
71
                batch["rejected_response_mask"],
72
            )
73

74
            # === 2. Reference model forward pass (no grad) ===
75
            with torch.no_grad():
76
                ref_chosen_lps = compute_sequence_log_prob(
77
                    ref_model,
78
                    batch["chosen_input_ids"],
79
                    batch["chosen_attention_mask"],
80
                    batch["chosen_response_mask"],
81
                )
82
                ref_rejected_lps = compute_sequence_log_prob(
83
                    ref_model,
84
                    batch["rejected_input_ids"],
85
                    batch["rejected_attention_mask"],
86
                    batch["rejected_response_mask"],
87
                )
88

89
            # === 3. DPO 损失 ===
90
            loss, metrics = dpo_loss(
91
                policy_chosen_lps, policy_rejected_lps,
92
                ref_chosen_lps, ref_rejected_lps,
93
                beta=beta,
94
            )
95

96
            # === 4. 反向传播 ===
97
            optimizer.zero_grad()
98
            loss.backward()
99
            torch.nn.utils.clip_grad_norm_(policy_model.parameters(), max_grad_norm)
100
            optimizer.step()
101

102
            global_step += 1
103

104
            if step % log_interval == 0:
105
                print(
106
                    f"[Epoch {epoch+1}] Step {step:4d} | "
107
                    f"loss={metrics['loss']:.4f} | "
108
                    f"acc={metrics['accuracy']:.3f} | "
109
                    f"margin={metrics['reward_margin']:.3f} | "
110
                    f"chosen_r={metrics['chosen_reward_mean']:.2f} | "
111
                    f"rejected_r={metrics['rejected_reward_mean']:.2f}"
112
                )

1
训练初期（margin ≈ 0）：
2
  模型对 y_w 和 y_l 的判断模糊
3
  Δ_θ ≈ 0 → σ(βΔ_θ) ≈ 0.5 → 梯度最大
4

5
训练中期：
6
  模型逐渐学会区分偏好
7
  margin 上升，accuracy 上升
8

9
训练后期：
10
  margin 趋近饱和，梯度变小 → 自动早停效果
11

12
理想曲线：
13
  margin:    ────────────────
14
            /
15
  accuracy: ────────────
16
  loss:     ────────────

1
┌────────────────────────────────┬──────────────────────────────────────────┐
2
│ 现象                            │ 原因 / 缓解                              │
3
├────────────────────────────────┼──────────────────────────────────────────┤
4
│ ① Reward margin 不上升          │ 偏好数据太简单（模型已能区分）            │
5
│   (accuracy 卡在 ~50%)          │  → 过滤掉 margin > 0.5 的样本            │
6
│                                │  → 增加对抗性/困难样本                   │
7
├────────────────────────────────┼──────────────────────────────────────────┤
8
│ ② Verbosity bias               │ RM 偏好长回答（标注员累了就倾向长回答）    │
9
│   (越长越好)                    │  → 归一化 response 长度                  │
10
│                                │  → 在偏好数据中控制长度分布               │
11
├────────────────────────────────┼──────────────────────────────────────────┤
12
│ ③ 过拟合到偏好数据              │ 学习率太大 / epoch 太多                   │
13
│   (生成能力退化)                │  → 用 β 增大 KL 约束                      │
14
│                                │  → 监控 perplexity 是否上升               │
15
├────────────────────────────────┼──────────────────────────────────────────┤
16
│ ④ Sigmoid 饱和                  │ Δ_θ 绝对值太大时梯度消失                  │
17
│   (后期 loss 不下降)             │  → label smoothing: log σ(βΔ - α)        │
18
│                                │  → 使用 IPO 损失（无饱和问题）             │
19
├────────────────────────────────┼──────────────────────────────────────────┤
20
│ ⑤ 偏好过拟合（OOD 泛化差）      │ 偏好数据分布太窄                         │
21
│                                │  → 数据增强（同 prompt 替换 answer）       │
22
│                                │  → 使用 RLAIF 增加多样性                 │
23
├────────────────────────────────┼──────────────────────────────────────────┤
24
│ ⑥ 参考模型偏移                  │ 多次 DPO 迭代后，π_ref 已过时            │
25
│                                │  → 定期更新 ref_model = policy_model     │
26
│                                │  → 使用 iterative DPO / SPIN              │
27
└────────────────────────────────┴──────────────────────────────────────────┘

1
def create_adversarial_samples(model, prompt, num_candidates=8, max_new_tokens=256):
2
    """
3
    用模型生成多个候选回答，保留最接近但不完全相同的配对
4
    """
5
    candidates = model.generate(
6
        [prompt] * num_candidates,
7
        max_new_tokens=max_new_tokens,
8
        do_sample=True,
9
        temperature=0.8,
10
        top_p=0.9,
11
    )
12

13
    # 用 RM 排序
14
    scores = reward_model([prompt] * num_candidates, candidates)
15
    sorted_idx = scores.argsort(descending=True)
16

17
    # 取排序后相邻的 pair（最接近的困难样本）
18
    pairs = []
19
    for i in range(len(sorted_idx) - 1):
20
        pairs.append((candidates[sorted_idx[i]], candidates[sorted_idx[i+1]]))
21

22
    return pairs

1
def simpo_loss(
2
    policy_chosen_lps: torch.Tensor,  # (B,) 每个序列的总 log prob
3
    policy_rejected_lps: torch.Tensor,
4
    chosen_len: torch.Tensor,          # (B,) chosen 序列长度
5
    rejected_len: torch.Tensor,
6
    gamma: float = 0.5,
7
):
8
    """SimPO 损失"""
9
    # 长度归一化
10
    chosen_avg = policy_chosen_lps / chosen_len
11
    rejected_avg = policy_rejected_lps / rejected_len
12

13
    logits = chosen_avg - rejected_avg - gamma
14
    loss = -F.logsigmoid(logits).mean()
15
    return loss, {"accuracy": (logits > 0).float().mean().item()}

1
┌──────────────────────────────────────────────────────────────┐
2
│                  DPO 训练最佳流水线                            │
3
├──────────────────────────────────────────────────────────────┤
4
│                                                              │
5
│  Step 1: 准备 SFT 模型（作为 policy 和 ref 的起点）          │
6
│     → 加载 base model                                        │
7
│     → SFT 训练（高质量指令数据）                              │
8
│     → 保存 π_SFT                                             │
9
│                                                              │
10
│  Step 2: 收集偏好数据                                        │
11
│     → π_SFT 生成候选回答（diverse sampling）                  │
12
│     → 人类/AI 标注偏好                                       │
13
│     → 质量过滤（长度、格式、RM 得分）                         │
14
│                                                              │
15
│  Step 3: DPO 训练                                            │
16
│     → policy = π_SFT，ref = π_SFT                           │
17
│     → β = 0.1（默认）                                        │
18
│     → lr = 1e-6 ~ 5e-6，epoch = 1~3                        │
19
│                                                              │
20
│  Step 4: 评估                                                │
21
│     → Reward margin、Accuracy                                 │
22
│     → Win Rate vs SFT / vs 基线                              │
23
│     → Perplexity（确保没崩）                                 │
24
│     → Safety rate                                            │
25
│                                                              │
26
└──────────────────────────────────────────────────────────────┘

1
dpo_config = {
2
    # 优化器
3
    "learning_rate": 1e-6,        # 比 SFT 通常更小
4
    "per_device_batch_size": 4,    # 受限于 2× 模型显存
5
    "gradient_accumulation_steps": 4,  # 总 effective batch = 16
6
    "num_epochs": 1,              # DPO 容易过拟合，1 epoch 常用
7
    "max_grad_norm": 1.0,
8

9
    # DPO 特有
10
    "beta": 0.1,                  # KL 系数，[0.05, 0.2]
11

12
    # 参考模型
13
    "ref_model": {
14
        "type": "frozen",         # frozen | ema | periodic_update
15
        "update_interval": 1,      # 每 N 个 step 更新一次（periodic 时）
16
        "ema_decay": 0.99,         # EMA 衰减率
17
    },
18

19
    # 数据
20
    "max_prompt_len": 512,
21
    "max_response_len": 512,
22
    "filter_responses_by_length": True,
23
    "min_length_diff": 10,        # chosen/rejected 长度差至少 10 tokens
24
}

1
def evaluate_dpo_model(model, ref_model, eval_pairs, rm_model=None):
2
    """
3
    DPO 模型评估：多维度
4
    """
5
    results = {}
6

7
    # === 1. DPO 内在指标 ===
8
    margins = []
9
    accuracies = []
10
    for batch in eval_pairs:
11
        # ... (同训练时的损失计算)
12
        margins.append(reward_margin)
13
        accuracies.append(accuracy)
14
    results["margin_mean"] = np.mean(margins)
15
    results["accuracy_mean"] = np.mean(accuracies)
16

17
    # === 2. Perplexity（语言模型质量） ===
18
    ppl = compute_perplexity(model, eval_texts)
19
    results["perplexity"] = ppl
20

21
    # === 3. Win Rate ===
22
    if rm_model is not None:
23
        win_rates = []
24
        for prompt in eval_prompts:
25
            sft_response = sft_model.generate(prompt)
26
            dpo_response = model.generate(prompt)
27
            score_sft = rm_model(prompt, sft_response)
28
            score_dpo = rm_model(prompt, dpo_response)
29
            win_rates.append(score_dpo > score_sft)
30
        results["win_rate_vs_sft"] = np.mean(win_rates)
31

32
    # === 4. Safety ===
33
    safety_score = evaluate_safety(model, adversarial_prompts)
34
    results["safety_rate"] = safety_score
35

36
    return results