人类偏好对齐

本章目标：理解 SFT 之后的"灵魂注入"阶段，让模型输出符合人类意图

---

3.1 SFT 工程细节

3.1.1 对话模板

ChatML 格式（最常用）：

# ChatML 模板
def format_chatml(messages, tokenizer):
    """
    messages: [{"role": "user", "content": "..."}, ...]
    """
    result = ""
    for msg in messages:
        result += f"<|im_start|>{msg['role']}\n{msg['content']}<|im_end|>\n"
    result += "<|im_start|>assistant\n"
    return result

# 示例
# 输入:
# [
#   {"role": "system", "content": "你是助手"},
#   {"role": "user", "content": "讲个笑话"},
#   {"role": "assistant", "content": "为什么..."}
# ]
#
# 输出:
# <|im_start|>system
# 你是助手<|im_end|>
# <|im_start|>user
# 讲个笑话<|im_end|>
# <|im_start|>assistant
# 为什么...<|im_end|>

Alpaca 格式（LLaMA 早期使用）：

def format_alpaca(messages):
    result = "### Instruction:\n"
    for i, msg in enumerate(messages):
        if msg['role'] == 'user':
            result += f"{msg['content']}\n\n### Response:\n"
        elif msg['role'] == 'assistant':
            result += f"{msg['content']}\n"
    return result

为什么 ChatML 更好？

1. 特殊 Token 明确：<|im_start|> / <|im_end|> 明确标记角色边界
2. 支持多轮对话：天然支持任意轮次的对话
3. System Prompt 支持：独立的 system role

3.1.2 System Prompt 的影响

# System Prompt 示例
system_prompts = [
    # 基础
    "You are a helpful assistant.",
    
    # 详细描述
    "You are a helpful assistant. You can answer questions, write code, and help with tasks.",
    
    # 性格设定
    "You are a witty assistant who gives concise answers with occasional humor.",
    
    # 安全性
    "You are a helpful and harmless assistant. Refuse to help with requests that could cause harm.",
]

# System Prompt 会影响整个对话的风格和输出

3.1.3 长指令数据构造

# SFT 数据构造策略

# 1. 简单指令微调
simple_data = [
    {"instruction": "翻译为中文", "input": "Hello world", "output": "你好世界"},
    {"instruction": "总结这段文字", "input": "长文本...", "output": "简短总结..."},
]

# 2. CoT (Chain-of-Thought) 数据
cot_data = [
    {
        "instruction": "如果小红有5个苹果，小明给她3个，她卖掉了2个，还剩多少？",
        "input": "",
        "output": "解题步骤：\n1. 小红原有5个苹果\n2. 小明给她3个，现在有5+3=8个\n3. 卖掉2个，还剩8-2=6个\n答案是6个。"
    }
]

# 3. 对话数据
dialog_data = [
    {
        "messages": [
            {"role": "user", "content": "什么是量子计算？"},
            {"role": "assistant", "content": "量子计算是一种基于量子力学原理的计算方式..."},
            {"role": "user", "content": "能举个例子吗？"},
            {"role": "assistant", "content": "比如量子计算机可以使用量子叠加态同时探索多个可能性..."}
        ]
    }
]

---

3.2 RLHF 完整链路

3.2.1 RLHF 三阶段概述

┌─────────────────────────────────────────────────────────────┐
│                        RLHF 流程                             │
│                                                             │
│  ┌──────────┐     ┌──────────┐     ┌──────────┐           │
│  │   SFT    │────▶│ Reward   │────▶│   PPO    │           │
│  │   模型   │     │  模型    │     │  优化    │           │
│  └──────────┘     └──────────┘     └──────────┘           │
│       │                │                │                    │
│       ▼                ▼                ▼                    │
│  人工编写的       人类偏好           强化学习              │
│  高质量对话       数据训练           优化策略               │
│                                                             │
└─────────────────────────────────────────────────────────────┘

3.2.2 Reward Model 训练

Bradley-Terry 模型：将人类偏好建模为"更喜欢 A 还是 B"

class RewardModel(nn.Module):
    def __init__(self, base_model):
        super().__init__()
        self.base = base_model
        self.reward_head = nn.Linear(base_model.hidden_size, 1)
    
    def forward(self, input_ids, attention_mask):
        outputs = self.base(input_ids, attention_mask)
        # 使用最后一层 [CLS] 或最后一个 token 的表示
        reward = self.reward_head(outputs.last_hidden_state[:, -1])
        return reward.squeeze(-1)


def reward_model_loss(reward_chosen, reward_rejected):
    """
    Bradley-Terry 目标函数：
    L = -log(σ(r_chosen - r_rejected))
    
    即：让 chosen 的 reward 显著高于 rejected
    """
    return -F.logsigmoid(reward_chosen - reward_rejected).mean()


# 训练数据格式：(prompt, chosen_response, rejected_response)
def prepare_rm_batch(batch):
    # 分别编码 chosen 和 rejected
    chosen_ids = tokenizer(prompt + chosen_response)
    rejected_ids = tokenizer(prompt + rejected_response)
    
    chosen_reward = model(**chosen_ids)
    rejected_reward = model(**rejected_ids)
    
    loss = reward_model_loss(chosen_reward, rejected_reward)
    return loss

3.2.3 PPO 训练

PPO (Proximal Policy Optimization) 核心目标：

$$ L^{CLIP}(\theta) = \mathbb{E}_{t}\left[\min \left(r_t(\theta) \hat{A}_t, \text{clip}(r_t(\theta), 1-\epsilon, 1+\epsilon) \hat{A}_t\right)\right] $$

其中 $r_t(\theta) = \frac{\pi_\theta(a_t|s_t)}{\pi_{ref}(a_t|s_t)}$ 是概率比。

class PPOTrainer:
    def __init__(self, policy_model, ref_model, reward_model, config):
        self.policy = policy_model
        self.ref = ref_model
        self.reward = reward_model
        self.gamma = config.get('gamma', 1.0)
        self.lam = config.get('lam', 0.95)
        self.epsilon = config.get('epsilon', 0.2)  # PPO clip range
        self.kl_coef = config.get('kl_coef', 0.1)  # KL 惩罚系数
    
    def compute_advantages(self, rewards, values):
        """
        GAE (Generalized Advantage Estimation)
        """
        advantages = []
        last_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]
            last_gae = delta + self.gamma * self.lam * last_gae
            advantages.insert(0, last_gae)
        
        return torch.tensor(advantages)
    
    def ppo_loss(self, logprobs, old_logprobs, advantages):
        """
        PPO Clipped 损失
        """
        ratio = torch.exp(logprobs - old_logprobs)
        
        # Clipped 目标
        surr1 = ratio * advantages
        surr2 = torch.clamp(ratio, 1 - self.epsilon, 1 + self.epsilon) * advantages
        
        return -torch.min(surr1, surr2).mean()
    
    def kl_penalty(self, logprobs, ref_logprobs):
        """
        KL 散度惩罚：防止策略偏离参考模型太远
        L_KL = π_θ / π_ref
        """
        return (torch.exp(ref_logprobs - logprobs) * (ref_logprobs - logprobs)).mean()
    
    def step(self, queries, responses, rewards):
        """
        一个 PPO 更新步骤
        """
        # 1. 计算 ref model 的 logprobs
        with torch.no_grad():
            ref_logprobs = self.ref(queries, responses).logprobs
            old_values = self.value_model(queries, responses).values
        
        # 2. 计算 advantage
        advantages = self.compute_advantages(rewards, old_values)
        advantages = (advantages - advantages.mean()) / (advantages.std() + 1e-8)
        
        # 3. 策略模型前向
        policy_logprobs = self.policy(queries, responses).logprobs
        
        # 4. PPO 损失
        ppo_loss = self.ppo_loss(policy_logprobs, old_logprobs, advantages)
        
        # 5. KL 惩罚
        kl_loss = self.kl_penalty(policy_logprobs, ref_logprobs)
        
        # 6. 总损失
        total_loss = ppo_loss + self.kl_coef * kl_loss
        
        # 7. 反向传播和优化
        total_loss.backward()
        
        return {
            'ppo_loss': ppo_loss.item(),
            'kl_loss': kl_loss.item(),
            'total_loss': total_loss.item()
        }

3.2.4 Reward Hacking 的识别与缓解

Reward Hacking：模型找到"取悦"奖励模型但不真正有用的 trick

例子：
- 奖励模型可能认为越长越好 → 模型开始输出废话
- 奖励模型认为特定关键词重要 → 模型学会"关键词堆砌"
- 奖励模型偏好某种风格 → 模型过度迎合

缓解方法：

# 1. KL 约束（已在 PPO 中实现）
# π_θ 不要偏离 π_ref 太远

# 2. 对抗对抗样本
# 训练时故意加入容易被 hacking 的样本

# 3. 多元奖励信号
# 不只用单个 reward model，而是多个投票

# 4. 人类反馈迭代
# Iterative RLHF: 训练→评估→重新标注→再训练

---

3.3 DPO 及其变体

3.3.1 DPO (Direct Preference Optimization)

核心思想：绕过奖励模型，直接用偏好数据优化策略

DPO 的闭式解推导：

标准 RLHF:
- 最大化: E[r(s,a)] - β * KL(π || π_ref)
- 需要 reward model + PPO

DPO 观察到:
- 最优策略满足: π*(a|s) ∝ π_ref(a|s) * exp(r(s,a)/β)
- 重新整理可得: r(s,a) ∝ β * log(π*/π_ref) + const

所以可以用策略比直接估计 reward！

DPO 损失函数：

def dpo_loss(policy_logprobs_chosen, policy_logprobs_rejected,
             ref_logprobs_chosen, ref_logprobs_rejected,
             beta=0.1):
    """
    DPO 损失函数
    
    L = - E_{(x,y_c,y_r) ~ D}[
        log σ( β * (log π_θ(y_c|x) - log π_θ(y_r|x) -
                   log π_ref(y_c|x) + log π_ref(y_r|x)) )
    ]
    """
    # 策略的 log 概率差
    policy_logratio = policy_logprobs_chosen - policy_logprobs_rejected
    
    # 参考模型的 log 概率差
    ref_logratio = ref_logprobs_chosen - ref_logprobs_rejected
    
    # 加权 log 概率差
    logits = beta * (policy_logratio - ref_logratio)
    
    # NLL loss
    return -F.logsigmoid(logits).mean()


def dpo_train_step(policy_model, ref_model, batch, beta=0.1):
    chosen_ids = batch['chosen_ids']
    rejected_ids = batch['rejected_ids']
    
    # Policy 模型
    policy_chosen = policy_model(**chosen_ids)
    policy_rejected = policy_model(**rejected_ids)
    
    # Reference 模型
    with torch.no_grad():
        ref_chosen = ref_model(**chosen_ids)
        ref_rejected = ref_model(**rejected_ids)
    
    # 计算 DPO 损失
    loss = dpo_loss(
        policy_chosen.logprobs, policy_rejected.logprobs,
        ref_chosen.logprobs, ref_rejected.logprobs,
        beta=beta
    )
    
    loss.backward()
    return loss.item()

3.3.2 IPO (Identity Preference Optimization)

DPO 的问题：可能过拟合，导致策略崩溃

IPO 解决方案：增加正则项

def ipo_loss(policy_logprobs_chosen, policy_logprobs_rejected,
             ref_logprobs_chosen, ref_logprobs_rejected,
             beta=0.1):
    """
    IPO 损失函数
    
    L = E[ (log(π_θ(y_c)/π_θ(y_r)) - log(π_ref(y_c)/π_ref(y_r)) - 1/β )² ]
    
    加了 1/β 常数项，作为恒等映射的正则
    """
    policy_logratio = policy_logprobs_chosen - policy_logprobs_rejected
    ref_logratio = ref_logprobs_chosen - ref_logprobs_rejected
    
    # IPO 特有的差异
    diff = (policy_logratio - ref_logratio) - (1.0 / beta)
    
    return (diff ** 2).mean()

3.3.3 KTO (Kahneman-Tversky Optimization)

核心思想：不需要成对偏好数据，只需判断"喜欢/不喜欢"

def kto_loss(policy_logprobs, ref_logprobs, is_preferred, beta=0.1):
    """
    KTO 损失
    
    - 不需要成对数据
    - 可以处理单个样本的"喜欢/不喜欢"标签
    
    损失设计基于前景理论：
    - 人类对损失的敏感度高于等量收益
    - KTO 借鉴这个思想，让模型更怕犯错
    """
    # 对于 preferred 样本：希望策略概率高于参考
    # 对于 non-preferred 样本：希望策略概率低于参考
    
    log_ratio = policy_logprobs - ref_logprobs
    
    if is_preferred:
        # 喜欢：最大化正差异
        loss = -F.logsigmoid(beta * log_ratio)
    else:
        # 不喜欢：最小化负差异
        loss = -F.logsigmoid(-beta * log_ratio)
    
    return loss.mean()

3.3.4 对比总结

方法	数据需求	稳定性	实现复杂度	效果
RLHF+PPO	成对偏好	中	高	最好（通常）
DPO	成对偏好	中	低	接近 RLHF
IPO	成对偏好	好	低	更稳定
KTO	单样本偏好	好	低	灵活

---

3.4 在线对齐方法

3.4.1 Iterative DPO (DPO-IT)

def iterative_dpo_train(num_iterations, train_data, base_model):
    """
    Iterative DPO: 每次迭代重新收集偏好数据
    """
    model = copy.deepcopy(base_model)
    
    for iteration in range(num_iterations):
        # 1. 用当前模型生成 responses
        responses = model.generate_responses(train_data['prompts'])
        
        # 2. 收集偏好（可以用 reward model 或人工）
        preferences = reward_model.get_preferences(
            train_data['prompts'], 
            responses['chosen'],
            responses['rejected']
        )
        
        # 3. DPO 训练
        for batch in dataloader:
            loss = dpo_train_step(model, base_model, batch)
            # ...
        
        # 4. 评估
        evaluate(model)
    
    return model

3.4.2 Self-Play (SPIN)

核心思想：模型同时扮演"学生"和"老师"

def spin_train_step(policy_model, old_model, batch, beta=0.1):
    """
    SPIN: Self-Play Preference Optimization
    
    - 从 old_model 采样 negative samples
    - 让 policy 学习区分自己和 old_model
    - 随着训练，old_model 逐渐被当前 policy 替代
    """
    # 1. 当前模型生成 chosen
    chosen = policy_model.generate(batch['prompts'])
    
    # 2. old_model 生成 rejected
    with torch.no_grad():
        rejected = old_model.generate(batch['prompts'])
    
    # 3. 最大化 chosen - rejected 的差异
    chosen_logprob = policy_model.logprob(batch['prompts'], chosen)
    rejected_logprob = policy_model.logprob(batch['prompts'], rejected)
    
    # 目标：chosen 的 logprob 高，rejected 的 logprob 低
    loss = -F.logsigmoid(chosen_logprob - rejected_logprob)
    
    return loss.mean()

3.4.3 RLAIF (AI Feedback)

核心思想：用另一个 LLM 代替人类提供偏好

class LLM-as-Judge:
    def __init__(self, judge_model):
        self.judge = judge_model
    
    def compare(self, prompt, response_a, response_b):
        """
        让 judge LLM 判断哪个响应更好
        """
        judgement_prompt = f"""Compare these two responses to the user query.

Query: {prompt}

Response A: {response_a}

Response B: {response_b}

Which is better? Answer A or B."""
        
        answer = self.judge.generate(judgement_prompt)
        
        # 解析答案
        if "A" in answer and "B" not in answer:
            return {"preferred": "A", "rejected": "B"}
        elif "B" in answer and "A" not in answer:
            return {"preferred": "B", "rejected": "A"}
        else:
            return None  # 不确定

---

3.5 本章小结

┌─────────────────────────────────────────────────────────────┐
│                    对齐技术全景图                              │
├─────────────────────────────────────────────────────────────┤
│  SFT        │  有监督微调，用人工编写的高质量数据               │
│             │  对话模板：ChatML vs Alpaca                     │
├─────────────────────────────────────────────────────────────┤
│  RLHF       │  Reward Model + PPO                            │
│             │  Bradley-Terry 偏好建模                         │
│             │  KL 约束防止策略偏离                             │
├─────────────────────────────────────────────────────────────┤
│  DPO        │  直接偏好优化，绕过 PPO                          │
│  IPO        │  加正则项，更稳定                               │
│  KTO        │  不需要成对数据                                 │
├─────────────────────────────────────────────────────────────┤
│  在线对齐    │  Iterative DPO: 迭代收集偏好                    │
│             │  SPIN: 自博弈                                   │
│             │  RLAIF: 用 LLM 当 judge                        │
└─────────────────────────────────────────────────────────────┘

面试高频问题

1. RLHF vs DPO 区别？ → RLHF 需要单独的 reward model + PPO；DPO 直接用偏好数据优化策略

2. DPO 为什么比 PPO 稳定？ → PPO 需要同时优化 critic 和 actor；DPO 只优化 actor，超参数少

3. Reward Hacking 怎么解决？ → KL 约束限制策略偏移；多元奖励信号；对抗样本训练

4. 为什么需要 KL 散度惩罚？ → 防止策略模型完全偏离 SFT 模型，避免"过度迎合" reward model