酷应用

如何在TensorFlow 2.0中构建强化学习智能体

百家作者：机器之心 2019-01-21 09:12:54

选自 inoryy

作者：Roman Ring

机器之心编译

机器之心编辑部

TensorFlow 2.0 即将在今年上半年推出正式版，虽然目前有预览版可供使用，但是对于大多数人来说，新版本能够带来什么还没有具体的概念。本文将简要介绍在 TensorFlow 2.0 上使用强化学习算法的体验。其中作者概述了 Keras 子类 API、Eager Execution、会话替换以及会让开发更加方便的技巧。

对此，Keras 提出者、谷歌科学家 François Chollet 表示，这是一份非常详尽的介绍。

在这一教程中，我们将会使用 TensorFlow 2.0 新特性，并借助深度强化学习中的 A2C 智能体解决经典 CartPole-v0 环境任务。虽然我们的目标是展示 TensorFlow2.0，但与此同时我们也会尽量详细解释深度强化学习（DRL）的概念，其中包括这一领域的简要概述。

TensorFlow 2.0 版的宗旨是让开发者们能够更轻松，在深度强化学习上这一理念显然也得到了发扬：在这个例子中，我们的智能体源代码不到 150 行！如果你想看看代码，Python 文件格式的在这里：https://github.com/inoryy/tensorflow2-deep-reinforcement-learning

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 模式，实际上它意味着计算是实时执行的——而不再是通过预编译的计算图来执行。你可以在 TensorFlow 的文档中找到很好的概述：https://www.tensorflow.org/tutorials/eager/eager_basics

强化学习

强化学习指的是面向目标的算法，这种算法学习如何在一些具体的步骤中达到一个目标或者最大化；例如，最大化一个游戏中通过一些行动而获得的得分。它们可以从一个空白状态开始，然后在合适的条件下达到超越人类水平的性能。就像被糖果和体罚刺激的小孩子一样，当它们做出错误的预测时，这些算法会受到惩罚，当它们做出正确的预测时，它们会得到奖励—这便是强化的意义所在。

大多数人第一次听闻强化学习概念是来自 AlphaGo，结合深度学习的强化算法可以在围棋和 Atari 游戏中打败人类冠军。尽管这听起来还不具有足够的说服力，但是这已经远远优于它们之前的成就了，而且目前最先进的进步是很迅速的。

两个强化学习的算法 Deep-Q learning 和 A3C 已经在 Deeplearning4j 库上实现了，现在，它已经可以玩《毁灭战士（Doom）》了。

有关强化学习概念，可参阅：

通过 TensorFlow 2.0 实现 Actor-Critic 的优势

这一部分主要介绍实现许多现代 DRL 算法的基础：Actor-Critic 智能体。为了简单起见，这里并不会实现并行的工作站。不过大多数代码已经可以支持并行化，所以这也可以作为读者自行练习的机会。

作为测试平台，我们会使用 CartPole-v0 环境。虽然这个环境很简单，但它仍然是一个很好的入门级选择，我们经常依赖它作为实现 RL 算法异常检查的环境。

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

首先，我们可以在单个 Model 类下定义策略和价值估计网络：

import numpy as npimport tensorflow as tfimport tensorflow.keras.layers as kl
class ProbabilityDistribution(tf.keras.Model):def call(self, logits):# sample a random categorical action from given logitsreturn 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]

这里需要注意的是：

模型的层级和执行路径是独立定义的
模型并没有「input」层，它将接收原始的 NumPy 数组
两个计算路径可以通过函数式 API 在一个模型中定义
模型可以包含动作采样等辅助性方法
在实时运行模式中，所有模块都从 NumPy 数组开始运行

随机智能体

现在我们可以开始编写更有意思的模块：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, 0while 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

当然随机初始化的权重肯定是不能获得最佳奖励的，现在就需要定义训练部分了。

损失或目标函数

一般而言，智能体会通过对某些损失函数或目标函数执行梯度下降而提升策略效果。在 Actor-Critic 中，我们需要训练三个目标函数：利用加权梯度最大化和信息熵最大化提升策略效果，并最小化价值估计误差。

import tensorflow.keras.losses as klsimport 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 returnsreturn 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 minimizesreturn policy_loss - self.params['entropy']*entropy_loss

上面已经完成对目标函数的定义，这样的代码非常紧凑，甚至注释行都要比代码本身多。

智能体训练循环

最后，我们需要定义一个训练循环，它会相对长一点，但同样也非常直观：采集样本、计算反馈奖励和梯度、最后训练并更新模型。

import tensorflow.keras.losses as klsimport 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 returnsreturn 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 minimizesreturn policy_loss - self.params['entropy']*entropy_loss

训练和结果

现在已经预备好在 CartPole-v0 上训练单工作站的 A2C 智能体了，训练过程也就需要几分钟。在训练完成后，我们应该能看到智能体成功实现了 200/200 的目标分值。

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

在源代码中，我们还实现了一些额外的辅助模块，包括打印运行 episode 的奖励值和损失值等，并绘制出所有奖励值的历史变化。

静态计算图

有了这些令人兴奋的实时执行模式，你可能会考虑以前的静态计算图还能行吗？当然静态计算图也是可以的，我们只需要额外的代码行就能启动它。

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

需要注意的是，在静态计算图执行的期间，我们不能只使用 Tensor。这也就是为什么在模型定义的过程中需要使用 CategoricalDistribution 技巧。

One More Thing…

还记得我说过 TensorFlow 默认使用 eager 模式，甚至还用代码展示了一下。然而，并不是这样的，不完全是。

如果你是用 Keras API 来构建和管理你的模型，那么它将会将模型编译成静态图。因此你最终将获得静态计算图的性能和 eager execution 的灵活性。

你可以通过 model.run_eagerly 标记来检查模型状态，你也可以通过将这个 flag 设置为 True 来强制使用 eager 模式。如果 Keras 检测到无法实现 eager 模式，就会使用默认的模式。

为了说明它确实以静态图运行，以下是一个简单的基准：

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

Eager 基准

%%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

静态基准

%%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

默认基准

%%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