RLHF 基础概念 #

┌─────────────────────────────────────────────────────────────┐
│                    强化学习框架                               │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│                    ┌─────────────┐                          │
│                    │   环境      │                          │
│                    │ Environment │                          │
│                    └─────────────┘                          │
│                      ↑         │                            │
│                      │ 状态 S  │ 奖励 R                      │
│                      │         ↓                            │
│                    ┌─────────────┐                          │
│                    │   智能体    │                          │
│                    │   Agent     │                          │
│                    └─────────────┘                          │
│                          │                                  │
│                          │ 动作 A                           │
│                          ↓                                  │
│                                                             │
│  核心要素：                                                  │
│  ├── 状态（State）：环境的当前情况                           │
│  ├── 动作（Action）：智能体可以执行的操作                    │
│  ├── 奖励（Reward）：环境对动作的反馈                        │
│  └── 策略（Policy）：智能体的行为策略                        │
│                                                             │
└─────────────────────────────────────────────────────────────┘

MDP 定义：
├── 状态空间 S：所有可能的状态
├── 动作空间 A：所有可能的动作
├── 转移概率 P(s'|s,a)：在状态 s 执行动作 a 后转移到 s' 的概率
├── 奖励函数 R(s,a)：在状态 s 执行动作 a 获得的奖励
└── 折扣因子 γ：未来奖励的折扣系数

目标：找到最优策略 π*，使得期望累积奖励最大

π* = argmax E[Σ γ^t R(s_t, a_t)]

┌─────────────────────────────────────────────────────────────┐
│                语言模型的 RL 建模                             │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  状态（State）：                                             │
│  ├── 当前已生成的 token 序列                                 │
│  └── 输入提示（prompt）                                      │
│                                                             │
│  动作（Action）：                                            │
│  └── 下一个要生成的 token                                    │
│                                                             │
│  奖励（Reward）：                                            │
│  ├── 奖励模型给出的评分                                      │
│  └── 通常在序列结束时给出                                    │
│                                                             │
│  策略（Policy）：                                            │
│  └── 语言模型本身 π(a|s) = P(token|context)                 │
│                                                             │
│  轨迹（Trajectory）：                                        │
│  └── 完整的生成序列 [token_1, token_2, ..., token_n]        │
│                                                             │
└─────────────────────────────────────────────────────────────┘

策略梯度定理：
────────────────────────
∇J(θ) = E[∇log π_θ(a|s) * Q^π(s,a)]

直观理解：
- 增加高奖励动作的概率
- 降低低奖励动作的概率
- 通过采样估计期望

REINFORCE 算法：
────────────────────────
θ = θ + α * ∇log π_θ(a|s) * G_t

其中 G_t 是从时刻 t 开始的累积奖励

┌─────────────────────────────────────────────────────────────┐
│                    偏好数据结构                               │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  单条偏好数据：                                              │
│  {                                                          │
│    "prompt": "用户输入的问题或指令",                         │
│    "chosen": "更好的回复",                                   │
│    "rejected": "较差的回复",                                 │
│    "metadata": {                                            │
│      "annotator_id": "标注者ID",                            │
│      "timestamp": "标注时间",                               │
│      "confidence": "标注置信度"                             │
│    }                                                        │
│  }                                                          │
│                                                             │
│  多选排序数据：                                              │
│  {                                                          │
│    "prompt": "用户输入",                                     │
│    "responses": [                                           │
│      {"text": "回复1", "rank": 1},                          │
│      {"text": "回复2", "rank": 2},                          │
│      {"text": "回复3", "rank": 3},                          │
│      {"text": "回复4", "rank": 4}                           │
│    ]                                                        │
│  }                                                          │
│                                                             │
└─────────────────────────────────────────────────────────────┘

步骤 1：提示收集
────────────────────────
├── 从用户交互中收集
├── 人工设计多样化提示
├── 覆盖不同任务类型
└── 确保分布合理

步骤 2：回复生成
────────────────────────
├── 使用 SFT 模型生成多个回复
├── 调整温度参数增加多样性
├── 通常生成 4-9 个候选
└── 记录生成参数

步骤 3：人工标注
────────────────────────
├── 标注者对回复排序
├── 提供标注指南
├── 多人标注取共识
└── 质量控制和审核

步骤 4：数据处理
────────────────────────
├── 清洗异常数据
├── 标注一致性检查
├── 构建训练数据集
└── 划分训练/验证/测试集

标注维度：
├── 有用性（Helpfulness）
│   └── 回复是否解决了用户的问题
│
├── 真实性（Truthfulness）
│   └── 信息是否准确、有无幻觉
│
├── 安全性（Safety）
│   └── 是否包含有害内容
│
├── 连贯性（Coherence）
│   └── 回复是否逻辑清晰、语言流畅
│
└── 指令遵循（Instruction Following）
    └── 是否正确理解和执行指令

标注规则：
├── 明确各维度的权重
├── 提供正反面示例
├── 定义边界情况处理
└── 定期校准标注标准

质量控制措施：
├── 标注者培训
│   ├── 详细培训材料
│   ├── 标注练习和测试
│   └── 定期反馈和校准
│
├── 多人标注
│   ├── 每条数据多人标注
│   ├── 计算标注一致性
│   └── 仲裁不一致样本
│
├── 自动检测
│   ├── 异常检测算法
│   ├── 标注者可信度评估
│   └── 自动过滤低质量数据
│
└── 持续监控
    ├── 监控标注质量指标
    ├── 定期审核
    └── 反馈改进标注流程

┌─────────────────────────────────────────────────────────────┐
│                    奖励函数的作用                             │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  引导优化方向：                                              │
│  ┌─────────────────────────────────────────────────────┐   │
│  │  高奖励 → 增加该行为的概率                           │   │
│  │  低奖励 → 降低该行为的概率                           │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
│  编码人类偏好：                                              │
│  ┌─────────────────────────────────────────────────────┐   │
│  │  奖励模型学习人类偏好模式                            │   │
│  │  将复杂的价值观编码为标量奖励                        │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
│  平衡多目标：                                                │
│  ┌─────────────────────────────────────────────────────┐   │
│  │  R_total = α * R_helpfulness                        │   │
│  │          + β * R_safety                             │   │
│  │          + γ * R_truthfulness                       │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
└─────────────────────────────────────────────────────────────┘

Bradley-Terry 模型：
────────────────────────
假设回复 y_w 比 y_l 好的概率：

P(y_w > y_l | x) = σ(r(x, y_w) - r(x, y_l))

其中：
├── x 是提示
├── y_w 是被选择的回复
├── y_l 是被拒绝的回复
├── r(x, y) 是奖励函数
└── σ 是 sigmoid 函数

损失函数：
────────────────────────
L = -E[log σ(r(x, y_w) - r(x, y_l))]

目标：最大化被选择回复获得更高奖励的概率

import torch
import torch.nn as nn
from transformers import AutoModelForCausalLM

class RewardModel(nn.Module):
    def __init__(self, base_model_name, hidden_size=4096):
        super().__init__()
        self.base_model = AutoModelForCausalLM.from_pretrained(
            base_model_name
        )
        self.value_head = nn.Sequential(
            nn.Linear(hidden_size, hidden_size // 2),
            nn.ReLU(),
            nn.Linear(hidden_size // 2, 1)
        )
    
    def forward(self, input_ids, attention_mask=None):
        outputs = self.base_model(
            input_ids=input_ids,
            attention_mask=attention_mask,
            output_hidden_states=True
        )
        last_hidden = outputs.hidden_states[-1]
        last_token_hidden = last_hidden[:, -1, :]
        reward = self.value_head(last_token_hidden)
        return reward

原始奖励：
────────────────────────
r(x, y) = 奖励模型输出

加入 KL 约束：
────────────────────────
r_total(x, y) = r(x, y) - β * KL(π_θ || π_ref)

其中：
├── π_θ 是当前策略
├── π_ref 是参考策略（SFT 模型）
├── β 是 KL 惩罚系数
└── 防止模型偏离太远

多目标奖励：
────────────────────────
r_total = α * r_preference
        + β * r_safety
        + γ * r_format
        - δ * KL_penalty

问题：奖励模型不完美
────────────────────────
├── 奖励模型可能存在漏洞
├── 模型可能学会"欺骗"奖励模型
├── 生成奖励高但实际质量差的输出
└── 这种现象称为"奖励黑客"（Reward Hacking）

问题：语言能力退化
────────────────────────
├── 过度优化可能导致语言能力下降
├── 生成不自然或重复的内容
├── 失去预训练获得的知识
└── 模型变得"过于安全"而失去有用性

解决方案：KL 约束
────────────────────────
├── 限制策略与参考模型的偏离程度
├── 保持语言能力的同时优化偏好
├── 平衡探索和利用
└── 提高训练稳定性

KL 散度定义：
────────────────────────
KL(P || Q) = Σ P(x) * log(P(x) / Q(x))

在 RLHF 中：
────────────────────────
KL(π_θ || π_ref) = E_x,y [log π_θ(y|x) - log π_ref(y|x)]

实际计算：
────────────────────────
对于序列 y = [y_1, y_2, ..., y_n]：

KL = Σ_t [log π_θ(y_t|y_<t, x) - log π_ref(y_t|y_<t, x)]

def compute_kl_penalty(policy_logprobs, ref_logprobs):
    kl = policy_logprobs - ref_logprobs
    return kl.mean()

def rlhf_loss(policy_model, ref_model, reward_model, 
              input_ids, beta=0.1):
    policy_outputs = policy_model(input_ids)
    ref_outputs = ref_model(input_ids)
    
    policy_logprobs = policy_outputs.log_probs
    ref_logprobs = ref_outputs.log_probs
    
    reward = reward_model(input_ids)
    kl_penalty = compute_kl_penalty(policy_logprobs, ref_logprobs)
    
    loss = -(reward - beta * kl_penalty).mean()
    return loss

状态价值函数 V(s)：
────────────────────────
从状态 s 开始，遵循策略 π 的期望累积奖励

V^π(s) = E[Σ γ^t r_t | s_0 = s]

动作价值函数 Q(s,a)：
────────────────────────
在状态 s 执行动作 a，然后遵循策略 π 的期望累积奖励

Q^π(s,a) = E[Σ γ^t r_t | s_0 = s, a_0 = a]

在 RLHF 中：
────────────────────────
├── 状态：已生成的 token 序列
├── 价值函数估计未来奖励的期望
├── 用于计算优势函数
└── 帮助减少方差

优势函数定义：
────────────────────────
A(s,a) = Q(s,a) - V(s)

直观理解：
────────────────────────
├── A > 0：该动作比平均好
├── A < 0：该动作比平均差
└── A = 0：该动作是平均水平

GAE（Generalized Advantage Estimation）：
────────────────────────
A_t = Σ_l (γλ)^l δ_{t+l}

其中 δ_t = r_t + γV(s_{t+1}) - V(s_t)

GAE 平衡偏差和方差：
├── λ = 0：低方差，高偏差
├── λ = 1：高方差，低偏差
└── 通常 λ = 0.95

class ValueModel(nn.Module):
    def __init__(self, base_model_name, hidden_size=4096):
        super().__init__()
        self.base_model = AutoModelForCausalLM.from_pretrained(
            base_model_name
        )
        self.value_head = nn.Linear(hidden_size, 1)
    
    def forward(self, input_ids, attention_mask=None):
        outputs = self.base_model(
            input_ids=input_ids,
            attention_mask=attention_mask,
            output_hidden_states=True
        )
        hidden_states = outputs.hidden_states[-1]
        values = self.value_head(hidden_states)
        return values.squeeze(-1)

温度采样：
────────────────────────
P(token) = softmax(logits / T)

├── T 高：更多探索，输出更随机
├── T 低：更多利用，输出更确定
└── 通常 T = 1.0 用于训练

Top-k 采样：
────────────────────────
只从概率最高的 k 个 token 中采样

├── 过滤掉低概率选项
├── 平衡多样性和质量
└── 通常 k = 50

Top-p（核）采样：
────────────────────────
从累积概率达到 p 的最小 token 集合中采样

├── 动态调整候选数量
├── 更灵活的探索控制
└── 通常 p = 0.9

训练早期：
├── 更多探索
├── 较高的温度
├── 发现更多可能的回复
└── 避免过早收敛

训练后期：
├── 更多利用
├── 较低的温度
├── 精细优化已发现的好的行为
└── 提高最终性能

实际策略：
├── 使用固定的中等温度
├── 或者逐渐降低温度
├── 结合 KL 约束自然限制探索范围
└── 监控奖励和 KL 散度变化