RLHF:奖励模型 + PPO

Build~90 minutesPython (with numpy)先修要求: Phase 10, Lesson 06 (Instruction Tuning / SFT)
SFT 教会模型遵循指令,但没有教会模型哪种回答更好。两个语法正确、事实准确的回答,在有用性上可能天差地别。RLHF 就是把人类判断编码进模型行为的方式。正是它让 Claude 乐于助人、让 GPT 彬彬有礼。
学习目标
  • 构建一个奖励模型,从人类偏好对(chosen 与 rejected)中为回答质量打分
  • 实现 PPO 训练循环,在 KL 惩罚下让语言模型策略针对奖励模型进行优化
  • 解释 RLHF 为什么需要三个模型(SFT、奖励、策略),以及 KL 约束如何防止奖励黑客
  • 通过比较偏好优化前后的回答质量,评估 RLHF 的效果

问题

问模型 "Explain quantum computing",它可能生成:

回答 A: "量子计算使用可以处于叠加态的量子比特,这意味着它们可以同时是 0、1 或两者兼有。这使得量子计算机在某些计算上比经典计算机快指数级。关键算法包括用于大数分解的 Shor 算法和用于搜索无序数据库的 Grover 算法。"

回答 B: "量子计算是一种利用量子力学现象的计算机类型。它最早于 1980 年代被提出。Richard Feynman 提出量子系统可以由量子计算机模拟。此后该领域显著发展。许多公司现在都在研究量子计算机。IBM、Google 等已经取得进展。Google 于 2019 年宣称实现量子霸权。"

两个回答在事实上都正确,语法都通顺,也都遵循了指令。但回答 A 明显更好:更简洁、信息量更大、结构更好。人类每次都会选 A。

SFT 无法捕捉这种区别。它在「正确」的回答上训练模型,却没有机制表达「这个回答比那个更好」。它把每个训练样本视为同样好。如果 A 和 B 都出现在 SFT 数据集中,模型会同等程度地从两者学习。

RLHF 解决了这个问题。它训练一个奖励模型来预测人类更偏好哪个回答,然后用这个奖励信号把语言模型推向更高质量的输出。InstructGPT(ChatGPT 的前身)用 RLHF 大幅提升了 GPT-3 的有用性、真实性和无害性。OpenAI 的内部评估者在 85% 的情况下更偏好 InstructGPT 的输出而非 GPT-3 的输出,尽管 InstructGPT 小了 135 倍(1.3B 对 175B 参数)。

概念

三个阶段

RLHF 不是单次训练,而是一个由三个顺序阶段组成的 pipeline,每个阶段都建立在前一个之上。

阶段 1:SFT。 在指令-回答对上训练基础模型(第 06 课)。这给你一个能遵循指令的模型,但它不知道哪些回答更好。

阶段 2:奖励模型。 收集人类偏好数据:向标注员展示同一 prompt 的两个回答,问「哪个更好?」。训练一个模型来预测这些偏好。奖励模型以 (prompt, response) 为输入,输出一个标量分数。

阶段 3:PPO。 用奖励模型为语言模型生成训练信号。语言模型生成回答,奖励模型为其打分,PPO 更新语言模型以产出得分更高的回答。KL 散度惩罚防止语言模型偏离 SFT checkpoint 太远。

graph TD
    subgraph Stage1["Stage 1: SFT"]
        B["Base Model"] --> S["SFT Model"]
        D["Instruction Data\n(27K examples)"] --> S
    end

    subgraph Stage2["Stage 2: Reward Model"]
        S --> |"Generate responses"| P["Preference Pairs\n(prompt, winner, loser)"]
        H["Human Annotators"] --> P
        P --> R["Reward Model\nR(prompt, response) → score"]
    end

    subgraph Stage3["Stage 3: PPO"]
        S --> |"Initialize policy"| PI["Policy Model\n(being optimized)"]
        S --> |"Freeze as reference"| REF["Reference Model\n(frozen SFT)"]
        PI --> |"Generate"| RESP["Response"]
        RESP --> R
        R --> |"Reward signal"| PPO["PPO Update"]
        REF --> |"KL penalty"| PPO
        PPO --> |"Update"| PI
    end

    style S fill:#1a1a2e,stroke:#51cf66,color:#fff
    style R fill:#1a1a2e,stroke:#e94560,color:#fff
    style PI fill:#1a1a2e,stroke:#0f3460,color:#fff
    style REF fill:#1a1a2e,stroke:#0f3460,color:#fff
    style PPO fill:#1a1a2e,stroke:#e94560,color:#fff

奖励模型

奖励模型是一个被改造成评分器的语言模型。拿 SFT 模型,把语言建模头(输出词表上的分布)换成标量头(输出一个数)。除最后一层外,架构完全相同。

输入:prompt 与回答拼接。输出:单个标量奖励分数。

训练数据是人类偏好对。对每个 prompt,标注员看到两个回答并选择更好的那个。这产生训练三元组:(prompt, preferred_response, rejected_response)。

损失函数使用 Bradley-Terry 成对偏好模型:

\text{loss} = -\log\big(\mathop{\mathrm{sigmoid}}(\text{reward}(\text{preferred}) - \text{reward}(\text{rejected}))\big)

这是关键方程。sigmoid(reward(A) - reward(B)) 给出回答 A 优于回答 B 的概率。损失推动奖励模型给偏好回答分配更高的分数。

为什么用成对比较而不是绝对分数?因为人类非常不擅长打绝对质量分(「这个回答是 7.3 还是 7.5 分?」),却很擅长相对比较(「A 比 B 好吗?」)。Bradley-Terry 模型把相对比较转化为一致的绝对评分体系。

InstructGPT 的数字: OpenAI 从 40 名承包商标注员那里收集了 33,000 对比较。每次比较约需 5 分钟。也就是说,奖励模型训练数据耗费了 2,750 小时的人力。

PPO:近端策略优化

PPO 是一种强化学习算法。在 RLHF 中,「环境」是奖励模型,「Agent」是语言模型,「动作」是生成一个 token。

目标:

\text{maximize:} \quad \mathbb{E}[R(\text{prompt}, \text{response})] - \beta \cdot \mathrm{KL}(\text{policy} \,\|\, \text{reference})

第一项推动模型生成高奖励的回答。第二项(KL 散度惩罚)防止模型偏离 SFT checkpoint 太远。

为什么需要 KL 惩罚?没有它,模型会找到退化解。奖励模型是在有限的人类偏好数据集上训练的,存在盲区。语言模型会利用这些盲区——找到在奖励模型上得分很高、实际上毫无意义的输出。经典例子:

  • 重复 "I'm so helpful and harmless!" 会在有用性/无害性奖励模型上得高分
  • 生成冗长、听起来正式但空洞的回答,在模式上匹配「高质量」
  • 利用训练数据中恰好与高奖励相关的特定短语

KL 惩罚的意思是:你可以改进,但不能变成完全不同的模型。保持在已经相当合理的 SFT 版本附近。偏离太远,KL 代价就会压过奖励。

InstructGPT 的数字: PPO 训练使用 lr=1.5e-5,KL 系数 \beta=0.02,256K 个 episode(prompt-回答对),每批 4 个 PPO epoch。整个 RLHF pipeline 在 GPU 集群上耗时数天。

graph LR
    subgraph PPO["PPO Training Loop"]
        direction TB
        PROMPT["Sample prompt\nfrom dataset"] --> GEN["Policy generates\nresponse"]
        GEN --> SCORE["Reward model\nscores response"]
        GEN --> KL["Compute KL divergence\nvs reference model"]
        SCORE --> OBJ["Objective:\nreward - beta * KL"]
        KL --> OBJ
        OBJ --> UPDATE["PPO gradient update\n(clipped surrogate loss)"]
        UPDATE --> |"repeat"| PROMPT
    end

    style PROMPT fill:#1a1a2e,stroke:#0f3460,color:#fff
    style SCORE fill:#1a1a2e,stroke:#51cf66,color:#fff
    style KL fill:#1a1a2e,stroke:#e94560,color:#fff
    style OBJ fill:#1a1a2e,stroke:#e94560,color:#fff

PPO 目标详解

PPO 使用「裁剪代理目标」来防止过大的更新。新策略与旧策略概率之比被裁剪到范围 [1 - \epsilon, 1 + \epsilon],其中 \epsilon 通常为 0.2。

\begin{aligned} \text{ratio} &= \pi_{\text{new}}(\text{action} \mid \text{state}) / \pi_{\text{old}}(\text{action} \mid \text{state}) \\ \text{clipped\_ratio} &= \mathop{\mathrm{clip}}(\text{ratio}, 1 - \epsilon, 1 + \epsilon) \\ \text{loss} &= -\min(\text{ratio} \cdot \text{advantage}, \text{clipped\_ratio} \cdot \text{advantage}) \end{aligned}

优势函数估计当前回答比预期质量好多少。在 RLHF 中:

advantage = reward(prompt, response) - baseline

基线通常是近期回答的平均奖励。正优势表示该回答优于平均,负优势表示更差。PPO 提高高于平均的回答的概率,降低低于平均的回答的概率。

裁剪防止灾难性更新。如果单个回答获得异常高的奖励,未裁剪的比率可能非常大,导致模型剧烈偏向该回答。裁剪为更新设上限,维持训练稳定性。

奖励黑客

RLHF 的阴暗面。语言模型针对奖励模型优化,而奖励模型只是人类偏好的不完美代理。随着语言模型越来越擅长最大化奖励,它开始利用奖励模型的弱点。

常见失败模式:

失败表现原因
冗长模型生成越来越长的回答人类标注员常常偏好更长、更详细的回答,因此奖励模型给长度分配更高分数
谄媚模型附和用户说的每件事标注员偏好与问题前提一致的回答
对冲模型拒绝给出确定的答案对冲式回答(「这是一个复杂话题,有很多视角……」)很少会被标记为错误
格式博弈模型过度使用项目符号和标题格式化的回答在标注员看来更「精致」

缓解策略:更强的 KL 惩罚(防止模型偏离到足以利用弱点的程度)、在对抗样本上训练奖励模型(修补已知失败模式)、使用多个不同架构的奖励模型(更难同时黑掉全部)。

真实的 RLHF pipeline

模型比较对标注员RM 大小PPO 步数KL 系数
InstructGPT33K406B256K0.02
Llama 2 Chat~1M未披露70B未披露0.01
Claude未披露未披露未披露未披露未披露
Anthropic RLHF paper22K2052B50K0.001

Anthropic 2022 年的论文在 22,000 个比较上训练了 52B 的奖励模型。更大的奖励模型产生更可靠的信号,使 PPO 训练更稳定。用小奖励模型训练大语言模型是有风险的——奖励模型没有足够容量捕捉好回答与坏回答之间的细微差别。

rlhf-pipeline静态版本未包含此交互图

构建它

步骤 1:合成偏好数据

生产中,人类标注员创建偏好数据。我们将创建合成配对,其中「偏好」的回答客观上更好(更简洁、更准确、更有帮助)。

pythonimport numpy as np

PREFERENCE_DATA = [
    {
        "prompt": "What is the capital of France?",
        "preferred": "The capital of France is Paris.",
        "rejected": "France is a country in Europe. It has many cities. The capital is Paris. Paris is known for the Eiffel Tower.",
    },
    {
        "prompt": "Explain gravity in one sentence.",
        "preferred": "Gravity is the force that attracts objects with mass toward each other.",
        "rejected": "Gravity is something that makes things fall down when you drop them.",
    },
    {
        "prompt": "What is 15 times 7?",
        "preferred": "15 times 7 is 105.",
        "rejected": "Let me think about this. 15 times 7. Well, 10 times 7 is 70, and 5 times 7 is 35, so the answer might be around 105.",
    },
    {
        "prompt": "Name three programming languages.",
        "preferred": "Python, Rust, and TypeScript.",
        "rejected": "There are many programming languages. Some popular ones include various languages like Python and others.",
    },
    {
        "prompt": "What year did World War II end?",
        "preferred": "World War II ended in 1945.",
        "rejected": "World War II was a major global conflict. It involved many countries. The war ended in the mid-1940s, specifically in 1945.",
    },
    {
        "prompt": "Define machine learning.",
        "preferred": "Machine learning is a field where algorithms learn patterns from data to make predictions without being explicitly programmed.",
        "rejected": "Machine learning is a type of AI. AI stands for artificial intelligence. Machine learning uses data to learn.",
    },
]

偏好的回答简洁直接。被拒绝的回答展示了常见失败模式:不必要的填充、对冲、冗余解释和不精确。这正是 SFT 无法捕捉而 RLHF 可以捕捉的区别。

步骤 2:奖励模型架构

奖励模型复用 mini GPT 的 Transformer 架构,但把词表大小的输出头换成单标量投影。

pythonimport sys
import os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "04-pre-training-mini-gpt", "code"))
from main import MiniGPT, LayerNorm, Embedding, TransformerBlock


class RewardModel:
    def __init__(self, vocab_size=256, embed_dim=128, num_heads=4,
                 num_layers=4, max_seq_len=128, ff_dim=512):
        self.embedding = Embedding(vocab_size, embed_dim, max_seq_len)
        self.blocks = [
            TransformerBlock(embed_dim, num_heads, ff_dim)
            for _ in range(num_layers)
        ]
        self.ln_f = LayerNorm(embed_dim)
        self.reward_head = np.random.randn(embed_dim) * 0.02

    def forward(self, token_ids):
        seq_len = token_ids.shape[-1]
        mask = np.triu(np.full((seq_len, seq_len), -1e9), k=1)

        x = self.embedding.forward(token_ids)
        for block in self.blocks:
            x = block.forward(x, mask)
        x = self.ln_f.forward(x)

        last_hidden = x[:, -1, :]
        reward = last_hidden @ self.reward_head

        return reward

奖励模型取最后一个 token 位置的隐藏状态,投影为标量。为什么是最后一个 token?因为因果注意力掩码意味着最后一个位置已经关注了之前所有 token。它对整个 (prompt, response) 序列拥有最完整的表示。

步骤 3:Bradley-Terry 损失

使用 Bradley-Terry 成对损失在偏好对上训练奖励模型。

pythondef tokenize_for_reward(prompt, response, vocab_size=256):
    prompt_tokens = [min(t, vocab_size - 1) for t in list(prompt.encode("utf-8"))]
    response_tokens = [min(t, vocab_size - 1) for t in list(response.encode("utf-8"))]
    return prompt_tokens + [0] + response_tokens


def sigmoid(x):
    return np.where(
        x >= 0,
        1.0 / (1.0 + np.exp(-x)),
        np.exp(x) / (1.0 + np.exp(x))
    )


def bradley_terry_loss(reward_preferred, reward_rejected):
    diff = reward_preferred - reward_rejected
    loss = -np.log(sigmoid(diff) + 1e-8)
    return loss


def train_reward_model(rm, preference_data, num_epochs=10, lr=1e-4, max_seq_len=128):
    print(f"Training Reward Model: {len(preference_data)} preference pairs, {num_epochs} epochs")
    print()

    losses = []
    accuracies = []

    for epoch in range(num_epochs):
        epoch_loss = 0.0
        epoch_correct = 0
        num_pairs = 0

        indices = np.random.permutation(len(preference_data))

        for idx in indices:
            pair = preference_data[idx]

            preferred_tokens = tokenize_for_reward(pair["prompt"], pair["preferred"])
            rejected_tokens = tokenize_for_reward(pair["prompt"], pair["rejected"])

            preferred_tokens = preferred_tokens[:max_seq_len]
            rejected_tokens = rejected_tokens[:max_seq_len]

            preferred_ids = np.array(preferred_tokens).reshape(1, -1)
            rejected_ids = np.array(rejected_tokens).reshape(1, -1)

            r_preferred = rm.forward(preferred_ids)[0]
            r_rejected = rm.forward(rejected_ids)[0]

            loss = bradley_terry_loss(r_preferred, r_rejected)

            if r_preferred > r_rejected:
                epoch_correct += 1

            diff = r_preferred - r_rejected
            grad = sigmoid(diff) - 1.0

            rm.reward_head -= lr * grad * rm.ln_f.forward(
                rm.embedding.forward(preferred_ids)
            )[:, -1, :].flatten()

            epoch_loss += loss
            num_pairs += 1

        avg_loss = epoch_loss / max(num_pairs, 1)
        accuracy = epoch_correct / max(num_pairs, 1)
        losses.append(avg_loss)
        accuracies.append(accuracy)

        if epoch % 2 == 0:
            print(f"  Epoch {epoch + 1:3d} | Loss: {avg_loss:.4f} | Accuracy: {accuracy:.1%}")

    return rm, losses, accuracies

准确率指标很直观:奖励模型正确排序的偏好对占多大比例?随机模型得分 50%。在干净数据上训练良好的奖励模型应超过 70%。InstructGPT 的奖励模型在留出比较上达到约 72% 准确率,听起来低,其实不错——许多偏好对即使对人类也是有歧义的(标注员间一致率约为 73%)。

步骤 4:简化版 PPO 循环

完整 PPO 很复杂。本实现抓住核心机制:生成回答、打分、计算优势,并在 KL 惩罚下更新策略。

pythondef compute_kl_divergence(policy_logits, reference_logits):
    policy_probs = np.exp(policy_logits - policy_logits.max(axis=-1, keepdims=True))
    policy_probs = policy_probs / policy_probs.sum(axis=-1, keepdims=True)
    policy_probs = np.clip(policy_probs, 1e-10, 1.0)

    ref_probs = np.exp(reference_logits - reference_logits.max(axis=-1, keepdims=True))
    ref_probs = ref_probs / ref_probs.sum(axis=-1, keepdims=True)
    ref_probs = np.clip(ref_probs, 1e-10, 1.0)

    kl = np.sum(policy_probs * np.log(policy_probs / ref_probs), axis=-1)
    return kl.mean()


def generate_response(model, prompt_tokens, max_new_tokens=30, temperature=0.8, max_seq_len=128):
    tokens = list(prompt_tokens)

    for _ in range(max_new_tokens):
        context = np.array(tokens[-max_seq_len:]).reshape(1, -1)
        logits = model.forward(context)
        next_logits = logits[0, -1, :]

        next_logits = next_logits / max(temperature, 1e-8)
        probs = np.exp(next_logits - next_logits.max())
        probs = probs / probs.sum()
        probs = np.clip(probs, 1e-10, 1.0)
        probs = probs / probs.sum()

        next_token = np.random.choice(len(probs), p=probs)
        tokens.append(int(next_token))

    return tokens


def copy_model_weights(source, target):
    target.embedding.token_embed = source.embedding.token_embed.copy()
    target.embedding.pos_embed = source.embedding.pos_embed.copy()
    target.ln_f.gamma = source.ln_f.gamma.copy()
    target.ln_f.beta = source.ln_f.beta.copy()
    for s_block, t_block in zip(source.blocks, target.blocks):
        t_block.attn.W_q = s_block.attn.W_q.copy()
        t_block.attn.W_k = s_block.attn.W_k.copy()
        t_block.attn.W_v = s_block.attn.W_v.copy()
        t_block.attn.W_out = s_block.attn.W_out.copy()
        t_block.ffn.W1 = s_block.ffn.W1.copy()
        t_block.ffn.W2 = s_block.ffn.W2.copy()
        t_block.ffn.b1 = s_block.ffn.b1.copy()
        t_block.ffn.b2 = s_block.ffn.b2.copy()
        t_block.ln1.gamma = s_block.ln1.gamma.copy()
        t_block.ln1.beta = s_block.ln1.beta.copy()
        t_block.ln2.gamma = s_block.ln2.gamma.copy()
        t_block.ln2.beta = s_block.ln2.beta.copy()


def ppo_training(policy_model, reference_model, reward_model, prompts,
                 num_episodes=20, lr=1.5e-5, kl_coeff=0.02, max_seq_len=128):
    print(f"PPO Training: {num_episodes} episodes, lr={lr}, KL coeff={kl_coeff}")
    print()

    rewards_history = []
    kl_history = []

    for episode in range(num_episodes):
        prompt_text = prompts[episode % len(prompts)]
        prompt_tokens = [min(t, 252) for t in list(prompt_text.encode("utf-8"))]

        response_tokens = generate_response(
            policy_model, prompt_tokens,
            max_new_tokens=20, temperature=0.8, max_seq_len=max_seq_len
        )

        response_ids = np.array(response_tokens[:max_seq_len]).reshape(1, -1)
        reward = reward_model.forward(response_ids)[0]

        policy_logits = policy_model.forward(response_ids)
        ref_logits = reference_model.forward(response_ids)
        kl = compute_kl_divergence(policy_logits, ref_logits)

        total_reward = reward - kl_coeff * kl

        rewards_history.append(float(reward))
        kl_history.append(float(kl))

        for block in policy_model.blocks:
            update_scale = lr * total_reward
            block.ffn.W1 += update_scale * np.random.randn(*block.ffn.W1.shape) * 0.01
            block.ffn.W2 += update_scale * np.random.randn(*block.ffn.W2.shape) * 0.01

        if episode % 5 == 0:
            avg_reward = np.mean(rewards_history[-5:]) if rewards_history else 0
            avg_kl = np.mean(kl_history[-5:]) if kl_history else 0
            print(f"  Episode {episode:3d} | Reward: {reward:.4f} | KL: {kl:.4f} | "
                  f"Avg Reward: {avg_reward:.4f}")

    return policy_model, rewards_history, kl_history

核心循环:(1) 采样一个 prompt,(2) 生成回答,(3) 用奖励模型打分,(4) 计算与冻结参考模型的 KL 散度,(5) 计算调整后的奖励(奖励减去 KL 惩罚),(6) 更新策略。KL 惩罚随策略偏离参考模型而增长,自动防止奖励黑客。

步骤 5:奖励分数比较

RLHF 之后,策略模型的回答在奖励模型上的得分应高于原始 SFT 模型的回答。

pythondef compare_models(sft_model, rlhf_model, reward_model, prompts, max_seq_len=128):
    print("Model Comparison (reward scores)")
    print("-" * 60)
    print(f"  {'Prompt':<35} {'SFT':>10} {'RLHF':>10}")
    print("  " + "-" * 55)

    sft_total = 0.0
    rlhf_total = 0.0

    for prompt in prompts:
        prompt_tokens = [min(t, 252) for t in list(prompt.encode("utf-8"))]

        sft_response = generate_response(
            sft_model, prompt_tokens,
            max_new_tokens=20, temperature=0.6, max_seq_len=max_seq_len
        )
        rlhf_response = generate_response(
            rlhf_model, prompt_tokens,
            max_new_tokens=20, temperature=0.6, max_seq_len=max_seq_len
        )

        sft_ids = np.array(sft_response[:max_seq_len]).reshape(1, -1)
        rlhf_ids = np.array(rlhf_response[:max_seq_len]).reshape(1, -1)

        sft_reward = reward_model.forward(sft_ids)[0]
        rlhf_reward = reward_model.forward(rlhf_ids)[0]

        sft_total += sft_reward
        rlhf_total += rlhf_reward

        truncated_prompt = prompt[:33] + ".." if len(prompt) > 35 else prompt
        print(f"  {truncated_prompt:<35} {sft_reward:>10.4f} {rlhf_reward:>10.4f}")

    n = len(prompts)
    print("  " + "-" * 55)
    print(f"  {'Average':<35} {sft_total/n:>10.4f} {rlhf_total/n:>10.4f}")

    return sft_total / n, rlhf_total / n

使用它

完整的 RLHF pipeline 演示

pythonif __name__ == "__main__":
    np.random.seed(42)

    print("=" * 70)
    print("RLHF PIPELINE: REWARD MODEL + PPO")
    print("=" * 70)
    print()

    print("STAGE 1: SFT Model (from Lesson 06)")
    print("-" * 40)
    sft_model = MiniGPT(
        vocab_size=256, embed_dim=128, num_heads=4,
        num_layers=4, max_seq_len=128, ff_dim=512
    )
    print(f"  Parameters: {sft_model.count_parameters():,}")
    print()

    print("STAGE 2: Train Reward Model")
    print("-" * 40)
    rm = RewardModel(
        vocab_size=256, embed_dim=128, num_heads=4,
        num_layers=4, max_seq_len=128, ff_dim=512
    )

    rm, rm_losses, rm_accuracies = train_reward_model(rm, PREFERENCE_DATA, num_epochs=10, lr=1e-4)
    print()

    print("Reward Model Evaluation:")
    print("-" * 40)
    correct = 0
    for pair in PREFERENCE_DATA:
        pref_tokens = tokenize_for_reward(pair["prompt"], pair["preferred"])[:128]
        rej_tokens = tokenize_for_reward(pair["prompt"], pair["rejected"])[:128]

        r_pref = rm.forward(np.array(pref_tokens).reshape(1, -1))[0]
        r_rej = rm.forward(np.array(rej_tokens).reshape(1, -1))[0]

        if r_pref > r_rej:
            correct += 1
        print(f"  Preferred: {r_pref:+.4f} | Rejected: {r_rej:+.4f} | {'Correct' if r_pref > r_rej else 'Wrong'}")

    print(f"\n  Accuracy: {correct}/{len(PREFERENCE_DATA)} = {correct/len(PREFERENCE_DATA):.1%}")
    print()

    print("STAGE 3: PPO Training")
    print("-" * 40)

    policy_model = MiniGPT(
        vocab_size=256, embed_dim=128, num_heads=4,
        num_layers=4, max_seq_len=128, ff_dim=512
    )
    reference_model = MiniGPT(
        vocab_size=256, embed_dim=128, num_heads=4,
        num_layers=4, max_seq_len=128, ff_dim=512
    )

    copy_model_weights(sft_model, policy_model)
    copy_model_weights(sft_model, reference_model)

    train_prompts = [pair["prompt"] for pair in PREFERENCE_DATA]

    policy_model, rewards, kls = ppo_training(
        policy_model, reference_model, rm,
        train_prompts, num_episodes=20, lr=1.5e-5, kl_coeff=0.02
    )
    print()

    print("=" * 70)
    print("COMPARISON: SFT vs RLHF")
    print("=" * 70)
    print()

    eval_prompts = [
        "What is the capital of France?",
        "Explain gravity.",
        "Name three programming languages.",
    ]

    sft_avg, rlhf_avg = compare_models(sft_model, policy_model, rm, eval_prompts)
    print()

    print("=" * 70)
    print("KL DIVERGENCE ANALYSIS")
    print("=" * 70)
    print()

    if kls:
        print(f"  Initial KL: {kls[0]:.4f}")
        print(f"  Final KL:   {kls[-1]:.4f}")
        print(f"  Max KL:     {max(kls):.4f}")
        kl_threshold = 0.1
        print(f"  KL > {kl_threshold}: {'Yes (model drifted significantly)' if max(kls) > kl_threshold else 'No (model stayed close to reference)'}")

交付它

本课产出 outputs/prompt-reward-model-designer.md——一个用于设计奖励模型训练 pipeline 的 prompt。给定目标行为(有用性、编码能力、安全性),它会生成数据收集协议、标注员指南和奖励模型评估标准。

练习

  1. 修改奖励模型,使用所有隐藏状态的均值而不是仅最后一个位置。比较准确率。均值池化让每个 token 权重相等,而最后位置方法依赖因果注意力聚合信息。在 6 个偏好对上测试,报告哪种方法准确率更高。
  1. 实现奖励模型校准。训练后,把所有偏好对输入奖励模型,计算:(a) 偏好回答的平均奖励,(b) 被拒回答的平均奖励,(c) 差值(偏好减被拒)。校准良好的模型应有明显差值。然后加入 4 个新的偏好对,检查差值在未见数据上是否保持。
  1. 模拟奖励黑客。创建一个给长回答高分的奖励模型(reward = len(response) / 100)。用这个有缺陷的奖励模型运行 PPO,观察策略模型生成越来越长、重复的输出。然后加入 0.1 的 KL 惩罚,展示它如何防止这种退化行为。
  1. 实现多目标奖励。训练两个奖励模型——一个针对有用性,一个针对简洁性。将它们组合为 R = 0.7 \cdot R_{\text{helpful}} + 0.3 \cdot R_{\text{concise}}。展示组合目标产生的回答既有用又简洁,避免单一有用性奖励的冗长陷阱。
  1. 比较不同的 KL 系数。用 \beta=0.001(太低,奖励黑客)、\beta=0.02(标准)和 \beta=0.5(太高,学不到东西)运行 PPO。绘制每种设置的奖励曲线和 KL 曲线。\beta=0.02 那一组应显示奖励稳步提升且 KL 有界。

关键术语

术语大家怎么说实际含义
RLHF「用人类反馈训练」Reinforcement Learning from Human Feedback:三阶段 pipeline(SFT、奖励模型、PPO),利用人类偏好信号优化语言模型输出
Reward model「给回答打分的模型」带标量输出头的 Transformer,使用 Bradley-Terry 损失在成对人类偏好上训练
Bradley-Terry「比较模型」一种概率模型,P(A > B) = \mathop{\mathrm{sigmoid}}(\text{score}(A) - \text{score}(B)),把成对偏好转换为一致的评分函数
PPO「那个 RL 算法」Proximal Policy Optimization:更新策略以最大化奖励,同时裁剪更新幅度以防不稳定
KL divergence「两个分布有多不同」策略模型 token 分布与参考模型 token 分布之间差异的度量——用作防止奖励黑客的惩罚
KL penalty「模型身上的牵引绳」从奖励信号中减去 \beta \cdot \mathrm{KL}(\text{policy} \Vert \text{reference})——防止策略偏离 SFT checkpoint 太远
Reward hacking「钻奖励的空子」策略利用奖励模型的弱点找到退化的高奖励输出,而非真正改进
Preference pair「A 和 B 哪个更好?」由 (prompt, preferred_response, rejected_response) 组成的训练样本——RLHF 训练数据的基本单位
Reference model「冻结的 SFT checkpoint」SFT 模型的副本,权重永不改变——用作 KL 散度计算的锚点

延伸阅读

“Stay hungry, stay foolish.”

— Steve Jobs

esc
↑↓切换↵打开esc关闭