RLHF 高级技术 #
概述 #
随着 RLHF 技术的发展，研究者们提出了多种变体和改进方法，旨在解决传统 RLHF 的局限性，提高训练效率和模型效果。
text
┌─────────────────────────────────────────────────────────────┐
│ RLHF 技术演进 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 传统方法： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ RLHF (2022) → 奖励模型 + PPO │ │
│ └─────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ 简化方法： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ DPO (2023) → 直接偏好优化 │ │
│ │ IPO (2023) → 恒等偏好优化 │ │
│ │ KTO (2023) → 卡尼曼-特沃斯基优化 │ │
│ │ ORPO (2024) → 几率比偏好优化 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ 扩展方法： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Constitutional AI → 规则驱动对齐 │ │
│ │ RLAIF → AI 反馈强化学习 │ │
│ │ Iterative DPO → 迭代式 DPO │ │
│ │ Multi-objective RLHF → 多目标 RLHF │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘
IPO（Identity Preference Optimization） #
原理 #
IPO 使用恒等函数替代 DPO 中的 sigmoid 函数，提供更稳定的训练：
text
DPO 损失：
────────────────────────
L_DPO = -log σ(β * (log_ratio_chosen - log_ratio_rejected))
IPO 损失：
────────────────────────
L_IPO = (log_ratio_chosen - log_ratio_rejected - 1/(2β))²
优势：
────────────────────────
├── 避免了 sigmoid 的饱和问题
├── 梯度更稳定
├── 对超参数更鲁棒
└── 理论上更优雅
实现 #
python
import torch
import torch.nn.functional as F
class IPOTrainer:
def __init__(self, model, ref_model, beta=0.1, learning_rate=1e-6):
self.model = model
self.ref_model = ref_model
self.beta = beta
self.optimizer = torch.optim.AdamW(model.parameters(), lr=learning_rate)
def compute_ipo_loss(self, log_probs_chosen, log_probs_rejected,
ref_log_probs_chosen, ref_log_probs_rejected):
log_ratio_chosen = log_probs_chosen - ref_log_probs_chosen
log_ratio_rejected = log_probs_rejected - ref_log_probs_rejected
diff = log_ratio_chosen - log_ratio_rejected
loss = (diff - 1 / (2 * self.beta)) ** 2
return loss.mean()
def train_step(self, batch):
log_probs_chosen = self.get_log_probs(
self.model, batch["input_ids_chosen"], batch["attention_mask_chosen"]
)
log_probs_rejected = self.get_log_probs(
self.model, batch["input_ids_rejected"], batch["attention_mask_rejected"]
)
with torch.no_grad():
ref_log_probs_chosen = self.get_log_probs(
self.ref_model, batch["input_ids_chosen"], batch["attention_mask_chosen"]
)
ref_log_probs_rejected = self.get_log_probs(
self.ref_model, batch["input_ids_rejected"], batch["attention_mask_rejected"]
)
loss = self.compute_ipo_loss(
log_probs_chosen, log_probs_rejected,
ref_log_probs_chosen, ref_log_probs_rejected
)
self.optimizer.zero_grad()
loss.backward()
self.optimizer.step()
return {"loss": loss.item()}
KTO（Kahneman-Tversky Optimization） #
原理 #
KTO 基于前景理论（Prospect Theory），不需要成对的偏好数据：
text
前景理论洞察：
────────────────────────
├── 人们对损失比收益更敏感
├── 损失厌恶：损失的痛苦大于同等收益的快乐
└── 可以单独处理好/坏样本
KTO 损失：
────────────────────────
对于好样本（desirable）：
L = λ_D * (1 - σ(β * log_ratio))
对于坏样本（undesirable）：
L = λ_U * σ(β * log_ratio)
其中 λ_D 和 λ_U 是权重系数
实现 #
python
class KTOTrainer:
def __init__(self, model, ref_model, beta=0.1,
desirable_weight=1.0, undesirable_weight=0.5):
self.model = model
self.ref_model = ref_model
self.beta = beta
self.desirable_weight = desirable_weight
self.undesirable_weight = undesirable_weight
def compute_kto_loss(self, log_probs, ref_log_probs, is_desirable):
log_ratio = log_probs - ref_log_probs
reward = self.beta * log_ratio
if is_desirable:
loss = self.desirable_weight * (1 - torch.sigmoid(reward))
else:
loss = self.undesirable_weight * torch.sigmoid(reward)
return loss.mean()
def train_step(self, batch):
log_probs = self.get_log_probs(
self.model, batch["input_ids"], batch["attention_mask"]
)
with torch.no_grad():
ref_log_probs = self.get_log_probs(
self.ref_model, batch["input_ids"], batch["attention_mask"]
)
is_desirable = batch["is_desirable"]
loss = self.compute_kto_loss(log_probs, ref_log_probs, is_desirable)
return {"loss": loss.item()}
ORPO（Odds Ratio Preference Optimization） #
原理 #
ORPO 将 SFT 和偏好学习统一到一个目标函数中：
text
ORPO 损失：
────────────────────────
L_ORPO = L_SFT + λ * L_preference
其中：
├── L_SFT = -log π_θ(y_w|x) （标准语言模型损失）
├── L_preference = -log σ(log_odds_ratio)
├── log_odds_ratio = log(π_θ(y_w|x) / π_θ(y_l|x))
└── λ 是偏好损失权重
优势：
────────────────────────
├── 无需参考模型
├── SFT 和对齐同时进行
├── 训练更简单
└── 效果与 DPO 相当
实现 #
python
class ORPOTrainer:
def __init__(self, model, lambda_orpo=0.1, learning_rate=1e-6):
self.model = model
self.lambda_orpo = lambda_orpo
self.optimizer = torch.optim.AdamW(model.parameters(), lr=learning_rate)
def compute_orpo_loss(self, log_probs_chosen, log_probs_rejected):
sft_loss = -log_probs_chosen.mean()
log_odds_ratio = log_probs_chosen - log_probs_rejected
preference_loss = -torch.log(torch.sigmoid(log_odds_ratio)).mean()
total_loss = sft_loss + self.lambda_orpo * preference_loss
return total_loss, sft_loss, preference_loss
def train_step(self, batch):
log_probs_chosen = self.get_log_probs(
self.model, batch["input_ids_chosen"], batch["attention_mask_chosen"]
)
log_probs_rejected = self.get_log_probs(
self.model, batch["input_ids_rejected"], batch["attention_mask_rejected"]
)
total_loss, sft_loss, preference_loss = self.compute_orpo_loss(
log_probs_chosen, log_probs_rejected
)
self.optimizer.zero_grad()
total_loss.backward()
self.optimizer.step()
return {
"total_loss": total_loss.item(),
"sft_loss": sft_loss.item(),
"preference_loss": preference_loss.item(),
}
Constitutional AI #
原理 #
Constitutional AI 使用规则（宪法）来指导模型对齐，减少对人工标注的依赖：
text
┌─────────────────────────────────────────────────────────────┐
│ Constitutional AI 流程 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 阶段 1：监督学习（Critique -> Revision） │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 1. 模型生成初始回复 │ │
│ │ 2. 根据"宪法"规则进行自我批评 │ │
│ │ 3. 根据批评修改回复 │ │
│ │ 4. 使用修改后的数据进行 SFT │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 阶段 2：RLHF（使用 AI 反馈） │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 1. 使用 AI 根据宪法评估回复 │ │
│ │ 2. 生成偏好数据 │ │
│ │ 3. 训练奖励模型 │ │
│ │ 4. 使用 RL 优化模型 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 宪法示例： │
│ ├── "选择最无害的回复" │
│ ├── "选择最乐于助人的回复" │
│ ├── "避免生成有害、非法或不道德的内容" │
│ └── "保持诚实，不要编造信息" │
│ │
└─────────────────────────────────────────────────────────────┘
实现 #
python
class ConstitutionalAI:
def __init__(self, model, tokenizer, constitution):
self.model = model
self.tokenizer = tokenizer
self.constitution = constitution
def generate_initial_response(self, prompt):
inputs = self.tokenizer(prompt, return_tensors="pt").to(self.model.device)
outputs = self.model.generate(
**inputs,
max_new_tokens=256,
temperature=0.7,
)
return self.tokenizer.decode(outputs[0], skip_special_tokens=True)
def critique_response(self, prompt, response, principle):
critique_prompt = f"""
原始问题：{prompt}
模型回复：{response}
请根据以下原则批评这个回复：
{principle}
请指出回复中的问题：
"""
return self.generate_initial_response(critique_prompt)
def revise_response(self, prompt, response, critique):
revision_prompt = f"""
原始问题：{prompt}
原始回复：{response}
批评意见：{critique}
请根据批评意见修改回复：
"""
return self.generate_initial_response(revision_prompt)
def generate_revised_data(self, prompts):
revised_data = []
for prompt in prompts:
initial_response = self.generate_initial_response(prompt)
for principle in self.constitution:
critique = self.critique_response(
prompt, initial_response, principle
)
revised = self.revise_response(
prompt, initial_response, critique
)
revised_data.append({
"prompt": prompt,
"response": revised,
})
return revised_data
RLAIF（RL from AI Feedback） #
原理 #
RLAIF 使用 AI 模型（而非人类）提供偏好反馈：
text
RLAIF 流程：
────────────────────────
1. 使用 SFT 模型生成多个回复
2. 使用 AI 评估器（如 GPT-4）对回复排序
3. 使用 AI 偏好数据训练奖励模型
4. 使用 PPO 优化策略模型
优势：
────────────────────────
├── 降低人工标注成本
├── 可扩展性强
├── 标注一致性高
└── 可以利用更强的模型
挑战：
────────────────────────
├── AI 评估器可能有偏见
├── 可能继承评估器的缺陷
├── 需要高质量的评估器
└── 评估成本仍然存在
实现 #
python
class RLAIF:
def __init__(self, policy_model, evaluator_model, tokenizer):
self.policy_model = policy_model
self.evaluator_model = evaluator_model
self.tokenizer = tokenizer
def generate_responses(self, prompt, num_responses=4):
inputs = self.tokenizer(prompt, return_tensors="pt").to(self.policy_model.device)
responses = []
for _ in range(num_responses):
outputs = self.policy_model.generate(
**inputs,
max_new_tokens=256,
temperature=1.0,
do_sample=True,
)
response = self.tokenizer.decode(
outputs[0][inputs["input_ids"].shape[1]:],
skip_special_tokens=True
)
responses.append(response)
return responses
def evaluate_with_ai(self, prompt, responses):
eval_prompt = f"""
问题：{prompt}
请对以下回复按质量从高到低排序（1最好，{len(responses)}最差）：
"""
for i, response in enumerate(responses):
eval_prompt += f"回复 {i+1}: {response}\n\n"
eval_prompt += "请只输出排序结果，格式如：3,1,4,2"
inputs = self.tokenizer(
eval_prompt, return_tensors="pt"
).to(self.evaluator_model.device)
outputs = self.evaluator_model.generate(
**inputs,
max_new_tokens=50,
temperature=0.0,
)
ranking_text = self.tokenizer.decode(
outputs[0][inputs["input_ids"].shape[1]:],
skip_special_tokens=True
)
return self._parse_ranking(ranking_text, len(responses))
def _parse_ranking(self, ranking_text, num_responses):
import re
numbers = re.findall(r'\d+', ranking_text)
ranking = [int(n) - 1 for n in numbers if 1 <= int(n) <= num_responses]
return ranking
def create_preference_pairs(self, prompt, responses, ranking):
pairs = []
for i in range(len(ranking)):
for j in range(i + 1, len(ranking)):
chosen_idx = ranking[i]
rejected_idx = ranking[j]
pairs.append({
"prompt": prompt,
"chosen": responses[chosen_idx],
"rejected": responses[rejected_idx],
})
return pairs
多目标 RLHF #
原理 #
多目标 RLHF 同时优化多个目标（如有用性、安全性、真实性）：
text
多目标优化：
────────────────────────
总奖励 = Σ w_i * r_i(x, y)
其中：
├── r_i 是第 i 个目标的奖励函数
├── w_i 是第 i 个目标的权重
└── 需要平衡不同目标之间的权衡
挑战：
────────────────────────
├── 目标之间可能冲突
├── 权重选择困难
├── 需要多维奖励模型
└── 评估更复杂
实现 #
python
class MultiObjectiveRLHF:
def __init__(self, policy_model, reward_models, weights):
self.policy_model = policy_model
self.reward_models = reward_models
self.weights = weights
def compute_multi_objective_reward(self, input_ids, attention_mask):
total_reward = 0
for reward_model, weight in zip(self.reward_models, self.weights):
reward = reward_model(input_ids, attention_mask)
total_reward += weight * reward
return total_reward
def compute_pareto_front(self, rewards_list):
pareto_front = []
for i, rewards in enumerate(rewards_list):
is_dominated = False
for j, other_rewards in enumerate(rewards_list):
if i != j:
if all(r <= o for r, o in zip(rewards, other_rewards)):
if any(r < o for r, o in zip(rewards, other_rewards)):
is_dominated = True
break
if not is_dominated:
pareto_front.append(i)
return pareto_front
迭代式 RLHF #
原理 #
迭代式 RLHF 通过多轮训练持续改进模型：
text
迭代流程：
────────────────────────
第 1 轮：
├── 使用初始模型生成回复
├── 收集偏好数据
├── 训练奖励模型
└── 使用 PPO/DPO 优化
第 2 轮：
├── 使用优化后的模型生成回复
├── 收集新的偏好数据
├── 更新奖励模型
└── 继续优化
...重复多轮
优势：
────────────────────────
├── 持续改进
├── 适应新的偏好
├── 提高数据效率
└── 更好的最终效果
实现 #
python
class IterativeRLHF:
def __init__(self, base_model, tokenizer, num_iterations=3):
self.base_model = base_model
self.tokenizer = tokenizer
self.num_iterations = num_iterations
self.current_model = base_model
def run_iteration(self, iteration, prompts):
print(f"Running iteration {iteration + 1}")
responses = self.generate_responses(prompts)
preference_data = self.collect_preferences(prompts, responses)
reward_model = self.train_reward_model(preference_data)
self.current_model = self.train_policy(
self.current_model, reward_model, prompts
)
return self.current_model
def run(self, prompts):
for i in range(self.num_iterations):
self.current_model = self.run_iteration(i, prompts)
return self.current_model
方法对比总结 #
text
┌─────────────────────────────────────────────────────────────┐
│ RLHF 变体方法对比 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 方法 │ 复杂度 │ 效果 │ 数据需求 │ 适用场景 │
│ ───────────────────────────────────────────────────────── │
│ RLHF │ 高 │ 优秀 │ 成对偏好 │ 大规模生产 │
│ DPO │ 低 │ 良好 │ 成对偏好 │ 快速原型 │
│ IPO │ 低 │ 良好 │ 成对偏好 │ 稳定训练 │
│ KTO │ 低 │ 良好 │ 单样本 │ 数据不完整 │
│ ORPO │ 最低 │ 良好 │ 成对偏好 │ 无参考模型 │
│ CAI │ 中 │ 良好 │ 规则 │ 减少人工 │
│ RLAIF │ 中 │ 良好 │ AI偏好 │ 成本敏感 │
│ │
└─────────────────────────────────────────────────────────────┘
下一步 #
现在你已经了解了 RLHF 的高级技术，接下来学习 实践指南，获取实际项目中的最佳实践经验！
最后更新：2026-04-05