大模型手撕¶

给定一对偏好数据 (chosen 和 rejected)，以及策略模型和参考模型的对数概率，用 Python 实现 DPO 损失函数的计算（基于公式）。¶

实现思路

给定一个 batch 的数据，我们已经计算好了：

chosen_logps：策略模型 πθ 对 chosen 回答的对数概率（形状 [B]）。
rejected_logps：策略模型对 rejected 回答的对数概率。
ref_chosen_logps：参考模型对 chosen 回答的对数概率。
ref_rejected_logps：参考模型对 rejected 回答的对数概率。

则 DPO 损失可写为：

def dpo_loss(chosen_logps, rejected_logps, ref_chosen_logps, ref_rejected_logps, beta):
    # 计算隐含奖励差
    chosen_rewards = beta * (chosen_logps - ref_chosen_logps)      # [B]
    rejected_rewards = beta * (rejected_logps - ref_rejected_logps) # [B]
    reward_diff = chosen_rewards - rejected_rewards                 # [B]
    # 损失：- log(sigmoid(reward_diff))
    loss = -torch.nn.functional.logsigmoid(reward_diff).mean()
    return loss

更完整的代码（含注释）：

import torch
import torch.nn.functional as F

def compute_dpo_loss(
    policy_chosen_logps: torch.Tensor,
    policy_rejected_logps: torch.Tensor,
    reference_chosen_logps: torch.Tensor,
    reference_rejected_logps: torch.Tensor,
    beta: float
) -> torch.Tensor:
    """
    计算 DPO 损失。

    Args:
        policy_chosen_logps: (B,) 策略模型在 chosen 回答上的 log probs。
        policy_rejected_logps: (B,) 策略模型在 rejected 回答上的 log probs。
        reference_chosen_logps: (B,) 参考模型在 chosen 回答上的 log probs（应 detach）。
        reference_rejected_logps: (B,) 参考模型在 rejected 回答上的 log probs（应 detach）。
        beta: 温度系数。

    Returns:
        loss: 标量损失值。
    """
    # 计算策略模型相对于参考模型的 log-ratio
    chosen_log_ratios = policy_chosen_logps - reference_chosen_logps
    rejected_log_ratios = policy_rejected_logps - reference_rejected_logps

    # 隐含奖励：r = beta * log_ratio
    chosen_rewards = beta * chosen_log_ratios
    rejected_rewards = beta * rejected_log_ratios

    # 奖励差
    reward_diff = chosen_rewards - rejected_rewards

    # 损失：- log(sigmoid(reward_diff))  =  -F.logsigmoid(reward_diff)
    loss = -F.logsigmoid(reward_diff).mean()
    return loss

写出 DPO 损失的完整数学公式，并解释公式中每个符号的含义（π_θ, π_ref, x, y_w, y_l, β）。¶

DPO 损失的完整公式

DPO 的目标是直接在人类偏好数据上优化策略模型，使得模型对“优选”回答（chosen）的生成概率相对于“拒绝”回答（rejected）更高，同时尽量不偏离参考模型太远。其损失函数定义为：

或者更简洁地写成：

符号解释

推导 DPO 损失函数的梯度，写出对策略模型参数的梯度表达式。¶

推导过程

定义隐含奖励差：

DPO 损失对单个样本为：

我们需要计算 ∇θL。

首先，利用链式法则：

实际上更方便的是直接使用 logsigmoid 的梯度：

或者用另一种常见形式：

解释 DPO 损失中隐式奖励 (implicit reward) 的定义，并用代码实现一个函数，给定策略和参考模型的 log probabilities，计算 chosen 和 rejected 的隐式奖励。¶

定义

在 DPO 中，隐含奖励（implicit reward）定义为：

代码实现

def compute_implicit_rewards(
    policy_logps: torch.Tensor,
    reference_logps: torch.Tensor,
    beta: float
) -> torch.Tensor:
    """
    计算隐含奖励 r = beta * (log π_θ - log π_ref)
    Args:
        policy_logps: (B,) 或 (B, L) 策略模型的 log probs
        reference_logps: 对应参考模型的 log probs
        beta: 温度系数
    Returns:
        rewards: 同形状的奖励张量
    """
    return beta * (policy_logps - reference_logps)

在 DPO 训练循环中，对 chosen 和 rejected 分别调用此函数，得到 rw 和 rl，然后损失函数优化 rw−rlrw−rl。

为什么 DPO 损失可以等效于在 Bradley-Terry 模型下的奖励最大化问题？写出等效性推导过程。¶

Bradley-Terry 模型

Bradley-Terry 模型用于描述成对比较的概率：对于两个项目 yw 和 yl，yw 被选中的概率为：

这个带 KL 约束的奖励最大化问题有一个闭式解（详见 policy gradient 相关推导）：

其中 Z(x) 是配分函数。这个方程可以反解出奖励：

DPO 的等价推导

DPO 的核心洞察是：在 Bradley-Terry 偏好模型中，只需要奖励的相对差 r(x,yw)−r(x,yl)，而配分函数项‘

β logZ(x) 在相减时会被消去。因此，如果我们直接用策略模型定义一个“隐式奖励”：

这正是 DPO 损失中的 σ(Δ)。因此，最大化 DPO 损失（即最大化 logP (yw≻yl)）等价于在 Bradley-Terry 模型下最大化偏好概率，而无需显式训练奖励模型，也无需处理难以计算的配分函数 Z(x)。这正是 DPO 简洁而高效的根本原因。

在 PyTorch 中，实现一个 DPO 损失函数模块，输入 chosen_logps、rejected_logps、ref_chosen_logps、ref_rejected_logps 和 β，返回标量损失值。¶

实现我们实现一个 torch.nn.Module 子类，方便集成到训练流程中。

import torch
import torch.nn as nn
import torch.nn.functional as F

class DPOLoss(nn.Module):
    def __init__(self, beta: float = 0.1):
        super().__init__()
        self.beta = beta

    def forward(
        self,
        policy_chosen_logps: torch.Tensor,
        policy_rejected_logps: torch.Tensor,
        reference_chosen_logps: torch.Tensor,
        reference_rejected_logps: torch.Tensor,
    ) -> torch.Tensor:
        """
        Args:
            policy_chosen_logps: (B,) 策略模型对 chosen 的 log prob
            policy_rejected_logps: (B,) 策略模型对 rejected 的 log prob
            reference_chosen_logps: (B,) 参考模型对 chosen 的 log prob (detached)
            reference_rejected_logps: (B,) 参考模型对 rejected 的 log prob (detached)
        Returns:
            scalar loss
        """
        # 计算 log-ratio
        chosen_log_ratios = policy_chosen_logps - reference_chosen_logps
        rejected_log_ratios = policy_rejected_logps - reference_rejected_logps

        # 奖励差
        reward_diff = self.beta * (chosen_log_ratios - rejected_log_ratios)

        # 损失
        loss = -F.logsigmoid(reward_diff).mean()
        return loss

DPO 损失中 β 参数的作用是什么？如果 β 增大或减小，对模型训练有什么影响？并用公式说明。¶

β 的作用

β 是一个正的温度系数，控制两件事：

β 直接缩放了 chosen 和 rejected 的 log-ratio 之差，进而影响 sigmoid 输入的幅度。

β 增大：

奖励差 Δ 放大 → sigmoid 输出更接近 0 或 1 → 损失对偏好数据的拟合更“硬”，模型更快地提高 chosen 的概率、降低 rejected 的概率。
允许更大的策略偏移，可能更好地对齐偏好，但也可能过度拟合偏好数据，导致性能退化（如生成多样性下降、过拟合标注噪声）。
训练可能更不稳定。

β 减小：

奖励差缩小 → sigmoid 输出更接近 0.5 → 损失梯度较小，模型更新更保守。
策略模型更贴近参考模型，保持更好的基础能力，但可能无法充分拟合偏好数据，对齐效果不足。
训练更稳定，但收敛较慢。

公式说明

分析 DPO 损失中 log-sigmoid 项的计算，如何在数值上保持稳定？写出健壮的实现。¶

数值稳定问题

当 Δ（即 reward_diff）非常大时，σ(Δ)≈1，log_sigmoid(Δ) 接近 0，计算 log(1)=0 没有问题。当 Δ 非常小时（强负值），σ(Δ)≈0，直接计算 log(sigmoid(Δ)) 可能会因为 sigmoid(Δ) 下溢到 0 而导致 log(0) = -inf，出现数值问题。

健壮实现

PyTorch 提供了 F.logsigmoid，它在内部已经实现了数值稳定的计算：利用恒等式 logσ(x)=−softplus(−x)，或者分段处理。因此直接使用 F.logsigmoid(reward_diff) 是安全且推荐的。

如果我们手动实现一个健壮版本（例如在不支持 logsigmoid 的环境），可以如下：

def stable_logsigmoid(x: torch.Tensor) -> torch.Tensor:
    """数值稳定的 log-sigmoid"""
    # 对于 x >= 0: log(sigmoid(x)) = -log(1 + exp(-x)) = -softplus(-x)
    # 对于 x < 0:  log(sigmoid(x)) = x - log(1 + exp(x))  = x - softplus(x)
    # 或者使用 logsigmoid 的内置实现
    return torch.where(
        x >= 0,
        -F.softplus(-x),
        x - F.softplus(x)
    )

在标准 DPO 损失中，直接使用 -F.logsigmoid(reward_diff).mean() 即可，因为 F.logsigmoid 已处理数值稳定性。需要注意的是，reward_diff 是 chosen 奖励减 rejected 奖励，理论上可以很大或很小，PyTorch 的实现能够覆盖。

在 DPO 训练过程中，通常需要使用 reference model 的 log probabilities，这些值需要 detach 吗？为什么？在代码中如何实现？¶

是否需要 detach？必须 detach。

更具体地说，在损失函数中，ref_chosen_logps 和 ref_rejected_logps 应当被视为常数（尽管它们是从参考模型计算得来的）。因此，必须在将这些值传入 DPO 损失函数之前，调用 .detach() 来切断与参考模型计算图的连接。

代码实现

# 假设 reference_model 和 policy_model 是两个相同的结构，但 reference_model 参数冻结
with torch.no_grad():
    ref_chosen_logps = reference_model(chosen_inputs).logprobs  # 已经是 no_grad，不需要额外 detach
    # 或者如果在 torch.enable_grad() 上下文中计算，则需要：
    # ref_chosen_logps = reference_model(chosen_inputs).logprobs.detach()

# 但在实际中，通常直接让 reference_model 在 eval 模式且不计算梯度：
reference_model.eval()
for param in reference_model.parameters():
    param.requires_grad = False

# 计算参考模型的 log probs 时，不需要 with torch.no_grad() 也行，因为参数 requires_grad=False，
# 但为了明确和性能，最好用 torch.no_grad()。
with torch.no_grad():
    ref_chosen_logps = reference_model(chosen_inputs).logprobs
    ref_rejected_logps = reference_model(rejected_inputs).logprobs

当传递给 DPO 损失函数时，这些张量的 requires_grad 属性已经是 False，因此不会对参考模型产生梯度。

在 DPO 损失的内部，尽管 ref_chosen_logps 参与了计算（如 chosen_log_ratios = policy_chosen_logps - reference_chosen_logps），但由于 reference_chosen_logps 没有梯度，梯度只会通过 policy_chosen_logps 反向传播。这是完全正确的。

总结：一定要保证参考模型的输出在送入损失函数前是 detached 的，最简单的方式是使用 torch.no_grad() 上下文或显式调用 .detach()。同时冻结参考模型的所有参数。

如果偏好数据中只有正向样本（chosen）而没有负向样本（rejected），DPO 损失还能用吗？如果不能，如何修改损失或数据构建方式来适应此情况？¶

直接使用原始 DPO 损失是不可行的。 DPO 损失的核心是对比 chosen 和 rejected 的隐含奖励。如果数据中只包含 chosen 回答（例如，只收集了“点赞”或“优质”回答，而没有“点踩”或对比对），则无法构造 reward margin rw−rl，损失函数将无法计算。

解决方案：

可以通过以下两种途径将单向数据转化为成对偏好数据，从而适配 DPO：

方案一：构造“拒绝”样本（Synthetic Rejected）

利用参考模型（或策略模型自身）为每个 prompt 生成多个回答，然后根据某种准则（如人工评分、规则）选出较好的作为 chosen，较差的作为 rejected。常见方法：

对同一 prompt，用初始 SFT 模型采样多条回答。
使用奖励模型或人工标注对这些回答打分。
选择得分最高的作为 chosen，得分最低的（或随机选一个低分）作为 rejected。
用这些 (prompt, chosen, rejected) 三元组训练 DPO。这种方式被称为 "Rejection Sampling" 或 "Synthetic Preference"，已被广泛使用（例如 UltraFeedback 数据集）。

方案二：设计仅使用正向样本的损失变体如果无法获得 rejected 样本，可以尝试将 DPO 损失修改为仅优化正向概率，同时加入正则项防止模型偏离参考模型太远。例如，最大化 log π_θ(y_chosen|x) - β KL(π_θ || π_ref)，但这实际上等同于监督微调（SFT）加上 KL 惩罚，已经退化为 SFT，失去了对比学习的优势。

另一种思路是构造一个虚拟的 rejected 样本：将参考模型在该 prompt 下的生成结果视为“rejected”，而将人工标注的优质回答作为 chosen。但这要求参考模型生成的回答质量较低且与 chosen 存在明显差距，否则训练信号会很弱。

方案三：迭代 DPO（Iterative DPO）

先用现有的 chosen 数据对模型进行 SFT，得到一个初始策略。
用该策略为训练集中的每个 prompt 生成多个回答。
用外部的奖励模型（或人工）对这些回答评分，选出 chosen 和 rejected，构造偏好对。
用这些偏好对训练 DPO，得到新的策略。
重复步骤 2-4 多次。这种方法能够自我生成高质量的偏好数据，已在实践中被证明非常有效。

总结：原始 DPO 必须依赖成对的偏好数据。当只有正向样本时，需要通过生成对比样本（如采样+评分）来构造 rejected 样本，从而形成三元组。

实现一个 batch 级别的 DPO 损失计算函数，要求处理 padding：输入 batch 中包含 chosen 和 rejected 的序列，以及对应的 attention mask，正确计算每个序列的平均 log probability（或 sum）。¶

背景：

在实际训练中，chosen 和 rejected 序列的长度通常不同。我们需要正确计算每个 token 的对数概率，并通过 attention mask 来忽略 padding token。DPO 损失中使用的是序列的总对数概率（或平均对数概率）。常见的做法是取平均 token 对数概率（即除以有效 token 数），这样可以使不同长度的序列在损失中的贡献更加均衡。

实现：

import torch
import torch.nn.functional as F

def compute_log_probs(logits, labels, attention_mask):
    """
    计算序列的 token 级对数概率，并根据 attention mask 取平均。
    Args:
        logits: (B, L, V) 模型输出的 logits
        labels: (B, L) 目标 token id
        attention_mask: (B, L) 1 表示有效 token，0 表示 padding
    Returns:
        mean_log_probs: (B,) 每个序列的平均 token 对数概率
    """
    # 计算每个 token 的交叉熵损失（reduction='none'）
    loss_per_token = F.cross_entropy(
        logits.permute(0, 2, 1),  # (B, V, L)
        labels,
        reduction='none'
    )  # (B, L)
    # 对有效 token 取平均
    loss_per_token = loss_per_token * attention_mask
    mean_log_probs = -loss_per_token.sum(dim=1) / attention_mask.sum(dim=1)
    return mean_log_probs

def batch_dpo_loss(
    policy_chosen_logps: torch.Tensor,
    policy_rejected_logps: torch.Tensor,
    ref_chosen_logps: torch.Tensor,
    ref_rejected_logps: torch.Tensor,
    beta: float
) -> torch.Tensor:
    """
    计算 batch 级别的 DPO 损失。
    假设输入已经是每个序列的平均 token 对数概率（已经过 padding 处理）。
    """
    chosen_ratio = policy_chosen_logps - ref_chosen_logps
    rejected_ratio = policy_rejected_logps - ref_rejected_logps
    reward_diff = beta * (chosen_ratio - rejected_ratio)
    loss = -F.logsigmoid(reward_diff).mean()
    return loss

# 完整的训练步骤示例：
# 1. 将 chosen 和 rejected 序列分别输入策略模型和参考模型，获取 logits。
# 2. 使用 compute_log_probs 得到每个序列的平均 log prob。
# 3. 调用 batch_dpo_loss 计算损失。

关于 sum 和 mean 的选择：

Sum：序列越长，其总对数概率绝对值越大，在损失中占的权重也越大。这可能导致模型偏向于优化长序列，但长序列天然包含更多 token，其 log prob 值域更大，可能主导梯度。
Mean：将每个序列的对数概率归一化到 token 级别，使不同长度的序列在损失中拥有相近的尺度。这是目前 DPO 的主流实现（例如 HuggingFace TRL 库中默认使用平均 log probability）。建议使用 mean，并在计算时通过 attention mask 准确统计有效 token 数。

DPO 训练时，除了最终的损失值，一般还需要监控哪些指标（如 chosen 和 rejected 的准确率、reward margin 等）？写出计算这些指标的代码。¶

需要监控的核心指标：

Reward Accuracy（奖励准确率）：chosen 的隐含奖励大于 rejected 的比例。直接反映模型对偏好数据的区分能力。
Reward Margin：chosen 奖励与 rejected 奖励之差，衡量模型的区分度。
Chosen 和 Rejected 的平均 Log Probability：监控策略模型在 chosen 和 rejected 上的对数概率变化。
KL Divergence：策略模型与参考模型之间的 KL 散度（可近似为 log_ratio 的期望），确保模型不过度偏离。
Loss 本身：DPO 损失的均值。

代码实现：

def compute_metrics(
    policy_chosen_logps, policy_rejected_logps,
    ref_chosen_logps, ref_rejected_logps, beta
):
    with torch.no_grad():
        chosen_rewards = beta * (policy_chosen_logps - ref_chosen_logps)
        rejected_rewards = beta * (policy_rejected_logps - ref_rejected_logps)
        reward_margin = chosen_rewards - rejected_rewards
        accuracy = (reward_margin > 0).float().mean().item()
        avg_margin = reward_margin.mean().item()
        avg_chosen_logp = policy_chosen_logps.mean().item()
        avg_rejected_logp = policy_rejected_logps.mean().item()
        # 近似 KL 散度：chosen 和 rejected 上的平均 log-ratio
        kl_approx = 0.5 * ((policy_chosen_logps - ref_chosen_logps).mean() +
                           (policy_rejected_logps - ref_rejected_logps).mean()).item()
    return {
        "loss": None,  # 在训练循环中单独计算
        "reward_accuracy": accuracy,
        "reward_margin": avg_margin,
        "chosen_logp": avg_chosen_logp,
        "rejected_logp": avg_rejected_logp,
        "kl_approx": kl_approx
    }

给定 DPO 损失对策略模型 logits 的梯度，推导出该梯度如何影响模型对 chosen 和 rejected 标记的预测概率调整方向。¶

从第3题的梯度公式出发：

自适应权重： σ(−Δ) 起到了自适应调节作用：

当模型已经很好地偏好 chosen 时（Δ 很大，σ(−Δ)很小），梯度幅度减小，模型不再大幅调整。
当模型还不能区分或偏好 rejected 时（Δ很小甚至为负），σ(−Δ) 较大（接近 0.5 或更高），梯度较大，模型会迅速提升 chosen 的概率并压低 rejected 的概率。

token 级别的影响：

比较 DPO 与 RLHF（PPO）在损失函数设计上的关键区别，用公式与伪代码阐述。¶

RLHF（PPO）流程：

PPO 需要不断从策略模型采样，计算奖励，并估计优势函数，过程复杂且不稳定。

DPO 流程：

直接利用偏好数据优化策略模型，无需显式奖励模型，也无需在线采样。损失函数为：

关键区别：

查看内嵌表格

伪代码对比：

RLHF PPO 训练步：

for batch in preference_data:
    # 1. 采样 (x, y_w, y_l) 训练奖励模型
    loss_rm = -log(sigmoid(r(x,y_w) - r(x,y_l)))
    # 2. 用 PPO 微调策略
    for ppo_step in range(ppo_epochs):
        y_sample = sample(policy, x)
        r = reward_model(x, y_sample)
        kl = log(policy(y_sample)) - log(ref(y_sample))
        advantage = r - beta * kl
        loss_ppo = -advantage
        optimizer.step()

DPO 训练步：

for batch in preference_data:
    # 一次前向同时得到 chosen 和 rejected 的 log prob
    loss_dpo = -log_sigmoid(beta * (log_pi(y_w) - log_ref(y_w) - log_pi(y_l) + log_ref(y_l)))
    optimizer.step()

可以看出，DPO 极为简洁。

如果 reference model 和 policy model 是同一个模型（比如训练初期），DPO 损失会发生什么？如何避免潜在的数值问题？¶

初始状态：

潜在的数值问题：

尽管初始时 log-ratio 为 0，数值上没有问题。但在训练过程中，如果参考模型和策略模型完全相同，则 log-ratio 始终为 0，损失恒定，无法学习。但这通常不会发生，因为策略模型一旦更新，就与参考模型不同了。

需要注意的真正数值风险：

如果 reference model 和 policy model 完全相同且我们使用了 detach 不当，或者参考模型的输出参与了梯度计算，则会导致参考模型被更新，从而破坏 DPO 的理论基础。所以必须确保参考模型的输出被 detach。

写一个函数，根据 chosen 和 rejected 的隐式奖励，计算 DPO 的奖励准确率（即 chosen 奖励大于 rejected 奖励的比例）。¶

def compute_reward_accuracy(chosen_rewards, rejected_rewards):
    """
    chosen_rewards: (B,) tensor
    rejected_rewards: (B,) tensor
    Returns: accuracy float
    """
    correct = (chosen_rewards > rejected_rewards).float().sum()
    total = chosen_rewards.size(0)
    return (correct / total).item()

可直接嵌入训练循环的监控代码中。

DPO 损失在实现时，通常对 chosen 和 rejected 的 log-prob 进行长度平均还是求和？分析长度平均对训练的影响，并写出两种实现方式的代码片段。¶

两种方式：

求和：total_log_prob = sum(token_log_probs)，不除以长度。
平均：mean_log_prob = total_log_prob / num_valid_tokens。

对训练的影响：

求和：长序列的总对数概率数值更大（更负），在损失中的贡献权重也更大。这会导致模型更关注长序列的优化，可能忽略短序列的偏好信号。此外，长序列的梯度方差可能更大。
平均：将每个序列的对数概率归一化到 token 级别，使得不同长度的序列在损失函数中具有可比性。这更公平，因为偏好通常不是由长度决定的。目前主流 DPO 实现（如 HuggingFace TRL）采用平均 log probability。

代码片段：

求和方式：

total_logp = (log_probs * attention_mask).sum(dim=1)  # (B,)

平均方式：

num_tokens = attention_mask.sum(dim=1)
mean_logp = (log_probs * attention_mask).sum(dim=1) / num_tokens

建议：使用平均方式，并确保在计算时排除 padding token 的影响。

如何将 DPO 损失扩展为支持多对偏好数据（如 n 个排列）？写出 k 个排序样本的 Listwise DPO 损失函数形式，并实现（假设输入为排序分数列表）。¶

背景：

原始 DPO 处理两两偏好。当有多个回答的排序（如 A > B > C > D）时，可以将其分解为多个两两比较对（A>B, A>C, A>D, B>C, B>D, C>D），并累加损失。更常见的 Listwise 方法是基于 Plackett-Luce 模型或直接扩展 DPO：

Listwise DPO 损失（基于所有排序对的 DPO 损失之和）：假设对于同一个 prompt x，我们有 K 个回答，并已知其偏好排序 y1≻y2≻⋯≻yK。则损失可以写为所有高阶对低阶的 DPO 损失之和：

或者可以采用更高效的 Listwise 损失，例如对排序列表应用 Plackett-Luce 模型，但上述成对分解简单且容易实现。

实现：

def listwise_dpo_loss(sorted_policy_logps, sorted_ref_logps, beta):
    """
    sorted_policy_logps: (B, K) 已按偏好降序排列（第一个是最优）
    sorted_ref_logps: (B, K)
    """
    B, K = sorted_policy_logps.shape
    loss = 0.0
    count = 0
    for i in range(K-1):
        for j in range(i+1, K):
            chosen_logp = sorted_policy_logps[:, i]
            rejected_logp = sorted_policy_logps[:, j]
            ref_chosen = sorted_ref_logps[:, i]
            ref_rejected = sorted_ref_logps[:, j]
            chosen_ratio = chosen_logp - ref_chosen
            rejected_ratio = rejected_logp - ref_rejected
            reward_diff = beta * (chosen_ratio - rejected_ratio)
            loss += -F.logsigmoid(reward_diff).sum()
            count += 1
    return loss / (B * count)  # 平均到每个样本和每个对

解释为什么 DPO 损失中的 log-sigmoid 函数相当于对 reward difference 施加一个软间隔，并与 hinge loss 对比。¶

DPO 损失本质： DPO 最小化 −log⁡σ(βΔ)，其中 Δ=rw−rl是隐含奖励差。这个损失函数可以视为对 Δ 的一个软间隔损失（soft margin loss）。

对比 hinge loss： Hinge loss 常用于 SVM，定义为 max⁡(0,1−y⋅f(x))。对于偏好学习，可以定义 hinge loss 为 max⁡(0,margin−Δ)，即希望 Δ 大于某个间隔 margin。但 hinge loss 在间隔达到后梯度为零，且不可导。

DPO 损失的软间隔特性：

当 Δ 很大（正值），σ(Δ)≈1，损失接近 0。模型已经很好地偏好 chosen。
当 Δ 很小甚至负值，σ(Δ) 接近 0，损失 −log⁡σ(Δ) 很大，产生强烈的梯度推动模型增大 Δ。
与 hinge loss 不同，DPO 损失处处可导，且对 Δ 的惩罚是平滑的：即使 Δ 已经为正值，损失仍会随着 Δ 增大而缓慢降低，这使得模型可以继续拉开 chosen 和 rejected 的差距，但梯度逐渐减小，形成“软间隔”。

数学形式对比：

DPO 的软间隔特性带来了更好的优化性质，避免 hinge loss 在间隔边界处的梯度不连续问题，同时通过 β 控制间隔的“硬度”：β 越大，曲线越陡峭，越接近硬间隔。

写出 DPO 训练一步的完整伪代码：数据加载、前向计算 chosen/rejected logps、ref logps、计算损失、反向传播、参数更新。¶

整体流程

DPO 训练的一个迭代步包含以下关键阶段：

从偏好数据集中采样一个 batch，包含提示（prompt）、chosen 回答和 rejected 回答。
分别用策略模型（policy model）和参考模型（reference model）对 chosen 和 rejected 序列进行前向传播，获取每个 token 的对数概率。
利用 attention mask 对对数概率进行有效长度归一化，得到每个序列的平均对数概率。
根据 DPO 损失公式计算损失值。
对策略模型参数执行反向传播和优化器更新。参考模型的参数始终保持冻结。

伪代码

Algorithm: DPO Training Step
Input: policy_model, reference_model, optimizer, beta, batch
Output: updated policy_model

1. 解包 batch 得到:
   - prompt_ids, chosen_ids, rejected_ids
   - chosen_mask, rejected_mask  (attention mask)

2. 将 chosen_ids 和 rejected_ids 分别送入 policy_model，计算 logits。
   使用 logits 和标签计算每个 token 的交叉熵损失（reduction='none'），
   再结合 attention mask 得到 chosen 和 rejected 的平均对数概率：
   policy_chosen_logps = average_log_prob(logits_chosen, chosen_ids, chosen_mask)
   policy_rejected_logps = average_log_prob(logits_rejected, rejected_ids, rejected_mask)

3. 在 torch.no_grad() 上下文中，将相同序列送入 reference_model，
   以相同方式计算 ref_chosen_logps 和 ref_rejected_logps。
   注意：这些张量必须 detach 或直接在 no_grad 下得到，保证不参与梯度计算。

4. 计算 DPO 损失:
   chosen_ratio = policy_chosen_logps - ref_chosen_logps
   rejected_ratio = policy_rejected_logps - ref_rejected_logps
   reward_diff = beta * (chosen_ratio - rejected_ratio)
   loss = -log_sigmoid(reward_diff).mean()

5. 反向传播：
   optimizer.zero_grad()
   loss.backward()
   optimizer.step()

6. 返回 loss 值用于日志。

完整 Python 实现（关键部分）

import torch
import torch.nn.functional as F

def average_log_prob(logits, labels, mask):
    """计算每个序列的平均 token 对数概率，mask 忽略 padding"""
    # logits: (B, L, V)
    # labels: (B, L)
    # mask: (B, L)
    loss_per_token = F.cross_entropy(
        logits.permute(0, 2, 1),  # (B, V, L)
        labels,
        reduction='none'
    )
    loss_per_token = loss_per_token * mask
    total_loss = loss_per_token.sum(dim=1)      # (B,)
    num_tokens = mask.sum(dim=1)                # (B,)
    mean_log_prob = -total_loss / num_tokens    # 平均对数概率 = -平均交叉熵
    return mean_log_prob

def dpo_training_step(policy_model, ref_model, batch, optimizer, beta):
    """
    执行一个 DPO 训练步。
    batch: {
        'chosen_input_ids': (B, Lc),
        'chosen_attention_mask': (B, Lc),
        'rejected_input_ids': (B, Lr),
        'rejected_attention_mask': (B, Lr),
        'chosen_labels': (B, Lc),       # 通常与 input_ids 相同（自回归）
        'rejected_labels': (B, Lr)
    }
    """
    # 策略模型前向 chosen
    policy_chosen_logits = policy_model(
        input_ids=batch['chosen_input_ids'],
        attention_mask=batch['chosen_attention_mask']
    ).logits
    policy_chosen_logps = average_log_prob(
        policy_chosen_logits, batch['chosen_labels'], batch['chosen_attention_mask']
    )

    # 策略模型前向 rejected
    policy_rejected_logits = policy_model(
        input_ids=batch['rejected_input_ids'],
        attention_mask=batch['rejected_attention_mask']
    ).logits
    policy_rejected_logps = average_log_prob(
        policy_rejected_logits, batch['rejected_labels'], batch['rejected_attention_mask']
    )

    # 参考模型前向（无梯度）
    with torch.no_grad():
        ref_chosen_logits = ref_model(
            input_ids=batch['chosen_input_ids'],
            attention_mask=batch['chosen_attention_mask']
        ).logits
        ref_chosen_logps = average_log_prob(
            ref_chosen_logits, batch['chosen_labels'], batch['chosen_attention_mask']
        )

        ref_rejected_logits = ref_model(
            input_ids=batch['rejected_input_ids'],
            attention_mask=batch['rejected_attention_mask']
        ).logits
        ref_rejected_logps = average_log_prob(
            ref_rejected_logits, batch['rejected_labels'], batch['rejected_attention_mask']
        )

    # 计算 DPO 损失
    chosen_ratio = policy_chosen_logps - ref_chosen_logps
    rejected_ratio = policy_rejected_logps - ref_rejected_logps
    reward_diff = beta * (chosen_ratio - rejected_ratio)
    loss = -F.logsigmoid(reward_diff).mean()

    # 反向传播
    optimizer.zero_grad()
    loss.backward()
    optimizer.step()

    return loss.item()

要点说明

使用 average_log_prob 将序列的对数概率归一化到 token 级别，使不同长度的序列在损失中公平对待。
参考模型的 logits 必须在 torch.no_grad() 下计算，且后续使用的 ref_*_logps 已经不需要梯度，这天然实现了 detach。
在实际代码中，为节省显存，可对 chosen 和 rejected 序列合并为一个 batch 同时计算，但这里为清晰性分开。

在 DPO 损失基础上，增加一个 SFT 正则项（对 chosen 序列的交叉熵），修改损失函数并实现。¶

动机

纯粹的 DPO 损失只优化 chosen 和 rejected 之间的相对顺序，可能忽略 chosen 本身的绝对质量，导致模型虽然偏好 chosen，但 chosen 的概率本身可能下降（例如模型整体输出概率变得分散）。加入对 chosen 的监督微调（SFT）正则项可以鼓励模型保持或提升 chosen 的生成概率，稳定训练。这在一些研究（如 DPO+、RSO）中被证明有益。

修改后的损失函数

其中 LSFT 是 chosen 序列的标准交叉熵损失（即平均负对数概率）。λ 是正则化系数，控制 SFT 项的强度。通常 λ 取较小的值（如 0.01~0.1）。

实现

def dpo_with_sft_loss(
    policy_chosen_logits,
    policy_rejected_logits,
    ref_chosen_logps,
    ref_rejected_logps,
    chosen_labels,
    chosen_mask,
    beta,
    sft_lambda=0.1
):
    # 1. 计算 DPO 损失部分
    policy_chosen_logps = average_log_prob(policy_chosen_logits, chosen_labels, chosen_mask)
    policy_rejected_logps = average_log_prob(policy_rejected_logits, rejected_labels, rejected_mask)

    chosen_ratio = policy_chosen_logps - ref_chosen_logps
    rejected_ratio = policy_rejected_logps - ref_rejected_logps
    reward_diff = beta * (chosen_ratio - rejected_ratio)
    dpo_loss = -F.logsigmoid(reward_diff).mean()

    # 2. 计算 SFT 损失（chosen 序列的交叉熵，即负平均对数概率）
    # 注意：这里的 sft_loss 就是 -policy_chosen_logps.mean()，因为 average_log_prob 返回的是平均对数概率
    # 若要保持与 DPO 内部相同的计算方式，可以直接用 policy_chosen_logps 的负均值。
    sft_loss = -policy_chosen_logps.mean()

    # 3. 组合损失
    total_loss = dpo_loss + sft_lambda * sft_loss
    return total_loss

说明

在某些实现中，SFT 损失可能直接复用 policy_chosen_logps（已经过归一化），因为 policy_chosen_logps = average_log_prob(...) 已经是平均对数概率，取负即交叉熵。
λ 需要调整，过大会使训练退化为普通 SFT，DPO 对比作用减弱；过小则正则效果不明显。
SFT 项只加在 chosen 上，因为我们的目标是保持 chosen 的质量，不应该增强 rejected。
该组合损失可以在一个 backward 中直接优化，无需分步。

实践建议在训练早期，DPO 尚未稳定，可以适当增大 λ；后期可以逐渐减小甚至退化为纯 DPO。也可以使用动态的 λ 衰减策略。

DPO 中 reference model 的输出 log probabilities 在训练中保持不变，但实践中可能由于数值精度导致梯度泄露，编写代码显式地 detach 并验证无梯度流动。¶

问题分析

虽然我们在计算参考模型的 log probabilities 时使用了 torch.no_grad()，该上下文中的张量不会有梯度。但如果在损失计算时，这些张量与策略模型的参数有某种间接联系（例如，参考模型和策略模型共享了部分 buffer 或由于 in-place 操作），可能会导致梯度意外泄露。为了绝对安全，最佳实践是：

参考模型在 eval 模式下运行，且其所有参数的 requires_grad 设置为 False。
在获得参考模型的 logits 后，显式调用 .detach() 再计算 log probabilities。
编写单元测试验证参考模型参数在反向传播后是否保持不变。

代码实现（包含验证）

def compute_ref_logps_with_detach(ref_model, input_ids, attention_mask, labels):
    """安全地计算参考模型的平均对数概率，确保完全切断梯度"""
    ref_model.eval()
    # 确保参考模型参数不参与梯度
    for param in ref_model.parameters():
        param.requires_grad = False

    with torch.no_grad():
        logits = ref_model(input_ids=input_ids, attention_mask=attention_mask).logits
        # 显式 detach，虽然 no_grad 已足够，但这里是双保险
        logits = logits.detach()

    # 计算平均对数概率（这里需要梯度吗？不需要，但计算本身可能不需要detach）
    # average_log_prob 内部使用 cross_entropy，它不会产生对 logits 的梯度因为我们已经 detach
    mean_logp = average_log_prob(logits, labels, attention_mask)
    # 再次 detach，确保返回的张量没有梯度
    return mean_logp.detach()

def verify_no_gradient_leak(ref_model, policy_model):
    """验证 DPO 损失不会对参考模型产生梯度"""
    # 保存参考模型参数初始值
    initial_params = [p.clone() for p in ref_model.parameters()]

    # 执行一个模拟训练步（需要 batch 数据）
    # ... 计算 loss 并 backward()

    # 检查参考模型参数是否变化
    for i, param in enumerate(ref_model.parameters()):
        if not torch.equal(param, initial_params[i]):
            print(f"Reference model parameter {i} changed! Gradient leak detected.")
            return False
    print("No gradient leak. Reference model parameters unchanged.")
    return True

解释

torch.no_grad() 已经阻止了梯度计算，但为了绝对安全，我们对 logits 调用 .detach()，并在返回 logps 时再次 .detach()，这样即使在后续的计算中误操作，也不会意外地将梯度传播回参考模型。
单元测试验证是工程中的好习惯，可以放在 CI 流程中。

使用 DPO 损失训练时，如何动态调整 β 系数？比如根据 reward margin 自适应调整，写出一个简单的自适应 β 调整策略的实现。¶

动机

β 控制着 DPO 中偏好信号的强度。固定 β 可能导致训练初期 chosen 和 rejected 的 reward margin 很小，学习缓慢；后期 margin 已经很大，继续使用相同的 β 可能导致过拟合。自适应 β 可以根据当前模型的状态动态调整：当 margin 较小时增大 β 以加强信号；当 margin 过大时减小 β 以防止过度优化，同时可以保持与参考模型的合理 KL 距离。

简单的自适应策略

一种启发式策略：计算最近几个 batch 的平均 reward margin，并与一个目标 margin 比较，通过比例积分（PI）控制器调整 β。例如：

其中 γ 是调整步长。当实际 margin 低于目标时，β 增加；反之减小。同时需要 clamp β 在一个合理范围（如 0.01~1.0）。

实现

class AdaptiveBetaScheduler:
    def __init__(self, initial_beta=0.1, target_margin=1.0, adjust_rate=0.01,
                 min_beta=0.01, max_beta=1.0):
        self.beta = initial_beta
        self.target_margin = target_margin
        self.adjust_rate = adjust_rate
        self.min_beta = min_beta
        self.max_beta = max_beta
        self.margin_history = []  # 存储最近的 reward margin

    def update(self, current_margin: float):
        """根据当前 margin 更新 beta，并返回新 beta"""
        self.margin_history.append(current_margin)
        # 使用最近 N 个 margin 的均值，这里简单取最近一次
        avg_margin = current_margin
        # 指数调整
        self.beta *= torch.exp(torch.tensor(self.adjust_rate * (self.target_margin - avg_margin)))
        # clamp
        self.beta = max(self.min_beta, min(self.max_beta, self.beta))
        return self.beta

    def get_beta(self):
        return self.beta

训练循环中使用

scheduler = AdaptiveBetaScheduler(initial_beta=0.1, target_margin=1.0)

for step, batch in enumerate(dataloader):
    # 计算当前 reward margin（可以在计算 loss 时顺便获得，或单独计算）
    with torch.no_grad():
        # 假设我们已经得到了 chosen_rewards 和 rejected_rewards
        margin = (chosen_rewards - rejected_rewards).mean().item()
    beta = scheduler.update(margin)

    # 使用 beta 计算 DPO 损失
    loss = dpo_loss(..., beta=beta)
    loss.backward()
    optimizer.step()

更高级的策略

也可以基于模型在 chosen 上的 log-prob 变化、KL 散度等指标来调整 β。核心思想是维持一个健康的 reward margin，避免过大或过小。