深度学习算法(第37期)----如何用强化学习玩游戏？
上期我们一起学习了强化学习中的时间差分学习和近似Q学习的相关知识，
深度学习算法(第36期)----强化学习之时间差分学习与近似Q学习
今天我们一起用毕生所学来训练一个玩游戏的AI智能体。
由于我们将使用 Atari 环境，我们必须首先安装 OpenAI gym 的 Atari 环境依赖项。当需要玩其他游戏的时候，我们也安装其他的 OpenAI gym 环境依赖项。在 macOS 上，假设你已经安装了 Homebrew 程序，你需要运行：
$ brew install cmake boost boost-python sdl2 swig wget
在 Ubuntu 上，输入以下命令（如果使用 Python 2，用 Python 替换 Python 3）：
$ apt-get install -y python3-numpy python3-dev cmake zlib1g-dev libjpeg-dev\ xvfb libav-tools xorg-dev python3-opengl libboost-all-dev libsdl2-dev swig
然后安装python模块：
$ pip3 install --upgrade 'gym[all]'
如果一切顺利，应该可以按照如下创建一个Ms.Pac-Man环境：
>>> env = gym.make("MsPacman-v0")
>>> obs = env.reset()
>>> obs.shape # [长，宽，通道]
(210, 160, 3)
>>> env.action_space
Discrete(9)
如上，该环境的行动空间中有9个离散的行动，分别相对应于操纵杆上的九个操作（左，右，上，下，中，左上，右上，左下，右下），观测值就简单的是游戏截屏，以一个图片三维数组的形式呈现，如下左图：
原始图像有些大，这里写了一个预处理函数，将原图进行裁剪压缩以及灰度化等一系列操作，来降低训练的计算量，进而加速训练过程，如下：
mspacman_color = np.array([210, 164, 74]).mean()
def preprocess_observation(obs):
img = obs[1:176:2, ::2] # 裁剪
img = img.mean(axis=2) # 灰度化
img[img==mspacman_color] = 0 # 提升对比度
img = (img - 128) / 128 - 1 # 正则化为-1到1.
return img.reshape(88, 80, 1)
图像预处理的结果如上图（右）。
接下来，我们就创建一个DQN，该网络的输入为状态行动对(s,a)，输出为相应Q(s,a)的估计值，但是由于所有的行动都是离散的，所以可以简化为输入为状态s，输出为每一个行动的Q估计值。如下:
该DQN网络由三个卷积层，两个全连接层（包含输出层）组成。
如下所示，训练算法将使用两个相同结构，参数不同的DQN网络：一个用来驱动训练（actor），另一个用来观察actor并且从尝试和错误中学习（critic）。每隔一段时间，将critic网络复制给actor网络。因为我们需要两个相同的 DQN，所以我们将创建一个q_network()函数来构建它们：
from tensorflow.contrib.layers import convolution2d,fully_connected
input_height = 88
input_width = 80
input_channels = 1
conv_n_maps = [32, 64, 64]
conv_kernel_sizes = [(8,8), (4,4), (3,3)]
conv_strides = [4, 2, 1]
conv_paddings = ["SAME"]*3
conv_activation = [tf.nn.relu]*3
n_hidden_in = 64 * 11 * 10 # conv3 has 64 maps of 11x10 each
n_hidden = 512
hidden_activation = tf.nn.relu
n_outputs = env.action_space.n # 9 discrete actions are available
initializer = tf.contrib.layers.variance_scaling_initializer()
def q_network(X_state, scope):
prev_layer = X_state
conv_layers = []
with tf.variable_scope(scope) as scope:
for n_maps, kernel_size, stride, padding, activation in zip(conv_n_maps, conv_kernel_sizes,conv_strides,conv_paddings, conv_activation):
prev_layer = convolution2d(prev_layer, num_outputs=n_maps,kernel_size=kernel_size,stride=stride, padding=padding,activation_fn=activation,weights_initializer=initializer)
conv_layers.append(prev_layer)
last_conv_layer_flat = tf.reshape(prev_layer, shape=[-1, n_hidden_in])
hidden = fully_connected(last_conv_layer_flat, n_hidden,activation_fn=hidden_activation,weights_initializer=initializer)
outputs = fully_connected(hidden, n_outputs,activation_fn=None,weights_initializer=initializer)
trainable_vars = tf.get_collection(tf.GraphKeys.TRAINABLE_VARIABLES,scope=scope.name)
trainable_vars_by_name = {var.name[len(scope.name):]: var for var in trainable_vars}
return outputs, trainable_vars_by_name
上述代码第一部分为定义了DQN网络架构的超参数，然后就是q_network()函数创建DQN，环境的状态X_state为输入，以及变量范围的名称。请注意，我们将只使用一个观察来表示环境的状态，因为几乎没有隐藏的状态（除了闪烁的物体和ghost的方向）。
trainable_vars_by_name字典收集了所有 DQN 的可训练变量。当我们创建操作以将critic DQN 复制到actor DQN 时，比较有用。字典的关键字是变量的名称，去掉与范围名称相对应的前缀的一部分。看起来像这样：
>>> trainable_vars_by_name
{'/Conv/biases:0': <tensorflow.python.ops.variables.Variable at 0x121cf7b50>,
'/Conv/weights:0': <tensorflow.python.ops.variables.Variable...>,
'/Conv_1/biases:0': <tensorflow.python.ops.variables.Variable...>,
'/Conv_1/weights:0': <tensorflow.python.ops.variables.Variable...>,
'/Conv_2/biases:0': <tensorflow.python.ops.variables.Variable...>,
'/Conv_2/weights:0': <tensorflow.python.ops.variables.Variable...>,
'/fully_connected/biases:0': <tensorflow.python.ops.variables.Variable...>,
'/fully_connected/weights:0': <tensorflow.python.ops.variables.Variable...>,
'/fully_connected_1/biases:0': <tensorflow.python.ops.variables.Variable...>,
'/fully_connected_1/weights:0': <tensorflow.python.ops.variables.Variable...>}
接下来，我们创建输入的placeholder，两个DQN，以及从critic到actor的复制操作：
X_state = tf.placeholder(tf.float32, shape=[None, input_height, input_width,input_channels])
actor_q_values, actor_vars = q_network(X_state, scope="q_networks/actor")
critic_q_values, critic_vars = q_network(X_state, scope="q_networks/critic")
copy_ops = [actor_var.assign(critic_vars[var_name])
for var_name, actor_var in actor_vars.items()]
copy_critic_to_actor = tf.group(*copy_ops)
我们回顾一下网络，现在我们有了两个网络，每一个都以环境状态为输入，以状态中每个行动的Q估计值为输出，以及我们有了copy_critic_to_actor的复制函数。
actor DQN 可以用来扮演 Ms.Pac-Man（最初很差劲）。正如前面所讨论的，我们希望它足够深入地探究游戏，所以通常情况下我们会将它用 ε 贪婪策略或另一种探索策略相结合。
但是critic DQN 呢？它如何去学习玩游戏？简而言之，它将试图使其预测的 Q 值去匹配actor通过游戏经验估计出来的 Q 值。具体来说，我们将让actor玩一段时间，把所有的经验保存在回放记忆存储器中。每个记忆将是一个 5 元组（状态、动作、下一状态、奖励、继续），其中“继续”项在游戏结束时等于 0，否则为 1。接下来，我们定期地从回放存储器中采样一批记忆，并且我们将估计这些存储器中的 Q 值。最后，我们将使用监督学习技术训练critic DQN 去预测这些 Q 值。每隔几个训练周期，我们会把critic DQN 复制到actor DQN。就这样！下公式显示了用于训练critic DQN 的损失函数：
其中：
s(i), a(i), r(i) 和 s′(i)分别为状态，行为，回报，和下一状态，均从存储器中第i次采样得到
m是记忆批处理的长度
θ_critic和θ_actor为critic和actor的参数
Q(s'(i),a',θactor)是critic DQN 对第i条记忆中的状态行为对 Q 值的预测
Q(s(i),a(i),θcritic)是actor DQN 在当前状态选择动作a'时的下一状态S' Q 值的预测的期望
y(i)是第i条记忆的目标 Q 值，注意，它等同于actor实际观察到的奖励，再加上actor在行动最优的情况下，未来回报的预测值
J(θ_critic)为训练actor DQN 的损失函数。正如所示，这只是由actor DQN 估计的目标 Q 值y和critic DQN 对这些 Q 值的预测之间的均方误差
接下来，我们把critic DQN的训练部分也加上，首先，我们需要计算存储器批处理中的每一个状态行动对中的Q预测值。由于 DQN 为每一个可能的动作输出一个 Q 值，所以我们只需要保持与在该存储器中实际选择的动作相对应的 Q 值。为此，我们将把动作转换成one-hot向量（记住这是一个满是 0 的向量，除了第i个索引中的1），并乘以 Q 值：这将删除所有与记忆动作对应的 Q 值外的 Q 值。然后只对第一轴求和，以获得每个存储器所需的 Q 值预测。
X_action = tf.placeholder(tf.int32, shape=[None])
q_value = tf.reduce_sum(critic_q_values * tf.one_hot(X_action, n_outputs), axis=1, keep_dims=True)
接下来，添加训练操作，这里目标Q值将通过placeholder输入。我们还创建了一个不是用来训练的变量global_step。优化器的minimize()操作将负责增加它。另外，我们创建了init操作和Saver。
y = tf.placeholder(tf.float32, shape=[None, 1])
cost = tf.reduce_mean(tf.square(y - q_value))
global_step = tf.Variable(0, trainable=False, name='global_step')
optimizer = tf.train.AdamOptimizer(learning_rate)
training_op = optimizer.minimize(cost, global_step=global_step)
init = tf.global_variables_initializer()
saver = tf.train.Saver()
这就是训练阶段的情况。在执行阶段之前，我们需要一些工具。首先，让我们从回放记忆开始。我们将使用一个deque列表，因为在将数据推送到队列中并在达到最大内存大小时从列表的末尾弹出它们使是非常有效的。我们还将编写一个小函数来随机地从回放记忆中采样一批处理：
from collections import deque
replay_memory_size = 10000
replay_memory = deque([], maxlen=replay_memory_size)
def sample_memories(batch_size):
indices = rnd.permutation(len(replay_memory))[:batch_size]
cols = [[], [], [], [], []] # state, action, reward, next_state, continue
for idx in indices:
memory = replay_memory[idx]
for col, value in zip(cols, memory):
col.append(value)
cols = [np.array(col) for col in cols]
return (cols[0], cols[1], cols[2].reshape(-1, 1), cols[3],cols[4].reshape(-1, 1))
接下来，就让actor去探索游戏吧，这里我们使用ε 贪婪策略，并在 50000 个训练步骤中逐步将ε从 1 降低到 0.05：
eps_min = 0.05
eps_max = 1.0
eps_decay_steps = 50000
def epsilon_greedy(q_values, step):
epsilon = max(eps_min, eps_max - (eps_max-eps_min) * step/eps_decay_steps)
if rnd.rand() < epsilon:
return rnd.randint(n_outputs) # 随机动作
else:
return np.argmax(q_values) # 最优动作
到这里，我们准备好了，开始训练吧，执行阶段没有太复杂的东西，首先，让我们初始化几个变量：
n_steps = 100000 # 总的训练步长
training_start = 1000 # 在游戏1000次迭代后开始训练
training_interval = 3 # 每3次迭代训练一次
save_steps = 50 # 每50训练步长保存模型
copy_steps = 25 # 每25训练步长后复制评论家Q值到行动者
discount_rate = 0.95
skip_start = 90 # 跳过游戏开始(只是等待时间)
batch_size = 50
iteration = 0 # 游戏迭代
checkpoint_path = "./my_dqn.ckpt"
done = True # env 需要被重置
接下来，开一个session，并且run起来：
with tf.Session() as sess:
if os.path.isfile(checkpoint_path):
saver.restore(sess, checkpoint_path)
else:
init.run()
while True:
step = global_step.eval()
if step >= n_steps:
break
iteration += 1
if done: # 游戏结束，重来
obs = env.reset()
for skip in range(skip_start): # 跳过游戏开头
obs, reward, done, info = env.step(0)
state = preprocess_observation(obs)
# actor评估要干什么
q_values = actor_q_values.eval(feed_dict={X_state: [state]})
action = epsilon_greedy(q_values, step)
# actor开始玩游戏
obs, reward, done, info = env.step(action)
next_state = preprocess_observation(obs)
# 让我们记下来刚才发生了啥
replay_memory.append((state, action, reward, next_state, 1.0 - done)) state = next_state
if iteration < training_start or iteration % training_interval != 0: continue
# critic学习
X_state_val, X_action_val, rewards, X_next_state_val, continues = ( sample_memories(batch_size))
next_q_values = actor_q_values.eval( feed_dict={X_state: X_next_state_val})
max_next_q_values = np.max(next_q_values, axis=1, keepdims=True)
y_val = rewards + continues * discount_rate * max_next_q_values
training_op.run(feed_dict={X_state: X_state_val,X_action: X_action_val, y: y_val})
# 复制critic Q值到actor
if step % copy_steps == 0:
copy_critic_to_actor.run()
# 保存模型
if step % save_steps == 0:
saver.save(sess, checkpoint_path)
如果checkpoint文件存在，我们就可以恢复模型，否则我们只能初始化变量。然后，主循环开始，其中iteration记下了从程序开始以来游戏步骤的总数，同时step计算从训练开始的训练步骤的总数（如果恢复了checkpoint，也恢复全局步骤）。然后代码重置游戏（跳过第一个无聊的等待游戏的步骤，这步骤啥都没有）。接下来，actor评估该做什么，然后玩游戏，并且它的经验被存储在记忆中。然后，每隔一段时间，critic开始一个训练步骤。它采样一批记忆，并要求actor估计下一状态的所有动作的Q值，并计算目标 Q 值y_val.这里唯一棘手的部分是，我们必须将下一个状态的 Q 值乘以continues向量，以将对应于游戏结束的记忆 Q 值清零。接下来，我们进行训练操作，以提高critic预测 Q 值的能力。最后，我们定期将critic的 Q 值复制给actor，然后保存模型。
好了，至此我们今天使用深度 Q 学习来训练一个智能体去玩 Ms. Pac-Man，就像 DeepMind 在 2013年所做的那样。代码可以很容易地调整，调整后学习去玩大多数 Atari 游戏的效果都相当好。在大多数动作游戏中，它可以达到超人的技能，但它在长时运行的游戏中却不太好。希望有些收获，欢迎留言或进社区共同交流，喜欢的话，就点个赞吧，您也可以置顶公众号，第一时间接收最新内容。