Roman Ring作者inoryy选自

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

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

正如你所看到的,Eager 模式是在静态模式之后的,在默认情况下,模型确实是静态执行的,所以也匹配显示静态图执行的形式。

结论

希望本文可以让你了解深度强化学习及其在 TensorFlow 2.0 中的实现方式。请注意,在文中使用的仍然是「每晚预览版本」,它甚至还不是正式版的候选版本。一切都可能会发生改变,不过这也意味着如果你对新版本的 TensorFlow 有什么不喜欢的地方,可以尽情地去提意见。

还有一个经常出现的问题:TensorFlow 和 PyTorch 比谁好?现在我们还无法确定,这两个深度学习框架都非常流行,我们很难说哪个更好。不过如果你很熟悉 PyTorch,你应该可以看得出 TenrorFlow 2.0 不仅补齐了缺点,而且还避免了 PyTorch API 的一些短板。

不论如何,深度学习框架的竞争对于用户来说都是好事,我们可以期待未来它们会变成什么样子。

原文链接:http://inoryy.com/post/tensorflow2-deep-reinforcement-learning/

工程深度强化学习TensorFlow
8
相关数据
DeepMind机构

DeepMind是一家英国的人工智能公司。公司创建于2010年,最初名称是DeepMind科技(DeepMind Technologies Limited),在2014年被谷歌收购。在2010年由杰米斯·哈萨比斯,谢恩·列格和穆斯塔法·苏莱曼成立创业公司。继AlphaGo之后,Google DeepMind首席执行官杰米斯·哈萨比斯表示将研究用人工智能与人类玩其他游戏,例如即时战略游戏《星际争霸II》(StarCraft II)。深度AI如果能直接使用在其他各种不同领域,除了未来能玩不同的游戏外,例如自动驾驶、投资顾问、音乐评论、甚至司法判决等等目前需要人脑才能处理的工作,基本上也可以直接使用相同的神经网上去学而习得与人类相同的思考力。

深度学习技术

深度学习(deep learning)是机器学习的分支,是一种试图使用包含复杂结构或由多重非线性变换构成的多个处理层对数据进行高层抽象的算法。 深度学习是机器学习中一种基于对数据进行表征学习的算法,至今已有数种深度学习框架,如卷积神经网络和深度置信网络和递归神经网络等已被应用在计算机视觉、语音识别、自然语言处理、音频识别与生物信息学等领域并获取了极好的效果。

深度强化学习技术

强化学习(Reinforcement Learning)是主体(agent)通过与周围环境的交互来进行学习。强化学习主体(RL agent)每采取一次动作(action)就会得到一个相应的数值奖励(numerical reward),这个奖励表示此次动作的好坏。通过与环境的交互,综合考虑过去的经验(exploitation)和未知的探索(exploration),强化学习主体通过试错的方式(trial and error)学会如何采取下一步的动作,而无需人类显性地告诉它该采取哪个动作。强化学习主体的目标是学习通过执行一系列的动作来最大化累积的奖励(accumulated reward)。 一般来说,真实世界中的强化学习问题包括巨大的状态空间(state spaces)和动作空间(action spaces),传统的强化学习方法会受限于维数灾难(curse of dimensionality)。借助于深度学习中的神经网络,强化学习主体可以直接从原始输入数据(如游戏图像)中提取和学习特征知识,然后根据提取出的特征信息再利用传统的强化学习算法(如TD Learning,SARSA,Q-Learnin)学习控制策略(如游戏策略),而无需人工提取或启发式学习特征。这种结合了深度学习的强化学习方法称为深度强化学习。

权重技术

线性模型中特征的系数,或深度网络中的边。训练线性模型的目标是确定每个特征的理想权重。如果权重为 0,则相应的特征对模型来说没有任何贡献。

基准技术

一种简单的模型或启发法,用作比较模型效果时的参考点。基准有助于模型开发者针对特定问题量化最低预期效果。

损失函数技术

在数学优化,统计学,计量经济学,决策理论,机器学习和计算神经科学等领域,损失函数或成本函数是将一或多个变量的一个事件或值映射为可以直观地表示某种与之相关“成本”的实数的函数。

TensorFlow技术

TensorFlow是一个开源软件库,用于各种感知和语言理解任务的机器学习。目前被50个团队用于研究和生产许多Google商业产品,如语音识别、Gmail、Google 相册和搜索,其中许多产品曾使用过其前任软件DistBelief。

信息熵技术

在信息论中,熵是接收的每条消息中包含的信息的平均量,又被称为信息熵、信源熵、平均自信息量。这里,“消息”代表来自分布或数据流中的事件、样本或特征。熵的单位通常为比特,但也用Sh、nat、Hart计量,取决于定义用到对数的底。

梯度下降技术

梯度下降是用于查找函数最小值的一阶迭代优化算法。 要使用梯度下降找到函数的局部最小值,可以采用与当前点的函数梯度(或近似梯度)的负值成比例的步骤。 如果采取的步骤与梯度的正值成比例,则接近该函数的局部最大值,被称为梯度上升。

目标函数技术

目标函数f(x)就是用设计变量来表示的所追求的目标形式,所以目标函数就是设计变量的函数,是一个标量。从工程意义讲,目标函数是系统的性能标准,比如,一个结构的最轻重量、最低造价、最合理形式;一件产品的最短生产时间、最小能量消耗;一个实验的最佳配方等等,建立目标函数的过程就是寻找设计变量与目标的关系的过程,目标函数和设计变量的关系可用曲线、曲面或超曲面表示。

强化学习技术

强化学习是一种试错方法,其目标是让软件智能体在特定环境中能够采取回报最大化的行为。强化学习在马尔可夫决策过程环境中主要使用的技术是动态规划(Dynamic Programming)。流行的强化学习方法包括自适应动态规划(ADP)、时间差分(TD)学习、状态-动作-回报-状态-动作(SARSA)算法、Q 学习、深度强化学习(DQN);其应用包括下棋类游戏、机器人控制和工作调度等。

推荐文章
暂无评论
暂无评论~