详解深度强化学习展现TensorFlow 2.0新特性(代码)

Posted 新智元

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了详解深度强化学习展现TensorFlow 2.0新特性(代码)相关的知识,希望对你有一定的参考价值。



  新智元报道  

来源:Reddit

编辑:三石、肖琴

【新智元导读】自TensorFlow官方发布其2.0版本新性能以来,不少人可能对此会有些许困惑。因此博主Roman Ring写了一篇概述性的文章,通过实现深度强化学习算法来具体的展示了TensorFlow 2.0的特性。


正所谓实践出真知。


TensorFlow 2.0的特性公布已经有一段时间了,但很多人对此应当还是一头雾水。


在本教程中,作者通过深度强化学习(DRL)来展示即将到来的TensorFlow 2.0的特性,具体来讲就是通过实现优势actor-critic(演员-评判家,A2C)智能体来解决经典的CartPole-v0环境。


虽然作者本文的目标是展示TensorFlow 2.0,但他先介绍了DRL方面的内容,包括对该领域的简要概述。


事实上,由于2.0版本的主要关注点是简化开发人员的工作,即易用性,所以现在正是使用TensorFlow进入DRL的好时机。


本文完整代码资源链接
GitHub:https://github.com/inoryy/tensorflow2-deep-reinforcement-learning

Google Colab:https://colab.research.google.com/drive/12QvW7VZSzoaF-Org-u-N6aiTdBN5ohNA


安装


由于TensorFlow 2.0仍处于试验阶段,建议将其安装在一个独立的(虚拟)环境中。我比较倾向于使用Anaconda,所以以此来做说明:


> conda create -n tf2 python=3.6
> source activate tf2
> pip install tf-nightly-2.0-preview # tf-nightly-gpu-2.0-preview for GPU version


让我们来快速验证一下,一切是否按着预测正常工作:


>>> import tensorflow as tf
>>> print(tf.__version__)
1.13.0-dev20190117
>>> print(tf.executing_eagerly())
True


不必担心1.13.x版本,这只是一个早期预览。此处需要注意的是,默认情况下我们是处于eager模式的!


>>> print(tf.reduce_sum([12345]))
tf.Tensor(15, shape=(), dtype=int32)


如果读者对eager模式并不熟悉,那么简单来讲,从本质上它意味着计算是在运行时(runtime)被执行的,而不是通过预编译的图(graph)来执行。读者也可以在TensorFlow文档中对此做深入了解:


https://www.tensorflow.org/tutorials/eager/eager_basics


深度强化学习


一般来说,强化学习是解决顺序决策问题的高级框架。RL智能体通过基于某些观察采取行动来导航环境,并因此获得奖励。大多数RL算法的工作原理是最大化智能体在一个轨迹中所收集的奖励的总和。


基于RL的算法的输出通常是一个策略—一个将状态映射到操作的函数。有效的策略可以像硬编码的no-op操作一样简单。随机策略表示为给定状态下行为的条件概率分布。


详解深度强化学习展现TensorFlow 2.0新特性(代码)


Actor-Critic方法


RL算法通常根据优化的目标函数进行分组。基于值的方法(如DQN)通过减少预期状态-动作值(state-action value)的误差来工作。


策略梯度(Policy Gradient)方法通过调整其参数直接优化策略本身,通常是通过梯度下降。完全计算梯度通常是很困难的,所以通常用蒙特卡洛(monte-carlo)方法来估计梯度。


最流行的方法是二者的混合:actor- critical方法,其中智能体策略通过“策略梯度”进行优化,而基于值的方法则用作期望值估计的引导。


深度actor- critical方法


虽然很多基础的RL理论是在表格案例中开发的,但现代RL几乎完全是用函数逼近器完成的,例如人工神经网络。 具体来说,如果策略和值函数用深度神经网络近似,则RL算法被认为是“深度的”。


详解深度强化学习展现TensorFlow 2.0新特性(代码)


异步优势(asynchronous advantage) actor- critical


多年来,为了解决样本效率和学习过程的稳定性问题,已经为此做出了一些改进。


首先,梯度用回报(return)来进行加权:折现的未来奖励,这在一定程度上缓解了信用(credit)分配问题,并以无限的时间步长解决了理论问题。


其次,使用优势函数代替原始回报。收益与基线(如状态行动估计)之间的差异形成了优势,可以将其视为与某一平均值相比某一给定操作有多好的衡量标准。


第三,在目标函数中使用额外的熵最大化项,以确保智能体充分探索各种策略。本质上,熵以均匀分布最大化,来测量概率分布的随机性。


最后,并行使用多个worker来加速样品采集,同时在训练期间帮助将它们去相关(decorrelate)。


将所有这些变化与深度神经网络结合起来,我们得到了两种最流行的现代算法:异步优势actor- critical算法,或简称A3C/A2C。两者之间的区别更多的是技术上的而不是理论上的:顾名思义,它归结为并行worker如何估计其梯度并将其传播到模型中。


详解深度强化学习展现TensorFlow 2.0新特性(代码)


有了这些,我将结束我们的DRL方法之旅,因为这篇博客文章的重点是TensorFlow 2.0特性。如果您仍然不确定主题,不要担心,通过代码示例,一切都会变得更加清晰明了。


使用TensorFlow 2.0实现Advantage Actor-Critic


让我们看看实现各种现代DRL算法的基础是什么:是actor-critic agent,如前一节所述。为了简单起见,我们不会实现并行worker,尽管大多数代码都支持它。感兴趣的读者可以将这作为一个练习机会。


作为一个测试平台,我们将使用CartPole-v0环境。虽然有点简单,但它仍然是一个很好的选择。


通过Keras模型API实现的策略和价值


首先,让我们在单个模型类下创建策略和价值预估神经网络:


import numpy as np
import tensorflow as tf
import tensorflow.keras.layers as kl

class ProbabilityDistribution(tf.keras.Model):
    def call(self, logits):
        # sample a random categorical action from given logits
        return tf.squeeze(tf.random.categorical(logits, 1), axis=-1)

class Model(tf.keras.Model):
    def __init__(self, num_actions):
        super().__init__('mlp_policy')
        # no tf.get_variable(), just simple Keras API
        self.hidden1 = kl.Dense(128, activation='relu')
        self.hidden2 = kl.Dense(128, activation='relu')
        self.value = kl.Dense(1, name='value')
        # logits are unnormalized log probabilities
        self.logits = kl.Dense(num_actions, name='policy_logits')
        self.dist = ProbabilityDistribution()

    def call(self, inputs):
        # inputs is a numpy array, convert to Tensor
        x = tf.convert_to_tensor(inputs, dtype=tf.float32)
        # separate hidden layers from the same input tensor
        hidden_logs = self.hidden1(x)
        hidden_vals = self.hidden2(x)
        return self.logits(hidden_logs), self.value(hidden_vals)

    def action_value(self, obs):
        # executes call() under the hood
        logits, value = self.predict(obs)
        action = self.dist.predict(logits)
        # a simpler option, will become clear later why we don't use it
        # action = tf.random.categorical(logits, 1)
        return np.squeeze(action, axis=-1), np.squeeze(value, axis=-1)


然后验证模型是否如预期工作:


import gym

env = gym.make('CartPole-v0')
model = Model(num_actions=env.action_space.n)

obs = env.reset()
# no feed_dict or tf.Session() needed at all
action, value = model.action_value(obs[None, :])
print(action, value) # [1] [-0.00145713]


这里需要注意的是:

  • 模型层和执行路径是分别定义的

  • 没有“输入”层,模型将接受原始numpy数组

  • 通过函数API可以在一个模型中定义两个计算路径

  • 模型可以包含一些辅助方法,比如动作采样

  • 在eager模式下,一切都可以从原始numpy数组中运行


Random Agent


现在让我们转到 A2CAgent 类。首先,让我们添加一个 test 方法,该方法运行完整的episode并返回奖励的总和。


class A2CAgent:
    def __init__(self, model):
        self.model = model

    def test(self, env, render=True):
        obs, done, ep_reward = env.reset(), False0
        while not done:
            action, _ = self.model.action_value(obs[None, :])
            obs, reward, done, _ = env.step(action)
            ep_reward += reward
            if render:
                env.render()
        return ep_reward


让我们看看模型在随机初始化权重下的得分:


agent = A2CAgent(model)
rewards_sum = agent.test(env)
print("%d out of 200" % rewards_sum) # 18 out of 200


离最佳状态还很远,接下来是训练部分!


损失/目标函数


正如我在DRL概述部分中所描述的,agent通过基于某些损失(目标)函数的梯度下降来改进其策略。在 actor-critic 中,我们针对三个目标进行训练:利用优势加权梯度加上熵最大化来改进策略,以及最小化价值估计误差。


import tensorflow.keras.losses as kls
import tensorflow.keras.optimizers as ko

class A2CAgent:
    def __init__(self, model):
        # hyperparameters for loss terms
        self.params = {'value'0.5'entropy'0.0001}
        self.model = model
        self.model.compile(
            optimizer=ko.RMSprop(lr=0.0007),
            # define separate losses for policy logits and value estimate
            loss=[self._logits_loss, self._value_loss]
        )

    def test(self, env, render=True):
        # unchanged from previous section
        ...

    def _value_loss(self, returns, value):
        # value loss is typically MSE between value estimates and returns
        return self.params['value']*kls.mean_squared_error(returns, value)

    def _logits_loss(self, acts_and_advs, logits):
        # a trick to input actions and advantages through same API
        actions, advantages = tf.split(acts_and_advs, 2, axis=-1)
        # polymorphic CE loss function that supports sparse and weighted options
        # from_logits argument ensures transformation into normalized probabilities
        cross_entropy = kls.CategoricalCrossentropy(from_logits=True)
        # policy loss is defined by policy gradients, weighted by advantages
        note: we only calculate the loss on the actions we've actually taken
        # thus under the hood a sparse version of CE loss will be executed
        actions = tf.cast(actions, tf.int32)
        policy_loss = cross_entropy(actions, logits, sample_weight=advantages)
        # entropy loss can be calculated via CE over itself
        entropy_loss = cross_entropy(logits, logits)
        # here signs are flipped because optimizer minimizes
        return policy_loss - self.params['entropy']*entropy_loss


我们完成了目标函数!注意代码非常紧凑:注释行几乎比代码本身还多。


Agent Training Loop 


最后,还有训练环路。它有点长,但相当简单:收集样本,计算回报和优势,并在其上训练模型。


class A2CAgent:
    def __init__(self, model):
        # hyperparameters for loss terms
        self.params = {'value'0.5'entropy'0.0001'gamma'0.99}
        # unchanged from previous section
        ...

   def train(self, env, batch_sz=32, updates=1000):
        # storage helpers for a single batch of data
        actions = np.empty((batch_sz,), dtype=np.int32)
        rewards, dones, values = np.empty((3, batch_sz))
        observations = np.empty((batch_sz,) + env.observation_space.shape)
        # training loop: collect samples, send to optimizer, repeat updates times
        ep_rews = [0.0]
        next_obs = env.reset()
        for update in range(updates):
            for step in range(batch_sz):
                observations[step] = next_obs.copy()
                actions[step], values[step] = self.model.action_value(next_obs[None, :])
                next_obs, rewards[step], dones[step], _ = env.step(actions[step])

                ep_rews[-1] += rewards[step]
                if dones[step]:
                    ep_rews.append(0.0)
                    next_obs = env.reset()

            _, next_value = self.model.action_value(next_obs[None, :])
            returns, advs = self._returns_advantages(rewards, dones, values, next_value)
            # a trick to input actions and advantages through same API
            acts_and_advs = np.concatenate([actions[:, None], advs[:, None]], axis=-1)
            # performs a full training step on the collected batch
            note: no need to mess around with gradients, Keras API handles it
            losses = self.model.train_on_batch(observations, [acts_and_advs, returns])
        return ep_rews

    def _returns_advantages(self, rewards, dones, values, next_value):
        # next_value is the bootstrap value estimate of a future state (the critic)
        returns = np.append(np.zeros_like(rewards), next_value, axis=-1)
        # returns are calculated as discounted sum of future rewards
        for t in reversed(range(rewards.shape[0])):
            returns[t] = rewards[t] + self.params['gamma'] * returns[t+1] * (1-dones[t])
        returns = returns[:-1]
        # advantages are returns - baseline, value estimates in our case
        advantages = returns - values
        return returns, advantages

    def test(self, env, render=True):
        # unchanged from previous section
        ...

    def _value_loss(self, returns, value):
        # unchanged from previous section
        ...

    def _logits_loss(self, acts_and_advs, logits):
        # unchanged from previous section
        ...


训练&结果


我们现在已经准备好在CartPole-v0上训练这个single-worker A2C agent!训练过程应该只用几分钟。训练结束后,你应该看到一个智能体成功地实现了200分的目标。


rewards_history = agent.train(env)
print("Finished training, testing...")
print("%d out of 200" % agent.test(env)) # 200 out of 200

详解深度强化学习展现TensorFlow 2.0新特性(代码)

在源代码中,我包含了一些额外的帮助程序,可以打印出正在运行的episode的奖励和损失,以及rewards_history。



静态计算图


eager mode效果这么好,你可能会想知道静态图执行是否也可以。当然是可以!而且,只需要多加一行代码就可以启用静态图执行。


with tf.Graph().as_default():
    print(tf.executing_eagerly()) # False

    model = Model(num_actions=env.action_space.n)
    agent = A2CAgent(model)

    rewards_history = agent.train(env)
    print("Finished training, testing...")
    print("%d out of 200" % agent.test(env)) # 200 out of 200


有一点需要注意的是,在静态图执行期间,我们不能只使用 Tensors,这就是为什么我们需要在模型定义期间使用CategoricalDistribution的技巧。


One More Thing…


还记得我说过TensorFlow在默认情况下以eager 模式运行,甚至用一个代码片段来证明它吗?好吧,我骗了你。


如果你使用Keras API来构建和管理模型,那么它将尝试在底层将它们编译为静态图。所以你最终得到的是静态计算图的性能,它具有eager execution的灵活性。


你可以通过model.run_eager标志检查模型的状态,还可以通过将此标志设置为True来强制使用eager mode,尽管大多数情况下可能不需要这样做——如果Keras检测到没有办法绕过eager mode,它将自动退出。


为了说明它确实是作为静态图运行的,这里有一个简单的基准测试:


# create a 100000 samples batch
env = gym.make('CartPole-v0')
obs = np.repeat(env.reset()[None, :], 100000, axis=0)


Eager Benchmark


%%time

model = Model(env.action_space.n)
model.run_eagerly = True

print("Eager Execution:  ", tf.executing_eagerly())
print("Eager Keras Model:", model.run_eagerly)

_ = model(obs)

######## Results #######

Eager Execution:   True
Eager Keras Model: True
CPU times: user 639 ms, sys: 736 ms, total: 1.38 s


Static Benchmark


%%time

with tf.Graph().as_default():
    model = Model(env.action_space.n)

    print("Eager Execution:  ", tf.executing_eagerly())
    print("Eager Keras Model:", model.run_eagerly)

    _ = model.predict(obs)

######## Results #######

Eager Execution:   False
Eager Keras Model: False
CPU times: user 793 ms, sys: 79.7 ms, total: 873 ms


Default Benchmark


%%time

model = Model(env.action_space.n)

print("Eager Execution:  ", tf.executing_eagerly())
print("Eager Keras Model:", model.run_eagerly)

_ = model.predict(obs)

######## Results #######

Eager Execution:   True
Eager Keras Model: False
CPU times: user 994 ms, sys: 23.1 ms, total: 1.02 s


正如你所看到的,eager模式位于静态模式之后,默认情况下,模型确实是静态执行的。


结论


希望本文对理解DRL和即将到来的TensorFlow 2.0有所帮助。请注意,TensorFlow 2.0仍然只是预览版的,一切都有可能发生变化,如果你对TensorFlow有什么特别不喜欢(或喜欢:))的地方,请反馈给开发者。


一个总被提起的问题是,TensorFlow是否比PyTorch更好?也许是,也许不是。两者都是很好的库,所以很难说是哪一个更好。如果你熟悉PyTorch,你可能会注意到TensorFlow 2.0不仅赶上了它,而且还避免了PyTorch API的一些缺陷。


无论最后谁胜出,对于开发者来说,这场竞争给双方都带来了净积极的结果,我很期待看到这些框架未来会变成什么样子。



【加入社群】


以上是关于详解深度强化学习展现TensorFlow 2.0新特性(代码)的主要内容,如果未能解决你的问题,请参考以下文章

深度强化学习动手,第 7 章。无法让 tensorflow 工作

教程 | 深度强化学习入门:用TensorFlow构建你的第一个游戏AI

AI大事件 | 李飞飞接受访谈, TensorFlow: 发布 1.4.0,深度强化学习Bootcamp-视频讲座&实验

谷歌官方解读 TensorFlow 2.0 所有新特性

解读:TensorFlow 2.0中即将到来的所有新特性

官方解读:TensorFlow 2.0中即将到来的所有新特性