图解ChatGPT强化学习过程：从PPO算法到奖励模型设计

3次阅读

共计 2947 个字符，预计需要花费 8 分钟才能阅读完成。

传统监督微调（Supervised Fine-Tuning, SFT）虽然能教会模型模仿人类回答，但存在三个致命缺陷：

缺乏全局优化目标：逐句匹配标准答案时，模型无法理解整体对话质量（如连贯性、信息量）
数据天花板效应：标注数据量有限时，模型容易过拟合到表面语言模式
价值观对齐困难：无法通过简单分类处理安全性、偏见等复杂语义问题

这时候就需要引入强化学习（Reinforcement Learning），让模型通过试错学习更接近人类偏好的对话策略。而 RLHF（Reinforcement Learning from Human Feedback）正是结合人类判断的强化学习框架。

用 Mermaid 绘制的三阶段训练流程如下：

graph TD
    A[预训练语言模型] -->| 监督微调 | B[SFT 模型]
    B -->| 人类排序数据 | C[奖励模型训练]
    C -->|PPO 优化 | D[RL 微调模型]

    subgraph 阶段 1: 监督微调
    A --> B
    end

    subgraph 阶段 2: 奖励建模
    B --> C
    style C fill:#ffe6e6
    end

    subgraph 阶段 3: RL 微调
    C --> D
    style D fill:#e6ffe6
    end

关键数据流说明：
1. 阶段 1 输入：海量对话语料（如 OpenAI 的 WebText）
2. 阶段 2 输入：人工标注的答案对（标注员选择更优回答）
3. 阶段 3 输入：SFT 模型生成 + 奖励模型打分

直接优化奖励会导致模型生成乱码（因为乱码可能意外获得高分）。解决方案是在损失函数中加入 KL 惩罚项：

$$\mathcal{L}^{\text{CLIP}}(\theta) = \mathbb{E}\left[\min\left(r_t(\theta)\hat{A}t, \text{clip}(r_t(\theta), 1-\epsilon, 1+\epsilon)\hat{A}_t\right)\right] – \beta \text{KL}[\pi\theta|\pi_{\text{SFT}}]$$

其中：
– $r_t(\theta) = \frac{\pi_\theta(a_t|s_t)}{\pi_{\text{old}}(a_t|s_t)}$ 为策略比例
– $\beta$ 一般取 0.1~0.2

对比学习框架：
输入：(好回答, 差回答) 对
输出：确保 $R(\text{好}) – R(\text{差}) > \text{margin}$
人工数据清洗：
剔除标注不一致样本（Inter-annotator disagreement）
平衡不同主题领域分布
分数标准化：
对每批数据做 Z -score 归一化
防止不同 batch 间奖励尺度漂移

import torch
from transformers import GPT2LMHeadModel

class PPOTrainer:
    def __init__(self, sft_model):
        self.policy = GPT2LMHeadModel.from_pretrained(sft_model)
        self.old_policy = GPT2LMHeadModel.from_pretrained(sft_model)
        self.kl_coef = 0.15  # KL 惩罚系数

    def compute_loss(self, batch):
        # batch 输入: input_ids, attention_mask, rewards
        outputs = self.policy(input_ids=batch['input_ids'],
            attention_mask=batch['attention_mask'],
            return_dict=True
        )
        logits = outputs.logits

        # 计算新旧策略概率比
        old_log_probs = self.old_policy(...).log_softmax(dim=-1)
        new_log_probs = torch.log_softmax(logits, dim=-1)
        ratio = (new_log_probs - old_log_probs).exp()

        # PPO-Clip 目标
        surr1 = ratio * batch['advantages']
        surr2 = torch.clamp(ratio, 1.0-0.2, 1.0+0.2) * batch['advantages']
        policy_loss = -torch.min(surr1, surr2).mean()

        # KL 散度惩罚
        kl = (old_log_probs - new_log_probs).mean()
        total_loss = policy_loss + self.kl_coef * kl

        return total_loss

from torch import nn

class RewardModel(nn.Module):
    def __init__(self, base_model):
        super().__init__()
        self.transformer = base_model.transformer  # 共享底层参数
        self.head = nn.Linear(base_model.config.n_embd, 1)

    def forward(self, input_ids):
        hidden = self.transformer(input_ids)[0]
        # 取 EOS token 对应的隐藏状态作为句子表示
        eos_pos = (input_ids == 50256).nonzero()[:, 1]
        pooled = hidden[torch.arange(hidden.size(0)), eos_pos]
        return self.head(pooled)

    def contrastive_loss(self, good_ids, bad_ids, margin=1.0):
        good_rewards = self(good_ids)
        bad_rewards = self(bad_ids)
        return torch.relu(margin - (good_rewards - bad_rewards)).mean()