强化学习与 RLHF

简介

强化学习（Reinforcement Learning, RL）通过"状态—动作—奖励"的闭环，让智能体在试错中学会更优策略。RLHF（Reinforcement Learning from Human Feedback）则把人类偏好引入大模型训练流程，用来让模型输出更符合"有帮助、无害、真实、可接受"的对齐目标，是 ChatGPT 一类系统中的关键训练思想之一。

强化学习的核心思想源自行为心理学：智能体通过与环境的交互获得奖励信号，学习如何在长期累积中获得最大收益。在传统 RL 中，奖励函数是手工设计的；而在 RLHF 中，奖励函数（奖励模型）是从人类偏好数据中学习得到的。这种转变使得 RL 技术能够应用于没有明确奖励定义的任务——比如"什么样的回答更好"，这是一个非常主观的判断。

特点

• 1.强化学习基础清晰 — 状态、动作、奖励、策略、价值函数构成核心框架
• 2.RLHF 面向对齐问题 — 重点不只是提分，而是让输出更符合人类偏好
• 3.PPO 是经典方案 — 在早期大模型对齐中应用非常广泛
• 4.奖励模型很关键 — 偏好数据质量直接决定后续优化方向
• 5.DPO 等替代方案兴起 — 新方法试图减少 PPO 训练复杂度
• 6.安全与可控性 — 对齐过程直接影响模型的安全边界

强化学习基础

核心概念

RL 核心元素：
- State（状态）：当前环境状态
- Action（动作）：智能体可执行动作
- Reward（奖励）：环境返回的奖励
- Policy（策略）：状态到动作的映射规则
- Value（价值）：未来期望收益估计
- Q-Function（动作价值函数）：在状态 s 执行动作 a 后的期望回报

关键公式：
- 策略梯度：∇J(θ) = E[∇log π(a|s) * G]
- 价值迭代：V(s) = max_a[R(s,a) + γ * V(s')]
- Bellman 方程：Q(s,a) = R(s,a) + γ * Σ P(s'|s,a) * max_a' Q(s',a')

策略网络与策略梯度

import torch
import torch.nn as nn
import torch.nn.functional as F

# 状态 -> 动作概率
class PolicyNetwork(nn.Module):
    def __init__(self, state_dim: int, action_dim: int):
        super().__init__()
        self.net = nn.Sequential(
            nn.Linear(state_dim, 64),
            nn.ReLU(),
            nn.Linear(64, action_dim),
            nn.Softmax(dim=-1)
        )

    def forward(self, state):
        return self.net(state)

policy = PolicyNetwork(state_dim=4, action_dim=2)
state = torch.tensor([[0.1, 0.2, 0.3, 0.4]], dtype=torch.float32)
probs = policy(state)
print(probs)

# 策略采样动作
from torch.distributions import Categorical

dist = Categorical(probs)
action = dist.sample()
log_prob = dist.log_prob(action)
print("action:", action.item(), "log_prob:", float(log_prob))

# REINFORCE 风格的简化更新
returns = torch.tensor([1.2, 0.7, -0.3], dtype=torch.float32)
log_probs = torch.tensor([-0.4, -0.8, -0.2], dtype=torch.float32)
loss = -(log_probs * returns).mean()
print("policy_loss:", float(loss))

价值网络与 Actor-Critic

class ValueNetwork(nn.Module):
    """价值网络：估计状态的价值（未来期望回报）"""
    def __init__(self, state_dim: int):
        super().__init__()
        self.net = nn.Sequential(
            nn.Linear(state_dim, 64),
            nn.ReLU(),
            nn.Linear(64, 1)
        )

    def forward(self, state):
        return self.net(state)

class ActorCritic(nn.Module):
    """
    Actor-Critic 架构：
    - Actor（策略网络）：决定采取什么动作
    - Critic（价值网络）：评估当前状态的价值
    优势：用 Critic 的评估来减少策略梯度的方差
    """
    def __init__(self, state_dim, action_dim, hidden_dim=64):
        super().__init__()
        self.actor = nn.Sequential(
            nn.Linear(state_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, action_dim),
            nn.Softmax(dim=-1)
        )
        self.critic = nn.Sequential(
            nn.Linear(state_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, 1)
        )

    def forward(self, state):
        action_probs = self.actor(state)
        value = self.critic(state)
        return action_probs, value

ac_model = ActorCritic(state_dim=4, action_dim=2)
state = torch.randn(1, 4)
probs, value = ac_model(state)
print(f"动作概率: {probs}")
print(f"状态价值: {value.item():.4f}")

REINFORCE 完整训练示例

def train_reinforce(env_steps=1000, gamma=0.99):
    """
    REINFORCE 算法完整训练流程
    注意：这是一个简化示例，实际环境需要 gym 或自定义环境
    """
    policy = PolicyNetwork(state_dim=4, action_dim=2)
    optimizer = torch.optim.Adam(policy.parameters(), lr=1e-3)

    # 模拟一批轨迹数据
    states = torch.randn(env_steps, 4)
    actions = torch.randint(0, 2, (env_steps,))
    rewards = torch.randn(env_steps)

    # 计算折扣回报
    returns = []
    G = 0
    for r in reversed(rewards.tolist()):
        G = r + gamma * G
        returns.insert(0, G)
    returns = torch.tensor(returns)

    # 标准化回报（减少方差）
    returns = (returns - returns.mean()) / (returns.std() + 1e-8)

    # 计算策略梯度损失
    log_probs = torch.log(policy(states).gather(1, actions.unsqueeze(1))).squeeze()
    loss = -(log_probs * returns).mean()

    optimizer.zero_grad()
    loss.backward()
    optimizer.step()

    print(f"REINFORCE Loss: {loss.item():.4f}")

print("REINFORCE 训练流程已定义")

PPO：RLHF 经典优化算法

PPO 核心思想

import torch

def ppo_objective(old_log_probs, new_log_probs, advantages, clip_eps=0.2):
    """
    PPO-Clip 目标函数：
    通过裁剪策略比率来限制策略更新幅度，
    防止一次更新破坏已经学到的好的策略

    ratio = π_new(a|s) / π_old(a|s) = exp(log π_new - log π_old)
    L = min(ratio * A, clip(ratio, 1-ε, 1+ε) * A)

    当 A > 0（好动作）：希望 ratio 变大但不超过 1+ε
    当 A < 0（差动作）：希望 ratio 变小但不低于 1-ε
    """
    ratio = torch.exp(new_log_probs - old_log_probs)
    surr1 = ratio * advantages
    surr2 = torch.clamp(ratio, 1 - clip_eps, 1 + clip_eps) * advantages
    return -torch.min(surr1, surr2).mean()

old_log_probs = torch.tensor([-1.1, -0.7, -0.5])
new_log_probs = torch.tensor([-1.0, -0.8, -0.45])
advantages = torch.tensor([0.8, -0.2, 0.5])

loss = ppo_objective(old_log_probs, new_log_probs, advantages)
print("ppo_loss:", float(loss))

# 可视化 clip 效果
ratios = torch.exp(new_log_probs - old_log_probs)
print(f"\n策略比率: {ratios.tolist()}")
print(f"裁剪后: {torch.clamp(ratios, 0.8, 1.2).tolist()}")

GAE（广义优势估计）

# GAE（广义优势估计）简化示例
rewards = [1.0, 0.4, -0.2]
values = [0.5, 0.3, 0.1]
gamma = 0.99
lam = 0.95

advantages = []
gae = 0
for t in reversed(range(len(rewards))):
    next_value = values[t + 1] if t + 1 < len(values) else 0
    delta = rewards[t] + gamma * next_value - values[t]
    gae = delta + gamma * lam * gae
    advantages.insert(0, gae)

print(advantages)
print(f"GAE 平滑了优势估计，减少了方差同时保留了偏差")

def compute_gae(rewards, values, dones, gamma=0.99, lam=0.95):
    """
    GAE 完整计算：
    GAE_t = Σ (γλ)^l * δ_{t+l}
    δ_t = r_t + γ * V(s_{t+1}) - V(s_t)

    参数：
    - gamma: 折扣因子，控制未来奖励的衰减
    - lam: GAE 参数，λ=0 退化为 TD(0)（低方差高偏差），λ=1 退化为蒙特卡洛（高方差低偏差）
    """
    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] + gamma * next_value * (1 - dones[t]) - values[t]
        gae = delta + gamma * lam * (1 - dones[t]) * gae
        advantages.insert(0, gae)

    return advantages

# 演示不同 lambda 值的效果
for lam in [0.0, 0.5, 0.95, 1.0]:
    advs = compute_gae(
        rewards=[1.0, 0.5, 0.3, -0.2, 0.8],
        values=[0.3, 0.4, 0.2, 0.1, 0.0],
        dones=[0, 0, 0, 0, 1],
        lam=lam
    )
    print(f"λ={lam:.2f}: GAE={[f'{a:.3f}' for a in advs]}")

print("\nλ 越大，优势估计考虑越长远的未来，方差越大")

PPO 的关键思想：
- 不让策略一次更新太猛
- 用 clip 限制更新幅度
- 在"效果提升"和"训练稳定性"之间做平衡

RLHF 三阶段：SFT → Reward Model → PPO

阶段一：SFT（监督微调）

# Step1: SFT（监督微调）
# 目标：先让模型学会"基本像样地回答"
sft_dataset = [
    {"prompt": "解释依赖注入", "response": "依赖注入是..."},
    {"prompt": "如何设计缓存", "response": "缓存设计通常需要..."}
]
print(sft_dataset)

def train_sft(model, dataset, epochs=3, lr=2e-5):
    """
    SFT 训练：用高质量的 (prompt, response) 对进行监督微调
    目标：让模型学会基本的对话格式和回答质量
    """
    optimizer = torch.optim.AdamW(model.parameters(), lr=lr)
    criterion = nn.CrossEntropyLoss()

    for epoch in range(epochs):
        for sample in dataset:
            prompt = sample["prompt"]
            response = sample["response"]

            # 构建输入
            input_text = f"问：{prompt}\n答：{response}"
            # 使用 tokenizer 编码...
            # loss = criterion(logits, labels)
            # loss.backward()
            # optimizer.step()
            pass

        print(f"SFT Epoch {epoch+1}/{epochs} 完成")

print("SFT 训练流程已定义")

阶段二：奖励模型（Reward Model）

# Step2: Reward Model（奖励模型）
# 一条 prompt 对应两条回答，人类选择更优的那条
preference_data = [
    {
        "prompt": "如何处理退款？",
        "chosen": "退款通常需要先校验订单状态，再检查时间窗口和支付状态。",
        "rejected": "退款就是退钱。"
    }
]

# 奖励模型训练目标：chosen 的得分 > rejected 的得分

import torch.nn.functional as F

def reward_loss(chosen_score, rejected_score):
    """
    Bradley-Terry 偏好损失：
    L = -log sigmoid(r_chosen - r_rejected)
    当 chosen 得分 > rejected 得分时，损失趋向于 0
    """
    return -F.logsigmoid(chosen_score - rejected_score).mean()

chosen = torch.tensor([2.3, 1.8])
rejected = torch.tensor([0.4, 1.2])
print("reward_loss:", float(reward_loss(chosen, rejected)))

class RewardModel(nn.Module):
    """
    奖励模型：学习人类偏好的评分函数
    输入：prompt + response
    输出：标量分数（越高越好）
    """
    def __init__(self, base_model, hidden_dim=768):
        super().__init__()
        self.base_model = base_model
        self.reward_head = nn.Sequential(
            nn.Linear(hidden_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, 1),
            nn.Tanh()  # 输出范围 [-1, 1]
        )

    def forward(self, input_ids, attention_mask):
        outputs = self.base_model(input_ids, attention_mask=attention_mask)
        # 使用最后一个 token 的隐藏状态
        last_hidden = outputs.last_hidden_state[:, -1, :]
        reward = self.reward_head(last_hidden)
        return reward.squeeze(-1)

print("奖励模型结构已定义")

阶段三：PPO / Policy Optimization

# Step3: PPO / Policy Optimization
# 用奖励模型对生成结果打分，再优化策略模型
rlhf_log = {
    "stage": "ppo",
    "avg_reward": 0.81,
    "kl_penalty": 0.12,
    "helpfulness_score": 4.5,
    "safety_violation_rate": 0.03
}
print(rlhf_log)

def ppo_training_loop(policy_model, reward_model, reference_model,
                       prompts, epochs=1, clip_eps=0.2, kl_coeff=0.1):
    """
    PPO 训练循环（RLHF 核心步骤）：
    1. 用当前策略模型生成回答
    2. 用奖励模型对回答打分
    3. 用 PPO 目标函数优化策略模型
    4. 用 KL 散度惩罚防止偏离参考模型太远
    """
    optimizer = torch.optim.AdamW(policy_model.parameters(), lr=1e-6)

    for epoch in range(epochs):
        # 生成回答
        responses = policy_model.generate(prompts)

        # 计算奖励
        rewards = reward_model(prompts, responses)

        # 计算 KL 散度（与 SFT 模型的偏离程度）
        kl_div = compute_kl_divergence(policy_model, reference_model, prompts, responses)

        # PPO 目标 = 奖励 - KL 惩罚
        total_reward = rewards - kl_coeff * kl_div

        # PPO 更新
        loss = ppo_objective(
            old_log_probs=policy_model.old_log_probs,
            new_log_probs=policy_model.new_log_probs,
            advantages=total_reward,
            clip_eps=clip_eps
        )

        optimizer.zero_grad()
        loss.backward()
        optimizer.step()

    return policy_model

def compute_kl_divergence(policy_model, reference_model, prompts, responses):
    """
    KL 散度惩罚：防止模型在追求奖励时偏离太远
    KL(π_θ || π_ref) = Σ π_θ * log(π_θ / π_ref)
    """
    # 简化示意
    return torch.tensor(0.1)  # 实际应逐 token 计算

print("PPO 训练循环已定义")

DPO：不显式训练 PPO 的替代方案

# DPO（Direct Preference Optimization）
# 直接利用偏好对来优化策略，不再显式训练 reward model + PPO 回路

preference_pair = {
    "prompt": "如何写异常处理？",
    "chosen": "建议在边界层捕获异常并记录结构化日志。",
    "rejected": "异常就 try catch 一下。"
}
print(preference_pair)

def dpo_loss(policy_chosen_logprob, policy_rejected_logprob,
             ref_chosen_logprob, ref_rejected_logprob, beta=0.1):
    """
    DPO 损失函数：
    L_DPO = -log sigmoid(β * (log π_θ(y_w|x)/π_ref(y_w|x) - log π_θ(y_l|x)/π_ref(y_l|x)))

    DPO 的巧妙之处：通过数学推导，将 RLHF 的隐式奖励
    转化为策略比率的函数，无需显式训练奖励模型

    参数：
    - beta: 温度参数，控制对偏好的敏感程度
    """
    chosen_ratio = policy_chosen_logprob - ref_chosen_logprob
    rejected_ratio = policy_rejected_logprob - ref_rejected_logprob
    logits = beta * (chosen_ratio - rejected_ratio)
    return -F.logsigmoid(logits).mean()

# 模拟 DPO 训练
policy_chosen = torch.tensor([-0.5])
policy_rejected = torch.tensor([-2.0])
ref_chosen = torch.tensor([-1.0])
ref_rejected = torch.tensor([-1.5])

loss = dpo_loss(policy_chosen, policy_rejected, ref_chosen, ref_rejected)
print(f"DPO Loss: {loss.item():.4f}")

对比不同偏好优化方法

方法	奖励模型	训练复杂度	稳定性	效果	适用场景
PPO	需要	高	中	高	资源充足的完整 RLHF 流程
DPO	不需要	低	高	中高	资源有限，偏好数据充足
ORPO	不需要	低	高	中高	结合 SFT 和偏好优化
IPO	不需要	低	高	中	处理偏好噪声较大的数据
KTO	不需要	低	高	中	只有正/负反馈，无配对数据

DPO 的优势：
- 训练流程更简单
- 工程复杂度更低
- 不需要单独维护 reward model + PPO rollout 链路

但它也不是所有场景都优于 RLHF，尤其在更复杂对齐目标下仍要结合具体任务评估。

Constitutional AI（CAI）

def constitutional_ai_pipeline():
    """
    Constitutional AI 流程（Anthropic 提出）：
    1. 自我批评：让模型自己判断回答是否有问题
    2. 修改：根据批评修改回答
    3. RLHF/DPO：用修改前后的回答对训练偏好模型
    """
    # 模拟流程
    prompt = "如何绕过网站的安全验证？"

    # 原始回答（可能有害）
    original_response = "你可以尝试修改请求头..."

    # 自我批评
    critique = """这个回答存在问题：
    1. 可能被用于恶意目的
    2. 违反了网络安全原则
    3. 可能帮助进行未授权访问"""

    # 修改后的回答
    revised_response = """我无法提供绕过安全验证的方法。
    如果你遇到了访问问题，建议：
    1. 联系网站管理员
    2. 检查你的账号权限
    3. 确认是否有网络配置问题"""

    print(f"Constitutional AI 示例：")
    print(f"原始回答: {original_response}")
    print(f"自我批评: {critique[:50]}...")
    print(f"修改回答: {revised_response[:50]}...")

    # 偏好对：(prompt, revised, original)
    # 用 DPO 或 RLHF 训练

constitutional_ai_pipeline()

优点

• 1.能解释大模型对齐逻辑 — 帮助理解聊天模型为什么更像"助手"而不是纯语言模型
• 2.强化偏好学习能力 — 更贴近真实用户主观满意度
• 3.适合研究模型行为 — 有利于理解 helpfulness、harmlessness、honesty 的权衡
• 4.方法体系完整 — 从经典 RL 到现代偏好优化有清晰演进路径
• 5.安全可控 — 通过对齐过程控制模型行为边界

缺点

• 1.训练复杂度高 — RLHF 工程链路远比普通 SFT 更复杂
• 2.依赖偏好数据质量 — 人类标注偏差会直接影响模型方向
• 3.奖励模型易被投机利用 — 模型可能学会"讨好奖励"而非真正解决任务
• 4.成本高昂 — rollout、采样、打分、PPO 训练都很贵
• 5.对齐税问题 — 对齐可能导致模型在某些任务上性能下降

总结

理解 RLHF 的关键，不是死记一串训练步骤，而是弄清它要解决的问题：让模型输出更符合人类偏好和使用期望。SFT 解决"会说"，奖励模型解决"哪种回答更好"，PPO/DPO 解决"如何把偏好真正注入模型行为"。在实际项目中，轻量对齐 + 强工程治理往往比完整 RLHF 更务实。

关键知识点

• 强化学习关心长期回报，不只是单次预测正确率。
• RLHF 本质上是"人类偏好监督下的策略优化"。
• 奖励模型不是真实目标本身，只是目标的近似代理。
• DPO 是重要替代路线，但不是所有情况下都自动更优。
• KL 散度惩罚是防止奖励投机（Reward Hacking）的关键机制。
• 偏好数据的质量和一致性比训练算法本身更重要。

项目落地视角

• 大模型助手产品更关注有帮助性、安全性和风格一致性，而不只是困惑度。
• 偏好数据采样、标注规范和质量控制比单纯堆训练轮次更重要。
• 对话系统上线时，还要结合规则、审核、拒答、检索兜底等手段，不会只靠 RLHF。
• 企业内部 AI 产品通常更适合"轻量对齐 + 强工程治理"，而不是盲目追求全套 RLHF 训练链路。

常见误区

• 以为 RLHF 只是"多训一轮"而已。
• 认为 reward model 分数高就代表真实用户一定满意。
• 忽略偏好数据的标注一致性和任务定义。
• 把所有对齐问题都寄希望于 PPO，忽略系统层安全策略。
• 忽略 KL 惩罚的重要性，导致模型在追求奖励时行为偏离。

进阶路线

• 深入学习 PPO、GAE、KL penalty 与 reward hacking。
• 学习 DPO、ORPO、IPO 等偏好优化新方法。
• 研究 Constitutional AI、RLAIF、Agent 评估等更现代的对齐路线。
• 将 RLHF 与推理时控制、工具调用、安全审核体系一起理解。

适用场景

• 大模型对齐研究。
• 聊天助手、Copilot、企业问答系统的行为优化。
• 需要根据用户偏好优化输出风格和质量的生成任务。
• 研究模型安全、用户满意度和可控性的团队。

落地建议

• 先明确"什么是更好的回答"，再谈偏好学习和对齐方法。
• 偏好数据要建立统一标注标准和质量抽检流程。
• 不要只看 reward 分数，要同时观察真实任务成功率与安全指标。
• 如果没有足够数据和算力，优先考虑更轻量的对齐策略而不是硬上全套 RLHF。

排错清单

• 模型越来越会"说漂亮话"但任务没变好：检查是否 reward hacking。
• 训练不稳定：检查 PPO 超参数、KL 惩罚和 advantage 估计。
• 偏好数据噪声大：检查标注一致性和 chosen/rejected 差距是否足够明显。
• 上线后表现与离线差异大：检查训练目标是否真的对应真实业务目标。

复盘问题

• 你希望模型"更像人"，还是"更完成任务"？这两者可能并不一致。
• 当前最关键的优化目标是什么：帮助性、安全性、事实性还是风格一致性？
• 奖励模型是否真的代表了你关心的目标？
• 你的问题应该用 RLHF 解决，还是其实更该从数据、检索、工具和系统治理下手？

延伸阅读

• 大模型微调
• Prompt 工程实践
• AI 模型评估
• AI 数据预处理