直接偏好优化：数据准备 - 从零开始大模型

上一章，我们用 SFT 让模型学会了按照指令对话。但是 SFT 有一个局限：它只能告诉模型怎么回答，不能告诉模型什么是更好的答案。

比如两个答案都正确，但是一个简洁、一个啰嗦；一个礼貌、一个生硬。SFT 无法区分。

实践中，研究者们发现，可以在微调中告诉模型哪个答案更合理，让模型学会以更准确、更诚实、更安全的方式回答问题。这种微调技术称为偏好优化（Preference Optimization，PO），通常叠加在 SFT 的基础上，进一步优化模型输出。

当前的偏好优化主要有以下三类方法：

基于人类反馈的强化学习（Reinforcement Learning from Human Feedback，RLHF）：训练一个奖励模型，使用强化学习算法来优化模型；
直接偏好优化（Direct Preference Optimization，DPO）：直接使用带偏好的数据训练模型；
推理时偏好优化（Test-Time Preference Optimization，TPO）：在推理阶段通过特定策略和算法（比如引导解码、拒绝采样）来调整模型输出。

直接偏好优化¶

直接偏好优化（DPO）是让模型学会比较的技术：给模型看同一个提示词的两个答案，并告诉它哪个更好。

数据格式¶

DPO 的数据格式是三元组：

提示词（Prompt）：用户的提问或者命令；
接受（Chosen）：更合理的答案；
拒绝（Rejected）：较差的答案。

例如：

提示词：ROMEO:
       Hello, Juliet!
接受：JULIET:
     Hello, Romeo! Nice to meet you!
拒绝：JULIET:
     Hi!

数据来源¶

和 SFT 数据一样，我们依然从微型莎士比亚剧本（tinyshakespeare）数据集的测试集中提取对话样本，作为提示词和接受答案。而拒绝答案则采用相同的提示词，由模型续写的结果。

💡 提示：为了方便教学，本章的样本和第 15 章一样取自莎士比亚剧本：接受答案是真实台词，拒绝答案是预训练模型的续写，这样就不需要人工标注偏好。实际项目中，偏好数据通常由人类或者更强的模型，对同一个提示词的多个回答进行比较和标注。

import pickle
import re
from itertools import pairwise

import numpy as np

from src.core import Tensor
from src.gpt import *

np.random.seed(42)

数据提取¶

提取接受答案¶

和上一章相同的提取函数 extract_turns，从莎士比亚剧本中提取测试集（最后 10% 的台词段，位于预训练的测试集之内）的所有台词。

def extract_turns(filename, split=0.9, max_turn_chars=100):
    with open(filename, encoding="utf-8") as f:
        text = f.read()

    blocks = re.split(r"\n\s*\n", text.strip())
    turns = []

    split = int(len(blocks) * split)
    for block in blocks[split:]:
        lines = block.strip("\n").split("\n")
        if not lines:
            continue

        m = re.compile(r"^([A-Z][A-Za-z' ]{0,30}):\s*$").match(lines[0].strip())
        if not m:
            continue

        speaker = m.group(1).strip()
        content_lines = []
        length = 0
        for line in lines[1:]:
            line = line.strip()
            if not line:
                continue
            if length + len(line) + 1 > max_turn_chars:
                break
            content_lines.append(line)
            length += len(line) + 1
        if not content_lines:
            continue

        turns.append((speaker, "\n".join(content_lines)))

    return turns

生成拒绝答案¶

函数 generate_rejected 用预训练模型生成拒绝答案：

温度采样：设置采样温度为 0.8，控制随机性，值越大越随机；
Top-k 采样：设置 top-k 为 20，只从概率最高的 20 个词元中采样，避免生成乱码；
长度：生成的长度与接受答案一致。

# 使用模型生成的回答作为拒绝答案
def generate_rejected(model, prompt, length, temperature=0.8, top_k=20):
    tokens = list(prompt)

    with Tensor.no_grad():
        for _ in range(length):
            # 推理
            logits = model(Tensor([tokens[-CONTEXT_SIZE:]]))
            last_logits = logits.data[0, -1] / temperature

            # 采样
            k = min(top_k, last_logits.size)
            threshold = np.partition(last_logits, -k)[-k]
            last_logits = np.where(last_logits < threshold, -np.inf, last_logits)
            probs = np.exp(last_logits - np.max(last_logits))
            probs /= np.sum(probs)
            tokens.append(int(np.random.choice(len(probs), p=probs)))

    return tokens[len(prompt):]

超参数¶

模型参数¶

所有用到的超参数保持和预训练一致。

BATCH_SIZE = 16
CONTEXT_SIZE = 256
EMBEDDING_SIZE = 384
HEADS = 6
BLOCKS = 6

构建样本¶

加载模型¶

拒绝答案由预训练模型生成，所以这里加载预训练模型，并切换到评估模式。

DATA_FILE = "../../../../data/tinyshakespeare.txt"
MODEL_FILE = "../../../../data/tinyshakespeare-gpt.npz"

dataset = CharDataset(DATA_FILE, BATCH_SIZE, CONTEXT_SIZE)
model = GPT(dataset.vocab_size, CONTEXT_SIZE, EMBEDDING_SIZE, HEADS, BLOCKS)
trainer = GPTTrainer(model, None, None)
trainer.load(MODEL_FILE)
model.eval()

提取样本¶

从数据集的测试集构建（提示词，接受答案，拒绝答案）三元组：

配对：和 SFT 一样，提取相邻的台词对，作为提示词和接受答案；
生成拒绝答案：用预训练模型为每个提示词生成一个输出作为拒绝答案；
过滤：去掉拒绝答案和接受答案一致的样本。

pairs = list(pairwise(extract_turns(DATA_FILE)))
np.random.shuffle(pairs)
pairs = pairs[:512]

samples = []
for i, ((speaker_a, content_a), (speaker_b, content_b)) in enumerate(pairs):
    # 提示词
    prompt = dataset.encode(f"{speaker_a}:\n{content_a}\n\n")
    # 接受：真实台词
    chosen = dataset.encode(f"{speaker_b}:\n{content_b}\n\n")
    # 拒绝：模型生成的回答
    rejected = generate_rejected(model, prompt, len(chosen))

    if rejected != chosen:
        samples.append((prompt, chosen, rejected))

    if (i + 1) % 64 == 0:
        print(f"generated {len(samples)}/{len(pairs)} DPO samples")

generated 64/512 DPO samples
generated 128/512 DPO samples
generated 192/512 DPO samples
generated 256/512 DPO samples
generated 320/512 DPO samples
generated 384/512 DPO samples
generated 448/512 DPO samples
generated 512/512 DPO samples

保存样本¶

同样用 pickle.dump() 保存到文件。

DPO_SAMPLES = "../../../../data/dpo-samples.pkl"

with open(DPO_SAMPLES, "wb") as f:
    pickle.dump(samples, f)
print(f"saved {len(samples)} DPO samples to {DPO_SAMPLES}")

saved 512 DPO samples to ../../../../data/dpo-samples.pkl

验证样本¶

查看第一个样本，确认格式正确。

sample = samples[0]
print(f"{dataset.decode(sample[0])}\n{dataset.decode(sample[1])}\n{dataset.decode(sample[2])}")

SEBASTIAN:
Foul weather?


ANTONIO:
Very foul.


ISABELLA:
I hope, sir

本章，我们仍然依靠微型莎士比亚剧本数据集的测试集，构建了 512 个 DPO 训练样本：每个样本包括一个提示词、一个接受答案（真实台词）和一个拒绝答案（模型输出）。

下一章，我们将用这些样本对 SFT 模型进行直接偏好优化训练，让模型学会从多个答案中选出更好的一个。

课后练习¶

试着把生成拒绝答案时的温度从 0.8 调到 1.5，观察拒绝答案有什么变化？这会怎样影响偏好数据的质量？