Python# RLHF 三阶段训练流程 def rlhf_pipeline(base_model, pref_data): # Stage 1: SFT — 监督微调 sft_model = supervised_finetune( base_model, instruction_data ) # Stage 2: RM — 奖励模型训练 reward_model = train_reward_model( sft_model, pref_data ) # Bradley-Terry loss # Stage 3: PPO — 强化学习优化 aligned_model = ppo_optimize( policy=sft_model, reward_model=reward_model, ref_model=copy(sft_model), kl_coef=0.02 ) return aligned_model

Python# Bradley-Terry 偏好模型 import torch import torch.nn.functional as F def bt_loss(reward_model, chosen_ids, rejected_ids): r_chosen = reward_model(chosen_ids) # r(x_chosen) r_rejected = reward_model(rejected_ids) # r(x_rejected) # P(chosen > rejected) = sigmoid(r_c - r_r) logits = r_chosen - r_rejected loss = -F.logsigmoid(logits).mean() return loss # 奖励分数排序准确率 def accuracy(r_c, r_r): return (r_c > r_r).float().mean().item()

Python# PPO Clipped Surrogate Objective def ppo_loss(policy, ref, reward, values, old_logp, states, actions, beta=0.02, eps=0.2): # 比率 ratio = pi(a|s) / pi_old(a|s) logp = policy.log_prob(states, actions) ratio = torch.exp(logp - old_logp) # 优势函数 (GAE) advantages = compute_gae(reward, values) advantages = (advantages - advantages.mean()) / \ (advantages.std() + 1e-8) # 裁剪目标 surr1 = ratio * advantages surr2 = torch.clamp(ratio, 1-eps, 1+eps) * advantages policy_loss = -torch.min(surr1, surr2).mean() # KL 惩罚 kl = kl_divergence(policy, ref, states).mean() return policy_loss + beta * kl

Python# Direct Preference Optimization import torch.nn.functional as F def dpo_loss(policy, ref, chosen, rejected, beta=0.1): # 隐式奖励: r = beta * log(pi / pi_ref) pi_c = policy.log_prob(chosen) pi_r = policy.log_prob(rejected) ref_c = ref.log_prob(chosen).detach() ref_r = ref.log_prob(rejected).detach() log_ratio_c = pi_c - ref_c # log(pi_c/pi_ref_c) log_ratio_r = pi_r - ref_r # DPO 损失 = -log sigmoid(beta * (r_c - r_r)) logits = beta * (log_ratio_c - log_ratio_r) loss = -F.logsigmoid(logits).mean() return loss # 无需奖励模型, 直接从偏好学习

JSON// pairwise 偏好数据格式 { "prompt": "解释什么是量子纠缠", "chosen": [ {"role": "user", "content": "..."}, {"role": "assistant", "content": "量子纠缠是..."} ], "rejected": [ {"role": "user", "content": "..."}, {"role": "assistant", "content": "纠缠就是..."} ], "metadata": { "annotator_id": 42, "agreement": 0.92, "source": "human" } } // listwise: prompt + ranked_responses[] // pointwise: prompt + response + score

Python# KL 散度惩罚 import torch.nn.functional as F def kl_penalty(policy, ref_model, states, beta=0.02): # 逐 token KL 散度 logp = policy.log_prob(states) # log pi_theta logp_ref = ref_model.log_prob(states) # log pi_ref p = torch.exp(logp) # KL(pi || pi_ref) = sum p * (logp - logp_ref) kl = (p * (logp - logp_ref)).sum(dim=-1).mean() # 总目标 = reward - beta * KL return beta * kl # reward_hacking 防御: # 高 reward + 低 KL = 安全优化 # 高 reward + 高 KL = 可能 hacking

Python# 安全分类与拦截 from transformers import AutoModelForSequenceClassification HARMFUL_CATEGORIES = [ "violence", "self_harm", "hate_speech", "sexual", "illegal", "deception", ] def safety_check(prompt, response, threshold=0.85): # 多层防御: 输入检测 + 输出检测 input_risk = classifier.predict(prompt) output_risk = classifier.predict(response) if input_risk.max() > threshold: return refuse( "抱歉，我无法协助此类请求。" ) if output_risk.max() > threshold: return rewrite(response) return response # 安全通过

Python# RLHF 评估基准 def evaluate(model, benchmarks): results = {} for bench in benchmarks: if bench == "mt_bench": # 多轮对话, GPT-4 评分 results[bench] = mt_bench_score( model, judge="gpt-4" ) elif bench == "alpaca_eval": # 胜率 vs reference results[bench] = alpaca_winrate( model, ref="text-davinci-003" ) elif bench == "arena_elo": # 人群盲评 Elo 排名 results[bench] = compute_elo( model, human_votes ) elif bench == "rm_acc": # 奖励模型排序准确率 results[bench] = rm_accuracy( reward_model, test_set ) return results # 人类评估 vs 自动指标 # correlation = pearson(human, auto)