用PyTorch从零实现REINFORCE算法一个完整的CartPole游戏训练实例在强化学习领域REINFORCE算法作为经典的策略梯度方法常被用作理解深度强化学习的入门案例。不同于Q-learning等基于价值的方法REINFORCE直接优化策略函数更适合处理连续动作空间和高维状态表示。本文将带您从零开始用PyTorch框架完整实现REINFORCE算法并在OpenAI Gym的CartPole环境中进行实战训练。1. 环境准备与基础配置首先需要安装必要的Python库。推荐使用conda创建虚拟环境以避免依赖冲突conda create -n reinforce python3.8 conda activate reinforce pip install gym torch matplotlibCartPole环境的目标是控制小车移动使顶部的杆子保持直立不倒下。状态空间包含4个连续变量小车位置、速度、杆子角度和角速度动作空间是离散的左移或右移。我们可以先测试环境的基本交互import gym env gym.make(CartPole-v1) state env.reset() for _ in range(100): action env.action_space.sample() # 随机选择动作 state, reward, done, _ env.step(action) if done: break env.close()2. 策略网络设计与实现REINFORCE算法的核心是策略网络它将状态映射到动作概率分布。我们使用简单的两层全连接网络import torch import torch.nn as nn import torch.nn.functional as F class PolicyNetwork(nn.Module): def __init__(self, state_dim, action_dim, hidden_size128): super().__init__() self.fc1 nn.Linear(state_dim, hidden_size) self.fc2 nn.Linear(hidden_size, action_dim) def forward(self, x): x F.relu(self.fc1(x)) x self.fc2(x) return F.softmax(x, dim-1)网络输出两个动作的概率值采样时使用这些概率进行随机选择def select_action(policy_net, state): state torch.FloatTensor(state).unsqueeze(0) probs policy_net(state) action torch.multinomial(probs, 1).item() return action, torch.log(probs[0, action])3. 训练循环实现REINFORCE的训练过程包含完整的episode收集和策略更新两个阶段。以下是关键训练代码def train(policy_net, optimizer, gamma0.99): rewards [] log_probs [] # 收集一个完整episode的数据 state env.reset() while True: action, log_prob select_action(policy_net, state) state, reward, done, _ env.step(action) log_probs.append(log_prob) rewards.append(reward) if done: break # 计算折扣回报 R 0 returns [] for r in reversed(rewards): R r gamma * R returns.insert(0, R) # 策略梯度更新 policy_loss [] for log_prob, R in zip(log_probs, returns): policy_loss.append(-log_prob * R) optimizer.zero_grad() policy_loss torch.stack(policy_loss).sum() policy_loss.backward() optimizer.step() return sum(rewards)4. 训练过程优化与调试实际训练中我们需要考虑以下几个关键点学习率选择策略梯度方法对学习率敏感建议从3e-4开始尝试折扣因子γ通常设置在0.9到0.99之间控制未来奖励的重要性基线减除可以添加价值函数作为基线减少方差训练监控可以使用滑动平均奖励import numpy as np from collections import deque def moving_average(data, window_size): return np.convolve(data, np.ones(window_size)/window_size, modevalid) # 训练主循环 policy_net PolicyNetwork(4, 2) optimizer torch.optim.Adam(policy_net.parameters(), lr3e-4) episode_rewards deque(maxlen100) for episode in range(1000): reward train(policy_net, optimizer) episode_rewards.append(reward) if episode % 50 0: avg_reward np.mean(episode_rewards) print(fEpisode {episode}, Avg Reward: {avg_reward:.1f}) if np.mean(episode_rewards) 195: # CartPole的解决标准 print(fSolved at episode {episode}!) break5. 结果可视化与分析训练完成后我们可以通过以下方式评估模型表现训练曲线可视化观察奖励随训练的变化趋势策略可视化在不同状态下查看动作概率分布实际运行演示渲染环境观察智能体表现import matplotlib.pyplot as plt def plot_training(rewards): plt.figure(figsize(10,5)) plt.plot(rewards, labelEpisode Reward) plt.plot(moving_average(rewards, 100), labelMoving Average (100)) plt.xlabel(Episode) plt.ylabel(Reward) plt.legend() plt.show() # 运行训练好的策略 def run_episode(policy_net, renderFalse): state env.reset() total_reward 0 while True: if render: env.render() action, _ select_action(policy_net, state) state, reward, done, _ env.step(action) total_reward reward if done: break return total_reward # 评估10次取平均 eval_rewards [run_episode(policy_net) for _ in range(10)] print(fAverage evaluation reward: {np.mean(eval_rewards):.1f})6. 常见问题与解决方案在实际实现REINFORCE时开发者常遇到以下挑战训练不稳定奖励曲线出现剧烈波动解决方案减小学习率增加批量大小使用多个episode的平均梯度收敛速度慢需要大量episode才能学到有效策略解决方案实现优势函数Advantage替代原始回报探索不足策略过早收敛到次优解解决方案在损失函数中添加熵正则项改进后的策略梯度计算可以这样实现def compute_advantages(rewards, values, gamma0.99, lam0.95): # 实现GAE(Generalized Advantage Estimation) deltas [r gamma * v_next - v for r, v_next, v in zip(rewards, values[1:], values[:-1])] advantages [] advantage 0 for delta in reversed(deltas): advantage delta gamma * lam * advantage advantages.insert(0, advantage) return advantages7. 进阶扩展方向掌握了基础REINFORCE实现后可以考虑以下扩展连续动作空间修改策略网络输出高斯分布的均值和方差并行采样使用多个环境实例加速数据收集结合价值函数实现Actor-Critic架构分布式训练使用多个worker同时收集经验一个简单的并行环境实现示例from multiprocessing import Process, Queue def worker(env_name, policy_net, queue): env gym.make(env_name) while True: state env.reset() episode_data [] while True: action, log_prob select_action(policy_net, state) next_state, reward, done, _ env.step(action) episode_data.append((state, action, log_prob, reward)) state next_state if done: break queue.put(episode_data)在实现REINFORCE的过程中最深的体会是策略梯度方法对超参数选择的敏感性。学习率、折扣因子和网络结构的小幅变化都可能导致训练结果显著不同。建议在正式训练前先进行小规模的超参数搜索找到合适的配置后再进行完整训练。