24/12/8 算法笔记＜强化学习＞ AC：actor-critic

Actor网络是根据状态来学习策略来指导动作。

Critic网络是用于评估状态价值。

PG的优化

回顾PG(策略梯度) 它是利用带权重的梯度下降方法更新策略，而获得权重的方法是蒙塔卡洛计算G值。

而蒙特卡洛回溯是需要完成整个游戏，直到最终状态才能回溯。这使得PG的效率被限制，蒙特卡洛的效率是相对比较低的，因为要走完整个过程，所以我们想到用TD代替MC,可以不用回溯，每走一步计算出G值。

解决网络更新问题

Critic网络通过拿到这一时刻和下一时刻的状态，来计算这一时刻和下一时刻的V.

Actor中用Q值和V值指导更新，但是如果我们用两个神经网络分别去更新两个值就会造成两倍的发现估测不准，我们可以统一成V,,

Critic，Actor网络怎么更新呢，我们用下一时刻的V值乘上gamma减去这一时刻的V值作为TD-error,这个TD-error就可以作为loss值来更新Critic网络，然后TD-error就可以放入我们目标函数的导函数里面，通过求导来调整Actor网络。

修改reward使得智能体玩游戏更久。

如果已经结束，那么奖励扣20，在某个濒临死亡的状态下，选择动作A会使得智能体死亡，游戏结束，我们不希望游戏结束，我们希望智能体能在濒临死亡的状态下力往狂澜。

于是我们把reward减去20,相当于在濒临死亡的状态下对动作A的不认同，通过reward减少20，能大幅度减少动作A出现的概率。

我们来看下代码

定义两个网络

import torch
import torch.nn as nn
import torch.optim as optim
import numpy as np
import gym


# 定义Actor网络
class Actor(nn.Module):
    def __init__(self, state_size, action_size):
        super(Actor, self).__init__()
        self.state_size = state_size
        self.action_size = action_size

        self.fc1 = nn.Linear(state_size, 128)
        self.fc2 = nn.Linear(128, 64)
        self.fc3 = nn.Linear(64, action_size)

    def forward(self, state):
        x = torch.relu(self.fc1(state))
        x = torch.relu(self.fc2(x))
        x = torch.softmax(self.fc3(x), dim=-1)
        return x


# 定义Critic网络
class Critic(nn.Module):
    def __init__(self, state_size):
        super(Critic, self).__init__()
        self.state_size = state_size

        self.fc1 = nn.Linear(state_size, 128)
        self.fc2 = nn.Linear(128, 64)
        self.fc3 = nn.Linear(64, 1)

    def forward(self, state):
        x = torch.relu(self.fc1(state))
        x = torch.relu(self.fc2(x))
        x = self.fc3(x)
        return x

构建训练网络

def train(env_name, num_episodes=1000, gamma=0.99, lr=0.001):
    env = gym.make(env_name)
    state_size = env.observation_space.shape[0]
    action_size = env.action_space.n

    actor = Actor(state_size, action_size)
    critic = Critic(state_size)

    actor_optimizer = optim.Adam(actor.parameters(), lr=lr)
    critic_optimizer = optim.Adam(critic.parameters(), lr=lr)

    for episode in range(num_episodes):
        state = env.reset()
        state = torch.FloatTensor(state).unsqueeze(0)
        done = False
        total_reward = 0

        # 用于存储每个时间步的奖励、下一个状态以及是否结束等信息，方便后续计算TD-error
        rewards = []
        next_states = []
        dones = []

        while not done:
            # Actor根据当前状态选择动作（基于概率分布采样动作）
            action_probs = actor(state)
            action_dist = torch.distributions.Categorical(action_probs)
            action = action_dist.sample()

            next_state, reward, done, _ = env.step(action.item())
            next_state = torch.FloatTensor(next_state).unsqueeze(0)

            rewards.append(reward)
            next_states.append(next_state)
            dones.append(done)

            state = next_state
            total_reward += reward

        

计算TD-error,更新网络

TD-error是通过Critic网络计算得到的

        # 计算每个时间步的TD-error
        returns = []
        R = torch.tensor(0.0) if done else critic(next_states[-1])
        for r, d in reversed(list(zip(rewards, dones))):
            R = r + gamma * R * (1 - d)
            returns.insert(0, R)
        returns = torch.stack(returns).detach()

        states = torch.cat([state.unsqueeze(0) for state in [env.reset()] + next_states[:-1]])
        values = critic(states).squeeze(-1)
        td_errors = returns - values

        # 更新Critic网络
        critic_loss = (td_errors ** 2).mean()
        critic_optimizer.zero_grad()
        critic_loss.backward()
        critic_optimizer.step()

        # 更新Actor网络（使用优势函数，这里使用TD-error作为优势函数的近似）
        action_probs = actor(states)
        action_log_probs = torch.log(action_probs.gather(1, torch.tensor([[a] for a in range(action_size)])))
        actor_loss = -(action_log_probs * td_errors.detach()).mean()

        actor_optimizer.zero_grad()
        actor_loss.backward()
        actor_optimizer.step()

        if (episode + 1) % 100 == 0:
            print(f'Episode: {episode + 1}, Total Reward: {total_reward}')

    env.close()


if __name__ == "__main__":
    train('CartPole-v1')

改进

我们可以通过求解期望价值计算，再最后用期望价值和TD-error来更新网络

        with tf.variable_scope('exp_v'):
            log_prob = tf.log(self.acts_prob[0, self.a])
            self.exp_v = tf.reduce_mean(log_prob * self.td_error)

A3C

强化学习的一个难点是智能体需要不断与环境交互获得数据才能更新，与有监督学习相比数据数量太少。有人想出我用多个智能体和环境交互，每个智能体产出的数据就可以一起给模型训练了。

基于这个思路和AC架构就有了A3C.

A3C中worker不仅要和环境互动，产生数据，而且要自己从这些数据中学习到”心得“。这里所谓心得，其实就是计算出来的梯度。需要强调的是，worker向全局网络汇总的是1梯度，而不是自己探索出来的数据。

A3C和DPPO

DPPO也是一个分布式架构,但DPPO的work自己不学习，而是提交数据让全局网络学习。

A3C的worker向全局网络汇总之后，并应用在全局网络的参数后，全局网络会把当前学习到的最新版本的参数，直接给worker。worker按照最新的网络继续跟环境做互动，互动后，再把梯度提交，获取最小的参数。

A3C的代码

导入库

import tensorflow as tf
import numpy as np
import gym
import threading
import time

Python 标准库中的多线程模块，用于创建和管理多个并行的工作线程，实现 A3C 算法中多个智能体异步并行训练的机制。

定义全局优化器

global_actor_optimizer = tf.train.RMSPropOptimizer(learning_rate=0.0007)
global_critic_optimizer = tf.train.RMSPropOptimizer(learning_rate=0.0007)

定义了两个全局共享的 RMSProp 优化器，分别用于更新 Actor 网络和 Critic 网络的参数

定义 Actor 网络

class ActorNetwork(tf.keras.Model):
    def __init__(self, n_features, n_actions):
        super(ActorNetwork, self).__init__()
        self.dense1 = tf.keras.layers.Dense(128, activation='relu', kernel_initializer='he_normal')
        self.dense2 = tf.keras.layers.Dense(64, activation='relu', kernel_initializer='he_normal')
        self.dense3 = tf.keras.layers.Dense(n_actions, activation='softmax', kernel_initializer='he_normal')

    def call(self, state):
        x = self.dense1(state)
        x = self.dense2(x)
        return self.dense3(x)

定义 Critic 网络

class CriticNetwork(tf.keras.Model):
    def __init__(self, n_features):
        super(CriticNetwork, self).__init__()
        self.dense1 = tf.keras.layers.Dense(128, activation='relu', kernel_initializer='he_normal')
        self.dense2 = tf.keras.layers.Dense(64, activation='relu', kernel_initializer='he_normal')
        self.dense3 = tf.keras.layers.Dense(1, kernel_initializer='he_normal')

    def call(self, state):
        x = self.dense1(state)
        x = self.dense2(x)
        return self.dense3(x)

计算优势函数（基于 GAE）

def compute_advantage(rewards, values, next_value, gamma, lamda):
    advantages = []
    returns = []
    gae = 0
    for step in reversed(range(len(rewards))): #反向遍历时间步计算相关量
        delta = rewards[step] + gamma * (next_value if step == len(rewards) - 1 else values[step + 1]) - values[step]
        gae = delta + gamma * lamda * gae
        advantages.insert(0, gae)
        returns.insert(0, gae + values[step])
    return np.array(advantages), np.array(returns)

这个函数实现了基于广义优势估计（Generalized Advantage Estimation，GAE）来计算优势函数以及回报的逻辑。GAE 是一种在强化学习中用于更准确地估计优势的方法，它结合了多步的时间差分信息，在偏差和方差之间进行了较好的平衡。

反向遍历时间步计算：

        从回合的最后一个时间步开始，逆向遍历整个回合的所有时间步。这种反向遍历的方式是符合 GAE 计算逻辑的，因为 GAE 需要基于后续时间步的信息来逐步往前推算每个时间步的优势估计和回报值。

计算时间差分(TD)误差delta:

        它衡量了当前实际获得的奖励加上未来预期价值与当前状态价值估计之间的差距。

计算广义优势估计（GAE）值 gae 并更新 advantages 列表:

        gae = delta + gamma * lamda * gae
        advantages.insert(0, gae)

        根据 GAE 的计算公式来更新 gae 的值，它是在当前时间步的 TD 误差 delta 的基础上，加上考虑了折扣因子 gamma、GAE 参数 lamda 以及上一步计算得到的 gae 值（代表了后续时间步累积过来的优势估计信息）的加权和，这样就逐步将多步的信息融合进来，得到当前时间步更全面的优势估计。然后将计算得到的 gae 值插入到 advantages 列表的开头，使得列表中的元素顺序与时间步顺序保持一致（虽然是反向遍历计算，但通过插入到开头来恢复正向顺序）。

计算回报（Returns）并更新 returns 列表：

        returns.insert(0, gae + values[step])

        它是当前时间步的广义优势估计值 gae 与当前状态价值估计 values[step] 之和。直观理解上，回报就是综合考虑了当前状态本身的价值以及采取动作后相对于平均水平的优势所带来的总体价值量，将这个计算得到的回报值插入到 returns 列表的开头，同样保持与时间步正向顺序一致。

定义工作线程执行的训练函数

def worker_thread(worker_id, global_actor, global_critic, env_name, num_episodes, gamma, lamda):
    env = gym.make(env_name)
    n_features = env.observation_space.shape[0]
    n_actions = env.action_space.n

    local_actor = ActorNetwork(n_features, n_actions)
    local_critic = CriticNetwork(n_features)

    local_actor.set_weights(global_actor.get_weights())
    local_critic.set_weights(global_critic.get_weights())

        每个线程都拥有自己独立的本地 Actor 和 Critic 网络，并且会与全局的网络参数进行交互和同步。

        在每个回合里，智能体根据当前状态选择动作，在环境中执行该动作，获取相应的奖励和下一个状态信息，同时记录下本回合内的奖励、状态以及状态价值估计等数据，以便后续用于计算优势函数、更新网络等操作。

    for episode in range(num_episodes):
        state = env.reset()
        state = np.expand_dims(state, axis=0)
        done = False
        episode_rewards = []
        episode_values = []
        episode_states = []

        while not done:
            action_probs = local_actor(tf.convert_to_tensor(state, dtype=tf.float32))
            action_dist = tf.distributions.Categorical(action_probs)
            action = action_dist.sample().numpy()[0]

            next_state, reward, done, _ = env.step(action)
            next_state = np.expand_dims(next_state, axis=0)

            episode_rewards.append(reward)
            episode_states.append(state)

            value = local_critic(tf.convert_to_tensor(state, dtype=tf.float32)).numpy()[0][0]

A3C三个A的特点：

Asynchronous（异步）：

        它利用多个工作线程（worker threads）同时在多个环境实例中进行训练。这些线程独立地与各自的环境副本进行交互，收集经验数据（如状态、动作、奖励等）。每个线程都有自己的局部 Actor - Critic 网络副本，并且会定期将计算得到的梯度更新应用到全局的 Actor - Critic 网络。这种异步的方式大大提高了数据收集和训练的效率，因为多个线程可以并行地探索不同的状态 - 动作空间区域，减少了训练时间。

Advantage（优势）：

        优势函数（Advantage Function）起到关键作用。优势函数用于衡量在某个状态下采取某个动作相较于平均水平的好坏程度。具体计算通常基于时间差分（TD）误差或者广义优势估计（GAE）等方法。在计算优势函数时，会考虑即时奖励、未来状态价值估计以及折扣因子等因素。通过优势函数，Actor 网络可以学习到更优的策略，即选择那些能够带来更高优势的动作。这种基于优势的学习方式有助于智能体更快地收敛到较好的策略，提高学习效率。

Actor - Critic（演员 - 评论家）：

        A3C 是基于 Actor - Critic 架构的算法。Actor 网络负责生成动作策略，根据输入的状态输出动作的概率分布，智能体基于这个概率分布来选择动作与环境进行交互。Critic 网络则负责评估状态的价值，它会根据输入的状态输出一个状态价值估计。在训练过程中，Critic 网络的输出（状态价值估计）和实际获得的奖励一起用于计算优势函数，然后 Actor 网络根据优势函数来调整自己的策略，以最大化长期累积奖励。这种 Actor - Critic 架构结合了基于策略（Actor）和基于价值（Critic）的学习方法的优点，使得算法能够更有效地学习到复杂环境中的最优策略。