深度强化学习(DRL)

强化学习 (Reinforcement learning,RL)强调如何基于环境而行动,以取得最大化的预期利益。结合了深度学习技术后的强化学习更是如虎添翼。这两年广为人知的 AlphaGo 即是深度强化学习的典型应用。

基础知识和原理

这里,我们使用深度强化学习玩 CartPole(平衡杆)游戏。简单说,我们需要让模型控制杆的左右运动,以让其一直保持竖直平衡状态。

../../_images/cartpole.gifCartPole 游戏

我们使用 OpenAI 推出的 Gym 环境库 中的 CartPole 游戏环境,具体安装步骤和教程可参考 官方文档这里 。Gym 的基本调用方法如下:

  1. import gym
  2.  
  3. env = gym.make('CartPole-v1') # 实例化一个游戏环境,参数为游戏名称
  4. state = env.reset() # 初始化环境,获得初始状态
  5. while True:
  6. env.render() # 对当前帧进行渲染,绘图到屏幕
  7. action = model.predict(state) # 假设我们有一个训练好的模型,能够通过当前状态预测出这时应该进行的动作
  8. next_state, reward, done, info = env.step(action) # 让环境执行动作,获得执行完动作的下一个状态,动作的奖励,游戏是否已结束以及额外信息
  9. if done: # 如果游戏结束则退出循环
  10. break

那么,我们的任务就是训练出一个模型,能够根据当前的状态预测出应该进行的一个好的动作。粗略地说,一个好的动作应当能够最大化整个游戏过程中获得的奖励之和,这也是强化学习的目标。

以下代码展示了如何使用深度强化学习中的 Deep Q-Learning 方法来训练模型。

  1. import tensorflow as tf
  2. import numpy as np
  3. import gym
  4. import random
  5. from collections import deque
  6.  
  7. num_episodes = 500
  8. num_exploration_episodes = 100
  9. max_len_episode = 1000
  10. batch_size = 32
  11. learning_rate = 1e-3
  12. gamma = 1.
  13. initial_epsilon = 1.
  14. final_epsilon = 0.01
  15.  
  16.  
  17. # Q-network用于拟合Q函数,和前节的多层感知机类似。输入state,输出各个action下的Q-value(CartPole下为2维)。
  18. class QNetwork(tf.keras.Model):
  19. def __init__(self):
  20. super().__init__()
  21. self.dense1 = tf.keras.layers.Dense(units=24, activation=tf.nn.relu)
  22. self.dense2 = tf.keras.layers.Dense(units=24, activation=tf.nn.relu)
  23. self.dense3 = tf.keras.layers.Dense(units=2)
  24.  
  25. def call(self, inputs):
  26. x = self.dense1(inputs)
  27. x = self.dense2(x)
  28. x = self.dense3(x)
  29. return x
  30.  
  31. def predict(self, inputs):
  32. q_values = self(inputs)
  33. return tf.argmax(q_values, axis=-1)
  34.  
  35.  
  36. if __name__ == '__main__':
  37. env = gym.make('CartPole-v1') # 实例化一个游戏环境,参数为游戏名称
  38. model = QNetwork()
  39. optimizer = tf.keras.optimizers.Adam(learning_rate=learning_rate)
  40. replay_buffer = deque(maxlen=10000)
  41. epsilon = initial_epsilon
  42. for episode_id in range(num_episodes):
  43. state = env.reset() # 初始化环境,获得初始状态
  44. epsilon = max(
  45. initial_epsilon * (num_exploration_episodes - episode_id) / num_exploration_episodes,
  46. final_epsilon)
  47. for t in range(max_len_episode):
  48. env.render() # 对当前帧进行渲染,绘图到屏幕
  49. if random.random() < epsilon: # epsilon-greedy探索策略
  50. action = env.action_space.sample() # 以epsilon的概率选择随机动作
  51. else:
  52. action = model.predict(
  53. tf.constant(np.expand_dims(state, axis=0), dtype=tf.float32)).numpy()
  54. action = action[0]
  55.  
  56. # 让环境执行动作,获得执行完动作的下一个状态,动作的奖励,游戏是否已结束以及额外信息
  57. next_state, reward, done, info = env.step(action)
  58. # 如果游戏Game Over,给予大的负奖励
  59. reward = -10. if done else reward
  60. # 将(state, action, reward, next_state)的四元组(外加done标签表示是否结束)放入经验重放池
  61. replay_buffer.append((state, action, reward, next_state, 1 if done else 0))
  62. # 更新当前state
  63. state = next_state
  64.  
  65. if done: # 游戏结束则退出本轮循环,进行下一个episode
  66. print("episode %d, epsilon %f, score %d" % (episode_id, epsilon, t))
  67. break
  68.  
  69. if len(replay_buffer) >= batch_size:
  70. # 从经验回放池中随机取一个批次的四元组,并分别转换为NumPy数组
  71. batch_state, batch_action, batch_reward, batch_next_state, batch_done = zip(
  72. *random.sample(replay_buffer, batch_size))
  73. batch_state, batch_reward, batch_next_state, batch_done = \
  74. [np.array(a, dtype=np.float32) for a in [batch_state, batch_reward, batch_next_state, batch_done]]
  75. batch_action = np.array(batch_action, dtype=np.int32)
  76.  
  77. q_value = model(tf.constant(batch_next_state, dtype=tf.float32))
  78. y = batch_reward + (gamma * tf.reduce_max(q_value, axis=1)) * (1 - batch_done) # 按照论文计算y值
  79. with tf.GradientTape() as tape:
  80. loss = tf.keras.losses.mean_squared_error( # 最小化y和Q-value的距离
  81. y_true=y,
  82. y_pred=tf.reduce_sum(model(tf.constant(batch_state)) * tf.one_hot(batch_action, depth=2), axis=1)
  83. )
  84. grads = tape.gradient(loss, model.variables)
  85. optimizer.apply_gradients(grads_and_vars=zip(grads, model.variables)) # 计算梯度并更新参数

对于不同的任务(或者说环境),我们需要根据任务的特点,设计不同的状态以及采取合适的网络来拟合 Q 函数。例如,如果我们考虑经典的打砖块游戏(Gym 环境库中的 Breakout-v0 ),每一次执行动作(挡板向左、向右或不动),都会返回一个 210 160 3 的 RGB 图片,表示当前屏幕画面。为了给打砖块游戏这个任务设计合适的状态表示,我们有以下分析:

  • 砖块的颜色信息并不是很重要,画面转换成灰度也不影响操作,因此可以去除状态中的颜色信息(即将图片转为灰度表示);

  • 小球移动的信息很重要,如果只知道单帧画面而不知道小球往哪边运动,即使是人也很难判断挡板应当移动的方向。因此,必须在状态中加入表征小球运动方向的信息。一个简单的方式是将当前帧与前面几帧的画面进行叠加,得到一个 210 160 X (X 为叠加帧数)的状态表示;

  • 每帧的分辨率不需要特别高,只要能大致表征方块、小球和挡板的位置以做出决策即可,因此对于每帧的长宽可做适当压缩。

而考虑到我们需要从图像信息中提取特征,使用 CNN 作为拟合 Q 函数的网络将更为适合。由此,将上面的 QNetwork 更换为 CNN 网络,并对状态做一些修改,即可用于玩一些简单的视频游戏。

强化学习原理初探

强化学习