【大模型】- RLHF：基于人类反馈的强化学习

RLHF：基于人类反馈的强化学习

使用人类偏好训练语言模型

类型： 学习　｜　语言： Python　｜　🏷 前置：《RLHF》（本系列第 6 篇）

学习目标

理解 RLHF 的三个阶段
训练一个奖励模型来预测人类偏好
使用 PPO 优化策略模型
处理 RLHF 训练的不稳定性
评估 RLHF 模型的质量

RLHF 概述

RLHF 通过三个阶段使模型与人类价值观对齐：

监督微调（SFT）：在高质量的指令-回答对上微调模型
奖励模型训练：训练模型预测人类对回答的偏好
强化学习（PPO）：使用奖励模型通过强化学习优化策略

阶段 1：监督微调

使用 SFT 数据微调基础模型：

# 在指令数据上微调
model = AutoModelForCausalLM.from_pretrained("base_model")
# ... SFT 训练代码 ...
model.save_pretrained("sft_model")

阶段 2：奖励模型训练

偏好数据格式

偏好数据由一个 prompt 和两个回答组成，一个被人类偏好：

{
  "prompt": "解释量子力学",
  "chosen": "量子力学是物理学的一个分支...",
  "rejected": "我不知道量子力学是什么..."
}

奖励模型架构

奖励模型基于语言模型，但输出一个标量奖励：

from transformers import AutoModelForSequenceClassification

class RewardModel(nn.Module):
    def __init__(self, model_name):
        super().__init__()
        self.model = AutoModelForSequenceClassification.from_pretrained(
            model_name,
            num_labels=1  # 输出一个标量
        )
    
    def forward(self, input_ids, attention_mask):
        outputs = self.model(input_ids, attention_mask)
        return outputs.logits.squeeze(-1)

训练奖励模型

def train_reward_model(model, train_dataloader, config):
    optimizer = torch.optim.AdamW(model.parameters(), lr=config["lr"])
    
    for epoch in range(config["epochs"]):
        model.train()
        total_loss = 0
        
        for batch in train_dataloader:
            # 获取 chosen 和 rejected 的输入
            chosen_input = batch["chosen_input"].to(config["device"])
            chosen_mask = batch["chosen_mask"].to(config["device"])
            rejected_input = batch["rejected_input"].to(config["device"])
            rejected_mask = batch["rejected_mask"].to(config["device"])
            
            # 计算奖励
            chosen_rewards = model(chosen_input, chosen_mask)
            rejected_rewards = model(rejected_input, rejected_mask)
            
            # Bradley-Terry 损失
            loss = -torch.log(
                torch.sigmoid(chosen_rewards - rejected_rewards)
            ).mean()
            
            loss.backward()
            optimizer.step()
            optimizer.zero_grad()
            
            total_loss += loss.item()
        
        print(f"Epoch {epoch + 1}: Loss = {total_loss / len(train_dataloader):.4f}")
    
    return model

阶段 3：PPO 优化

PPO 算法概述

PPO（近端策略优化）是一种策略梯度方法：

1	策略模型 → 生成回答 → 奖励模型 → 奖励 → PPO 更新

PPO 实现

from transformers import PPOTrainer, PPOConfig
from trl import PPOTrainer, PPOConfig, AutoModelForCausalLMWithValueHead

def train_ppo(policy_model, ref_model, reward_model, train_dataloader, config):
    # 配置
    ppo_config = PPOConfig(
        learning_rate=config["lr"],
        batch_size=config["batch_size"],
        mini_batch_size=config["mini_batch_size"],
        ppo_epochs=config["ppo_epochs"],
        kl_penalty="kl",  # KL 散度惩罚
        init_kl_coef=0.2,
        target_kl=6.0,
    )
    
    # 价值头
    model = AutoModelForCausalLMWithValueHead.from_pretrained(policy_model)
    
    # PPO 训练器
    ppo_trainer = PPOTrainer(
        config=ppo_config,
        model=model,
        ref_model=ref_model,
        tokenizer=tokenizer
    )
    
    for batch in train_dataloader:
        query_tensors = batch["input_ids"]
        
        # 生成回答
        response_tensors = ppo_trainer.generate(
            query_tensors,
            max_new_tokens=256,
            temperature=0.7
        )
        
        # 计算奖励
        rewards = []
        for query, response in zip(query_tensors, response_tensors):
            input_text = tokenizer.decode(query)
            response_text = tokenizer.decode(response)
            
            reward_input = tokenizer(
                input_text + response_text,
                return_tensors="pt"
            ).to(config["device"])
            
            reward = reward_model(
                reward_input["input_ids"],
                reward_input["attention_mask"]
            )
            rewards.append(reward)
        
        # PPO 更新
        stats = ppo_trainer.step(
            query_tensors,
            response_tensors,
            rewards
        )
        
        print(f"Reward: {np.mean([r.item() for r in rewards]):.4f}")
        print(f"KL: {stats['ppo/kl']:.4f}")

KL 散度惩罚

防止策略偏离参考模型太远：

def compute_kl_penalty(policy_logprobs, ref_logprobs):
    # KL(π || π_ref)
    kl = policy_logprobs - ref_logprobs
    return kl.mean()

# 在 PPO 中使用
kl_penalty = kl_coef * compute_kl_penalty(policy_logprobs, ref_logprobs)
reward = reward - kl_penalty

训练稳定性

常见问题

奖励攻击：模型学会利用奖励模型的弱点
- 解决方案：使用 KL 惩罚，定期更新奖励模型
模式崩溃：模型只生成一种类型的回答
- 解决方案：增加多样性奖励，使用熵正则化
训练不稳定：奖励和损失波动剧烈
- 解决方案：使用较小的学习率，增加预热步骤

监控指标

metrics = {
    "reward_mean": np.mean(rewards),
    "reward_std": np.std(rewards),
    "kl_divergence": kl.item(),
    "policy_loss": policy_loss.item(),
    "value_loss": value_loss.item(),
    "response_length": np.mean([len(r) for r in response_tensors]),
}

wandb.log(metrics)

评估

def evaluate_rlhf(model, tokenizer, test_prompts):
    model.eval()
    results = []
    
    for prompt in test_prompts:
        inputs = tokenizer(prompt, return_tensors="pt").to(model.device)
        
        with torch.no_grad():
            outputs = model.generate(
                **inputs,
                max_new_tokens=512,
                temperature=0.7,
                do_sample=True
            )
        
        response = tokenizer.decode(outputs[0], skip_special_tokens=True)
        results.append({
            "prompt": prompt,
            "response": response
        })
    
    return results

完整训练流程

# 1. 加载 SFT 模型
sft_model = AutoModelForCausalLM.from_pretrained("sft_model")
ref_model = AutoModelForCausalLM.from_pretrained("sft_model")

# 2. 训练奖励模型
reward_model = RewardModel("sft_model")
reward_model = train_reward_model(reward_model, reward_dataloader, config)

# 3. PPO 训练
ppo_model = train_ppo(
    sft_model,
    ref_model,
    reward_model,
    ppo_dataloader,
    config
)

# 4. 保存最终模型
ppo_model.save_pretrained("rlhf_model")

总结

RLHF 通过三个阶段使模型与人类价值观对齐：SFT、奖励模型训练和 PPO 优化。关键是平衡奖励最大化和 KL 散度惩罚，以防止模式崩溃和奖励攻击。

下一步

下一课将介绍 DPO（直接偏好优化），一种简化的对齐方法。

📚 本文改编自 AI Engineering from Scratch（MIT License · 作者 Rohit Ghumare），中文内容来自官方中文镜像。原课程共 503 课 · 20 阶段 · 免费开源，教程网站见 aiengineeringfromscratch.com。

📝 自我检查（课程配套测验）

Q1（学前） RLHF 中的 reward model 从什么数据学习？

A. 原始文本文档
B. 人类偏好对：给定两个回答，人类更偏好哪一个
C. benchmark 分数
D. 模型 loss 曲线

答案： B　解析： reward model 在偏好数据上训练：同一 prompt 的两个回答，人类标注哪个更好。它学习给更符合人类偏好的回答打更高分。

Q2（学前） 为什么在 RLHF 的 PPO 训练中使用 KL 散度惩罚？

A. 加快训练
B. 防止 policy 偏离 SFT 模型过远，否则会导致 reward hacking
C. 减少内存占用
D. 改进 tokenization

答案： B　解析： 没有 KL 惩罚，模型会找到最大化 reward 分数的退化方式（例如产生重复文本以 exploit reward model 弱点）。KL 将模型约束在行为良好的 SFT baseline 附近。

Q3（学后） 完整 RLHF pipeline 需要几个独立模型？

A. 一个
B. 两个
C. 三个：SFT 模型、reward model、以及正在优化的 policy model
D. 四个

答案： C　解析： RLHF 需要：(1) SFT 模型作为起点和 KL 参考；(2) 在偏好数据上训练的 reward model；(3) 用 PPO 优化的 policy model。这种复杂性正是 DPO（第 08 课）被开发的原因。

Q4（学后） RLHF 中的「reward hacking」是什么？

A. reward model 被 adversary 攻击
B. policy 找到最大化 reward 分数的方式，但实际并未提升回答质量
C. 训练数据被污染
D. learning rate 过高

答案： B　解析： reward model 是人类判断的不完美代理。policy 可能发现高 reward 模式（例如冗长回答、过度 hedging）而并非真正更有帮助。KL 惩罚限制这种行为。

Q5（学后） PPO 的 clipping 机制防止什么？

A. gradient 溢出
B. 过大的 policy 更新导致训练不稳定
C. 内存溢出
D. 数据泄漏

答案： B　解析： PPO 将新旧 policy 的概率比裁剪到如 [0.8, 1.2] 的范围。这防止单次更新使 policy 变化过大，比 vanilla policy gradient 更稳定。