【深度学习精通】第23章 | 强化学习与深度学习 - DQN、PPO与ChatGPT的RLHF
本文系统介绍了深度强化学习（DRL）的核心算法与技术发展。首先回顾强化学习基础概念，包括马尔可夫决策过程和价值函数。重点讲解了深度Q网络（DQN）及其改进版本，分析其创新点如经验回放和目标网络。随后深入探讨策略梯度方法与Actor-Critic架构，详细解析PPO算法的裁剪机制和广义优势估计。针对连续控制问题，介绍了SAC、TD3等先进算法。最后阐述了AlphaGo的蒙特卡洛树搜索原理。全文涵盖了
·
环境声明
- Python版本：Python 3.10+
- PyTorch版本：PyTorch 2.0+
- 开发工具：PyCharm 或 VS Code
- 操作系统：Windows / macOS / Linux
- 依赖库：gymnasium, numpy, matplotlib
学习目标和摘要
本章将带你深入理解深度强化学习（Deep Reinforcement Learning, DRL）的核心算法，从经典的DQN到近端策略优化PPO，再到ChatGPT背后的RLHF技术。你将掌握：
- 深度Q网络（DQN）的原理与实现
- 策略梯度方法与Actor-Critic架构
- PPO算法的核心机制
- SAC、TD3等连续控制算法
- AlphaGo/AlphaZero的蒙特卡洛树搜索
- RLHF在大型语言模型中的应用
- 2025年最新进展：GRPO、World Models等
补充：强化学习被誉为"通用人工智能的必经之路"，因为它让智能体能够通过与环境的交互自主学习最优策略。
1. 强化学习基础回顾
1.1 马尔可夫决策过程（MDP）
强化学习的数学基础是马尔可夫决策过程，由五元组 (S, A, P, R, gamma) 定义：
- S：状态空间（State Space）
- A：动作空间（Action Space）
- P：状态转移概率 P(s’|s,a)
- R：奖励函数 R(s,a)
- gamma：折扣因子，通常取 0.99
1.2 价值函数与策略
强化学习的核心目标是学习一个最优策略 pi，使得累积奖励最大化：
状态价值函数：
V^pi(s) = E[sum_{t=0}^{infty} gamma^t * r_t | s_0 = s]
动作价值函数（Q函数）：
Q^pi(s,a) = E[sum_{t=0}^{infty} gamma^t * r_t | s_0 = s, a_0 = a]
2. 深度Q网络（DQN）
2.1 从Q-Learning到DQN
传统的Q-Learning使用表格存储Q值，无法处理高维状态空间。DQN（Deep Q-Network）使用神经网络近似Q函数，突破了这一限制。
一句话总结：DQN就像是给Q-Learning装上了"深度大脑"，让它能够处理图像等高维输入。
2.2 经验回放（Experience Replay）
经验回放是DQN的关键创新之一。智能体将经验 (s, a, r, s’, done) 存储在回放缓冲区中，训练时随机采样小批量数据。
为什么需要经验回放？
- 打破数据相关性：连续的经验高度相关，随机采样使数据更接近独立同分布
- 提高样本效率：同一条经验可以被多次用于训练
- 平滑数据分布：避免训练过程中的震荡
2.3 目标网络（Target Network）
DQN使用两个网络：
- 在线网络（Online Network）：用于选择动作和计算当前Q值
- 目标网络（Target Network）：用于计算目标Q值
目标网络定期从在线网络复制参数，而不是每一步都更新，这样可以稳定训练目标。
2.4 DQN算法流程
1. 初始化在线网络 Q 和目标网络 Q_target
2. 初始化回放缓冲区 D
3. for episode = 1 to M:
4. 获取初始状态 s
5. for t = 1 to T:
6. 以 epsilon 概率随机选择动作，否则 a = argmax Q(s,a)
7. 执行动作 a，观察奖励 r 和下一状态 s'
8. 存储经验 (s,a,r,s',done) 到 D
9. 从 D 中随机采样小批量经验
10. 计算目标值：y = r + gamma * max Q_target(s',a') * (1-done)
11. 最小化损失：L = (y - Q(s,a))^2
12. 每 C 步更新目标网络：Q_target = Q
2.5 DQN改进版本
| 算法 | 核心改进 | 主要优势 |
|---|---|---|
| Double DQN | 解耦动作选择与动作评估 | 解决Q值过估计问题 |
| Dueling DQN | 分离状态价值和动作优势 | 更好地学习哪些状态有价值 |
| Prioritized Experience Replay | 按TD误差优先级采样 | 提高重要经验的采样概率 |
| Noisy Nets | 参数化探索噪声 | 替代 epsilon-greedy，自适应探索 |
| Categorical DQN (C51) | 学习值分布而非期望值 | 建模Q值的完整分布 |
| Rainbow | 整合六种改进 | 达到Atari游戏SOTA |
3. 策略梯度与Actor-Critic
3.1 策略梯度定理
策略梯度方法直接参数化策略 pi_theta(a|s)，通过梯度上升优化策略。
策略梯度定理：
grad J(theta) = E[grad log pi_theta(a|s) * Q^pi(s,a)]
直观理解：如果某个动作导致了高奖励，就增加选择该动作的概率；反之则减少。
3.2 REINFORCE算法
REINFORCE是最基础的策略梯度算法，使用蒙特卡洛估计回报：
1. 初始化策略网络参数 theta
2. for episode = 1 to M:
3. 根据当前策略生成一条完整轨迹
4. 计算每个时间步的累积回报 G_t
5. 更新参数：theta += alpha * grad log pi_theta(a_t|s_t) * G_t
REINFORCE的缺陷：
- 高方差：蒙特卡洛估计回报方差大
- 只能用于回合制任务
- 需要完整轨迹才能更新
3.3 Actor-Critic架构
Actor-Critic结合了值函数方法和策略梯度方法的优点：
- Actor（演员）：策略网络 pi_theta(a|s)，决定采取什么动作
- Critic（评论家）：价值网络 V_w(s) 或 Q_w(s,a)，评估动作的好坏
优势：
- Critic提供低方差的基准，减少梯度估计方差
- 可以在线学习，不需要完整回合
- 比纯策略梯度更稳定
3.4 A2C与A3C
A2C（Advantage Actor-Critic）：
使用优势函数 A(s,a) = Q(s,a) - V(s) 代替Q值，进一步减少方差。
A3C（Asynchronous Advantage Actor-Critic）：
使用多个并行环境同时训练，异步更新全局网络。优势包括：
- 数据去相关：不同环境处于不同状态
- 无需经验回放
- 训练速度快
4. PPO：近端策略优化
4.1 从TRPO到PPO
策略梯度方法面临一个核心问题：策略更新步长过大可能导致性能崩溃，步长过小则收敛缓慢。
**TRPO（Trust Region Policy Optimization）**使用KL散度约束更新步长，但计算复杂。
**PPO（Proximal Policy Optimization）**简化了TRPO，使用裁剪目标函数达到类似效果，成为目前最流行的RL算法之一。
4.2 PPO核心机制
重要性采样比率：
r_t(theta) = pi_theta(a_t|s_t) / pi_theta_old(a_t|s_t)
裁剪目标函数：
L^{CLIP}(theta) = E[min(r_t * A_t, clip(r_t, 1-eps, 1+eps) * A_t)]
裁剪的作用：
- 当 r_t > 1+eps 且 A_t > 0 时，停止增加该动作的概率
- 当 r_t < 1-eps 且 A_t < 0 时，停止减少该动作的概率
- 防止策略更新过大
4.3 广义优势估计（GAE）
GAE是PPO中常用的优势估计方法，平衡偏差与方差：
A^{GAE}_t = sum_{l=0}^{infty} (gamma*lambda)^l * delta_{t+l}
其中 delta_t = r_t + gamma*V(s_{t+1}) - V(s_t)
- lambda = 0：高偏差，低方差（TD误差）
- lambda = 1：无偏差，高方差（蒙特卡洛）
5. SAC与TD3：连续控制算法
5.1 连续动作空间的挑战
DQN和PPO最初设计用于离散动作空间。对于连续动作空间（如机器人控制），需要特殊处理。
5.2 DDPG与TD3
DDPG（Deep Deterministic Policy Gradient）：
- 使用确定性策略 mu(s) 直接输出动作
- Actor-Critic架构，Critic估计Q值
- 使用目标网络和软更新稳定训练
**TD3（Twin Delayed DDPG）**改进DDPG：
- 双Q网络：使用两个Critic，取较小Q值估计，减少过估计
- 延迟策略更新：Critic更新频率高于Actor
- 目标策略平滑：给目标动作添加噪声，平滑Q函数
5.3 SAC（Soft Actor-Critic）
SAC是基于最大熵框架的离线策略算法：
最大熵目标：
J(pi) = E[sum r_t + alpha * H(pi(.|s_t))]
特点：
- 鼓励探索：最大化奖励的同时最大化策略熵
- 自动调整温度参数 alpha
- 样本效率高
- 对超参数不敏感
| 算法 | 策略类型 | 在线/离线 | 主要特点 |
|---|---|---|---|
| DQN | 离散 | 离线 | 经验回放，目标网络 |
| PPO | 离散/连续 | 在线 | 裁剪目标，稳定训练 |
| DDPG | 连续 | 离线 | 确定性策略 |
| TD3 | 连续 | 离线 | 双Q网络，延迟更新 |
| SAC | 连续 | 离线 | 最大熵，自动调温 |
6. AlphaGo与AlphaZero
6.1 蒙特卡洛树搜索（MCTS）
AlphaGo的核心创新是将深度学习与蒙特卡洛树搜索结合。
MCTS四步循环：
- 选择（Selection）：从根节点选择子节点，直到叶子节点
- 扩展（Expansion）：扩展叶子节点，添加子节点
- 模拟（Simulation）：从新节点进行随机 rollout
- 反向传播（Backup）：将模拟结果反向传播更新节点统计
UCB1公式（节点选择）：
UCB1 = Q(s,a) + c * sqrt(log N(s) / N(s,a))
6.2 AlphaGo架构
AlphaGo使用三个神经网络：
- 策略网络：预测人类棋手的落子概率
- 价值网络：评估当前局面的胜率
- 快速 rollout策略：轻量级策略用于MCTS模拟
训练流程：
- 监督学习：在人类棋谱上训练策略网络
- 强化学习：自我对弈优化策略网络
- 训练价值网络：预测自我对弈结果
6.3 AlphaZero的简化
AlphaZero进一步简化，只使用一个神经网络：
- 输入：棋盘状态
- 输出：落子概率分布 p 和局面价值 v
- 完全通过自我对弈学习，不需要人类棋谱
损失函数：
l = (z - v)^2 - pi^T log p + c||theta||^2
7. RLHF：ChatGPT的强化学习秘密
7.1 从GPT到ChatGPT
ChatGPT的成功离不开RLHF（Reinforcement Learning from Human Feedback）技术。其核心流程包括：
- 预训练：在大规模文本上训练基础模型
- 监督微调（SFT）：在人类标注的对话数据上微调
- 奖励模型训练：学习人类偏好
- RL微调：使用PPO优化策略
7.2 奖励模型（Reward Model）
奖励模型学习人类的偏好判断：
训练数据：对于同一提示，人类标注者比较两个回答，选择更好的一个
Bradley-Terry模型：
P(y_w > y_l | x) = sigma(r(x, y_w) - r(x, y_l))
损失函数：
L = -E[log sigma(r(x, y_w) - r(x, y_l))]
7.3 PPO训练
使用训练好的奖励模型作为环境，对语言模型进行PPO训练：
KL散度惩罚：
为了防止模型偏离太远，添加KL散度惩罚项：
r(x,y) = r_RM(x,y) - beta * KL(pi_RL || pi_SFT)
完整目标：
max E[r(x,y)] - beta * KL(pi_RL || pi_SFT)
7.4 RLHF的挑战
- 奖励黑客（Reward Hacking）：模型可能找到欺骗奖励模型的方式
- 分布偏移：训练时使用的提示与真实场景不同
- 人类标注成本：高质量偏好数据获取昂贵
8. 2025年最新进展
8.1 GRPO：群体相对策略优化
GRPO（Group Relative Policy Optimization）是DeepSeek团队在2025年提出的强化学习算法，被用于训练DeepSeek-R1推理模型。
核心思想：
- 对于每个问题，采样一组回答（group）
- 使用组内相对奖励代替绝对奖励
- 无需额外的价值网络（Critic），降低内存和计算开销
GRPO优势：
- 无需Critic模型：节省约50%的显存
- 相对奖励：自动归一化，适应不同难度的问题
- 适合推理任务：在数学、代码等任务上表现优异
GRPO vs PPO：
| 特性 | PPO | GRPO |
|---|---|---|
| 价值网络 | 需要 | 不需要 |
| 奖励计算 | 绝对奖励 | 组内相对奖励 |
| 内存占用 | 高 | 低 |
| 适用场景 | 通用RL | 大模型推理优化 |
| 样本效率 | 中等 | 高 |
8.2 World Models与世界模型
World Models是Yann LeCun等研究者倡导的方向，目标是让AI像人类一样建立对世界的内部模型。
核心组件：
- 视觉感知模块：编码观察为紧凑表示
- 世界模型：预测未来状态
- 代价模块：评估状态的好坏
- 行动模块：选择最优动作
- 短期记忆模块：存储历史信息
JEPA架构：
LeCun提出的联合嵌入预测架构（Joint Embedding Predictive Architecture）：
- 在表示空间进行预测，而非像素空间
- 学习世界的抽象表示
- 更适合规划和推理
8.3 RLVR：可验证奖励的强化学习
RLVR（Reinforcement Learning with Verifiable Rewards）是2025年后代码模型的主流训练方法。
核心思想：
- 对于代码生成任务，奖励可以通过单元测试自动验证
- 无需人工标注偏好数据
- 可扩展性强
应用场景：
- 代码生成与修复
- 数学问题求解
- 形式化证明
9. 深度强化学习实战代码
9.1 DQN完整实现
import torch
import torch.nn as nn
import torch.optim as optim
import numpy as np
import random
from collections import deque
import gymnasium as gym
class ReplayBuffer:
"""经验回放缓冲区"""
def __init__(self, capacity=100000):
self.buffer = deque(maxlen=capacity)
def push(self, state, action, reward, next_state, done):
self.buffer.append((state, action, reward, next_state, done))
def sample(self, batch_size):
batch = random.sample(self.buffer, batch_size)
states, actions, rewards, next_states, dones = zip(*batch)
return (
np.array(states),
np.array(actions),
np.array(rewards, dtype=np.float32),
np.array(next_states),
np.array(dones, dtype=np.float32)
)
def __len__(self):
return len(self.buffer)
class DQN(nn.Module):
"""DQN网络"""
def __init__(self, state_dim, action_dim, hidden_dim=128):
super(DQN, self).__init__()
self.fc1 = nn.Linear(state_dim, hidden_dim)
self.fc2 = nn.Linear(hidden_dim, hidden_dim)
self.fc3 = nn.Linear(hidden_dim, action_dim)
self.relu = nn.ReLU()
def forward(self, x):
x = self.relu(self.fc1(x))
x = self.relu(self.fc2(x))
return self.fc3(x)
class DQNAgent:
"""DQN智能体"""
def __init__(self, state_dim, action_dim, lr=1e-3, gamma=0.99,
epsilon_start=1.0, epsilon_end=0.01, epsilon_decay=0.995):
self.state_dim = state_dim
self.action_dim = action_dim
self.gamma = gamma
# epsilon-贪婪探索
self.epsilon = epsilon_start
self.epsilon_end = epsilon_end
self.epsilon_decay = epsilon_decay
# 网络
self.policy_net = DQN(state_dim, action_dim)
self.target_net = DQN(state_dim, action_dim)
self.target_net.load_state_dict(self.policy_net.state_dict())
self.optimizer = optim.Adam(self.policy_net.parameters(), lr=lr)
self.memory = ReplayBuffer()
self.batch_size = 64
self.target_update = 100
self.step_count = 0
def select_action(self, state):
"""epsilon-贪婪策略选择动作"""
if random.random() < self.epsilon:
return random.randrange(self.action_dim)
with torch.no_grad():
state = torch.FloatTensor(state).unsqueeze(0)
q_values = self.policy_net(state)
return q_values.argmax().item()
def store_transition(self, state, action, reward, next_state, done):
"""存储经验"""
self.memory.push(state, action, reward, next_state, done)
def learn(self):
"""从经验中学习"""
if len(self.memory) < self.batch_size:
return
# 采样
states, actions, rewards, next_states, dones = self.memory.sample(self.batch_size)
states = torch.FloatTensor(states)
actions = torch.LongTensor(actions)
rewards = torch.FloatTensor(rewards)
next_states = torch.FloatTensor(next_states)
dones = torch.FloatTensor(dones)
# 当前Q值
current_q = self.policy_net(states).gather(1, actions.unsqueeze(1)).squeeze(1)
# 目标Q值（Double DQN）
with torch.no_grad():
next_actions = self.policy_net(next_states).argmax(1)
next_q = self.target_net(next_states).gather(1, next_actions.unsqueeze(1)).squeeze(1)
target_q = rewards + self.gamma * next_q * (1 - dones)
# 计算损失并更新
loss = nn.MSELoss()(current_q, target_q)
self.optimizer.zero_grad()
loss.backward()
self.optimizer.step()
# 更新目标网络
self.step_count += 1
if self.step_count % self.target_update == 0:
self.target_net.load_state_dict(self.policy_net.state_dict())
# 衰减epsilon
self.epsilon = max(self.epsilon_end, self.epsilon * self.epsilon_decay)
return loss.item()
def train_dqn():
"""训练DQN"""
env = gym.make('CartPole-v1')
state_dim = env.observation_space.shape[0]
action_dim = env.action_space.n
agent = DQNAgent(state_dim, action_dim)
episodes = 500
scores = []
for episode in range(episodes):
state, _ = env.reset()
score = 0
while True:
action = agent.select_action(state)
next_state, reward, terminated, truncated, _ = env.step(action)
done = terminated or truncated
# 调整奖励，让训练更稳定
reward = reward if not done else -10
agent.store_transition(state, action, reward, next_state, done)
agent.learn()
score += reward
state = next_state
if done:
break
scores.append(score)
avg_score = np.mean(scores[-100:])
if episode % 50 == 0:
print(f"Episode {episode}, Score: {score:.2f}, Avg Score: {avg_score:.2f}, Epsilon: {agent.epsilon:.3f}")
env.close()
return agent, scores
if __name__ == "__main__":
agent, scores = train_dqn()
9.2 PPO完整实现
import torch
import torch.nn as nn
import torch.optim as optim
from torch.distributions import Categorical
import numpy as np
import gymnasium as gym
class ActorCritic(nn.Module):
"""Actor-Critic网络"""
def __init__(self, state_dim, action_dim, hidden_dim=256):
super(ActorCritic, self).__init__()
# 共享特征提取层
self.feature = nn.Sequential(
nn.Linear(state_dim, hidden_dim),
nn.Tanh(),
nn.Linear(hidden_dim, hidden_dim),
nn.Tanh()
)
# Actor：输出动作概率
self.actor = nn.Linear(hidden_dim, action_dim)
# Critic：输出状态价值
self.critic = nn.Linear(hidden_dim, 1)
def forward(self, state):
features = self.feature(state)
action_probs = torch.softmax(self.actor(features), dim=-1)
state_value = self.critic(features)
return action_probs, state_value
def get_action(self, state):
"""采样动作"""
action_probs, state_value = self.forward(state)
dist = Categorical(action_probs)
action = dist.sample()
return action.item(), dist.log_prob(action), state_value
def evaluate(self, state, action):
"""评估动作"""
action_probs, state_value = self.forward(state)
dist = Categorical(action_probs)
log_prob = dist.log_prob(action)
entropy = dist.entropy()
return log_prob, state_value, entropy
class PPOAgent:
"""PPO智能体"""
def __init__(self, state_dim, action_dim, lr=3e-4, gamma=0.99,
lambda_gae=0.95, clip_epsilon=0.2, epochs=10):
self.gamma = gamma
self.lambda_gae = lambda_gae
self.clip_epsilon = clip_epsilon
self.epochs = epochs
self.policy = ActorCritic(state_dim, action_dim)
self.optimizer = optim.Adam(self.policy.parameters(), lr=lr)
# 存储轨迹
self.states = []
self.actions = []
self.rewards = []
self.log_probs = []
self.values = []
self.dones = []
def select_action(self, state):
"""选择动作"""
with torch.no_grad():
state = torch.FloatTensor(state)
action, log_prob, value = self.policy.get_action(state)
return action, log_prob.item(), value.item()
def store_transition(self, state, action, reward, log_prob, value, done):
"""存储转移"""
self.states.append(state)
self.actions.append(action)
self.rewards.append(reward)
self.log_probs.append(log_prob)
self.values.append(value)
self.dones.append(done)
def compute_gae(self, next_value):
"""计算广义优势估计"""
rewards = np.array(self.rewards)
values = np.array(self.values + [next_value])
dones = np.array(self.dones)
advantages = np.zeros_like(rewards)
gae = 0
for t in reversed(range(len(rewards))):
delta = rewards[t] + self.gamma * values[t+1] * (1-dones[t]) - values[t]
gae = delta + self.gamma * self.lambda_gae * (1-dones[t]) * gae
advantages[t] = gae
returns = advantages + np.array(self.values)
return advantages, returns
def update(self, next_state):
"""更新策略"""
# 计算下一个状态的价值
with torch.no_grad():
next_state = torch.FloatTensor(next_state)
_, next_value = self.policy(next_state)
next_value = next_value.item()
# 计算GAE和回报
advantages, returns = self.compute_gae(next_value)
# 转换为张量
old_states = torch.FloatTensor(np.array(self.states))
old_actions = torch.LongTensor(np.array(self.actions))
old_log_probs = torch.FloatTensor(np.array(self.log_probs))
advantages = torch.FloatTensor(advantages)
returns = torch.FloatTensor(returns)
# 归一化优势
advantages = (advantages - advantages.mean()) / (advantages.std() + 1e-8)
# 多轮更新
for _ in range(self.epochs):
log_probs, state_values, entropy = self.policy.evaluate(old_states, old_actions)
# 计算比率
ratios = torch.exp(log_probs - old_log_probs)
# 裁剪目标
surr1 = ratios * advantages
surr2 = torch.clamp(ratios, 1-self.clip_epsilon, 1+self.clip_epsilon) * advantages
actor_loss = -torch.min(surr1, surr2).mean()
# Critic损失
critic_loss = nn.MSELoss()(state_values.squeeze(), returns)
# 熵奖励（鼓励探索）
entropy_loss = -entropy.mean()
# 总损失
loss = actor_loss + 0.5 * critic_loss + 0.01 * entropy_loss
self.optimizer.zero_grad()
loss.backward()
self.optimizer.step()
# 清空轨迹
self.states = []
self.actions = []
self.rewards = []
self.log_probs = []
self.values = []
self.dones = []
def save(self, path):
torch.save(self.policy.state_dict(), path)
def load(self, path):
self.policy.load_state_dict(torch.load(path))
def train_ppo():
"""训练PPO"""
env = gym.make('CartPole-v1')
state_dim = env.observation_space.shape[0]
action_dim = env.action_space.n
agent = PPOAgent(state_dim, action_dim)
episodes = 1000
update_interval = 2000 # 每收集这么多步更新一次
scores = []
step_count = 0
for episode in range(episodes):
state, _ = env.reset()
score = 0
while True:
action, log_prob, value = agent.select_action(state)
next_state, reward, terminated, truncated, _ = env.step(action)
done = terminated or truncated
agent.store_transition(state, action, reward, log_prob, value, done)
score += reward
state = next_state
step_count += 1
# 定期更新
if step_count % update_interval == 0:
agent.update(next_state)
if done:
break
scores.append(score)
avg_score = np.mean(scores[-100:])
if episode % 50 == 0:
print(f"Episode {episode}, Score: {score:.2f}, Avg Score: {avg_score:.2f}")
env.close()
return agent, scores
if __name__ == "__main__":
agent, scores = train_ppo()
9.3 简单的RLHF示例
import torch
import torch.nn as nn
import torch.optim as optim
from transformers import GPT2LMHeadModel, GPT2Tokenizer
class RewardModel(nn.Module):
"""奖励模型"""
def __init__(self, base_model_name='gpt2'):
super(RewardModel, self).__init__()
self.model = GPT2LMHeadModel.from_pretrained(base_model_name)
# 修改输出层为标量奖励
self.model.lm_head = nn.Linear(self.model.config.n_embd, 1)
def forward(self, input_ids, attention_mask):
outputs = self.model(input_ids, attention_mask=attention_mask)
# 取最后一个token的隐藏状态作为奖励
reward = outputs.logits[:, -1, 0]
return reward
class RLHFTrainer:
"""RLHF训练器"""
def __init__(self, policy_model, ref_model, reward_model, tokenizer):
self.policy = policy_model
self.ref_model = ref_model # 参考模型（SFT模型）
self.reward_model = reward_model
self.tokenizer = tokenizer
self.optimizer = optim.Adam(self.policy.parameters(), lr=1e-5)
self.kl_coef = 0.2 # KL散度系数
def compute_rewards(self, prompts, responses):
"""计算奖励"""
# 合并prompt和response
full_texts = [p + r for p, r in zip(prompts, responses)]
# 编码
encodings = self.tokenizer(
full_texts,
return_tensors='pt',
padding=True,
truncation=True
)
# 获取奖励模型分数
with torch.no_grad():
rewards = self.reward_model(
encodings['input_ids'],
encodings['attention_mask']
)
return rewards
def compute_kl_penalty(self, prompts, responses):
"""计算KL散度惩罚"""
full_texts = [p + r for p, r in zip(prompts, responses)]
encodings = self.tokenizer(
full_texts,
return_tensors='pt',
padding=True,
truncation=True
)
# 策略模型和参考模型的log概率
with torch.no_grad():
ref_outputs = self.ref_model(**encodings)
ref_log_probs = torch.log_softmax(ref_outputs.logits, dim=-1)
policy_outputs = self.policy(**encodings)
policy_log_probs = torch.log_softmax(policy_outputs.logits, dim=-1)
# KL散度
kl_div = (policy_log_probs - ref_log_probs).sum(dim=-1).mean()
return kl_div
def ppo_step(self, prompts, old_responses, old_log_probs, advantages):
"""PPO更新步骤"""
# 重新计算当前策略的log概率
full_texts = [p + r for p, r in zip(prompts, old_responses)]
encodings = self.tokenizer(
full_texts,
return_tensors='pt',
padding=True,
truncation=True
)
outputs = self.policy(**encodings)
log_probs = torch.log_softmax(outputs.logits, dim=-1)
# 计算比率
ratios = torch.exp(log_probs - old_log_probs)
# 裁剪目标
surr1 = ratios * advantages
surr2 = torch.clamp(ratios, 0.8, 1.2) * advantages
policy_loss = -torch.min(surr1, surr2).mean()
# KL惩罚
kl_penalty = self.compute_kl_penalty(prompts, old_responses)
# 总损失
loss = policy_loss + self.kl_coef * kl_penalty
self.optimizer.zero_grad()
loss.backward()
torch.nn.utils.clip_grad_norm_(self.policy.parameters(), 1.0)
self.optimizer.step()
return loss.item()
# 简化的RLHF训练流程示例
def train_rlhf_simple():
"""简化的RLHF训练流程"""
tokenizer = GPT2Tokenizer.from_pretrained('gpt2')
tokenizer.pad_token = tokenizer.eos_token
# 加载模型
policy_model = GPT2LMHeadModel.from_pretrained('gpt2')
ref_model = GPT2LMHeadModel.from_pretrained('gpt2')
reward_model = RewardModel()
# 冻结参考模型
for param in ref_model.parameters():
param.requires_grad = False
trainer = RLHFTrainer(policy_model, ref_model, reward_model, tokenizer)
# 示例数据
prompts = [
"人工智能是",
"深度学习的未来",
"强化学习的应用包括"
]
print("RLHF训练流程初始化完成")
print("在实际应用中，需要：")
print("1. 收集人类偏好数据训练奖励模型")
print("2. 使用PPO算法优化策略模型")
print("3. 监控KL散度防止模型崩溃")
if __name__ == "__main__":
train_rlhf_simple()
10. 避坑小贴士
10.1 训练不稳定问题
问题：DQN/PPO训练过程中奖励剧烈波动
解决方案：
- 降低学习率，从 1e-4 或更小开始
- 增加批量大小，提高梯度估计稳定性
- 使用梯度裁剪（gradient clipping），限制梯度范数
- 检查奖励缩放，避免过大或过小的奖励值
10.2 探索与利用的平衡
问题：智能体陷入局部最优，无法发现更好的策略
解决方案：
- DQN：缓慢衰减epsilon，不要过快降低探索率
- PPO：调整熵系数，保持适当的探索
- 使用好奇心驱动（Curiosity-Driven）探索方法
- 尝试内在奖励（Intrinsic Rewards）
10.3 样本效率问题
问题：需要大量交互才能学会简单任务
解决方案：
- 优先使用离线策略算法（DQN、SAC）
- 实现优先经验回放（Prioritized Replay）
- 使用多步回报（N-step Returns）
- 考虑模型基方法（World Models）
10.4 连续动作空间选择
建议：
- 低维连续控制（<10维）：SAC或TD3
- 高维连续控制：PPO更稳定
- 需要高样本效率：选择SAC
- 对超参数敏感：TD3比DDPG更鲁棒
10.5 神经网络架构
建议：
- 简单环境（CartPole）：2-3层MLN即可
- 图像输入：使用CNN提取特征
- 层归一化（LayerNorm）可以提高稳定性
- 对于PPO，Critic网络可以比Actor更大
11. 本章小结和知识点回顾
本章系统介绍了深度强化学习的核心算法和最新进展：
核心知识点：
- DQN：通过经验回放和目标网络稳定训练，是值函数方法的代表
- 策略梯度：直接优化策略参数，REINFORCE是基础，Actor-Critic是改进
- PPO：通过裁剪目标函数限制策略更新幅度，是目前最流行的在线策略算法
- SAC/TD3：适用于连续控制，SAC基于最大熵框架，TD3解决过估计问题
- AlphaGo/AlphaZero：深度学习+MCTS，在围棋等复杂游戏中取得突破
- RLHF：将人类反馈引入语言模型训练，是ChatGPT成功的关键
2025年最新进展：
- GRPO：DeepSeek提出的无Critic RL算法，大幅降低训练成本
- World Models：Yann LeCun倡导的方向，学习世界的内部模型
- RLVR：可验证奖励的强化学习，适用于代码生成等任务
学习建议：
- 从简单的DQN开始，理解值函数方法的核心思想
- 实现PPO，掌握策略梯度方法的精髓
- 阅读原始论文，深入理解算法设计动机
- 在Gym环境中实验，积累调参经验
参考资源
- Mnih et al. “Human-level control through deep reinforcement learning” (Nature, 2015)
- Schulman et al. “Proximal Policy Optimization Algorithms” (arXiv, 2017)
- Haarnoja et al. “Soft Actor-Critic: Off-Policy Maximum Entropy Deep RL” (ICML, 2018)
- Silver et al. “Mastering the game of Go without human knowledge” (Nature, 2017)
- Ouyang et al. “Training language models to follow instructions with human feedback” (NeurIPS, 2022)
- DeepSeek-AI. “DeepSeek-R1: Incentivizing Reasoning Capability in LLMs via RL” (2025)
- LeCun. “A Path Towards Autonomous Machine Intelligence” (2022)
本教程持续更新中，如有疑问欢迎在评论区交流讨论。
欢迎加入DeepSeek 技术社区。在这里，你可以找到志同道合的朋友，共同探索AI技术的奥秘。
更多推荐
9
0- 0
已为社区贡献1条内容
所有评论(0)