DPO 对齐训练实验

在 SFT 模型对话实验中，我们通过监督微调教会了模型遵循对话格式，实现了从续写到对话的转变。SFT 的训练数据是"用户问 → 模型答"的指令回答对，模型学到的是模仿训练数据中的回答模式。但同一个问题可以有多种合理的回答方式，SFT 无法告诉模型哪一种更符合人类的偏好，回答更准确、语气更友好、拒绝更礼貌，这些维度上的差异 SFT 都捕捉不到。

本实验使用直接偏好优化方法，在 SFT 模型的基础上进行对齐训练。DPO 的训练数据是同一个问题的两个回答，人类标注哪个更好，模型从中学会区分回答的优劣，在生成时更倾向于选择偏好的回答风格。

实验准备

在开始实验之前，请确保已完成以下准备工作：

已完成 SFT 模型对话实验，模型权重文件 full_sft_768.pth 已在数据目录中正确生成。
已挂载数据目录并下载好 DPO 偏好语料。

# 选择 "下载数据集" -> 选择 "MiniMind Alignment (LLM对齐语料)"
dmla data

MiniMind 项目的 DPO 偏好语料（dpo.jsonl）包含约 20K 条偏好对比数据，数据抽样自 DPO-En-Zh-20k，体积约 53 MB。数据集下载完成后，以下代码可验证 SFT 模型和 DPO 语料是否完整：

import os

# 检查 SFT 模型（由上一章实验生成）
sft_dir = os.path.join(DATA_DIR, 'models', 'minimind', 'sft')
sft_path = os.path.join(sft_dir, 'full_sft_768.pth')
if os.path.exists(sft_path):
    size_mb = os.path.getsize(sft_path) / (1024 ** 2)
    print(f"SFT 模型: 已存在 ({size_mb:.1f} MB)")
else:
    # 尝试 epoch checkpoint
    for epoch in [2, 1]:
        ckp = os.path.join(sft_dir, f'sft_epoch{epoch}.pth')
        if os.path.exists(ckp):
            size_mb = os.path.getsize(ckp) / (1024 ** 2)
            print(f"SFT 模型: 使用 epoch {epoch} checkpoint ({size_mb:.1f} MB)")
            break
    else:
        print("SFT 模型: 未找到！请先完成 SFT 实验")

# 检查 DPO 语料
dpo_dir = os.path.join(DATA_DIR, 'datasets', 'minimind-alignment')
if os.path.exists(dpo_dir):
    print(f"DPO 语料目录: 已存在")
    for f in os.listdir(dpo_dir):
        fpath = os.path.join(dpo_dir, f)
        if os.path.isfile(fpath):
            size_mb = os.path.getsize(fpath) / (1024 ** 2)
            print(f"  {f}: {size_mb:.1f} MB")
else:
    print("DPO 语料: 未下载，请运行 'dmla data' 下载 MiniMind Alignment 数据集")

# 检查 tokenizer（复用预训练的）
tokenizer_dir = os.path.join(DATA_DIR, 'datasets', 'minimind-pretrain')
tokenizer_json = os.path.join(tokenizer_dir, 'tokenizer.json')
print(f"Tokenizer: {'已存在' if os.path.exists(tokenizer_json) else '未找到'}")

点击 Run 按钮执行代码，点击代码区域可编辑

第一阶段：偏好对比数据集

DPO 的训练数据格式与 SFT 不同。SFT 的每条样本是一个指令回答对 $(x, y)$ ，DPO 的每条样本是一个偏好对比三元组 $(x, y_w, y_l)$ ，其中 $x$ 是用户指令， $y_w$ 是被选中的好回答（chosen）， $y_l$ 是被拒绝的差回答（rejected）。chosen 和 rejected 对应同一个用户指令，只有 assistant 的回答不同。

数据以 JSONL 格式存储，每行一条偏好对：

{
  "chosen": [
    {"role": "user", "content": "什么是机器学习"},
    {"role": "assistant", "content": "机器学习是人工智能的一个分支，它使计算机能够从数据中学习规律..."}
  ],
  "rejected": [
    {"role": "user", "content": "什么是机器学习"},
    {"role": "assistant", "content": "机器学习就是让电脑自己学东西"}
  ]
}

下面代码实现了 DPODataset，将偏好对比数据转换为模型可训练的格式，每条样本包含 chosen 和 rejected 两条对话，分别 tokenize 后生成对应的输入序列和掩码。掩码的作用是只在 assistant 回答部分计算对数概率，用户提问部分不参与 DPO 损失计算。这段代码会在训练阶段被调用，无需手动运行。

import os
import torch
from torch.utils.data import Dataset
from datasets import load_dataset, Features, Value
from datasets import logging as datasets_logging

class DPODataset(Dataset):
    """
    DPO 数据集：将偏好对比数据 tokenize 为模型可训练的格式

    每条样本格式：{"chosen": [{role, content}, ...], "rejected": [{role, content}, ...]}
    输出 chosen 和 rejected 的 input_ids、目标 ids 和 loss_mask
    loss_mask 仅在 assistant 回答部分为 1，其余为 0
    """
    CHATML_TEMPLATE = (
        "{% for message in messages %}<|im_start|>{{ message.role }}\n"
        "{{ message.content }}<|im_end|>\n"
        "{% endfor %}"
        "{% if add_generation_prompt %}<|im_start|>assistant\n{% endif %}"
    )

    def __init__(self, jsonl_path, tokenizer, max_length=768):
        super().__init__()
        os.environ["TOKENIZERS_PARALLELISM"] = "false"
        self.tokenizer = tokenizer
        if not tokenizer.chat_template:
            tokenizer.chat_template = self.CHATML_TEMPLATE
        self.max_length = max_length
        self.padding = tokenizer.pad_token_id if tokenizer.pad_token_id is not None else 0