RLHF（Reinforcement Learning from Human Feedback，人类反馈强化学习）是大语言模型对齐训练的核心技术路径，ChatGPT、Claude、Llama等主流模型均采用该方法实现模型输出与人类偏好的对齐。大模型开发中，单纯依靠预训练和监督微调（SFT）无法完全消除模型生成有害、偏颇或低质量内容的问题，RLHF通过引入人类偏好数据构建奖励模型，再使用PPO算法对策略模型进行优化，显著提升了模型输出的安全性和有用性。
RLHF三阶段训练流程与数据准备
RLHF训练分为三个阶段：监督微调（SFT）、奖励模型训练（RM）、强化学习优化（PPO）。每个阶段的数据格式和训练目标不同，需要分别准备。
阶段一：监督微调SFT数据格式
SFT阶段使用指令-响应对训练模型基础遵循能力。数据格式为JSONL，每条包含prompt和completion字段：
{"prompt": "解释什么是梯度消失问题", "completion": "梯度消失是指在深度神经网络反向传播过程中..."}
{"prompt": "用Python实现快速排序", "completion": "def quicksort(arr):\n if len(arr) <= 1:\n return arr..."}
SFT训练通常使用因果语言建模损失，只对completion部分计算loss：
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer, get_linear_schedule_with_warmup
model_name = "meta-llama/Llama-3-8B"
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForCausalLM.from_pretrained(model_name, torch_dtype=torch.bfloat16, device_map="auto")
# 构造SFT训练数据
def format_sft_sample(prompt, completion):
full_text = prompt + completion + tokenizer.eos_token
prompt_ids = tokenizer(prompt, return_tensors="pt").input_ids
full_ids = tokenizer(full_text, return_tensors="pt").input_ids
labels = full_ids.clone()
labels[:, :prompt_ids.shape[1]] = -100 # prompt部分不计算loss
return {"input_ids": full_ids[0], "labels": labels[0]}
# 训练配置
optimizer = torch.optim.AdamW(model.parameters(), lr=2e-5)
scheduler = get_linear_schedule_with_warmup(optimizer, num_warmup_steps=100, num_training_steps=10000)
for epoch in range(3):
for batch in dataloader:
outputs = model(input_ids=batch["input_ids"].to("cuda"), labels=batch["labels"].to("cuda"))
loss = outputs.loss
loss.backward()
torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0)
optimizer.step()
scheduler.step()
optimizer.zero_grad()奖励模型训练数据构造与Pairwise Ranking损失
奖励模型接收prompt和response作为输入，输出一个标量奖励分数。训练数据由标注员对同一prompt的多个模型输出进行排序，生成偏好对。
奖励模型使用Pairwise Ranking Loss，鼓励好的response获得更高分数：
from transformers import AutoModelForSequenceClassification
import torch.nn.functional as F
class RewardModel:
def __init__(self, model_name="meta-llama/Llama-3-8B"):
self.model = AutoModelForSequenceClassification.from_pretrained(
model_name, num_labels=1, torch_dtype=torch.bfloat16, device_map="auto"
)
self.tokenizer = AutoTokenizer.from_pretrained(model_name)
def compute_reward(self, prompt, response):
text = prompt + response
inputs = self.tokenizer(text, return_tensors="pt", truncation=True, max_length=2048)
with torch.no_grad():
score = self.model(input_ids=inputs["input_ids"].to("cuda")).logits
return score.item()
def train_step(self, prompt, chosen_response, rejected_response):
chosen_score = self.model(
input_ids=self.tokenizer(prompt + chosen_response, return_tensors="pt", truncation=True, max_length=2048)["input_ids"].to("cuda")
).logits
rejected_score = self.model(
input_ids=self.tokenizer(prompt + rejected_response, return_tensors="pt", truncation=True, max_length=2048)["input_ids"].to("cuda")
).logits
# Bradley-Terry pairwise loss
loss = -F.logsigmoid(chosen_score - rejected_score).mean()
return loss标注数据格式如下，每条包含prompt、chosen（好回答）、rejected（差回答）：
{"prompt": "如何防止SQL注入？", "chosen": "使用参数化查询和ORM框架，避免拼接SQL字符串...", "rejected": "在前端用正则过滤特殊字符就行了。"}PPO强化学习优化算法核心实现
PPO（Proximal Policy Optimization）是RLHF第三阶段的核心算法。训练过程中同时维护四个模型：策略模型（Actor）、参考模型（Reference，冻结的SFT模型）、奖励模型（Reward，冻结）、价值模型（Critic）。策略模型生成响应，奖励模型打分，通过KL散度惩罚防止策略模型偏离参考模型太远。
import torch
import torch.nn.functional as F
from copy import deepcopy
class PPOTrainer:
def __init__(self, actor_model, ref_model, reward_model, critic_model,
kl_coef=0.2, clip_ratio=0.2, gamma=1.0, lam=0.95):
self.actor = actor_model # 可训练策略模型
self.ref = ref_model # 冻结的参考模型
self.reward_model = reward_model # 冻结的奖励模型
self.critic = critic_model # 可训练价值模型
self.kl_coef = kl_coef
self.clip_ratio = clip_ratio
self.gamma = gamma
self.lam = lam
def compute_advantages(self, rewards, values):
"""GAE: Generalized Advantage Estimation"""
advantages = []
gae = 0
for t in reversed(range(len(rewards))):
if t == len(rewards) - 1:
next_value = 0
else:
next_value = values[t + 1]
delta = rewards[t] + self.gamma * next_value - values[t]
gae = delta + self.gamma * self.lam * gae
advantages.insert(0, gae)
advantages = torch.tensor(advantages)
advantages = (advantages - advantages.mean()) / (advantages.std() + 1e-8)
return advantages
def compute_kl_penalty(self, prompt, response):
"""计算策略模型与参考模型之间的KL散度"""
full_text = prompt + response
input_ids = self.tokenizer(full_text, return_tensors="pt").input_ids.to("cuda")
with torch.no_grad():
ref_logits = self.ref(input_ids).logits
ref_logprobs = F.log_softmax(ref_logits, dim=-1)
actor_logits = self.actor(input_ids).logits
actor_logprobs = F.log_softmax(actor_logits, dim=-1)
kl = (actor_logprobs.exp() * (actor_logprobs - ref_logprobs)).sum(dim=-1).mean()
return kl
def train_step(self, prompt, generated_response):
# 1. 奖励模型打分
with torch.no_grad():
reward = self.reward_model.compute_reward(prompt, generated_response)
kl = self.compute_kl_penalty(prompt, generated_response)
final_reward = reward - self.kl_coef * kl
# 2. 计算旧策略对数概率（用于PPO ratio）
with torch.no_grad():
old_logprobs = self._get_logprobs(self.actor, prompt, generated_response)
# 3. Critic预估价值
values = self.critic(prompt, generated_response)
# 4. GAE优势计算
advantages = self.compute_advantages([final_reward], [values])
# 5. PPO clipped objective
new_logprobs = self._get_logprobs(self.actor, prompt, generated_response)
ratio = torch.exp(new_logprobs - old_logprobs)
clipped_ratio = torch.clamp(ratio, 1 - self.clip_ratio, 1 + self.clip_ratio)
actor_loss = -torch.min(ratio * advantages, clipped_ratio * advantages).mean()
critic_loss = F.mse_loss(values, torch.tensor([final_reward]))
return actor_loss, critic_loss, final_reward.item()
def _get_logprobs(self, model, prompt, response):
full_text = prompt + response
input_ids = self.tokenizer(full_text, return_tensors="pt").input_ids.to("cuda")
logits = model(input_ids).logits
logprobs = F.log_softmax(logits, dim=-1)
# 只取response部分的logprob
prompt_len = self.tokenizer(prompt, return_tensors="pt").input_ids.shape[1]
response_logprobs = logprobs[0, prompt_len-1:-1, :]
# 取实际token的logprob
token_ids = input_ids[0, prompt_len:]
token_logprobs = response_logprobs.gather(-1, token_ids.unsqueeze(-1)).squeeze(-1)
return token_logprobs.sum()KL散度惩罚系数调参与训练稳定性
PPO训练中最常见的稳定性问题是奖励黑客（Reward Hacking）和模式崩溃（Mode Collapse）。KL散度惩罚系数kl_coef控制策略模型可以多大程度上偏离参考模型。kl_coef过小会导致模型为追求高奖励而生成不自然文本，过大则限制模型学习空间。
实际训练中常见的调参策略：
# 自适应KL系数调度
class AdaptiveKLScheduler:
def __init__(self, init_kl_coef=0.2, target_kl=10.0, kl_horizon=100):
self.kl_coef = init_kl_coef
self.target_kl = target_kl
self.kl_history = []
self.kl_horizon = kl_horizon
def step(self, current_kl):
self.kl_history.append(current_kl)
if len(self.kl_history) >= self.kl_horizon:
avg_kl = sum(self.kl_history[-self.kl_horizon:]) / self.kl_horizon
if avg_kl > self.target_kl * 1.5:
self.kl_coef *= 1.5 # KL过大，加大惩罚
elif avg_kl < self.target_kl * 0.5:
self.kl_coef *= 0.7 # KL过小，放松约束
self.kl_coef = max(0.05, min(self.kl_coef, 1.0))
# 混合精度训练减少显存占用
from torch.cuda.amp import autocast, GradScaler
scaler = GradScaler()
optimizer = torch.optim.AdamW(actor.parameters(), lr=1e-6)
for step, batch in enumerate(ppo_dataloader):
with autocast(dtype=torch.bfloat16):
actor_loss, critic_loss, reward = ppo_trainer.train_step(
batch["prompt"], batch["response"]
)
total_loss = actor_loss + 0.5 * critic_loss
scaler.scale(total_loss).backward()
scaler.unscale_(optimizer)
torch.nn.utils.clip_grad_norm_(actor.parameters(), 1.0)
scaler.step(optimizer)
scaler.update()
optimizer.zero_grad()
if step % 50 == 0:
print(f"Step {step} | Loss: {total_loss.item():.4f} | Reward: {reward:.4f}")DPO直接偏好优化作为RLHF替代方案
DPO（Direct Preference Optimization）绕过奖励模型和PPO训练，直接从偏好数据中优化策略模型，大幅简化了训练流程。DPO通过推导证明，最优奖励函数可以表示为策略模型与参考模型的对数概率之比，从而将RLHF问题转化为监督学习问题。
import torch.nn.functional as F
def dpo_loss(policy_model, ref_model, batch, beta=0.1):
"""DPO损失函数"""
prompt = batch["prompt"]
chosen = batch["chosen"]
rejected = batch["rejected"]
# 策略模型对chosen/rejected的对数概率
policy_chosen_logps = get_sequence_logps(policy_model, prompt, chosen)
policy_rejected_logps = get_sequence_logps(policy_model, prompt, rejected)
# 参考模型对chosen/rejected的对数概率
with torch.no_grad():
ref_chosen_logps = get_sequence_logps(ref_model, prompt, chosen)
ref_rejected_logps = get_sequence_logps(ref_model, prompt, rejected)
# DPO loss
chosen_logratios = policy_chosen_logps - ref_chosen_logps
rejected_logratios = policy_rejected_logps - ref_rejected_logps
loss = -F.logsigmoid(beta * (chosen_logratios - rejected_logratios)).mean()
# 计算准确率（监控指标）
chosen_rewards = beta * chosen_logratios.detach()
rejected_rewards = beta * rejected_logratios.detach()
acc = (chosen_rewards > rejected_rewards).float().mean()
return loss, acc.item()
def get_sequence_logps(model, prompt, response):
"""计算模型对response序列的对数概率"""
input_ids = tokenizer(prompt + response, return_tensors="pt").input_ids.to("cuda")
prompt_len = tokenizer(prompt, return_tensors="pt").input_ids.shape[1]
logits = model(input_ids).logits
logprobs = F.log_softmax(logits[:, :-1, :], dim=-1)
target_ids = input_ids[:, prompt_len:]
response_logps = logprobs[:, prompt_len-1:, :].gather(-1, target_ids.unsqueeze(-1)).squeeze(-1)
return response_logps.sum(dim=-1)DPO训练配置更简单，不需要奖励模型和价值模型，显存占用减少约60%。但PPO在复杂多轮对话和工具调用场景下仍有优势，两种方法的训练效果取决于具体任务和偏好数据质量。DeepSeek、Anthropic等团队在最新模型中使用RLHF的变体（如RLAIF、Constitutional AI）进一步扩展了人类反馈对齐的技术边界。
原创文章，作者：小编，如若转载，请注明出处：https://www.yunthe.com/rlhf-ren-lei-fan-kui-qiang-hua-xue-xi-dui-qi-xun-lian-quan/
