切换视频源:

Asynchronous Advantage Actor-Critic (A3C) (Tensorflow)

作者: rcdrones 编辑: rcdrones 2017-05-03

学习资料:

要点

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

因为这节内容是基于 Actor-Critic, 所以还不了解 Actor-Critic 的朋友们, 强烈推荐你在这个短视频这个 Python 教程中获得了解,

下面是这节内容的效果提前看:

算法

A3C 的算法实际上就是将 Actor-Critic 放在了多个线程中进行同步训练. 可以想象成几个人同时在玩一样的游戏, 而他们玩游戏的经验都会同步上传到一个中央大脑. 然后他们又从中央大脑中获取最新的玩游戏方法.

这样, 对于这几个人, 他们的好处是: 中央大脑汇集了所有人的经验, 是最会玩游戏的一个, 他们能时不时获取到中央大脑的必杀招, 用在自己的场景中.

对于中央大脑的好处是: 中央大脑最怕一个人的连续性更新, 不只基于一个人推送更新这种方式能打消这种连续性. 使中央大脑不必有用像 DQN, DDPG 那样的记忆库也能很好的更新.

Asynchronous Advantage Actor-Critic (A3C) (Tensorflow)

为了达到这个目的, 我们要有两套体系, 可以看作中央大脑拥有 global net 和他的参数, 每位玩家有一个 global net 的副本 local net, 可以定时向 global net 推送更新, 然后定时从 global net 那获取综合版的更新.

如果在 tensorboard 中查看我们今天要建立的体系, 这就是你会看到的.

Asynchronous Advantage Actor-Critic (A3C) (Tensorflow)

W_0 就是第0个 worker, 每个 worker 都可以分享 global_net.

Asynchronous Advantage Actor-Critic (A3C) (Tensorflow)

如果我们调用 sync 中的 pull, 这个 worker 就会从 global_net 中获取到最新的参数.

Asynchronous Advantage Actor-Critic (A3C) (Tensorflow)

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

这次我们使用一个连续动作的环境 Pendulum 举例. 如果直接看所有代码, 请看我的 Github, 如果你处理的是一个离散动作环境, 可以参考这个Github 中的这个文件.

接下来我们就开始定义连续动作的 A3C 啦.

主结构

我们用 Tensorflow 搭建神经网络, 对于我们的 Actor, tensorboard 中可以看清晰的看到我们是如果搭建的:

Asynchronous Advantage Actor-Critic (A3C) (Tensorflow)

我们使用了 Normal distribution 来选择动作, 所以在搭建神经网络的时候, actor 这边要输出动作的均值和方差. 然后放入 Normal distribution 去选择动作. 计算 actor loss 的时候我们还需要使用到 critic 提供的 TD error 作为 gradient ascent 的导向.

Asynchronous Advantage Actor-Critic (A3C) (Tensorflow)

critic 很简单啦, 只需要得到他对于 state 的价值就好了. 用于计算 TD error.

Actor Critic 网络

其搭建的代码部分在这, 因为写下来全部代码比较眼花, 所以会有点伪代码 (如果想一次性看全部, 请去我的Github):

我们将 ActorCritic 合并成一整套系统, 这样方便运行.

# 这个 class 可以被调用生成一个 global net.
# 也能被调用生成一个 worker 的 net, 因为他们的结构是一样的,
# 所以这个 class 可以被重复利用.
class ACNet(object):
    def __init__(self, globalAC=None):
        # 当创建 worker 网络的时候, 我们传入之前创建的 globalAC 给这个 worker
        if 这是 global:   # 判断当下建立的网络是 local 还是 global
            with tf.variable_scope('Global_Net'):
                self._build_net()
        else:
            with tf.variable_scope('worker'):
                self._build_net()

            # 接着计算 critic loss 和 actor loss
            # 用这两个 loss 计算要推送的 gradients

            with tf.name_scope('sync'):  # 同步
                with tf.name_scope('pull'):
                    # 更新去 global
                with tf.name_scope('push'):
                    # 获取 global 参数

    def _build_net(self):
        # 在这里搭建 Actor 和 Critic 的网络
        return 均值, 方差, state_value

    def update_global(self, feed_dict):
        # 进行 push 操作

    def pull_global(self):
        # 进行 pull 操作

    def choose_action(self, s):
        # 根据 s 选动作

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

Worker

每个 worker 有自己的 class, class 里面有他的工作内容 work, 看全部请来我的 Github.

class Worker(object):
    def __init__(self, name, globalAC):
        self.env = gym.make(GAME).unwrapped # 创建自己的环境
        self.name = name    # 自己的名字
        self.AC = ACNet(name, globalAC) # 自己的 local net, 并绑定上 globalAC

    def 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 的 value
                    if done:
                        v_s_ = 0   # terminal
                    else:
                        v_s_ = SESS.run(self.AC.v, {self.AC.s: s_[np.newaxis, :]})[0, 0]

                    buffer_v_target = []    # 下 state value 的缓存, 用于算 TD
                    for r in buffer_r[::-1]:    # 进行 n_steps forward view
                        v_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)    # 推送更新去 globalAC
                    buffer_s, buffer_a, buffer_r = [], [], []   # 清空缓存
                    self.AC.pull_global()   # 获取 globalAC 的最新参数

                s = s_
                if done:
                    GLOBAL_EP += 1  # 加一回合
                    break   # 结束这回合

Worker 并行工作

这里才是真正的重点! Worker 的并行计算.

with tf.device("/cpu:0"):
    GLOBAL_AC = ACNet(GLOBAL_NET_SCOPE)  # 建立 Global AC
    workers = []
    for i in range(N_WORKERS):  # 创建 worker, 之后在并行
        workers.append(Worker(GLOBAL_AC))   # 每个 worker 都有共享这个 global AC

COORD = 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 的线程调度

我的电脑里可以建立 4个 worker, 也就可以把它们放在4个线程中并行探索更新. 最后的学习结果可以用这个获取 moving average 的 reward 的图来概括.

Asynchronous Advantage Actor-Critic (A3C) (Tensorflow)

上面讲到的是一个 continuous action 的例子, 全部代码在这里清晰可见. 还有一个是 discrete action 的例子. 使用的是 Cartpole 的实验, 代码在这. 同时, 我还做了一个 A3C 加上 RNN 的例子, 同样是用 Pendulum 的例子, 代码在这.

机械手臂

我也用这套 A3C 测试过自己写的机器手臂的环境, 发现效果也还行. 有兴趣的朋友可以看到这里.

有很多人留言说想要我做一个关于这个机器手臂的教程, 不负众望, 你可以在这里 看到我怎么从零开始, 手写环境, debug 测试, 来制作一个强化学习的机器手臂.

multiprocessing + A3C

除此之外, 我心里一直有一个疙瘩, 因为这个 A3C 中, 我用的是 python 的 threading, 懂 python 的朋友知道, threading 有 GIL, 运算速度是问题, 我的 CPU 都不是满格的. 我一直想把这个 A3C 代码移植去 multiprocessing, 提高效率. 但是 Tensorflow 的 session 就是和 multiprocessing 不兼容, Global Net 做不好. 怎么办?

Distributed Tensorflow 是一个备选方案. 但是这个要求你是在计算机集群上做, 不然速度上还不如这个 threading 的 A3C. 这时, 我不爽了, 到在知乎上抱怨了一番. 和知友们聊了会, 然后我想出了下面这个方案.

和 Tensorflow 一样, 我做过一些 Pytorch 的教程, pytorch 也是做神经网络的. 但是它是支持 multiprocessing 的. 我专门开了一个 repo, 把 Pytorch + multiprocessing 的代码分享了出来. 这会儿, CPU 满格, 心情舒畅多了~

Asynchronous Advantage Actor-Critic (A3C) (Tensorflow)

分享到: Facebook 微博 微信 Twitter
如果你觉得这篇文章或视频对你的学习很有帮助, 请你也分享它, 让它能再次帮助到更多的需要学习的人. 莫烦没有正式的经济来源, 如果你也想支持 莫烦Python 并看到更好的教学内容, 赞助他一点点, 作为鼓励他继续开源的动力.

支持 让教学变得更优秀