学习去使用深度 Q 学习来玩 Ms.Pac-Man

由于我们将使用 Atari 环境，我们必须首先安装 OpenAI gym 的 Atari 环境依赖项。当需要玩其他的时候，我们也会为你想玩的其他 OpenAI gym 环境安装依赖项。在 macOS 上，假设你已经安装了 Homebrew 程序，你需要运行：

  1. $ brew install cmake boost boost-python sdl2 swig wget

在 Ubuntu 上，输入以下命令（如果使用 Python 2，用 Python 替换 Python 3）：

  1. $ 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 包：

  1. $ pip3 install --upgrade 'gym[all]'

如果一切顺利，你应该能够创造一个 Ms.Pac-Man 环境：

  1. >>> env = gym.make("MsPacman-v0") 
  2. >>> obs = env.reset() 
  3. >>> obs.shape  #  [长，宽，通道] 
  4. (210, 160, 3) 
  5. >>> env.action_space 
  6. Discrete(9)

正如你所看到的，有九个离散动作可用，它对应于操纵杆的九个可能位置（左、右、上、下、中、左上等），观察结果是 Atari 屏幕的截图（见图 16-9，左），表示为 3D Numpy 矩阵。这些图像有点大，所以我们将创建一个小的预处理函数，将图像裁剪并缩小到88×80像素，将其转换成灰度，并提高 Ms.Pac-Man 的对比度。这将减少 DQN 所需的计算量，并加快培训练。

  1. mspacman_color = np.array([210, 164, 74]).mean()
  3. def preprocess_observation(obs):    
  4.     img = obs[1:176:2, ::2] #  裁剪    
  5.     img = img.mean(axis=2) #  灰度化    
  6.     img[img==mspacman_color] = 0 #  提升对比度    
  7.     img = (img - 128) / 128 - 1 #  正则化为-1到1.   
  8.     return img.reshape(88, 80, 1)

过程的结果如图 16-9 所示（右）。

图16-9

接下来，让我们创建 DQN。它可以只取一个状态动作对(S,A)作为输入，并输出相应的 Q 值Q(s,a)的估计值，但是由于动作是离散的，所以使用只使用状态S作为输入并输出每个动作的一个 Q 值估计的神经网络是更方便的。DQN 将由三个卷积层组成，接着是两个全连接层，其中包括输出层（如图 16-10）。

图16-10

正如我们将看到的，我们将使用的训练算法需要两个具有相同架构（但不同参数）的 DQN：一个将在训练期间用于驱动 Ms.Pac-Man（the actor，行动者），另一个将观看行动者并从其试验和错误中学习（the critic，评判者）。每隔一定时间，我们把评判者网络复制给行动者网络。因为我们需要两个相同的 DQN，所以我们将创建一个q_network()函数来构建它们：

  1. from tensorflow.contrib.layers import convolution2d, fully_connected
  2. input_height = 88 
  3. input_width = 80 
  4. input_channels = 1 
  5. conv_n_maps = [32, 64, 64] 
  6. conv_kernel_sizes = [(8,8), (4,4), (3,3)] 
  7. conv_strides = [4, 2, 1] 
  8. conv_paddings = ["SAME"]*3 
  9. conv_activation = [tf.nn.relu]*3 
  10. n_hidden_in = 64 * 11 * 10  #  conv3 有 64 个 11x10 映射
  11. each n_hidden = 512 
  12. hidden_activation = tf.nn.relu 
  13. n_outputs = env.action_space.n  #  9个离散动作
  14. initializer = tf.contrib.layers.variance_scaling_initializer()
  16. def q_network(X_state, scope):    
  17.     prev_layer = X_state    
  18.     conv_layers = []    
  19.     with tf.variable_scope(scope) as scope:        
  20.         for n_maps, kernel_size, stride, padding, activation in zip(conv_n_maps,                                                                  conv_kernel_sizes, 
  21.                                                                     conv_strides,
  22.                                                                     conv_paddings,                                                                  conv_activation):           
  23.             prev_layer = convolution2d(prev_layer, 
  24.                                        num_outputs=n_maps, 
  25.                                        kernel_size=kernel_size,
  26.                                        stride=stride, padding=padding, 
  27.                                        activation_fn=activation,
  28.                                        weights_initializer=initializer)           
  29.             conv_layers.append(prev_layer)       
  30.         last_conv_layer_flat = tf.reshape(prev_layer, shape=[-1, n_hidden_in])     
  31.         hidden = fully_connected(last_conv_layer_flat, n_hidden,
  32.                                  activation_fn=hidden_activation,                                                  weights_initializer=initializer) 
  33.         outputs = fully_connected(hidden, n_outputs, 
  34.                                   activation_fn=None,
  35.                                   weights_initializer=initializer)  
  37.     trainable_vars = tf.get_collection(tf.GraphKeys.TRAINABLE_VARIABLES,
  38.                                        scope=scope.name)                
  39.     trainable_vars_by_name = {var.name[len(scope.name):]: var 
  40.                               for var in trainable_vars}    
  41.     return outputs, trainable_vars_by_name

该代码的第一部分定义了DQN体系结构的超参数。然后q_network()函数创建 DQN，将环境的状态X_state作为输入，以及变量范围的名称。请注意，我们将只使用一个观察来表示环境的状态，因为几乎没有隐藏的状态（除了闪烁的物体和鬼魂的方向）。

trainable_vars_by_name字典收集了所有 DQN 的可训练变量。当我们创建操作以将评论家 DQN 复制到行动者 DQN 时，这将是有用的。字典的键是变量的名称，去掉与范围名称相对应的前缀的一部分。看起来像这样：

  1. >>> trainable_vars_by_name 
  2. {'/Conv/biases:0': <tensorflow.python.ops.variables.Variable at 0x121cf7b50>, '/Conv/weights:0': <tensorflow.python.ops.variables.Variable...>, 
  3. '/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...>}

现在让我们为两个 DQN 创建输入占位符，以及复制评论家 DQN 给行动者 DQN 的操作：

  1. X_state = tf.placeholder(tf.float32, 
  2.                          shape=[None, input_height, input_width,input_channels])          
  3. actor_q_values, actor_vars = q_network(X_state, scope="q_networks/actor") 
  4. critic_q_values, critic_vars = q_network(X_state, scope="q_networks/critic")
  5. copy_ops = [actor_var.assign(critic_vars[var_name])  
  6.             for var_name, actor_var in actor_vars.items()] 
  7. copy_critic_to_actor = tf.group(*copy_ops)

让我们后退一步：我们现在有两个 DQN，它们都能够将环境状态（即预处理观察）作为输入，并输出在该状态下的每一个可能的动作的估计 Q 值。另外，我们有一个名为copy_critic_to_actor的操作，将评论家 DQN 的所有可训练变量复制到行动者 DQN。我们使用 TensorFlow 的tf.group()函数将所有赋值操作分组到一个方便的操作中。

行动者 DQN 可以用来扮演 Ms.Pac-Man（最初非常糟糕）。正如前面所讨论的，你希望它足够深入地探究游戏，所以通常情况下你想将它用 ε 贪婪策略或另一种探索策略相结合。

但是评论家 DQN 呢？它如何去学习玩游戏？简而言之，它将试图使其预测的 Q 值去匹配行动者通过其经验的游戏估计的 Q 值。具体来说，我们将让行动者玩一段时间，把所有的经验保存在回放记忆存储器中。每个记忆将是一个 5 元组（状态、动作、下一状态、奖励、继续），其中“继续”项在游戏结束时等于 0，否则为 1。接下来，我们定期地从回放存储器中采样一批记忆，并且我们将估计这些存储器中的 Q 值。最后，我们将使用监督学习技术训练评论家 DQN 去预测这些 Q 值。每隔几个训练周期，我们会把评论家 DQN 复制到行动者 DQN。就这样！公式 16-7 示出了用于训练评论家 DQN 的损失函数：

图E16-7

其中：

  • s^{(i)}, a^{(i)}, r^{(i)}和s{'(i)}分别为状态，行为，回报，和下一状态，均从存储器中第i次采样得到
  • m是记忆批处理的长度
  • θ critic和θactor为评论者和行动者的参数
  • Q(s^{(i)}, a^{(i)}, \theta_{critic})是评论家 DQN 对第i记忆状态行为 Q 值的预测
  • Q(s^{'(i)}, a^{'}, \theta_{actor})是演员 DQN 在选择动作A'时的下一状态S'的期望 Q 值的预测
  • y是第i记忆的目标 Q 值，注意，它等同于行动者实际观察到的奖励，再加上行动者对如果它能发挥最佳效果（据它所知），未来的回报应该是什么的预测。
  • J为训练评论家 DQN 的损失函数。正如你所看到的，这只是由行动者 DQN 估计的目标 Q 值y和评论家 DQN 对这些 Q 值的预测之间的均方误差。

回放记忆是可选的，但强烈推荐使它存在。没有它，你会训练评论家 DQN 使用连续的经验，这可能是相关的。这将引入大量的偏差并且减慢训练算法的收敛性。通过使用回放记忆，我们确保馈送到训练算法的存储器可以是不相关的。

让我们添加评论家 DQN 的训练操作。首先，我们需要能够计算其在存储器批处理中的每个状态动作的预测 Q 值。由于 DQN 为每一个可能的动作输出一个 Q 值，所以我们只需要保持与在该存储器中实际选择的动作相对应的 Q 值。为此，我们将把动作转换成一个热向量（记住这是一个满是 0 的向量，除了第i个索引中的1），并乘以 Q 值：这将删除所有与记忆动作对应的 Q 值外的 Q 值。然后只对第一轴求和，以获得每个存储器所需的 Q 值预测。

  1. X_action = tf.placeholder(tf.int32, shape=[None]) 
  2. q_value = tf.reduce_sum(critic_q_values * tf.one_hot(X_action, n_outputs), axis=1, keep_dims=True)

接下来，让我们添加训练操作，假设目标Q值将通过占位符馈入。我们还创建了一个不可训练的变量global_step。优化器的minimize()操作将负责增加它。另外，我们创建了init操作和Saver。

  1. y = tf.placeholder(tf.float32, shape=[None, 1]) 
  2. cost = tf.reduce_mean(tf.square(y - q_value)) 
  3. global_step = tf.Variable(0, trainable=False, name='global_step') 
  4. optimizer = tf.train.AdamOptimizer(learning_rate) 
  5. training_op = optimizer.minimize(cost, global_step=global_step)
  6. init = tf.global_variables_initializer() 
  7. saver = tf.train.Saver()

这就是训练阶段的情况。在我们查看执行阶段之前，我们需要一些工具。首先，让我们从回放记忆开始。我们将使用一个deque列表，因为在将数据推送到队列中并在达到最大内存大小时从列表的末尾弹出它们使是非常有效的。我们还将编写一个小函数来随机地从回放记忆中采样一批处理：

  1. from collections import deque
  3. replay_memory_size = 10000 
  4. replay_memory = deque([], maxlen=replay_memory_size)
  6. def sample_memories(batch_size):    
  7.     indices = rnd.permutation(len(replay_memory))[:batch_size]    
  8.     cols = [[], [], [], [], []] #  state, action, reward, next_state, continue    
  9.     for idx in indices:        
  10.         memory = replay_memory[idx]        
  11.         for col, value in zip(cols, memory):            
  12.             col.append(value)    
  13.     cols = [np.array(col) for col in cols]    
  14.     return (cols[0], cols[1], cols[2].reshape(-1, 1), cols[3],cols[4].reshape(-1, 1))

接下来，我们需要行动者来探索游戏。我们使用 ε 贪婪策略，并在 50000 个训练步骤中逐步将ε从 1 降低到 0.05。

  1. eps_min = 0.05 
  2. eps_max = 1.0 
  3. eps_decay_steps = 50000
  4. def epsilon_greedy(q_values, step):    
  5.     epsilon = max(eps_min, eps_max - (eps_max-eps_min) * step/eps_decay_steps)   
  6.      if rnd.rand() < epsilon:        
  7.         return rnd.randint(n_outputs) #  随机动作   
  8.     else:        
  9.         return np.argmax(q_values) #  最优动作

就是这样！我们准备好开始训练了。执行阶段不包含太复杂的东西，但它有点长，所以深呼吸。准备好了吗？来次够！首先，让我们初始化几个变量：

  1. n_steps = 100000  #  总的训练步长 
  2. training_start = 1000  #  在游戏1000次迭代后开始训练
  3. training_interval = 3  #  每3次迭代训练一次
  4. save_steps = 50  #  每50训练步长保存模型
  5. copy_steps = 25  #  每25训练步长后复制评论家Q值到行动者
  6. discount_rate = 0.95 
  7. skip_start = 90  #  跳过游戏开始(只是等待时间) 
  8. batch_size = 50 
  9. iteration = 0  #  游戏迭代
  10. checkpoint_path = "./my_dqn.ckpt" 
  11. done = True #  env 需要被重置

接下来，让我们打开会话并开始训练：

  1. with tf.Session() as sess:    
  2.     if os.path.isfile(checkpoint_path):        
  3.         saver.restore(sess, checkpoint_path)    
  4.     else:        
  5.         init.run()    
  6.     while True:        
  7.         step = global_step.eval()        
  8.         if step >= n_steps:            
  9.             break        
  10.         iteration += 1        
  11.         if done: #  游戏结束，重来            
  12.             obs = env.reset()            
  13.             for skip in range(skip_start): #  跳过游戏开头              
  14.                 obs, reward, done, info = env.step(0)            
  15.                 state = preprocess_observation(obs)
  17.         #  行动者评估要干什么        
  18.         q_values = actor_q_values.eval(feed_dict={X_state: [state]})        
  19.         action = epsilon_greedy(q_values, step)
  21.         #  行动者开始玩游戏       
  22.         obs, reward, done, info = env.step(action)        
  23.         next_state = preprocess_observation(obs)
  25.         #  让我们记下来刚才发生了啥        
  26.         replay_memory.append((state, action, reward, next_state, 1.0 - done))        state = next_state
  28.         if iteration < training_start or iteration % training_interval != 0:                continue
  30.         #  评论家学习        
  31.         X_state_val, X_action_val, rewards, X_next_state_val, continues = (            sample_memories(batch_size))        
  32.         next_q_values = actor_q_values.eval( feed_dict={X_state: X_next_state_val})    
  33.         max_next_q_values = np.max(next_q_values, axis=1, keepdims=True)        
  34.         y_val = rewards + continues * discount_rate * max_next_q_values        
  35.         training_op.run(feed_dict={X_state: X_state_val,X_action: X_action_val, y: y_val})
  37.         #  复制评论家Q值到行动者        
  38.         if step % copy_steps == 0:            
  39.             copy_critic_to_actor.run()
  41.         #  保存模型        
  42.         if step % save_steps == 0:            
  43.             saver.save(sess, checkpoint_path)

如果检查点文件存在，我们就开始恢复模型，否则我们只需初始化变量。然后，主循环开始，其中iteration计算从程序开始以来游戏步骤的总数，同时step计算从训练开始的训练步骤的总数（如果恢复了检查点，也恢复全局步骤）。然后代码重置游戏（跳过第一个无聊的等待游戏的步骤，这步骤啥都没有）。接下来，行动者评估该做什么，并且玩游戏，并且它的经验被存储在回放记忆中。然后，每隔一段时间（热身期后），评论家开始一个训练步骤。它采样一批回放记忆，并要求行动者估计下一状态的所有动作的Q值，并应用公式 16-7 来计算目标 Q 值y_val.这里唯一棘手的部分是，我们必须将下一个状态的 Q 值乘以continues向量，以将对应于游戏结束的记忆 Q 值清零。接下来，我们进行训练操作，以提高评论家预测 Q 值的能力。最后，我们定期将评论家的 Q 值复制给行动者，然后保存模型。

不幸的是，训练过程是非常缓慢的：如果你使用你的破笔记本电脑进行训练的话，想让 Ms. Pac-Man 变好一点点你得花好几天，如果你看看学习曲线，计算一下每次的平均奖励，你会发现到它是非常嘈杂的。在某些情况下，很长一段时间内可能没有明显的进展，直到智能体学会在合理的时间内生存。如前所述，一种解决方案是将尽可能多的先验知识注入到模型中（例如，通过预处理、奖励等），也可以尝试通过首先训练它来模仿基本策略来引导模型。在任何情况下，RL仍然需要相当多的耐心和调整，但最终结果是非常令人兴奋的。