灵魂拷问

SFT 之后的模型能听懂指令、能回答问题了。但同一个问题，它可能给出两种回答：

知识地图

Part 1：为何对齐

SFT 让模型"会说话"，但"说好话"是另一个维度的问题。这一节我们理解对齐的动机。

训练三阶段

SFT 的局限

SFT 教会模型"模仿标准答案"。但现实世界中，同一个问题存在多种回答方式，有好有坏。

有害输出示例

没有对齐的 SFT 模型，可能产生以下问题：

Part 2：RLHF→DPO

DPO 之前，业界标准方案是 RLHF。理解 RLHF 的痛点，才能体会 DPO 的精妙。

RLHF 三步流程

DPO 出现之前，业界标准方案是 RLHF。流程复杂：

RLHF 的痛点

DPO 一句话核心

RLHF vs DPO

Part 3：数据

DPO 的核心输入是"成对偏好数据"。理解数据的结构，是理解一切的基础。

成对偏好数据

DPO 的输入是成对的偏好数据。同一个问题，一个好回答，一个差回答：

JSONL 格式

{
  "chosen": [
    {"role": "user", "content": "什么是递归？"},
    {"role": "assistant", "content": "递归是一种函数调用自身的编程技术..."}
  ],
  "rejected": [
    {"role": "user", "content": "什么是递归？"},
    {"role": "assistant", "content": "递归就是递归，没什么好说的。"}
  ]
}

数据怎么来

数据加载与编码的核心流程

class DPODataset(Dataset):
    def __init__(self, file_path, tokenizer, max_length=4096):
        super().__init__()
        self.tokenizer = tokenizer
        self.max_length = max_length
        self.padding = tokenizer.pad_token_id if tokenizer.pad_token_id is not None else 0
        self.bos_id = tokenizer(f'{tokenizer.bos_token}assistant\n',
                                add_special_tokens=False).input_ids
        self.eos_id = tokenizer(f'{tokenizer.eos_token}\n',
                                add_special_tokens=False).input_ids
        self.samples = load_dataset('json', data_files=file_path, split='train')

数据编码流程

__getitem__ 将偏好数据转为训练张量

def __getitem__(self, index):
    sample = self.samples[index]
    chosen = sample['chosen']      # 好回答的对话列表
    rejected = sample['rejected']  # 差回答的对话列表

    # 用 chat_template 将对话列表转为文本
    chosen_prompt = self.tokenizer.apply_chat_template(
        chosen, tokenize=False, add_generation_prompt=False)
    rejected_prompt = self.tokenizer.apply_chat_template(
        rejected, tokenize=False, add_generation_prompt=False)

    # 编码 + 截断 + padding
    chosen_encoding = self.tokenizer(
        chosen_prompt, truncation=True, max_length=self.max_length, padding='max_length')
    rejected_encoding = self.tokenizer(
        rejected_prompt, truncation=True, max_length=self.max_length, padding='max_length')

    chosen_input_ids = chosen_encoding['input_ids']
    rejected_input_ids = rejected_encoding['input_ids']

张量构建技巧

左移一位，构建训练输入和目标

    # 生成损失掩码：只在 assistant 回答部分为 1
    chosen_loss_mask = self.generate_loss_mask(chosen_input_ids)
    rejected_loss_mask = self.generate_loss_mask(rejected_input_ids)

    # x = tokens[:-1], y = tokens[1:], mask = mask[1:]
    # 模型根据位置 i 的 token 预测位置 i+1 的 token
    x_chosen = torch.tensor(chosen_input_ids[:-1], dtype=torch.long)
    y_chosen = torch.tensor(chosen_input_ids[1:], dtype=torch.long)
    mask_chosen = torch.tensor(chosen_loss_mask[1:], dtype=torch.long)
    x_rejected = torch.tensor(rejected_input_ids[:-1], dtype=torch.long)
    y_rejected = torch.tensor(rejected_input_ids[1:], dtype=torch.long)
    mask_rejected = torch.tensor(rejected_loss_mask[1:], dtype=torch.long)

    return {
        'x_chosen': x_chosen, 'y_chosen': y_chosen, 'mask_chosen': mask_chosen,
        'x_rejected': x_rejected, 'y_rejected': y_rejected, 'mask_rejected': mask_rejected
    }

数据流全览

损失掩码生成

只在 assistant 回答部分计算损失

def generate_loss_mask(self, input_ids):
    loss_mask = [0] * len(input_ids)
    i = 0
    while i < len(input_ids):
        # 找到 assistant 回答的开头标记
        if input_ids[i:i + len(self.bos_id)] == self.bos_id:
            start = i + len(self.bos_id)
            end = start
            # 找到回答的结尾标记 eos_id
            while end < len(input_ids):
                if input_ids[end:end + len(self.eos_id)] == self.eos_id:
                    break
                end += 1
            # bos 和 eos 之间的位置标记为 1
            for j in range(start, min(end + len(self.eos_id), self.max_length)):
                loss_mask[j] = 1
            i = end + len(self.eos_id) if end < len(input_ids) else len(input_ids)
        else:
            i += 1
    return loss_mask

掩码可视化

# tokens: [system] 你好 [user] 什么是AI? [assistant\n] AI是... [eos\n]
# mask:   [   0   ]  0  [  0  ]   0    [    1      ]  1 1 1 [  0  ]
#                                                    ^^^^^^^^^^^^^^^^
#                                                 只有这部分参与损失计算

Part 4：数学

DPO 背后有一套优雅的数学。理解公式，才能正确调参和排错。

BT 模型

DPO 的理论根基

假设人类对回答 y_w（好）和 y_l（差）的偏好服从 Bradley-Terry 模型：

RLHF→DPO

DPO 的关键数学洞察

RLHF 的优化目标：最大化奖励，同时约束不偏离参考模型：

DPO 论文的核心贡献：这个优化问题存在闭式解。奖励函数可以用策略模型和参考模型的概率比来表示：

DPO 损失公式

公式四层分解

极端情况分析

logits 转概率

logits_to_log_probs 的作用

定义：将模型输出的 logits 转为每个位置上"真实 token"的 log 概率。

白话：模型给词汇表中每个词打了个分（logits）。这个函数问：对那个实际出现的词，模型打了多少分？

假设词汇表有 5 个词 ["我", "你", "他", "她", "它"]：

# 模型在某个位置输出的 log 概率：
# [-1.2, -0.3, -2.1, -3.5, -4.0]
#               ^^^^
# 实际下一个 token 是 "你" (index=1)
# 所以该位置的 log_prob = -0.3

logits→概率

模型输出的是 6400 维的原始分数，我们需要把它变成每个 token 的对数概率

这一步很关键：DPO 要比较"策略模型给 chosen 多少分"和"参考模型给 chosen 多少分"。但模型输出的是原始 logits，不是概率。

转换分三步：先 softmax 归一化 → 再取 log → 最后用 gather 按 label 索引挑出真实 token 对应的值。

log_probs = F.log_softmax(logits, dim=2)  # 归一化 + 取 log
log_probs_per_token = torch.gather(        # 按 label 挑出真实 token
    log_probs, dim=2, index=labels.unsqueeze(2)
).squeeze(-1)
return log_probs_per_token  # (batch, seq_len)

这个函数同时用于策略模型和参考模型。DPO 的核心比较就建立在这两组值上。

张量形状变换

dpo_loss

DPO 损失函数的完整实现

def dpo_loss(ref_log_probs, policy_log_probs, mask, beta):
    # 应用掩码，只对 assistant 回答部分求和
    ref_log_probs = (ref_log_probs * mask).sum(dim=1)
    policy_log_probs = (policy_log_probs * mask).sum(dim=1)

    batch_size = ref_log_probs.shape[0]
    # 前 N 条是 chosen，后 N 条是 rejected
    chosen_ref = ref_log_probs[:batch_size // 2]
    reject_ref = ref_log_probs[batch_size // 2:]
    chosen_policy = policy_log_probs[:batch_size // 2]
    reject_policy = policy_log_probs[batch_size // 2:]

    # 核心公式：与数学公式一一对应
    pi_logratios = chosen_policy - reject_policy      # log pi(y_w) - log pi(y_l)
    ref_logratios = chosen_ref - reject_ref            # log ref(y_w) - log ref(y_l)
    logits = pi_logratios - ref_logratios              # 两者的差
    loss = -F.logsigmoid(beta * logits)                # -log(sigmoid(beta * logits))
    return loss.mean()

逐步拆解

四步对应公式的四个层次

batch 拼接

一次前向传播搞定 chosen 和 rejected

# 在 train_epoch 中：
x = torch.cat([x_chosen, x_rejected], dim=0)  # (2N, seq_len)
y = torch.cat([y_chosen, y_rejected], dim=0)
mask = torch.cat([mask_chosen, mask_rejected], dim=0)

# 一次前向传播，得到 2N 条数据的 logits
# 所以 dpo_loss 用 [:batch_size//2] 和 [batch_size//2:] 拆分

Part 5：参考模型

DPO 需要两个模型。参考模型是理解 DPO 的关键之一。

为什么需要 ref

定义：参考模型是 SFT 模型的一个冻结副本，充当"锚点"，约束策略模型不能偏离太远。

白话：没有参考模型，模型可能发现一条捷径——把所有回答的概率都降低，rejected 降得更多。结果？模型学会了"什么都不说"。

参考模型就像一根安全绳。它保证模型只在已有能力上微调偏好，不会偏离到生成空回复或乱码的地步。

无 ref 退化

双模型初始化

# 初始化策略模型（会被训练）
model, tokenizer = init_model(
    lm_config, args.from_weight, device=args.device
)
Logger(f'策略模型总参数量：{sum(p.numel() for p in model.parameters()) / 1e6:.3f} M')

# 初始化参考模型（冻结，不参与训练）
ref_model, _ = init_model(
    lm_config, args.from_weight, device=args.device
)
ref_model.eval()
ref_model.requires_grad_(False)
Logger(f'参考模型总参数量：{sum(p.numel() for p in ref_model.parameters()) / 1e6:.3f} M')

GPU 显存分析

Part 6：训练

把前面的所有组件串起来。DPO 训练的核心循环和普通微调很像，只多了"双模型推理"和"dpo_loss 计算两步。

训练全流程图

单步训练时序

训练循环核心

train_epoch 核心逻辑

def train_epoch(epoch, loader, iters, ref_model, lm_config,
                start_step=0, wandb=None, beta=0.1):
    for step, batch in enumerate(loader, start=start_step + 1):
        # 1. 取出 chosen 和 rejected 数据，移到 GPU
        x_chosen = batch['x_chosen'].to(args.device)
        x_rejected = batch['x_rejected'].to(args.device)
        y_chosen = batch['y_chosen'].to(args.device)
        y_rejected = batch['y_rejected'].to(args.device)
        mask_chosen = batch['mask_chosen'].to(args.device)
        mask_rejected = batch['mask_rejected'].to(args.device)

        # 2. 沿 batch 维度拼接，一次前向传播搞定
        x = torch.cat([x_chosen, x_rejected], dim=0)
        y = torch.cat([y_chosen, y_rejected], dim=0)
        mask = torch.cat([mask_chosen, mask_rejected], dim=0)

损失与反向传播

前向计算和梯度更新的完整流程

        # 3. 动态学习率调整
        lr = get_lr(epoch * iters + step, args.epochs * iters,
                    args.learning_rate)
        for param_group in optimizer.param_groups:
            param_group['lr'] = lr

        with autocast_ctx:
            # 参考模型前向：不计算梯度
            with torch.no_grad():
                ref_outputs = ref_model(x)
            ref_log_probs = logits_to_log_probs(ref_outputs.logits, y)

            # 策略模型前向：计算梯度
            outputs = model(x)
            policy_log_probs = logits_to_log_probs(outputs.logits, y)

            # 计算 DPO 损失 + MoE 辅助损失
            dpo_loss_val = dpo_loss(ref_log_probs, policy_log_probs,
                                    mask, beta=beta)
            loss = dpo_loss_val + outputs.aux_loss
            loss = loss / args.accumulation_steps

        # 4. 梯度累积 + 裁剪 + 更新
        scaler.scale(loss).backward()
        if step % args.accumulation_steps == 0:
            scaler.unscale_(optimizer)
            torch.nn.utils.clip_grad_norm_(model.parameters(), args.grad_clip)
            scaler.step(optimizer)
            scaler.update()
            optimizer.zero_grad(set_to_none=True)

主循环完整逻辑

模型初始化、训练启动、检查点保存

# 初始化策略模型和参考模型
model, tokenizer = init_model(lm_config, args.from_weight, device=args.device)
ref_model, _ = init_model(lm_config, args.from_weight, device=args.device)
ref_model.eval()
ref_model.requires_grad_(False)

# 创建数据集和优化器
train_ds = DPODataset(args.data_path, tokenizer, max_length=args.max_seq_len)
optimizer = optim.AdamW(model.parameters(), lr=args.learning_rate)

# 开始训练
for epoch in range(start_epoch, args.epochs):
    setup_seed(42 + epoch)
    loader = DataLoader(train_ds, batch_sampler=..., num_workers=...)
    train_epoch(epoch, loader, len(loader), ref_model, lm_config,
                0, wandb, args.beta)

学习率为何极低

默认 4e-8，比 SFT 的 1e-5 小约 1000 倍

梯度累积与精度

# 不增加显存，模拟大 batch
if step % args.accumulation_steps == 0:
    scaler.step(optimizer)
    optimizer.zero_grad()

dtype = torch.bfloat16 if args.dtype == "bfloat16" else torch.float16
autocast_ctx = torch.cuda.amp.autocast(dtype=dtype)

Part 7：排错

动手构建 DPO 数据，启动训练，以及常见错误的诊断和修复。

准备 DPO 数据

创建 dataset/dpo.jsonl，每行一条偏好数据：

{"chosen": [{"role": "user", "content": "1+1等于几？"}, {"role": "assistant", "content": "1+1等于2。这是基础的加法运算。"}], "rejected": [{"role": "user", "content": "1+1等于几？"}, {"role": "assistant", "content": "1+1等于3吧，我不太确定。"}]}
{"chosen": [{"role": "user", "content": "介绍一下地球"}, {"role": "assistant", "content": "地球是太阳系第三颗行星，赤道半径约6371公里..."}], "rejected": [{"role": "user", "content": "介绍一下地球"}, {"role": "assistant", "content": "地球就是个球。"}]}

启动训练

# 确保已完成 SFT 训练，权重保存在 out/full_sft_*.pth
python trainer/train_dpo.py \
    --epochs 1 \
    --batch_size 4 \
    --learning_rate 4e-8 \
    --beta 0.15 \
    --from_weight full_sft \
    --data_path ../dataset/dpo.jsonl \
    --device cuda:0

超参数调优

错误诊断一

错误诊断二

错误诊断三

# train_epoch 末尾的清理代码
del x_chosen, x_rejected, y_chosen, y_rejected, mask_chosen, mask_rejected
del ref_outputs, ref_logits, ref_log_probs, outputs, logits, policy_log_probs

Part 8：练习

动手实践是最好的学习方式。通过练习巩固对 DPO 的理解。

动手练习

# 起始框架：创建 dpo.jsonl
import json
data = [
    {
        "chosen": [
            {"role": "user", "content": "请在此写问题"},
            {"role": "assistant", "content": "请在此写好回答"}
        ],
        "rejected": [
            {"role": "user", "content": "请在此写同样的问题"},
            {"role": "assistant", "content": "请在此写差回答"}
        ]
    }
    # 再写 4 条...
]
with open('dataset/dpo.jsonl', 'w', encoding='utf-8') as f:
    for item in data:
        f.write(json.dumps(item, ensure_ascii=False) + '\n')

练习：观察

# beta = 0.05（激进）
python trainer/train_dpo.py --beta 0.05 --save_weight dpo_005

# beta = 0.15（温和，默认）
python trainer/train_dpo.py --beta 0.15 --save_weight dpo_015

# beta = 0.5（保守）
python trainer/train_dpo.py --beta 0.5 --save_weight dpo_050

练习：验证效果

# 使用 1-inference.py 进行对比生成
# 训练前：加载 full_sft 权重
# 训练后：加载 dpo 权重

# 对比维度：
# 1. 回答的详细程度
# 2. 回答的准确性
# 3. 是否回避了 rejected 风格的敷衍回答

思考题

SFT vs DPO

学完本阶段后

延伸阅读