RLHF实现2026版：人类反馈强化学习的现代实践

1. SFT（监督微调）: 用高质量数据微调基础模型
2. RM（奖励模型）: 训练奖励模型预测人类偏好
3. RL（强化学习）: 用PPO等算法优化策略

from trl import SFTTrainer, SFTConfig

config = SFTConfig(
    output_dir="./sft",
    num_train_epochs=3,
    per_device_train_batch_size=4,
    learning_rate=2e-5,
    max_seq_length=2048,
    bf16=True,
)

trainer = SFTTrainer(
    model=base_model,
    args=config,
    train_dataset=sft_dataset,
)
trainer.train()

{"prompt": "...", "chosen": "好回答", "rejected": "差回答"}

from trl import RewardTrainer, RewardConfig

# 奖励模型通常用SFT模型初始化
rm_model = AutoModelForSequenceClassification.from_pretrained(
    "sft-model", num_labels=1
)

config = RewardConfig(
    output_dir="./reward-model",
    num_train_epochs=1,
    per_device_train_batch_size=16,
    learning_rate=5e-6,
    max_length=2048,
)

trainer = RewardTrainer(
    model=rm_model,
    args=config,
    train_dataset=preference_dataset,
)
trainer.train()

from trl import PPOTrainer, PPOConfig

config = PPOConfig(
    output_dir="./ppo",
    learning_rate=1.46e-5,
    batch_size=32,
    mini_batch_size=4,
    ppo_epochs=4,
    cliprange=0.2,
    beta=0.05,  # KL惩罚系数
)

ppo_trainer = PPOTrainer(
    model=sft_model,        # 策略模型
    ref_model=sft_model,    # 参考模型（冻结）
    reward_model=rm_model,  # 奖励模型
    args=config,
    tokenizer=tokenizer,
    train_dataset=ppo_dataset,
)

for batch in dataloader:
    # 1. 策略模型生成回答
    responses = ppo_trainer.generate(batch["prompt"])
    
    # 2. 奖励模型打分
    rewards = reward_model.score(batch["prompt"], responses)
    
    # 3. PPO更新
    stats = ppo_trainer.step(batch["prompt"], responses, rewards)

from trl import RLOOTrainer, RLOOConfig

config = RLOOConfig(
    output_dir="./rloo",
    learning_rate=1e-6,
    batch_size=16,
    rloo_epochs=4,
)

class OnlineRLHF:
    async def train_step(self):
        # 1. 生成回答
        responses = await self.model.generate(prompts)
        
        # 2. 实时获取人类反馈
        preferences = await self.collect_human_preferences(prompts, responses)
        
        # 3. 更新奖励模型
        await self.reward_model.train(preferences)
        
        # 4. 更新策略
        rewards = await self.reward_model.score(prompts, responses)
        await self.ppo_trainer.step(prompts, responses, rewards)

class RLAIF:
    async def get_ai_preference(self, prompt, response_a, response_b):
        """使用AI评判偏好"""
        prompt = f"""
        问题: {prompt}
        回答A: {response_a}
        回答B: {response_b}
        
        哪个回答更好？输出JSON: {{"winner": "A"/"B", "reason": "..."}}
        """
        return await self.judge_model.call(prompt)

# 监控关键指标
def monitor_training(stats):
    # KL散度不应太大
    if stats["kl"] > 10:
        logger.warning("KL divergence too high, increasing beta")
        config.beta *= 1.5
    
    # 奖励不应下降
    if stats["reward"] < previous_reward:
        logger.warning("Reward decreasing, reducing learning rate")
        config.learning_rate *= 0.5