自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([1,?2,?3,?4,?5]))tf。Tensor(15,?shape=(),?dtype=int32)

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

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

深度强化学习

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

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

Actor-Critic方法

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

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

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

深度actor- critical方法

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

异步优势(asynchronous advantage) actor- critical

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

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

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

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

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

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

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

使用TensorFlow 2.0实现Advantage Actor-Critic

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

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

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

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

import?numpy?as?npimport?tensorflow?as?tfimport?tensorflow.keras.layers?as?klclass?ProbabilityDistribution(tf.keras。Model):jQuery183013280445588751233_1548295081438??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?gymenv?=?gym。make('CartPole-v0')model?=?Model(num_actions=env.action_space。n)obs?=?env。reset()#?no?feed_dict?or?tf.Session()?needed?at?allaction,?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(),?False,?0????????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?klsimport?tensorflow.keras。optimizers?as?koclass?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

在源代码中，我包含了一些额外的帮助程序，可以打印出正在运行的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?batchenv?=?gym。make('CartPole-v0')obs?=?np。repeat(env。reset()[None,?:],?100000,?axis=0)

Eager Benchmark

%%timemodel?=?Model(env.action_space。n)model。run_eagerly?=?Trueprint("Eager?Execution:??",?tf。executing_eagerly())print("Eager?Keras?Model:",?model。run_eagerly)_?=?model(obs)########?Results?#######Eager?Execution:???TrueEager?Keras?Model:?TrueCPU?times:?user?639?ms,?sys:?736?ms,?total:?1.38?s

Static Benchmark

%%timewith?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:???FalseEager?Keras?Model:?FalseCPU?times:?user?793?ms,?sys:?79.7?ms,?total:?873?ms

Default Benchmark

%%timemodel?=?Model(env.action_space。n)print("Eager?Execution:??",?tf。executing_eagerly())print("Eager?Keras?Model:",?model。run_eagerly)_?=?model.predict(obs)########?Results?#######Eager?Execution:???TrueEager?Keras?Model:?FalseCPU?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的一些缺陷。

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