Actor Critic 方法初探

方差问题#

策略梯度（Policy Gradient）方法因其直观和有效性而备受关注。我们之前探讨过Reinforce算法，它在许多任务中表现良好。然而，Reinforce 方法依赖于蒙特卡洛（Monte Carlo）采样来估计回报，这意味着我们需要使用整个回合的数据来计算回报。这种方法带来了一个关键问题 ——策略梯度估计中的高方差。

Pasted image 20241010090847

PG 估计的核心在于找到回报增加最快的方向。换句话说，我们需要更新策略的权重，使得那些带来高回报的动作在未来被选中的概率更高。理想情况下，这样的更新会使得策略逐步优化，获得更高的总回报。

然而，使用蒙特卡洛方法估计回报时，由于依赖整个回合的数据计算实际回报（不估计回报），策略梯度的估计会有显著的方差（无偏但高方差）。高方差意味着我们的梯度估计不稳定，可能导致训练过程缓慢甚至不收敛。为了获得可靠的梯度估计，我们可能需要大量的样本，这在实际应用中往往代价高昂。

Pasted image 20241010091609

环境和策略的随机性导致相同初始状态可能产生截然不同的回报，造成高方差。因此，从同一状态开始的回报在不同情节中可能显著变化。使用大量轨迹可减少方差，提供更准确的回报估计。但大 batch size 会降低样本效率，因此需要寻找其他减少方差的机制。

Advantage Actor-Critic (A2C)#

通过 Actor-Critic 方法减少方差#

从前面所学知识和章节导语带来的一个直观感受是 “如果结合 Value-Based 和 Policy-Based，方差和训练的问题都会得到优化”。演员 - 评论家（Actor-Critic）方法 正是这样的混合架构，具体来说：

演员（Actor）：负责选择动作，基于当前策略生成动作概率分布。
评论家（Critic）：估计当前策略下的价值函数，提供对动作选择的反馈。

想象你和你的朋友都是菜鸟玩家。你负责操作（Actor），而你的朋友负责观察和评价（Critic）。一开始，你们都不太懂游戏。你瞎猫撞死耗子般地操作，而你的朋友也在摸索如何评价你的表现。随着时间推移，你通过实践不断改进操作技巧（Policy），同时你的朋友也在学习如何更准确地评估每个动作的好坏（Value）。

Pasted image 20241010092831

你们互相帮助，共同进步：你的操作为朋友提供了评估的基础，而朋友的反馈帮助你调整策略。

也就是说，我们会学习两个函数逼近（神经网络）：

控制 agent（Actor）操作的的策略函数： $\pi_\theta(s)$
衡量行动好坏辅助策略优化（Critic）的价值函数： $\hat{q}_w(s,a)$

算法流程#

在每个时间步 $t$ ，我们从环境中获取当前状态 $S_t$ ，作为输入传递给我们的 Actor 和 Critic。
Actor 根据状态输出一个动作 $A_t$ 。

Screenshot 2024-10-10 at 10.06.53

Critic 也将该动作作为输入，并使用 $S_t$ 和 $A_t$ ，计算在该状态下采取该动作的价值：即 Q 值。
在环境中执行动作 $A_t$ ，输出新状态 $S_{t+1}$ 和奖励 $R_{t+1}$ 。
Actor 利用 $Q$ 值更新其策略参数。

Screenshot 2024-10-10 at 10.08.30

Screenshot 2024-10-10 at 10.19.12

Actor 使用更新后的参数，在给定新状态 $S_{t+1}$ 时生成下一步要采取的动作 $A_{t+1}$ 。
Critic 随后更新其价值参数。

Screenshot 2024-10-10 at 10.37.56

Critic 起到了提供用于调整回报估计的 baseline 作用，从而使得梯度估计更加稳定。训练过程更为平滑，收敛速度更快，所需样本数量也大大减少。

添加 Advantage (A2C)#

可以通过使用Advantage Function 作为 Critic 而不是动作值函数来进一步稳定学习。

优势：让好动作脱颖而出#

核心想法是，将你的动作通过两个部分来评估：

你获得的即时奖励和下一个状态的价值。
你在当前状态中对价值的预期。

数学上，我们称之为优势：

A(s_t, a_t) = r_{t+1} + \gamma V(s_{t+1}) - V(s_t)

这表达的是：在状态 $s_t$ 中，你做出的动作 $a_t$ ，比你在状态 $s_t$ 中原本的期望（由 $V(s_t)$ 表示的 baseline 期望）好多少？

如果实际的奖励和未来状态的价值 $r_{t+1} + \gamma V(s_{t+1})$ 高于你对当前状态 $V(s_t)$ 的预期，那么这个动作是好的；如果低了，那…… 你可以做得更好。

这个优势不仅告诉你动作好不好，还告诉你有多好或者有多差（相对于 baseline 来说）。

策略更新#

当我们执行某个动作时，奖励本身并不足够指导策略的改进。奖励告诉我们某个动作好或不好，但它并没有告诉我们这个动作究竟有多好，或者比预期好多少。
所以改进策略时，与其盲目追逐奖励，不如关注调整动作，基于它们超越（或低于）期望的程度。现在就可以微调策略，向那些持续表现优于 baseline 的动作靠拢。

这就是我们得到的更新公式：

\nabla_\theta J(\theta) \sim \sum_{t=0}^{T-1} \nabla_\theta \log \pi_\theta(a_t | s_t) A(s_t, a_t)

$\nabla_\theta \log \pi_\theta(a_t | s_t)$ ：表示在每个时间步 $t$ ，我们计算策略 $\pi_\theta$ 选择动作 $a_t$ 的概率的对数梯度。这一步帮助我们找到在当前策略下，如何通过改变参数 $\theta$ 来提高选择 $a_t$ 的可能性。
$A(s_t, a_t)$ ：动作的优势函数，告诉我们这个动作 $a_t$ 在状态 $s_t$ 下相对于 baseline 有多好或多差。

简单来说：你的策略 $\pi_\theta$ 的梯度由优势 $A(s_t, a_t)$ 调整。你在更新你的策略，不仅基于这个动作是否带来了一些奖励，而是基于这个动作相比期望超越了多少。

更妙的是：你只需要一个神经网络来预测价值函数 $V(s)$ 。

现在来聊聊 TD 误差#

当然，计算这个优势函数很棒，但在线学习有一个妙处：你不需要等到最后再更新策略。于是，时序差分误差（TD Error） 就登场了：

\delta = r + \gamma V(s') - V(s)

这里的关键是：TD 误差实际上是优势函数的在线估计。它告诉你，就在此刻，你的动作是否让未来状态比你预期的要好。这个误差 $\delta$ 直接反映了优势的概念：

如果 $\delta > 0$ ：“嘿，这个动作比我想象中好！”（优势为正）。
如果 $\delta < 0$ ：“嗯，我本来以为会更好……”（优势为负）。

这让你可以逐步调整你的策略，不用等到一整 episode 结束才做改变。这对提高效率来说，简直是绝佳策略。

代码实现#

Actor-Critic 网络架构#

首先，我们需要构建一个神经网络。这是一个双头网络：一个用于 Actor（学习策略来选择动作），另一个用于 Critic（估计状态的价值）。

class ActorCritic(nn.Module):
    def __init__(self, num_inputs, num_actions, hidden_size, learning_rate=3e-4):
        super(ActorCritic, self).__init__()

        # Critic 网络（价值函数近似）
        # 这个网络用于预测 V(s)，也就是状态 s 的价值
        self.critic_linear1 = nn.Linear(num_inputs, hidden_size)
        self.critic_linear2 = nn.Linear(hidden_size, 1)  # 价值函数是标量输出

        # Actor 网络（策略函数近似）
        # 这个网络用于预测 π(a|s)，即在状态 s 下选择动作 a 的概率
        self.actor_linear1 = nn.Linear(num_inputs, hidden_size)
        self.actor_linear2 = nn.Linear(hidden_size, num_actions)  # 输出对所有动作的概率分布
    
    def forward(self, state):
        # 将状态转换为 torch tensor，并增加一个维度以支持 batch 处理
        state = Variable(torch.from_numpy(state).float().unsqueeze(0))
        
        # Critic 网络的前向传播
        value = F.relu(self.critic_linear1(state))
        value = self.critic_linear2(value)  # 输出状态的价值 V(s)
        
        # Actor 网络的前向传播
        policy_dist = F.relu(self.actor_linear1(state))
        policy_dist = F.softmax(self.actor_linear2(policy_dist), dim=1)  # 用 softmax 将原始的数值转换为动作的概率分布（策略）
        
        return value, policy_dist

A2C 算法的核心实现#

接下来进入 A2C 的核心：主循环和更新机制。每一 episode 中，agent 在环境中运行一定步数，收集状态、动作和奖励的轨迹。在每集结束时，更新 Actor（策略）和 Critic（价值函数）。

def a2c(env):
    # 从环境中获取输入和输出的维度
    num_inputs = env.observation_space.shape[0]
    num_outputs = env.action_space.n
    
    # 初始化 Actor-Critic 网络
    actor_critic = ActorCritic(num_inputs, num_outputs, hidden_size)
    ac_optimizer = optim.Adam(actor_critic.parameters(), lr=learning_rate)

    # 用于追踪性能的数据容器
    all_lengths = []  # 追踪每 episode 的长度
    average_lengths = []  # 追踪最近 10 episode 的平均长度
    all_rewards = []  # 追踪每 episode 的累计奖励
    entropy_term = 0  # 激励探索

    # 进入每一 episode 的循环
    for episode in range(max_episodes):
        log_probs = []  # 存储动作的对数概率
        values = []  # 存储 Critic 的价值估计（V(s)）
        rewards = []  # 存储获得的奖励

        state = env.reset()  # 重置环境，开始新的一集
        for steps in range(num_steps):
            # 前向传播通过网络
            value, policy_dist = actor_critic.forward(state)
            value = value.detach().numpy()[0,0]  # Critic 估计当前状态的价值
            dist = policy_dist.detach().numpy()
            
            # 从动作概率分布中采样一个动作
            action = np.random.choice(num_outputs, p=np.squeeze(dist))
            log_prob = torch.log(policy_dist.squeeze(0)[action])  # 记录选中动作的对数概率
            entropy = -np.sum(np.mean(dist) * np.log(dist))  # 使用熵衡量探索的多样性
            new_state, reward, done, _ = env.step(action)  # 执行动作，获取奖励和新状态

            # 记录这一段轨迹的数据
            rewards.append(reward)
            values.append(value)
            log_probs.append(log_prob)
            entropy_term += entropy
            state = new_state  # 更新为新的状态
            
            # 如果本 episode 结束，记录并退出本轮循环
            if done or steps == num_steps-1:
                Qval, _ = actor_critic.forward(new_state)  # 最后状态的价值估计
                Qval = Qval.detach().numpy()[0,0]
                all_rewards.append(np.sum(rewards))  # 记录本 episode 的总奖励
                all_lengths.append(steps)
                average_lengths.append(np.mean(all_lengths[-10:]))
                if episode % 10 == 0:
                    sys.stdout.write("episode: {}, reward: {}, total length: {}, average length: {} \n".format(
                        episode, np.sum(rewards), steps, average_lengths[-1]))
                break

        # 计算 Q 值（Critic 的目标值）
        Qvals = np.zeros_like(values)  # 初始化 Q 值数组
        for t in reversed(range(len(rewards))):
            Qval = rewards[t] + GAMMA * Qval  # Bellman 公式计算 Q 值
            Qvals[t] = Qval

        Qvals = torch.FloatTensor(Qvals)
        log_probs = torch.stack(log_probs)
        
        # 计算优势函数
        advantage = Qvals - values  # 动作的表现超出了 Critic 的期望多少？

        # 损失函数
        actor_loss = (-log_probs * advantage).mean()  # 策略损失（鼓励表现更好的动作）
        critic_loss = 0.5 * advantage.pow(2).mean()  # 价值函数损失（最小化预测误差）
        ac_loss = actor_loss + critic_loss + 0.001 * entropy_term  # 总损失，包括熵激励

        # 反向传播和优化
        ac_optimizer.zero_grad()
        ac_loss.backward()
        ac_optimizer.step()

Actor 更新基于策略梯度：我们将动作的对数概率与动作的优势相乘，若动作比预期表现好，优势为正，则鼓励该动作。
Critic 更新基于均方误差：Critic 比较其预测的 $V (s_t)$ 与实际的回报 $Q_t$，并最小化这个差异。
entropy_term：通过引入熵来鼓励探索行为，防止智能体过早收敛到某些动作而缺乏足够的探索。

总结#

A2C 的核心就是通过优势函数引导策略更新，同时借助TD 误差进行在线学习，让 agent 的每一步行动更加聪明和高效。

优势函数：不仅告诉你动作好坏，还告诉你相对 baseline 有多好。
TD 误差：一个实时计算优势的利器，帮助你快速高效地更新策略。

Asynchronous A2C(A3C)#

A3C 是在 Deepmind 的论文《Asynchronous Methods for Deep Reinforcement Learning》中提出的。本质上，A3C 相当于 并行化 的 A2C，多个并行的 agent （工作线程）在并行环境中独立更新全局的价值函数，因此被称为 “异步”，在现代的多核 CPU 上效率较高。

可以从前面看出来，A2C 是 A3C 的同步版本，在执行更新之前等待每个参与者完成其经验段，平均所有参与者的结果，优点是可以更高效地利用 GPU。并且在OpenAI Baselines: ACKTR & A2C中提到：
我们的同步 A2C 实现的性能优于我们的异步实现 —— 我们没有看到任何证据表明异步引入的噪声提供了任何性能优势。在使用单 GPU 机器时，这种 A2C 实现比 A3C 更具成本效益，并且在使用更大策略时比仅使用 CPU 的 A3C 实现更快。

Nagi-ovo