优势演员-评论家A2C详解：python从零实现

🧠 向所有学习者致敬！

“学习不是装满一桶水，而是点燃一把火。” —— 叶芝

---

我的博客主页： https://lizheng.blog.csdn.net

🌐 欢迎点击加入AI人工智能社区！

🚀 让我们一起努力，共创AI未来！ 🚀

---

引言

演员-评论家方法是一类强化学习算法，结合了基于策略的方法（如 REINFORCE）和基于价值的方法（如 Q-learning）的优点。它们维护两个独立的组件：一个演员，负责学习和执行策略；一个评论家，负责学习一个价值函数，以评估演员访问的状态或状态-动作对。优势演员-评论家（A2C）是这些方法的一种流行且有效的同步变体。

什么是演员-评论家？

核心思想是利用评论家的价值估计来改进演员的策略更新。与 REINFORCE 那种使用高方差的蒙特卡洛回报 $G_t$ 来加权策略梯度 ${\nabla }_{\theta }\log {\pi }_{\theta }(a_t|s_t)$ 不同，演员-评论家方法使用来自评论家的低方差信号。

• 演员：一个策略网络 $\pi (a|s;\theta )$，输出动作概率（或直接确定动作）。
• 评论家：一个价值网络（例如 $V(s;\varphi )$ 或 $Q(s,a;\varphi )$），估计一个状态或状态-动作对的好坏。

评论家基于收到的奖励学习时间差（TD）误差，而演员则根据评论家的评估（通常使用 TD 误差或优势函数作为策略梯度的缩放因子）更新其策略。

什么是 A2C？

优势演员-评论家（A2C）是演员-评论家方法的一种具体实现策略。它的关键特点如下：

1. 优势函数：它明确使用优势函数 $A(s_t,a_t)=Q(s_t,a_t)-V(s_t)$ 来更新策略。直观来说，$A(s_t,a_t)$ 衡量了在状态 $s_t$ 下采取动作 $a_t$ 比该状态下平均动作好多少。
2. 状态价值评论家：评论家通常学习状态价值函数 $V(s;\varphi )$。优势则通过这个评论家估计，通常使用 n 步回报或 GAE：${\hat{A}}_t\approx R_t-V(s_t;\varphi )$，其中 $R_t$ 是从状态 $s_t$ 开始的总回报估计。
3. 同步更新：与它的前身 A3C（异步优势演员-评论家）不同，A2C 执行同步更新。一个中央控制器等待一批经验被收集（通常来自多个并行工作者，尽管在这里是顺序实现的），然后计算梯度并更新共享的演员和评论家网络。

同步 vs. 异步（A3C）

• A3C：使用多个并行工作者，每个工作者都有自己的网络和环境副本。工作者独立计算梯度，并异步更新全局网络。这样可以避免使用回放缓冲区，并被认为可以去相关数据。
• A2C：等待所有工作者（或一个工作者收集一批数据）完成一段经验。基于完整批次计算梯度并同步应用。从经验上看，A2C 通常表现得和 A3C 一样好，甚至更好，实现起来更简单（尤其是在 GPU 上），并且更有效地利用硬件。

为什么使用优势？

使用优势函数 $A_t$ 而不是原始回报 $G_t$（像 REINFORCE 那样）或 TD 误差 ${\delta }_t$ 可以显著降低策略梯度估计的方差。减去状态价值 $V(s_t)$ 作为基线，将更新围绕零居中。比平均动作好的动作会得到正强化，比平均动作差的动作会得到负强化，从而实现更稳定和高效的学习。

A2C 的应用场景

A2C 是一种强大的基线算法，用于各种强化学习领域：

1. 经典控制与基准测试：在 CartPole、Pendulum、Acrobot 等任务中表现出色。
2. Atari 游戏：虽然在某些游戏中可以达到不错的表现，但通常会被更先进的方法（如 PPO 或 Rainbow）超越。
3. 连续控制：适用于适当的策略分布（例如高斯分布）。
4. 基础算法：作为 A3C 的更简单的同步对应物，并且与 PPO 共享许多概念。

A2C 适用于以下情况：

• 需要在 REINFORCE 的简单性和 PPO/TRPO 的复杂性/性能之间取得平衡。
• 可行的在线策略交互。
• 需要比 REINFORCE 更低方差的更新。
• 优先选择同步更新而不是异步更新（便于调试，更好地利用 GPU）。

A2C 的数学基础

A2C 的目标是优化演员（策略 ${\pi }_{\theta }$）和评论家（价值函数 $V_{\varphi }$）。

演员更新（带优势的策略梯度）

演员参数 $\theta$ 使用策略目标 $J(\theta )$ 的梯度上升进行更新。梯度通过评论家提供的优势进行估计：
$${\nabla }_{\theta }J(\theta )\approx {\mathbb{E}}_t[{\nabla }_{\theta }\log {\pi }_{\theta }(a_t|s_t){\hat{A}}_t]$$
其中 ${\hat{A}}_t$ 是在时间 $t$ 采取的状态-动作对 $(s_t,a_t)$ 的估计优势。
更新规则鼓励那些导致正优势的动作，而抑制那些导致负优势的动作。

评论家更新（价值函数估计）

评论家参数 $\varphi$ 被更新以最小化预测状态价值 $V_{\varphi }(s_t)$ 和目标值 $R_t$ 之间的误差。目标 $R_t$ 通常是真实价值 $V^{\pi }(s_t)$ 的估计值，通常使用 n 步回报或基于 GAE 计算。
损失函数通常是均方误差（MSE）：
$$L^{VF}(\varphi )={\mathbb{E}}_t[(R_t-V_{\varphi }(s_t){)}^2]$$
最小化这个损失可以使评论家更好地预测未来的回报。

优势估计（n 步或 GAE）

优势 ${\hat{A}}_t$ 可以用几种方式估计：

1. 一步（TD 误差）：${\hat{A}}_t=r_{t+1}+\gamma V_{\varphi }(s_{t+1})-V_{\varphi }(s_t)={\delta }_t$。简单但可能有偏差。
2. n 步：${\hat{A}}_t=({\sum }_{k=0}^{n-1}{\gamma }^k{r}_{t+k+1})+{\gamma }^n{V}_{\varphi }(s_{t+n})-V_{\varphi }(s_t)$。根据 $n$ 平衡偏差和方差。
3. GAE（广义优势估计）：与 A2C/PPO 一起常用，用于良好的偏差-方差权衡，与 TRPO/PPO 中的使用方式相同。
   $${\hat{A}}_t^{GAE}=\sum _{l=0}^{\infty }(\gamma \lambda {)}^l{\delta }_{t+l},\text{其中}{\delta }_t=r_t+\gamma V_{\varphi }(s_{t+1})-V_{\varphi }(s_t)$$
   使用 GAE 时，评论家更新的目标通常是 $R_t={\hat{A}}_t^{GAE}+V_{\varphi }(s_t)$。

熵奖励

与 PPO 类似，通常在演员的目标中添加熵奖励以鼓励探索：
$$J_{entropy}(\theta )={\mathbb{E}}_t[H({\pi }_{\theta }(\cdot |s_t))]={\mathbb{E}}_t[{\mathbb{E}}_{a\sim {\pi }_{\theta }(\cdot |s_t)}[-\log {\pi }_{\theta }(a|s_t)]]$$

综合损失

演员和评论家的更新通常同时进行，使用一个综合损失函数。假设对演员目标进行梯度上升，对评论家损失进行梯度下降，需要最小化的总体损失为：
$$L^{A2C}(\theta ,\varphi )={\mathbb{E}}_t[-\log {\pi }_{\theta }(a_t|s_t){\hat{A}}_t^{\text{detached}}-c_{e}H({\pi }_{\theta }(\cdot |s_t))+c_v(R_t-V_{\varphi }(s_t){)}^2]$$
其中 $c_e$ 和 $c_v$ 分别是熵奖励和价值损失的系数。注意 ${\hat{A}}_t$ 在计算策略梯度时被视为常数（分离），以避免干扰梯度。

A2C 的逐步解释

1. 初始化：演员网络 $\pi (a|s;\theta )$，评论家网络 $V(s;\varphi )$，超参数（$\gamma ,\lambda$，学习率 ${\alpha }_{\theta },{\alpha }_{\varphi }$，熵系数 $c_e$，价值系数 $c_v$，回滚长度 $N$）。
2. 对于每次迭代：
   a. 收集轨迹：使用当前策略 ${\pi }_{\theta }$ 运行 $N$ 步（或在多个并行环境中运行 $N$ 步），收集 $(s_t,a_t,r_{t+1},s_{t+1},d_t,\log {\pi }_{\theta }(a_t|s_t))$，$t=\mathrm{0...}N-1$。如果最后一个状态 $s_N$ 不是终止状态，则引导其价值 $V_{\varphi }(s_N)$。
   b. 估计优势和回报：使用收集的奖励和价值估计 $V_{\varphi }(s_t)$，计算 ${\hat{A}}_t$（例如使用 GAE）和目标回报 $R_t={\hat{A}}_t+V_{\varphi }(s_t)$，$t=\mathrm{0...}N-1$。
   c. 计算综合损失：使用当前网络输出（对数概率、熵、预测值）和计算的目标值（优势、回报）计算 A2C 损失 $L^{A2C}$。
   d. 更新网络：对 $L^{A2C}$ 执行单次梯度下降步骤，更新 $\theta$ 和 $\varphi$（如果适用，使用单独的优化器或共享优化器）。
3. 重复：直到收敛。

A2C 的关键组件

策略网络（演员）

• 学习策略 $\pi (a|s;\theta )$。根据优势估计进行更新。

价值网络（评论家）

• 学习状态价值函数 $V(s;\varphi )$。为优势计算和目标值提供基线。
• 根据 TD 误差或回报更新。

回滚收集（在线策略，批量）

• 使用当前策略收集一批经验（通常是固定步数 $N$）。

优势估计

• 计算 ${\hat{A}}_t=R_t-V(s_t)$，通常使用 GAE 估计 $R_t$（通过 ${\hat{A}}_t^{GAE}$ 隐式实现）。为策略更新提供低方差信号。

综合演员-评论家损失

• 将策略梯度损失（按优势加权）、价值函数损失（MSE）和熵奖励整合到一个单一目标中进行优化。

同步更新

• 批次中的所有梯度一起计算并应用，简化了实现和调试，与 A3C 相比。

超参数

• 演员和评论家的学习率。
• 折扣因子 $\gamma$。
• GAE $\lambda$（如果使用）。
• 回滚长度 $N$。
• 熵系数 $c_e$。
• 价值损失系数 $c_v$。

实践示例：自定义网格世界

我们继续使用自定义网格世界环境，以保持一致性并展示 A2C 的工作原理。

环境描述：（与之前相同）

• 网格大小：10x10。
• 状态：[row/9, col/9]
• 动作：4 个离散动作（上、下、左、右）。
• 起点：(0, 0)，终点：(9, 9)。
• 奖励：到达终点 +10，碰到墙壁 -1，每步 -0.1。
• 终止条件：到达终点或达到最大步数。

设置环境

导入必要的库。

# 导入必要的库
import numpy as np
import matplotlib.pyplot as plt
import random
import math
from collections import namedtuple, deque 
from itertools import count
from typing import List, Tuple, Dict, Optional, Callable# 导入 PyTorch
import torch
import torch.nn as nn
import torch.optim as optim
import torch.nn.functional as F
from torch.distributions import Categorical# 设置设备
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print(f"使用设备：{device}")# 设置随机种子以确保可重复性
seed = 42
random.seed(seed)
np.random.seed(seed)
torch.manual_seed(seed)
if torch.cuda.is_available():torch.cuda.manual_seed_all(seed)%matplotlib inline

使用设备：cpu

创建自定义环境

使用相同的 GridEnvironment 类。

# 自定义网格世界环境（与之前的笔记本相同）
class GridEnvironment:"""一个简单的 10x10 网格世界环境。状态：(row, col)，表示为归一化向量 [row/9, col/9]。动作：0（上），1（下），2（左），3（右）。奖励：到达终点 +10，碰到墙壁 -1，每步 -0.1。"""def __init__(self, rows: int = 10, cols: int = 10) -> None:self.rows: int = rowsself.cols: int = colsself.start_state: Tuple[int, int] = (0, 0)self.goal_state: Tuple[int, int] = (rows - 1, cols - 1)self.state: Tuple[int, int] = self.start_stateself.state_dim: int = 2self.action_dim: int = 4self.action_map: Dict[int, Tuple[int, int]] = {0: (-1, 0), 1: (1, 0), 2: (0, -1), 3: (0, 1)}def reset(self) -> torch.Tensor:self.state = self.start_statereturn self._get_state_tensor(self.state)def _get_state_tensor(self, state_tuple: Tuple[int, int]) -> torch.Tensor:norm_row = state_tuple[0] / (self.rows - 1) if self.rows > 1 else 0.0norm_col = state_tuple[1] / (self.cols - 1) if self.cols > 1 else 0.0normalized_state: List[float] = [norm_row, norm_col]return torch.tensor(normalized_state, dtype=torch.float32, device=device)def step(self, action: int) -> Tuple[torch.Tensor, float, bool]:if self.state == self.goal_state:return self._get_state_tensor(self.state), 0.0, Truedr, dc = self.action_map[action]current_row, current_col = self.statenext_row, next_col = current_row + dr, current_col + dcreward: float = -0.1if not (0 <= next_row < self.rows and 0 <= next_col < self.cols):next_row, next_col = current_row, current_colreward = -1.0self.state = (next_row, next_col)next_state_tensor: torch.Tensor = self._get_state_tensor(self.state)done: bool = (self.state == self.goal_state)if done:reward = 10.0return next_state_tensor, reward, donedef get_action_space_size(self) -> int:return self.action_dimdef get_state_dimension(self) -> int:return self.state_dim

实例化并测试环境。

custom_env = GridEnvironment(rows=10, cols=10)
n_actions_custom = custom_env.get_action_space_size()
n_observations_custom = custom_env.get_state_dimension()print(f"自定义网格环境：")
print(f"大小：{custom_env.rows}x{custom_env.cols}")
print(f"状态维度：{n_observations_custom}")
print(f"动作维度：{n_actions_custom}")
start_state_tensor = custom_env.reset()
print(f"示例状态张量 (0,0)：{start_state_tensor}")

自定义网格环境：
大小：10x10
状态维度：2
动作维度：4
示例状态张量 (0,0)：tensor([0., 0.])

实现 A2C 算法

定义演员和评论家网络，计算 GAE，以及综合 A2C 更新函数。

定义演员网络

输出动作概率。

# 定义策略网络（演员）
class PolicyNetwork(nn.Module):""" A2C 的 MLP 演员网络 """def __init__(self, n_observations: int, n_actions: int):super(PolicyNetwork, self).__init__()self.layer1 = nn.Linear(n_observations, 128)self.layer2 = nn.Linear(128, 128)self.layer3 = nn.Linear(128, n_actions)def forward(self, x: torch.Tensor) -> Categorical:""" 前向传播，返回一个 Categorical 分布。 """if not isinstance(x, torch.Tensor):x = torch.tensor(x, dtype=torch.float32, device=device)elif x.dtype != torch.float32:x = x.to(dtype=torch.float32)if x.dim() == 1:x = x.unsqueeze(0)x = F.relu(self.layer1(x))x = F.relu(self.layer2(x))action_logits = self.layer3(x)return Categorical(logits=action_logits)

定义评论家网络

输出单个状态价值估计。

# 定义价值网络（评论家）
class ValueNetwork(nn.Module):""" A2C 的 MLP 评论家网络 """def __init__(self, n_observations: int):super(ValueNetwork, self).__init__()self.layer1 = nn.Linear(n_observations, 128)self.layer2 = nn.Linear(128, 128)self.layer3 = nn.Linear(128, 1)def forward(self, x: torch.Tensor) -> torch.Tensor:""" 前向传播，返回估计的状态价值。 """if not isinstance(x, torch.Tensor):x = torch.tensor(x, dtype=torch.float32, device=device)elif x.dtype != torch.float32:x = x.to(dtype=torch.float32)if x.dim() == 1:x = x.unsqueeze(0)x = F.relu(self.layer1(x))x = F.relu(self.layer2(x))state_value = self.layer3(x)return state_value

计算回报和优势（使用 GAE）

重用 GAE 函数。注意，我们还需要回报到终点（$R_t$）作为评论家更新的目标。

def compute_gae_and_returns(rewards: torch.Tensor, values: torch.Tensor, next_values: torch.Tensor, dones: torch.Tensor, gamma: float, lambda_gae: float, standardize_adv: bool = True) -> Tuple[torch.Tensor, torch.Tensor]:"""计算广义优势估计（GAE）和回报到终点（价值目标）。参数：- rewards, values, next_values, dones：从回滚中收集的张量。- gamma (float)：折扣因子。- lambda_gae (float)：GAE 平滑参数。- standardize_adv (bool)：是否标准化优势。返回：- Tuple[torch.Tensor, torch.Tensor]：- advantages：GAE 优势的张量。- returns_to_go：评论家的目标值（优势 + V_old）。"""advantages = torch.zeros_like(rewards)last_advantage = 0.0n_steps = len(rewards)# 使用 GAE 公式反向计算优势for t in reversed(range(n_steps)):mask = 1.0 - dones[t]delta = rewards[t] + gamma * next_values[t] * mask - values[t]advantages[t] = delta + gamma * lambda_gae * last_advantage * masklast_advantage = advantages[t]# 计算回报到终点（评论家的目标）returns_to_go = advantages + values # R_t = A_t + V(s_t)# 标准化优势（可选，但通常推荐）if standardize_adv:mean_adv = torch.mean(advantages)std_adv = torch.std(advantages) + 1e-8advantages = (advantages - mean_adv) / std_advreturn advantages, returns_to_go

A2C 更新步骤

这个函数计算演员和评论家的综合损失，并执行一次优化步骤。

def update_a2c(actor: PolicyNetwork,critic: ValueNetwork,actor_optimizer: optim.Optimizer,critic_optimizer: optim.Optimizer, # 通常使用单独的优化器states: torch.Tensor,actions: torch.Tensor,advantages: torch.Tensor,returns_to_go: torch.Tensor,value_loss_coeff: float,entropy_coeff: float) -> Tuple[float, float, float]:"""对演员和评论家执行一次同步更新。参数：- actor, critic：网络。- actor_optimizer, critic_optimizer：优化器。- states, actions, advantages, returns_to_go：批次数据张量。- value_loss_coeff (float)：价值损失的系数。- entropy_coeff (float)：熵奖励的系数。返回：- Tuple[float, float, float]：策略损失、价值损失和熵。"""# --- 评估当前网络 ---policy_dist = actor(states)log_probs = policy_dist.log_prob(actions)entropy = policy_dist.entropy().mean()values_pred = critic(states).squeeze()# --- 计算损失 ---# 策略损失（演员）：- E[log_pi * A] - 熵奖励# 优势被分离，因为它们不应该向评论家传播梯度policy_loss = -(log_probs * advantages.detach()).mean() - entropy_coeff * entropy# 价值损失（评论家）：MSE(V_pred, R_target)# 回报到终点被分离，因为它们是目标value_loss = F.mse_loss(values_pred, returns_to_go.detach())# 综合损失（可选，也可以单独优化）# total_loss = policy_loss + value_loss_coeff * value_loss# --- 优化演员 ---actor_optimizer.zero_grad()policy_loss.backward() # 只计算演员参数的梯度actor_optimizer.step()# --- 优化评论家 ---critic_optimizer.zero_grad()# 如果使用综合损失和单次 backward()，需要在 backward() 之前对价值损失进行缩放# 如果单独优化，则使用未缩放的损失进行 backward()(value_loss_coeff * value_loss).backward() # 只计算评论家参数的梯度critic_optimizer.step()# 返回各个损失组件以便记录return policy_loss.item() + entropy_coeff * entropy.item(), value_loss.item(), entropy.item()# 返回策略目标部分（-log_pi*A），价值损失和熵

运行 A2C 算法

设置超参数，初始化网络/优化器，并运行 A2C 训练循环。

超参数设置

定义 A2C 超参数。

# A2C 在自定义网格世界的超参数
GAMMA_A2C = 0.99             # 折扣因子
GAE_LAMBDA_A2C = 0.95        # GAE 的 lambda 参数
ACTOR_LR_A2C = 3e-4          # 演员的学习率
CRITIC_LR_A2C = 1e-3         # 评论家的学习率
VALUE_LOSS_COEFF_A2C = 0.5   # 价值损失的系数
ENTROPY_COEFF_A2C = 0.01     # 熵奖励的系数
STANDARDIZE_ADV_A2C = True   # 是否标准化优势NUM_ITERATIONS_A2C = 400    # A2C 更新次数（迭代次数）
STEPS_PER_ITERATION_A2C = 256 # 每次迭代收集的步数（批次大小）
MAX_STEPS_PER_EPISODE_A2C = 200 # 每个回合的最大步数

初始化

初始化演员、评论家及其优化器。

# 重新实例化环境
custom_env: GridEnvironment = GridEnvironment(rows=10, cols=10)
n_actions_custom: int = custom_env.get_action_space_size()
n_observations_custom: int = custom_env.get_state_dimension()# 初始化演员和评论家
actor_a2c: PolicyNetwork = PolicyNetwork(n_observations_custom, n_actions_custom).to(device)
critic_a2c: ValueNetwork = ValueNetwork(n_observations_custom).to(device)# 初始化优化器
actor_optimizer_a2c: optim.Adam = optim.Adam(actor_a2c.parameters(), lr=ACTOR_LR_A2C)
critic_optimizer_a2c: optim.Adam = optim.Adam(critic_a2c.parameters(), lr=CRITIC_LR_A2C)# 用于绘图的列表
a2c_iteration_rewards = []
a2c_iteration_avg_ep_lens = []
a2c_iteration_policy_losses = []
a2c_iteration_value_losses = []
a2c_iteration_entropies = []

训练循环

A2C 训练循环：收集 $N$ 步的批次，计算优势/回报，对演员和评论家执行一次更新。

print("开始在自定义网格世界中训练 A2C...")# --- A2C 训练循环 ---
current_state = custom_env.reset() # 从初始状态开始for iteration in range(NUM_ITERATIONS_A2C):# --- 1. 收集轨迹（N 步） ---batch_states_list: List[torch.Tensor] = []batch_actions_list: List[int] = []batch_log_probs_list: List[torch.Tensor] = []batch_rewards_list: List[float] = []batch_values_list: List[torch.Tensor] = [] # 存储 V(s_t)batch_dones_list: List[float] = []episode_rewards_in_iter: List[float] = []episode_lengths_in_iter: List[int] = []current_episode_reward = 0.0current_episode_length = 0for step in range(STEPS_PER_ITERATION_A2C):state_tensor = current_state# 采样动作并获取价值估计with torch.no_grad():policy_dist = actor_a2c(state_tensor)value_estimate = critic_a2c(state_tensor)action_tensor = policy_dist.sample()action = action_tensor.item()log_prob = policy_dist.log_prob(action_tensor)# 存储数据batch_states_list.append(state_tensor)batch_actions_list.append(action)batch_log_probs_list.append(log_prob)batch_values_list.append(value_estimate)# 与环境交互next_state, reward, done = custom_env.step(action)batch_rewards_list.append(reward)batch_dones_list.append(float(done))current_state = next_statecurrent_episode_reward += rewardcurrent_episode_length += 1# 处理回合终止（在批次收集过程中）if done or current_episode_length >= MAX_STEPS_PER_EPISODE_A2C:episode_rewards_in_iter.append(current_episode_reward)episode_lengths_in_iter.append(current_episode_length)current_state = custom_env.reset() # 为下一个回合重置current_episode_reward = 0.0current_episode_length = 0# --- 回滚结束 ---# 如果回合没有结束，引导最后一个状态的价值with torch.no_grad():last_value = critic_a2c(current_state).squeeze() # 我们将从下一个步骤开始的状态的价值# 将列表转换为张量states_tensor = torch.stack(batch_states_list).to(device).squeeze(1) # 如果网络添加了额外维度，则移除actions_tensor = torch.tensor(batch_actions_list, dtype=torch.long, device=device)log_probs_tensor = torch.stack(batch_log_probs_list).squeeze().to(device)rewards_tensor = torch.tensor(batch_rewards_list, dtype=torch.float32, device=device)values_tensor = torch.cat(batch_values_list).squeeze().to(device)dones_tensor = torch.tensor(batch_dones_list, dtype=torch.float32, device=device)# 需要 next_values 用于 GAE 计算# 移动 values 并追加引导的 last_valuenext_values_tensor = torch.cat((values_tensor[1:], last_value.unsqueeze(0)))# --- 2. 估计优势和回报到终点 ---advantages_tensor, returns_to_go_tensor = compute_gae_and_returns(rewards_tensor, values_tensor, next_values_tensor, dones_tensor, GAMMA_A2C, GAE_LAMBDA_A2C, standardize_adv=STANDARDIZE_ADV_A2C)# --- 3. 执行 A2C 更新 ---avg_policy_loss, avg_value_loss, avg_entropy = update_a2c(actor_a2c, critic_a2c, actor_optimizer_a2c, critic_optimizer_a2c,states_tensor, actions_tensor, advantages_tensor, returns_to_go_tensor,VALUE_LOSS_COEFF_A2C, ENTROPY_COEFF_A2C)# --- 记录 ---avg_reward_iter = np.mean(episode_rewards_in_iter) if episode_rewards_in_iter else np.nanavg_len_iter = np.mean(episode_lengths_in_iter) if episode_lengths_in_iter else np.nana2c_iteration_rewards.append(avg_reward_iter)a2c_iteration_avg_ep_lens.append(avg_len_iter)a2c_iteration_policy_losses.append(avg_policy_loss)a2c_iteration_value_losses.append(avg_value_loss)a2c_iteration_entropies.append(avg_entropy)if (iteration + 1) % 50 == 0: # 减少打印频率，以便进行可能更长时间的训练print(f"迭代 {iteration+1}/{NUM_ITERATIONS_A2C} | 平均奖励：{avg_reward_iter:.2f} | 平均长度：{avg_len_iter:.1f} | 策略损失：{avg_policy_loss:.4f} | 价值损失：{avg_value_loss:.4f} | 熵：{avg_entropy:.4f}")print("自定义网格世界训练完成（A2C）。")

开始在自定义网格世界中训练 A2C...
迭代 50/400 | 平均奖励：-1.55 | 平均长度：60.2 | 策略损失：-0.0259 | 价值损失：14.3651 | 熵：1.3115
迭代 100/400 | 平均奖励：6.08 | 平均长度：22.2 | 策略损失：-0.0787 | 价值损失：20.6416 | 熵：0.9647
迭代 150/400 | 平均奖励：6.54 | 平均长度：19.7 | 策略损失：-0.0793 | 价值损失：9.2166 | 熵：0.8163
迭代 200/400 | 平均奖励：6.85 | 平均长度：19.3 | 策略损失：-0.0320 | 价值损失：1.6230 | 熵：0.7364
迭代 250/400 | 平均奖励：7.31 | 平均长度：18.2 | 策略损失：-0.0797 | 价值损失：1.5631 | 熵：0.6973
迭代 300/400 | 平均奖励：7.53 | 平均长度：18.0 | 策略损失：-0.0521 | 价值损失：0.7190 | 熵：0.6447
迭代 350/400 | 平均奖励：7.89 | 平均长度：18.3 | 策略损失：-0.0556 | 价值损失：0.4610 | 熵：0.6319
迭代 400/400 | 平均奖励：7.89 | 平均长度：18.2 | 策略损失：-0.0499 | 价值损失：0.2122 | 熵：0.5835
自定义网格世界训练完成（A2C）。

可视化学习过程

为 A2C 代理绘制结果。

# 绘制 A2C 在自定义网格世界的图表
plt.figure(figsize=(20, 8))# 每次迭代的平均奖励
plt.subplot(2, 3, 1)
valid_rewards_a2c = [r for r in a2c_iteration_rewards if not np.isnan(r)]
valid_indices_a2c = [i for i, r in enumerate(a2c_iteration_rewards) if not np.isnan(r)]
plt.plot(valid_indices_a2c, valid_rewards_a2c)
plt.title('A2C 自定义网格：每次迭代的平均奖励')
plt.xlabel('迭代次数')
plt.ylabel('平均奖励')
plt.grid(True)
if len(valid_rewards_a2c) >= 20: # 如果奖励数据较多，使用较大的窗口进行平滑rewards_ma_a2c = np.convolve(valid_rewards_a2c, np.ones(20)/20, mode='valid')plt.plot(valid_indices_a2c[19:], rewards_ma_a2c, label='20 次迭代的移动平均', color='orange')plt.legend()# 每次迭代的平均回合长度
plt.subplot(2, 3, 2)
valid_lens_a2c = [l for l in a2c_iteration_avg_ep_lens if not np.isnan(l)]
valid_indices_len_a2c = [i for i, l in enumerate(a2c_iteration_avg_ep_lens) if not np.isnan(l)]
plt.plot(valid_indices_len_a2c, valid_lens_a2c)
plt.title('A2C 自定义网格：每次迭代的平均回合长度')
plt.xlabel('迭代次数')
plt.ylabel('平均步数')
plt.grid(True)
if len(valid_lens_a2c) >= 20:lens_ma_a2c = np.convolve(valid_lens_a2c, np.ones(20)/20, mode='valid')plt.plot(valid_indices_len_a2c[19:], lens_ma_a2c, label='20 次迭代的移动平均', color='orange')plt.legend()# 每次迭代的评论家（价值）损失
plt.subplot(2, 3, 3)
plt.plot(a2c_iteration_value_losses)
plt.title('A2C 自定义网格：每次迭代的价值损失')
plt.xlabel('迭代次数')
plt.ylabel('MSE 损失')
plt.grid(True)
if len(a2c_iteration_value_losses) >= 20:vloss_ma_a2c = np.convolve(a2c_iteration_value_losses, np.ones(20)/20, mode='valid')plt.plot(np.arange(len(vloss_ma_a2c)) + 19, vloss_ma_a2c, label='20 次迭代的移动平均', color='orange')plt.legend()# 每次迭代的演员（策略）损失
plt.subplot(2, 3, 4)
plt.plot(a2c_iteration_policy_losses)
plt.title('A2C 自定义网格：每次迭代的策略损失')
plt.xlabel('迭代次数')
plt.ylabel('平均 (-log_pi*A - 熵)')
plt.grid(True)
if len(a2c_iteration_policy_losses) >= 20:ploss_ma_a2c = np.convolve(a2c_iteration_policy_losses, np.ones(20)/20, mode='valid')plt.plot(np.arange(len(ploss_ma_a2c)) + 19, ploss_ma_a2c, label='20 次迭代的移动平均', color='orange')plt.legend()# 每次迭代的策略熵
plt.subplot(2, 3, 5)
plt.plot(a2c_iteration_entropies)
plt.title('A2C 自定义网格：每次迭代的策略熵')
plt.xlabel('迭代次数')
plt.ylabel('熵')
plt.grid(True)
if len(a2c_iteration_entropies) >= 20:entropy_ma_a2c = np.convolve(a2c_iteration_entropies, np.ones(20)/20, mode='valid')plt.plot(np.arange(len(entropy_ma_a2c)) + 19, entropy_ma_a2c, label='20 次迭代的移动平均', color='orange')plt.legend()plt.tight_layout()
plt.show()

A2C 学习曲线分析（自定义网格世界）：

1. 每次迭代的平均奖励：
   代理成功学习，奖励在大约 20 到 100 次迭代之间迅速增加，然后稳定在接近最优值（约 8）的水平。曲线显示出比 REINFORCE 更少的方差，这表明评论家的基线显著稳定了学习过程，尽管在收敛速度上可能略逊于 PPO/REINFORCE。

2. 每次迭代的平均回合长度：
   与奖励趋势一致，回合长度在主要学习阶段（约 20 到 100 次迭代）迅速下降，并稳定在接近最优路径长度（约 18 步）的水平。这证实了代理在最大化奖励的同时学习到了高效的策略，实现了快速且一致的目标达成。

3. 每次迭代的平均价值损失：
   评论家的 MSE 损失在策略快速变化的初期增加，随后显著下降并收敛到一个非常低的值。这表明价值函数成功地学会了准确预测状态价值，为优势计算提供了良好的基线，从而稳定了学习过程。

4. 每次迭代的平均策略损失：
   策略损失（演员的目标，包括熵）在学习过程中大幅下降（变得更负），表明策略在优势估计的驱动下得到了有效改进。它表现出适度的方差，比 REINFORCE 更少，但可能比 PPO 的裁剪目标更多。后期的逐渐上升可能反映了接近收敛时优势的减小和熵的降低。

5. 每次迭代的平均策略熵：
   熵从高值开始，促进探索，并在整个训练过程中平稳且持续地减少，因为策略变得更加确定性。这种受控的减少表明代理在平衡探索和利用，最终收敛到一个更自信的最优策略，而没有过早崩溃。

总体结论：
A2C 成功解决了网格世界问题，展示了由于价值基线的方差降低而带来的演员-评论家方法的稳定学习特性。它高效地收敛到接近最优的策略，同时平衡了奖励最大化和路径效率。损失和熵的学习曲线表现出预期的行为，证实了算法各组成部分的正确性。

分析学到的策略（可选可视化）

可视化 A2C 演员网络学到的策略。

# 重用策略绘图函数（适用于输出 Categorical 的演员网络）
def plot_a2c_policy_grid(policy_net: PolicyNetwork, env: GridEnvironment, device: torch.device) -> None:"""绘制从 A2C 策略网络导出的贪婪策略。显示每个状态中最有可能的动作。（与 REINFORCE/TRPO/PPO 绘图函数相同）"""rows: int = env.rowscols: int = env.colspolicy_grid: np.ndarray = np.empty((rows, cols), dtype=str)action_symbols: Dict[int, str] = {0: '↑', 1: '↓', 2: '←', 3: '→'}fig, ax = plt.subplots(figsize=(cols * 0.6, rows * 0.6))for r in range(rows):for c in range(cols):state_tuple: Tuple[int, int] = (r, c)if state_tuple == env.goal_state:policy_grid[r, c] = 'G'ax.text(c, r, 'G', ha='center', va='center', color='green', fontsize=12, weight='bold')else:state_tensor: torch.Tensor = env._get_state_tensor(state_tuple)with torch.no_grad():action_dist: Categorical = policy_net(state_tensor)best_action: int = action_dist.probs.argmax(dim=1).item()policy_grid[r, c] = action_symbols[best_action]ax.text(c, r, policy_grid[r, c], ha='center', va='center', color='black', fontsize=12)ax.matshow(np.zeros((rows, cols)), cmap='Greys', alpha=0.1)ax.set_xticks(np.arange(-.5, cols, 1), minor=True)ax.set_yticks(np.arange(-.5, rows, 1), minor=True)ax.grid(which='minor', color='black', linestyle='-', linewidth=1)ax.set_xticks([])ax.set_yticks([])ax.set_title("A2C 学到的策略（最有可能的动作）")plt.show()# 绘制 A2C 演员学到的策略
print("\n绘制 A2C 学到的策略：")
plot_a2c_policy_grid(actor_a2c, custom_env, device)

绘制 A2C 学到的策略：

A2C 中常见的挑战和解决方案

挑战：批次内的相关更新

• 问题：在同步版本中（尤其是单个工作器时），$N$ 步的批次来自同一个轨迹片段。这些连续的样本高度相关，可能会降低梯度质量，导致学习不稳定。
• 解决方案：
  • 使用多个并行工作器：标准的 A2C 实现通常使用多个并行环境来收集 $N$ 步的批次。这为批次引入了多样性，去相关了数据，提高了稳定性。
  • 增加批次大小 ($N$)：较大的批次可以平均掉一些噪声，但会增加更新之间的时间。

挑战：对超参数的敏感性

• 问题：性能取决于调整学习率、熵/价值系数、回滚长度 ($N$) 和 GAE 参数。
• 解决方案：
  • 仔细调整：从常见的默认值开始进行实验。
  • 学习率调度：衰减学习率有助于稳定训练的后期阶段。
  • 熵调度：有时衰减熵系数 $c_e$ 也是有益的。

挑战：价值函数的准确性

• 问题：与所有使用优势估计的演员-评论家方法一样，不准确的评论家会导致嘈杂或有偏差的优势信号，阻碍演员学习。
  解决方案：
  • 调整评论家学习率 / 价值系数：确保评论家能够有效地学习，而不会压倒策略梯度。
  • 每个演员更新进行多次评论家更新：有时对评论家进行更多次更新有助于其跟上策略的变化（尽管在基本 A2C 中不如某些离策略方法常见）。

挑战：样本效率低下（在线策略）

• 问题：与其他在线策略方法一样，数据在更新后就会被丢弃。
  解决方案：
  • PPO：PPO 明确允许对相同数据进行多个时期的更新，显著提高了样本效率。
  • 离策略演员-评论家：算法如 DDPG、TD3、SAC 使用回放缓冲区，实现更高的样本效率。

结论

优势演员-评论家（A2C）是一种基础且有效的演员-评论家算法。通过使用评论家估计状态价值并计算优势，它显著降低了策略梯度的方差，从而实现了比 REINFORCE 更稳定、通常更快的学习。其同步特性使其比异步版本 A3C 更容易实现，并且能够更好地利用 GPU 等硬件。

A2C 是一个强大的基线算法，并且是通往更先进方法（如 PPO）的概念桥梁。尽管 PPO 通常通过其裁剪目标和多次更新时期进一步提高了稳定性和样本效率，但 A2C 仍然是演员-评论家家族中一个有价值的算法，其简单性使其成为展示如何使用价值函数改进策略梯度的一个很好的例子。

---