【强化学习】Asynchronous Advantage Actor-Critic（A3C）

1 A3C简介

A3C全称Asynchronous Advantage Actor-Critic，顾名思义，它采用的是Actor-Critic的形式（需要回顾Actor-Critic的，可以点击这里【强化学习】Actor-Critic（演员-评论家）算法详解）。为了训练一对Actor和Critic，我们将他复制多份红色的，然后同时放在不同的平行空间中，让他们各玩各的。然后每个红色副本都悄悄告诉黑色的Actor-Critic自己在那边的世界玩的怎么样，有哪些经验值得分享。然后还能从黑色的Actor-Critic这边再次获取综合考量所有副本经验后的通关秘籍。这样一来一回就形成了一种有效率的强化学习方式。
在这里插入图片描述
目前的计算机多半是双核，4核，甚至6核，8核。一般的学习方法，只能让机器人在一个核上面玩耍。但是如果使用A3C的方法，我们可以给他们安排去不同的核，并行运算。实验结果就是，这样的计算方式往往比传统的方式快上好多倍。

2 Asynchronous Advantage Actor-Critic (A3C)详解

2.1 要点

一句话概括A3C：Google DeepMind 提出的一种解决Actor-Critic不收敛问题的算法。它会创建多个并行的环境，让多个拥有副结构的agent同时在这些并行环境上更新主结构中的参数。并行中的agent们互不干扰，而主结构的参数更新收到副结构提交更新的不连续性干扰，所以更新的相关性被降低，收敛性提高。

2.2 算法

A3C的算法实际上就是将Actor-Critic放在了多个线程中进行同步训练。可以想象成几个人同时在玩一样的游戏，而他们玩游戏的经验都会同步上传的一个中央大脑。然后他们又从中央大脑中获取最新的玩游戏方法。
这样，对于这几个人，他们的好处是：中央大脑汇集了所有人的经验，是最会玩游戏的一个，他们能时不时获取到中央大脑的必杀招，用在自己的场景中。
对于中央大脑的好处是：中央大脑最怕一个人的连续性更新，不只基于一个人推送更新这种方式能打消这种连续性。使中央大脑不必有用像DQN，DDPG那样的记忆库也能很好的更新。
在这里插入图片描述
为了达到这个目的，我们要有两套体系，可以看作中央大脑拥有global net和他的参数，每位玩家有一个gobal net的副本local net，可以定时向global net推送更新，然后定时从glabal net那获取综合版的更新。
如果在tensorboard中查看我们要建立的体系：
在这里插入图片描述
w_0就是第0个worker，每个worker都可以分享global_net。

如果调用sync中的pull，这个worker就会从global_net中获取到最新的参数。

如果我们调用sync中的push，这个worker就会将自己的个人更新推送去global_net。

2.3 主结构

我们用Tensorflow搭建神经网络，对于我们的Actor，tensorboard中可以看清晰的看到我们是如何搭建的：
在这里插入图片描述
我们使用了Normal distribution 来选择动作，所以在搭建神经网络的时候，actor这边要输出动作的均值和方差。然后放入Normal distribution去选择动作。计算actor loss的时候我们还需要使用到critic提供的TD error作为gradient ascent 的导向。
在这里插入图片描述
critic很简单那，只需要得到他对于state的价值就好了，用于计算TD error。

2.4 Actor-Critic网络

我们将Actor和Critic合并成一整套系统，这样方便运行。

# 这个 class 可以被调用生成一个 global net.
# 也能被调用生成一个 worker 的 net, 因为他们的结构是一样的,
# 所以这个 class 可以被重复利用.
class ACNet(object):def __init__(self, globalAC=None):# 当创建 worker 网络的时候, 我们传入之前创建的 globalAC 给这个 workerif 这是 global:   # 判断当下建立的网络是 local 还是 globalwith tf.variable_scope('Global_Net'):self._build_net()else:with tf.variable_scope('worker'):self._build_net()# 接着计算 critic loss 和 actor loss# 用这两个 loss 计算要推送的 gradientswith tf.name_scope('sync'):  # 同步with tf.name_scope('pull'):# 更新去 globalwith tf.name_scope('push'):# 获取 global 参数def _build_net(self):# 在这里搭建 Actor 和 Critic 的网络return 均值, 方差, state_valuedef update_global(self, feed_dict):# 进行 push 操作def pull_global(self):# 进行 pull 操作def choose_action(self, s):# 根据 s 选动作

这些是在创建网络，worker还有属于自己的class，用来执行在每个线程里的工作。

2.5 Worker

每个woeker都有自己的class，class里面有他的工作内容work

class Worker(object):def __init__(self, name, globalAC):self.env = gym.make(GAME).unwrapped # 创建自己的环境self.name = name    # 自己的名字self.AC = ACNet(name, globalAC) # 自己的 local net, 并绑定上 globalACdef work(self):# s, a, r 的缓存, 用于 n_steps 更新buffer_s, buffer_a, buffer_r = [], [], []while not COORD.should_stop() and GLOBAL_EP < MAX_GLOBAL_EP:s = self.env.reset()for ep_t in range(MAX_EP_STEP):a = self.AC.choose_action(s)s_, r, done, info = self.env.step(a)buffer_s.append(s)  # 添加各种缓存buffer_a.append(a)buffer_r.append(r)# 每 UPDATE_GLOBAL_ITER 步 或者回合完了, 进行 sync 操作if total_step % UPDATE_GLOBAL_ITER == 0 or done:# 获得用于计算 TD error 的 下一 state 的 valueif done:v_s_ = 0   # terminalelse:v_s_ = SESS.run(self.AC.v, {self.AC.s: s_[np.newaxis, :]})[0, 0]buffer_v_target = []    # 下 state value 的缓存, 用于算 TDfor r in buffer_r[::-1]:    # 进行 n_steps forward viewv_s_ = r + GAMMA * v_s_buffer_v_target.append(v_s_)buffer_v_target.reverse()buffer_s, buffer_a, buffer_v_target = np.vstack(buffer_s), np.vstack(buffer_a), np.vstack(buffer_v_target)feed_dict = {self.AC.s: buffer_s,self.AC.a_his: buffer_a,self.AC.v_target: buffer_v_target,}self.AC.update_global(feed_dict)    # 推送更新去 globalACbuffer_s, buffer_a, buffer_r = [], [], []   # 清空缓存self.AC.pull_global()   # 获取 globalAC 的最新参数s = s_if done:GLOBAL_EP += 1  # 加一回合break   # 结束这回合

2.6 Worker并行工作

这里是真正的重点，Worker的并行计算。

with tf.device("/cpu:0"):GLOBAL_AC = ACNet(GLOBAL_NET_SCOPE)  # 建立 Global ACworkers = []for i in range(N_WORKERS):  # 创建 worker, 之后在并行workers.append(Worker(GLOBAL_AC))   # 每个 worker 都有共享这个 global ACCOORD = tf.train.Coordinator()  # Tensorflow 用于并行的工具worker_threads = []
for worker in workers:job = lambda: worker.work()t = threading.Thread(target=job)    # 添加一个工作线程t.start()worker_threads.append(t)
COORD.join(worker_threads)  # tf 的线程调度