1. 从“演员-评委”到代码A2C算法到底在干什么如果你玩过游戏尤其是那种需要你不断做选择、然后根据选择好坏得到即时反馈的游戏那你其实已经摸到了强化学习的门道。想象一下你是一个新手玩家面对一个复杂的游戏关卡一开始你完全不知道怎么操作只能瞎按。但每按一次游戏会给你一个分数比如加血、扣分、或者直接Game Over。你的目标就是通过不断尝试找到一套“操作秘籍”让你能稳定地拿到高分。A2C算法就是帮你自动生成这套“操作秘籍”的智能教练。它的全称是Advantage Actor-Critic中文可以叫“优势演员-评论家”。这个名字非常形象地揭示了它的核心架构。我们来拆解一下演员Actor这就是你要训练的策略网络。它的任务是根据当前看到的“游戏画面”状态决定下一步该做什么“动作”。比如在倒立摆CartPole游戏里状态就是小车的位置、速度、杆子的角度和角速度动作就是向左或向右推小车。演员一开始很笨动作是随机的但它会学习。评论家Critic这是一个价值网络。它不直接做动作而是扮演“评委”的角色。它的任务是评估在当前这个“游戏画面”下演员做出的那个动作到底有多好它能预测这个动作未来能带来多少总回报。评委的打分是演员学习的关键依据。那么优势Advantage又是什么呢这是A2C算法的精髓所在。评委Critic会给出一个“基础分”这个分是对当前状态整体价值的一个估计。而优势就是演员某个具体动作的得分减去这个基础分。简单说优势值衡量的是“选择这个动作比在这个状态下平均选择要好多少”。如果优势是正的说明这个动作很棒以后要多做如果是负的说明这个动作拖后腿了要少做。A2C就是利用这个“优势值”来更精准地指导演员Actor更新策略避免了早期策略梯度方法中方差大的问题让学习过程更稳定、更快。所以整个A2C的运作模式就像一场持续进行的表演课演员策略网络不断做出动作评委价值网络根据环境反馈的奖励游戏分数给演员的表现打分计算优势值然后演员根据这个评分来调整自己的表演方式更新网络参数目标是下次拿到更高的分。我们接下来要做的所有PyTorch代码实现就是把这个“表演课”的流程自动化、程序化。2. 搭建舞台PyTorch环境与核心网络模型设计理论懂了手就开始痒了对吧咱们直接开干。首先你得把舞台搭好。这里我们选择PyTorch因为它动态图的设计对研究和实验非常友好调试起来直观。确保你安装了torch、gym经典强化学习环境库和numpy。用pip install gym torch numpy就能搞定。我们以Gym库里的经典控制问题CartPole-v1倒立摆作为训练场。这个环境的目标是控制小车底部移动让连接在小车上的杆子保持竖直不倒。状态是4维的动作是2维左/右。搭建舞台的核心是创建我们的“演员”和“评委”也就是Actor-Critic网络模型。在PyTorch里我们会用一个类来同时定义这两个网络因为它们的前几层特征提取层通常可以共享。但为了清晰我们先实现一个基础版本让演员和评委有各自独立的网络。import torch import torch.nn as nn import torch.nn.functional as F from torch.distributions import Categorical class ActorCriticNetwork(nn.Module): A2C的核心网络模型包含独立的Actor和Critic。 输入状态 (state) 输出动作概率分布 (Actor) 和状态价值 (Critic) def __init__(self, state_dim, action_dim, hidden_dim256): super(ActorCriticNetwork, self).__init__() # 共享的特征提取层可选这里为简单起见Actor和Critic独立 # self.shared_base nn.Sequential( # nn.Linear(state_dim, hidden_dim), # nn.ReLU() # ) # 演员Actor网络输出每个动作的概率 self.actor nn.Sequential( nn.Linear(state_dim, hidden_dim), nn.ReLU(), nn.Linear(hidden_dim, action_dim), nn.Softmax(dim-1) # 将输出转换为概率分布 ) # 评论家Critic网络输出一个标量代表当前状态的价值 self.critic nn.Sequential( nn.Linear(state_dim, hidden_dim), nn.ReLU(), nn.Linear(hidden_dim, 1) ) def forward(self, state): 前向传播。 参数: state: 环境状态形状为 [batch_size, state_dim] 返回: action_distribution: 动作的概率分布Categorical对象 state_value: 当前状态的估计价值 action_probs self.actor(state) # 形状: [batch_size, action_dim] state_value self.critic(state) # 形状: [batch_size, 1] # 用概率分布创建一个分布对象方便采样和计算对数概率 action_distribution Categorical(action_probs) return action_distribution, state_value这个ActorCriticNetwork类就是我们的智能体大脑。self.actor输出的是每个动作的概率比如在CartPole中就是[向左的概率向右的概率]并且通过Softmax保证两者之和为1。self.critic输出一个数字代表评委认为当前这个状态“值多少钱”。forward函数一次前向传播同时得到了动作分布和状态价值。这里有个小细节我们使用了Categorical分布。这是因为CartPole的动作是离散的左/右。如果你的环境是连续动作空间比如控制机器人关节角度就需要用Normal高斯分布。Categorical帮我们封装好了采样sample()和计算对数概率log_prob(action)的方法非常方便。3. 核心引擎优势函数计算与策略更新网络搭好了接下来就是最关键的“学习引擎”部分如何用环境反馈的数据来更新我们的演员和评委这个过程就是A2C的训练循环我把它拆解成几个你可以跟着做的步骤。首先我们需要收集一段轨迹Trajectory数据。在A2C中我们通常不会像DQN那样使用经验回放池而是采用“同策略”的方式用当前策略网络与环境交互若干步比如5步收集一批(状态动作奖励下一个状态是否结束)这样的数据。这个过程在代码里通常是一个循环。数据有了关键的一步来了计算优势Advantage。还记得吗优势 这个动作的实际回报 - 评委认为的状态基础价值。但“实际回报”怎么算这里我们使用折扣累积回报。假设我们收集了5步的数据从最后一步倒着算def compute_returns_and_advantages(next_value, rewards, masks, gamma0.99): 计算回报Returns和优势Advantages。 参数: next_value: 轨迹结束后下一个状态的估计价值形状 [1, 1] rewards: 奖励列表每个元素形状 [n_envs, 1] masks: 非终止标志列表1-done终止为0非终止为1。形状同rewards。 gamma: 折扣因子 返回: returns: 累积回报 advantages: 优势值 R next_value returns [] # 从后向前计算累积回报 for step in reversed(range(len(rewards))): R rewards[step] gamma * R * masks[step] returns.insert(0, R) # 插入到列表开头 returns torch.cat(returns).detach() # 拼接成张量并断开计算图 # 注意这里我们需要有之前收集的每个状态对应的state_value # 假设我们已经将每个状态的state_value收集在 values 列表中 values torch.cat(values) advantages returns - values.detach() # 优势 回报 - 价值估计 return returns, advantagesmasks的作用是处理回合终止。如果某一步之后游戏结束了doneTrue那么mask就是0这意味着下一状态的回报R不会被累积到当前步因为游戏已经重置了。有了优势值我们就可以定义损失函数来更新网络了。A2C的损失函数是三部分的加权和策略损失Actor Loss目标是最大化优势 * 所选动作的对数概率。因为优势为正时我们希望这个动作的概率变大对数概率增大优势为负时希望其概率变小。在PyTorch中我们通常最小化负的该值。价值损失Critic Loss目标是让评委Critic的预测更准。我们用均方误差MSE来衡量评委预测的价值values和实际计算出的回报returns之间的差距。熵正则项Entropy Bonus这是一个技巧用于鼓励探索。它计算策略分布的熵熵越大表示动作越随机探索性强。我们在损失中加一个负的熵相当于鼓励熵变大防止策略过早收敛到局部最优、停止探索。# 假设我们已经收集了 # log_probs: 每一步动作的对数概率列表最终拼接成张量 # values: 每一步的状态价值估计列表最终拼接成张量 # returns: 计算出的累积回报 # advantages: 计算出的优势 # entropy: 每一步策略分布的熵的平均在收集数据时累加计算 policy_loss -(log_probs * advantages.detach()).mean() value_loss F.mse_loss(values, returns) entropy_loss -entropy.mean() # 因为我们希望熵增大所以加负号 # 总损失是三者加权和 loss policy_loss 0.5 * value_loss 0.01 * entropy_loss # 然后反向传播更新参数 optimizer.zero_grad() loss.backward() # 可以加一个梯度裁剪防止梯度爆炸 torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm0.5) optimizer.step()这里的权重系数0.5和0.01是超参数你可以根据实际情况调整。价值损失的权重通常小于1是为了防止价值网络更新太快影响策略的稳定性。熵正则项的权重一般很小只是一个微弱的探索激励。4. 效率加速多环境并行训练实战用单个环境训练强化学习智能体就像让一个学生做一张卷子交上来老师批改再发回去订正。效率很低。多环境并行训练就像是同时让几十个学生环境副本一起做卷子老师一次性批改一大摞然后统一讲解。这样单位时间内智能体看到的数据量大大增加训练速度能提升一个数量级。实现多环境并行我们不需要自己从零写多进程代码。可以借鉴OpenAI Baselines中的SubprocVecEnv设计思路。它的核心思想是为每个环境创建一个独立的子进程主进程通过管道Pipe向子进程发送指令如step、reset并接收结果。这样多个环境就能在CPU上真正同步运行。在代码中我们通常会先定义一个创建环境的函数make_env然后使用SubprocVecEnv来包装它。这里我给出一个简化版的实现和用法# 假设我们已经有了从baselines中导入的 SubprocVecEnv 类 # 或者使用了类似 stable-baselines3 库中的 VecEnv def make_env(env_name, seed0): 创建一个环境函数便于多进程复制 def _thunk(): env gym.make(env_name) env.seed(seed) return env return _thunk # 在主训练脚本中 num_envs 8 # 并行环境数量根据你的CPU核心数调整 envs SubprocVecEnv([make_env(CartPole-v1) for _ in range(num_envs)]) # 初始化状态注意现在状态形状是 [num_envs, state_dim] state envs.reset()当调用envs.step(actions)时actions是一个包含num_envs个动作的数组或列表。它会并行地在所有环境中执行并返回一个批量的next_states,rewards,dones,infos。这极大地提高了数据收集的效率。在训练循环中数据收集部分就变成了这样log_probs, values, rewards, masks [], [], [], [] entropy 0 # 收集 n_steps 的数据 for _ in range(n_steps): # 将状态转换为Tensor state_tensor torch.FloatTensor(state).to(device) # 前向传播得到所有并行环境的动作分布和价值 dist, value model(state_tensor) action dist.sample() # 采样动作形状 [num_envs] # 执行动作这里一步就完成了8个环境的交互 next_state, reward, done, _ envs.step(action.cpu().numpy()) # 计算对数概率和熵 log_prob dist.log_prob(action) entropy dist.entropy().mean() # 取平均熵 # 存储数据 log_probs.append(log_prob) values.append(value) rewards.append(torch.FloatTensor(reward).unsqueeze(1).to(device)) # 增加维度以匹配value形状 masks.append(torch.FloatTensor(1 - done).unsqueeze(1).to(device)) state next_state # 更新状态准备下一步你看循环n_steps次由于每次迭代都并行地在num_envs个环境中交互我们实际上收集了n_steps * num_envs条经验数据。这比单环境快了num_envs倍。后续的优势计算和反向更新流程与单环境完全一样只是数据量变成了批量形式。这是让A2C这类在线策略算法实现高效训练的关键技巧。5. 调参与调试让CartPole屹立不倒的秘诀代码都写好了一运行发现智能体学得一塌糊涂奖励曲线像过山车或者根本学不会。别急这太正常了。强化学习的训练充满了“玄学”很大程度上就是调参和调试的功夫。结合我在CartPole环境上的多次实验分享几个关键的调参点和调试技巧。核心超参数解析学习率Learning Rate这是最重要的参数之一。对于Actor和Critic有时甚至需要设置不同的学习率。通常Critic的学习率可以设得比Actor稍高一点因为它要快速拟合价值函数。一个常见的起点是lr3e-4或lr1e-3。如果训练不稳定奖励剧烈震荡尝试调低学习率如果学习速度太慢可以稍微调高。折扣因子Gamma决定了未来奖励的重要性。gamma0.99是控制问题的标准选择意味着智能体比较“远视”。如果环境奖励稀疏可以适当增大gamma如果环境奖励密集且即时可以稍微减小比如0.95。并行环境数量n_envs与步数n_steps这两个参数共同决定了每次参数更新前收集的数据量n_envs * n_steps。n_envs受限于你的CPU核心数通常4、8、16都是常用值。n_steps一般在5到20之间。更大的批次意味着更稳定的梯度估计但更新频率变慢。我常用的组合是n_envs8, n_steps5。熵系数Entropy Coefficient这个参数控制探索的强度。默认0.01是个不错的起点。如果你发现智能体过早地陷入某个固定行为比如小车一直朝一个方向走可以适当增大这个系数如0.02来鼓励探索。训练后期可以逐渐减小它让策略收敛。价值损失系数Value Loss Coefficient在总损失中价值损失的权重。通常设为0.5或1。如果Critic学得太快或太慢导致优势估计不准可以调整这个参数。实用的调试技巧监控关键指标不要只看总奖励。把policy_loss、value_loss、entropy的平均值、advantages的均值标准差都打印或记录下来。如果value_loss一直降不下去可能是Critic网络能力不足或学习率不对。如果entropy快速降到接近0说明探索不足需要增大熵系数。观察智能体行为定期比如每训练1000步让智能体在环境中“表演”几次不进行训练只是渲染出来看。直观感受它的策略是否合理。在CartPole中一个学会的策略会让小车在杆子倒向一边时快速向那边移动以保持平衡。梯度裁剪Gradient Clipping在loss.backward()之后、optimizer.step()之前加入torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm0.5)。这能防止梯度爆炸是训练RNN和很多RL算法的标配能让训练过程稳定很多。奖励归一化Reward Scaling有时环境原始奖励的尺度不合适太大或太小会导致梯度问题。一个简单的做法是对收集到的rewards进行减均值、除标准差的归一化。这能稳定Critic的学习。从简单环境开始CartPole-v1最大步数500比CartPole-v0最大步数200难。如果你的算法在v0上都学不好肯定实现有问题。先在v0上调通再挑战v1。我自己的踩坑经验是最初实现时忘了对优势advantages进行归一化比如减去均值除以标准差导致策略更新步长不稳定训练曲线锯齿非常严重。后来加上了advantages (advantages - advantages.mean()) / (advantages.std() 1e-8)这一行训练立刻平滑了很多。另一个坑是在多环境训练中dones的处理要小心。当一个环境提前结束时SubprocVecEnv会自动重置它并返回新的next_state。但在计算masks时我们用的(1 - done)这个done信号必须正确使用否则回报计算会出错。多花时间确保数据收集这部分逻辑正确是成功的一半。