Policy Gradient强化学习:从原理到代码实战
1. Policy Gradient 极简教程:从理论到实践
作为一名在强化学习领域摸爬滚打多年的从业者,我经常被问到:"Policy Gradient(策略梯度)到底怎么用?" 今天我就用最直白的语言,带你彻底搞懂这个强化学习的核心算法。不同于那些堆砌公式的教程,我会用实际代码和可视化案例,让你在30分钟内掌握Policy Gradient的精髓。
Policy Gradient是解决连续动作空间问题的利器,从机器人控制到游戏AI都有广泛应用。它的核心思想是直接优化策略函数,通过梯度上升让智能体获得更多奖励。相比Q-learning等价值迭代方法,它能更自然地处理高维连续动作,这也是为什么AlphaGo和自动驾驶系统都依赖这类算法。
2. 核心原理拆解
2.1 策略梯度定理
Policy Gradient的核心是策略梯度定理(Policy Gradient Theorem),它告诉我们如何通过采样得到的轨迹(trajectory)来估计梯度。用大白话说就是:如果一个动作导致了高回报,我们就增加这个动作的概率;反之则减少。
数学表达为: ∇J(θ) = E[∇logπ(a|s) * Q(s,a)]
其中:
- π(a|s) 是策略函数(比如神经网络)
- Q(s,a) 是状态-动作价值函数
- θ 是策略的参数
这个公式的美妙之处在于,我们不需要知道环境模型,只需要采样得到的轨迹就能计算梯度。这也是为什么PG属于model-free方法。
2.2 蒙特卡洛梯度估计
实际操作中,我们用蒙特卡洛方法估计期望值。假设我们收集了N条轨迹,每条轨迹的回报是R(τ),那么梯度可以近似为:
∇J(θ) ≈ (1/N) Σ [∇logπ(τ) * R(τ)]
这里有个重要技巧:减去基线(baseline)来降低方差。常用的是状态价值函数V(s),这样梯度变为:
∇J(θ) ≈ (1/N) Σ [∇logπ(a|s) * (Q(s,a)-V(s))]
这就是著名的Advantage Actor-Critic(A2C)算法的核心思想。
3. 代码实现详解
3.1 环境搭建
我们以OpenAI Gym的CartPole(倒立摆)环境为例。先安装必要库:
pip install gym numpy torch
3.2 策略网络设计
用PyTorch实现一个简单的策略网络:
import torch
import torch.nn as nn
import torch.optim as optim
class PolicyNetwork(nn.Module):
def __init__(self, state_dim, action_dim):
super().__init__()
self.fc1 = nn.Linear(state_dim, 64)
self.fc2 = nn.Linear(64, action_dim)
def forward(self, x):
x = torch.relu(self.fc1(x))
return torch.softmax(self.fc2(x), dim=-1)
这个网络输出的是动作的概率分布。对于CartPole,state_dim=4,action_dim=2(左/右)。
3.3 训练循环实现
完整的训练流程如下:
def train(env, policy, optimizer, n_episodes=1000):
for episode in range(n_episodes):
state = env.reset()
rewards = []
log_probs = []
# 收集轨迹
while True:
state = torch.FloatTensor(state)
action_probs = policy(state)
action = torch.multinomial(action_probs, 1).item()
next_state, reward, done, _ = env.step(action)
log_probs.append(torch.log(action_probs[action]))
rewards.append(reward)
state = next_state
if done:
break
# 计算梯度
returns = []
R = 0
for r in reversed(rewards):
R = r + 0.99 * R # 折扣因子
returns.insert(0, R)
returns = torch.FloatTensor(returns)
returns = (returns - returns.mean()) / (returns.std() + 1e-9)
policy_loss = []
for log_prob, R in zip(log_probs, returns):
policy_loss.append(-log_prob * R)
optimizer.zero_grad()
loss = torch.stack(policy_loss).sum()
loss.backward()
optimizer.step()
关键点说明:
- 使用折扣回报(discounted return)考虑长期收益
- 对回报进行标准化(减均值除标准差)以稳定训练
- 负号是因为PyTorch默认做梯度下降,而我们需要梯度上升
4. 实战技巧与调参经验
4.1 学习率选择
Policy Gradient对学习率非常敏感。根据我的经验:
- CartPole:1e-3 ~ 3e-3
- Pendulum:1e-4 ~ 5e-4
- Atari游戏:1e-4 ~ 2e-4
建议使用Adam优化器,它比SGD更稳定。
4.2 折扣因子γ
γ控制未来奖励的重要性:
- 简单任务(如CartPole):0.99
- 复杂任务(如机器人控制):0.9 ~ 0.95
- 稀疏奖励任务:可能需要0.8
4.3 基线(Baseline)技巧
原始PG算法方差很大,加入基线能显著提升性能。简单实现:
# 在训练循环中添加
baseline = 0
returns = torch.FloatTensor(returns)
baseline = 0.9 * baseline + 0.1 * returns.mean()
advantage = returns - baseline
更高级的做法是用一个神经网络来估计V(s),这就是Actor-Critic架构。
5. 常见问题排查
5.1 训练不稳定
症状:回报波动大,时好时坏 解决方案:
- 减小学习率
- 增加批量大小(收集更多轨迹再更新)
- 使用梯度裁剪(
torch.nn.utils.clip_grad_norm_(policy.parameters(), 0.5))
5.2 策略过早收敛
症状:智能体很快找到局部最优,不再探索 解决方案:
- 增加熵正则项:
entropy = -torch.sum(action_probs * torch.log(action_probs)) loss = loss - 0.01 * entropy # 系数通常0.01~0.1 - 降低折扣因子γ
5.3 回报不增长
症状:训练多轮后回报仍无提升 检查点:
- 网络结构是否足够复杂(尝试增加隐藏层)
- 奖励设计是否合理(稀疏奖励问题)
- 是否应该换更高级算法(如PPO)
6. 进阶技巧
6.1 重要性采样
当我们需要复用旧策略采集的数据时,可以使用重要性采样(Importance Sampling)来修正偏差:
# 新旧策略概率比
ratio = new_probs / old_probs.detach()
surr1 = ratio * advantage
surr2 = torch.clamp(ratio, 1-eps, 1+eps) * advantage
loss = -torch.min(surr1, surr2).mean()
这就是PPO(Proximal Policy Optimization)的核心思想。
6.2 多进程并行
用Python的multiprocessing加速数据收集:
from multiprocessing import Process, Queue
def worker(env_name, policy, queue, n_episodes):
env = gym.make(env_name)
for _ in range(n_episodes):
# 收集轨迹并放入queue
queue.put(collect_trajectory(env, policy))
主进程从queue中取出轨迹批量更新。
6.3 状态归一化
对于不同量纲的状态(如位置和速度),建议做归一化:
class RunningStats:
def __init__(self):
self.mean = 0
self.var = 1
self.count = 1e-4
def update(self, x):
batch_mean = x.mean()
batch_var = x.var()
delta = batch_mean - self.mean
self.mean += delta * len(x) / (self.count + len(x))
self.var = (self.count * self.var + len(x) * batch_var +
delta**2 * self.count * len(x) / (self.count + len(x))) / (self.count + len(x))
self.count += len(x)
在训练前先收集一些状态计算统计量。
7. 实际应用案例
7.1 机械臂控制
在UR5机械臂抓取任务中,我们使用PG算法训练端到端策略:
- 状态空间:7维(关节角度+末端位置)
- 动作空间:7维(关节角速度)
- 奖励函数:抓取成功+1,碰撞-0.1,每步-0.01
关键调整:
- 使用tanh激活限制动作范围
- 添加动作变化惩罚(避免抖动)
- 用LSTM处理时序依赖
7.2 游戏AI开发
在《星际争霸II》的微操场景中:
- 状态空间:单位位置、血量、敌人信息(约100维)
- 动作空间:移动、攻击、技能(离散+连续)
- 采用Hybrid PG处理混合动作空间
7.3 量化交易
在股票高频交易策略中:
- 状态:历史价格、成交量、技术指标(50维)
- 动作:买入/卖出/持有(离散)+ 仓位大小(连续)
- 奖励:夏普比率(考虑风险)
8. 与其他算法对比
8.1 vs Q-learning
| 特性 | Policy Gradient | Q-learning |
|---|---|---|
| 动作空间 | 连续/离散 | 通常离散 |
| 收敛性 | 局部最优 | 可能发散 |
| 方差 | 高 | 低 |
| 适用场景 | 高维动作 | 离散动作 |
8.2 vs Actor-Critic
PG是AC的特例(当Critic=0时)。AC通过引入价值函数降低方差,但增加了偏差。
8.3 现代变种
- PPO:带约束的策略优化,更稳定
- SAC:最大熵RL,探索能力更强
- MPO:基于期望最大化的策略搜索
9. 工程实现建议
9.1 代码架构
推荐分层设计:
project/
├── agents/
│ ├── pg.py
│ └── ppo.py
├── envs/
│ └── custom_env.py
├── networks/
│ └── policies.py
└── utils/
├── logger.py
└── replay_buffer.py
9.2 可视化工具
用TensorBoard监控训练:
from torch.utils.tensorboard import SummaryWriter
writer = SummaryWriter()
writer.add_scalar('reward', episode_reward, global_step)
9.3 分布式训练
用Ray框架实现分布式PG:
import ray
@ray.remote
class Worker:
def __init__(self):
self.env = gym.make('Pendulum-v0')
def rollout(self, policy):
return collect_trajectory(self.env, policy)
workers = [Worker.remote() for _ in range(8)]
results = ray.get([w.rollout.remote(policy) for w in workers])
10. 个人实战心得
在真实项目中应用PG算法时,我总结了这些经验:
- 先在小环境(如CartPole)验证算法正确性
- 奖励函数设计比模型结构更重要
- 添加足够多的监控指标(如动作熵、梯度范数)
- 随机种子对结果影响很大,建议多次运行
- 当标准PG效果不好时,优先尝试PPO
最后分享一个调试技巧:如果发现智能体总是选择同一个动作,可以在策略网络输出前添加少量噪声(如高斯噪声),强制其探索。
更多推荐
所有评论(0)