Actor-Critic算法实战：用PyTorch实现CartPole平衡（附完整代码）

Actor-Critic算法实战用PyTorch实现CartPole平衡附完整代码在强化学习领域Actor-Critic算法因其独特的架构设计而备受关注。它巧妙地将策略梯度方法与值函数估计相结合既避免了纯策略梯度方法的高方差问题又克服了纯值函数方法在连续动作空间中的局限性。本文将带您从零开始实现一个完整的Actor-Critic模型解决经典的CartPole平衡问题。1. 环境准备与算法原理CartPole是OpenAI Gym中最经典的测试环境之一目标是通过控制小车的左右移动来保持杆子的竖直平衡。这个看似简单的任务实际上包含了强化学习的核心挑战——如何在连续状态空间中找到最优策略。Actor-Critic算法的核心思想可以概括为Actor演员负责根据当前状态选择动作即策略函数π(a|s)Critic评论家评估当前状态的价值V(s)或状态-动作对的Q值两者协同工作的过程就像演员在表演而评论家则不断给出反馈帮助演员改进表演技巧。具体来说# 伪代码表示Actor-Critic的交互过程 for episode in episodes: state env.reset() while not done: action actor.select_action(state) # Actor决策 next_state, reward, done, _ env.step(action) td_error critic.compute_error(state, reward, next_state) # Critic评估 actor.update(td_error) # 根据评价调整策略 critic.update(td_error) # 更新价值估计 state next_state这种架构的优势在于能够实现单步更新而不像REINFORCE算法需要等到整个episode结束才能更新参数。2. 网络架构设计与实现我们将使用PyTorch构建两个神经网络一个策略网络Actor和一个价值网络Critic。这两个网络将共享部分特征提取层以提高学习效率。2.1 策略网络Actor策略网络的目标是学习一个从状态到动作的映射。对于CartPole环境输出是离散的左或右因此我们使用softmax输出层import torch import torch.nn as nn import torch.nn.functional as F class PolicyNet(nn.Module): def __init__(self, state_dim, hidden_dim, action_dim): super(PolicyNet, self).__init__() self.fc1 nn.Linear(state_dim, hidden_dim) self.fc2 nn.Linear(hidden_dim, action_dim) def forward(self, x): x F.relu(self.fc1(x)) return F.softmax(self.fc2(x), dim1)2.2 价值网络Critic价值网络的任务是估计当前状态的价值为策略网络提供学习信号class ValueNet(nn.Module): def __init__(self, state_dim, hidden_dim): super(ValueNet, self).__init__() self.fc1 nn.Linear(state_dim, hidden_dim) self.fc2 nn.Linear(hidden_dim, 1) def forward(self, x): x F.relu(self.fc1(x)) return self.fc2(x)2.3 网络参数对比下表展示了两个网络的关键参数设置参数策略网络价值网络输入维度状态维度 (4)状态维度 (4)隐藏层128神经元128神经元输出层动作维度 (2)标量值 (1)激活函数ReLU SoftmaxReLU优化器Adam (lr1e-3)Adam (lr1e-2)3. 完整算法实现现在我们将上述组件整合成一个完整的Actor-Critic算法类。这个类需要处理动作选择、经验收集和参数更新等核心功能。class ActorCritic: def __init__(self, state_dim, hidden_dim, action_dim, actor_lr, critic_lr, gamma, device): self.actor PolicyNet(state_dim, hidden_dim, action_dim).to(device) self.critic ValueNet(state_dim, hidden_dim).to(device) self.actor_optimizer torch.optim.Adam( self.actor.parameters(), lractor_lr) self.critic_optimizer torch.optim.Adam( self.critic.parameters(), lrcritic_lr) self.gamma gamma self.device device def take_action(self, state): state torch.tensor([state], dtypetorch.float).to(self.device) probs self.actor(state) action_dist torch.distributions.Categorical(probs) action action_dist.sample() return action.item() def update(self, transition_dict): states torch.tensor(transition_dict[states], dtypetorch.float).to(self.device) actions torch.tensor(transition_dict[actions]).view(-1, 1).to(self.device) rewards torch.tensor(transition_dict[rewards], dtypetorch.float).view(-1, 1).to(self.device) next_states torch.tensor(transition_dict[next_states], dtypetorch.float).to(self.device) dones torch.tensor(transition_dict[dones], dtypetorch.float).view(-1, 1).to(self.device) # 计算时序差分误差 td_target rewards self.gamma * self.critic(next_states) * (1 - dones) td_delta td_target - self.critic(states) # 更新Actor log_probs torch.log(self.actor(states).gather(1, actions)) actor_loss torch.mean(-log_probs * td_delta.detach()) self.actor_optimizer.zero_grad() actor_loss.backward() self.actor_optimizer.step() # 更新Critic critic_loss torch.mean(F.mse_loss(self.critic(states), td_target.detach())) self.critic_optimizer.zero_grad() critic_loss.backward() self.critic_optimizer.step()提示在实际实现中我们使用detach()方法切断了一些计算图的连接这是为了避免某些参数的梯度被错误计算。4. 训练过程与调参技巧训练一个稳定的Actor-Critic模型需要关注几个关键点学习率设置、折扣因子选择以及训练节奏控制。下面是我们推荐的训练流程初始化环境与参数import gym import numpy as np import matplotlib.pyplot as plt env gym.make(CartPole-v0) state_dim env.observation_space.shape[0] action_dim env.action_space.n actor_lr 1e-3 critic_lr 1e-2 num_episodes 1000 hidden_dim 128 gamma 0.98 device torch.device(cuda if torch.cuda.is_available() else cpu)训练循环实现def train_actor_critic(env, agent, num_episodes): return_list [] for i in range(num_episodes): episode_return 0 state env.reset() done False transition_dict { states: [], actions: [], next_states: [], rewards: [], dones: [] } while not done: action agent.take_action(state) next_state, reward, done, _ env.step(action) transition_dict[states].append(state) transition_dict[actions].append(action) transition_dict[next_states].append(next_state) transition_dict[rewards].append(reward) transition_dict[dones].append(done) state next_state episode_return reward agent.update(transition_dict) return_list.append(episode_return) if (i1) % 100 0: print(fEpisode {i1}, Return: {np.mean(return_list[-100:])}) return return_list关键参数调优经验学习率比例Critic的学习率通常设为Actor的10倍因为价值估计需要更快收敛折扣因子γ对于CartPole这类即时奖励密集的任务γ可以设得较高0.98-0.99批量训练虽然我们这里使用在线学习但实际中可以积累多个transition再更新更稳定熵正则化可以添加策略熵项防止过早收敛到次优策略训练结果可视化agent ActorCritic(state_dim, hidden_dim, action_dim, actor_lr, critic_lr, gamma, device) return_list train_actor_critic(env, agent, num_episodes) plt.plot(return_list) plt.xlabel(Episodes) plt.ylabel(Returns) plt.title(Actor-Critic on CartPole) plt.show()经过约500轮训练模型通常能够学会完美平衡杆子持续200步以上。如果训练曲线波动较大可以尝试减小学习率或增加批处理大小。5. 进阶优化与扩展基础Actor-Critic实现后我们可以考虑以下几个方向的改进5.1 优势函数A2C使用优势函数A(s,a)Q(s,a)-V(s)代替简单的TD误差能显著降低方差# 修改update方法中的优势估计部分 delta td_delta.detach() advantage delta - delta.mean() # 简单的优势估计 actor_loss torch.mean(-log_probs * advantage)5.2 并行环境采样通过同时运行多个环境实例加速数据收集from multiprocessing import Process, Queue def worker(env_id, task_queue, result_queue): env gym.make(env_id) while True: state task_queue.get() if state is None: break action agent.take_action(state) next_state, reward, done, _ env.step(action) result_queue.put((state, action, reward, next_state, done))5.3 与其他算法对比下表展示了Actor-Critic与几种常见算法的特性比较特性Actor-CriticDQNREINFORCEPPO动作空间离散/连续离散离散/连续离散/连续更新方式在线离线回合制小批量方差中低高低实现难度中等简单简单复杂样本效率中高低高在实际项目中我发现Actor-Critic的一个实用技巧是定期保存和加载模型参数。当训练过程中出现性能骤降时可以回滚到之前的稳定版本# 保存检查点 torch.save({ actor_state_dict: agent.actor.state_dict(), critic_state_dict: agent.critic.state_dict(), actor_optimizer_state_dict: agent.actor_optimizer.state_dict(), critic_optimizer_state_dict: agent.critic_optimizer.state_dict(), }, checkpoint.pth) # 加载检查点 checkpoint torch.load(checkpoint.pth) agent.actor.load_state_dict(checkpoint[actor_state_dict]) agent.critic.load_state_dict(checkpoint[critic_state_dict]) agent.actor_optimizer.load_state_dict(checkpoint[actor_optimizer_state_dict]) agent.critic_optimizer.load_state_dict(checkpoint[critic_optimizer_state_dict])