强化学习(DQN）教程

作者：Adam Paszke

译者：wutong Zhang

校验：wutong Zhang

本教程介绍了如何使用PyTorch训练一个Deep Q-learning(DQN）智能点(Agent）来完成OpenAI Gym中的CartPole-V0任务。

任务

智能点需要决定两种动作：向左或向右来使其上的杆保持直立。 你可以在OpenAI Gym找到一个有各种算法和可视化的官方排行榜。

cartpole

当智能点观察环境的当前状态并选择动作时，环境将转换为新状态，并返回指示动作结果的奖励。在这项任务中，每增加一个时间步，奖励+1，如果杆子掉得太远或大车移动距离中心超过2.4个单位，环境就会终止。这意味着更好的执行场景将持续更长的时间，积累更大的回报。

Cartpole任务的设计为智能点输入代表环境状态(位置、速度等）的4个实际值。然而，神经网络完全可以通过观察场景来解决这个任务，所以我们将使用以车为中心的一块屏幕作为输入。因此，我们的结果无法直接与官方排行榜上的结果相比——我们的任务更艰巨。不幸的是，这会减慢训练速度，因为我们必须渲染所有帧。

严格地说，我们将以当前帧和前一个帧之间的差异来呈现状态。这将允许代理从一张图像中考虑杆子的速度。

包

首先你需要导入必须的包。我们需要 gym 作为环境 (使用 pip install gym 安装). 我们也需要 PyTorch 的如下功能:

  • 神经网络(torch.nn）
  • 优化(torch.optim）
  • 自动微分(torch.autograd）
  • 对于视觉任务工具(torchvision- 一个单独的包）
  1. import gym
  2. import math
  3. import random
  4. import numpy as np
  5. import matplotlib
  6. import matplotlib.pyplot as plt
  7. from collections import namedtuple
  8. from itertools import count
  9. from PIL import Image
  11. import torch
  12. import torch.nn as nn
  13. import torch.optim as optim
  14. import torch.nn.functional as F
  15. import torchvision.transforms as T
  18. env = gym.make('CartPole-v0').unwrapped
  20. # set up matplotlib
  21. is_ipython = 'inline' in matplotlib.get_backend()
  22. if is_ipython:
  23.     from IPython import display
  25. plt.ion()
  27. # if gpu is to be used
  28. device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

回放内存

我们将使用经验回放内存来训练DQN。它存储智能点观察到的转换，允许我们稍后重用此数据。通过从中随机抽样，组成批对象的转换将被取消相关性。结果表明，这大大稳定和改进了DQN训练过程。

因此，我们需要两个类别：

  • Transition- 一个命名的元组，表示我们环境中的单个转换。它基本上将(状态、动作）对映射到它们的(下一个状态、奖励）结果，状态是屏幕差分图像，如后面所述。
  • ReplayMemory- 一个有界大小的循环缓冲区，用于保存最近观察到的转换。它还实现了一个.sample()方法，用于选择一批随机转换进行训练。
  1. Transition = namedtuple('Transition',
  2.                         ('state', 'action', 'next_state', 'reward'))
  5. class ReplayMemory(object):
  7.     def __init__(self, capacity):
  8.         self.capacity = capacity
  9.         self.memory = []
  10.         self.position = 0
  12.     def push(self, *args):
  13.         """Saves a transition."""
  14.         if len(self.memory) < self.capacity:
  15.             self.memory.append(None)
  16.         self.memory[self.position] = Transition(*args)
  17.         self.position = (self.position + 1) % self.capacity
  19.     def sample(self, batch_size):
  20.         return random.sample(self.memory, batch_size)
  22.     def __len__(self):
  23.         return len(self.memory)

现在我们来定义自己的模型。但首先来快速了解一下DQN。

DQN算法

我们的环境是确定的，所以这里提出的所有方程也都是确定性的，为了简单起见。在强化学习文献中，它们还包含对环境中随机转换的期望。

我们的目标是制定一项策略，试图最大化折扣、累积奖励

强化学习(DQN）教程 - 图2

，其中

强化学习(DQN）教程 - 图3

也被认为是返回值。 折扣，

强化学习(DQN）教程 - 图4

应该是介于

强化学习(DQN）教程 - 图5

和

强化学习(DQN）教程 - 图6

之间的常量，以确保和收敛。它使来自不确定的遥远未来的回报对我们的智能点来说比它在不久的将来相当有信心的回报更不重要。

Q-Learning背后的主要思想是，如果我们有一个函数

强化学习(DQN）教程 - 图7

，则如果我们在特定的状态下采取行动，那么我们可以很容易地构建一个最大化回报的策略：

强化学习(DQN）教程 - 图8

然而，我们并不了解世界的一切，因此我们无法访问

强化学习(DQN）教程 - 图9

。但是，由于神经网络是通用的函数逼近器，我们可以简单地创建一个并训练它类似于

强化学习(DQN）教程 - 图10

。

对对于我们的训练更新规则，我们将假设某些策略的每个

强化学习(DQN）教程 - 图11

函数都遵循Bellman方程：

强化学习(DQN）教程 - 图12

等式两边的差异被称为时间差误差，即

强化学习(DQN）教程 - 图13

：

强化学习(DQN）教程 - 图14

为了尽量减少这个错误，我们将使用Huber loss。Huber损失在误差很小的情况下表现为均方误差，但在误差较大的情况下表现为平均绝对误差 —— 这使得当对

强化学习(DQN）教程 - 图15

的估计噪音很大时，对异常值的鲁棒性更强。我们通过从重放内存中取样的一批转换来计算

强化学习(DQN）教程 - 图16

强化学习(DQN）教程 - 图17

强化学习(DQN）教程 - 图18

Q-网络

我们的模型将是一个卷积神经网络需要在当前和以前的屏幕补丁之间的差异。它具有两个输出端，表示

强化学习(DQN）教程 - 图19

和

强化学习(DQN）教程 - 图20

(其中

强化学习(DQN）教程 - 图21

是输入到网络)。实际上，网络正试图预测在给定电流输入的情况下采取每项行动的预期回报。

  1. class DQN(nn.Module):
  3. def __init__(self, h, w, outputs):
  4.     super(DQN, self).__init__()
  5.     self.conv1 = nn.Conv2d(3, 16, kernel_size=5, stride=2)
  6.     self.bn1 = nn.BatchNorm2d(16)
  7.     self.conv2 = nn.Conv2d(16, 32, kernel_size=5, stride=2)
  8.     self.bn2 = nn.BatchNorm2d(32)
  9.     self.conv3 = nn.Conv2d(32, 32, kernel_size=5, stride=2)
  10.     self.bn3 = nn.BatchNorm2d(32)
  12.     # 线性输入连接的数量取决于conv2d层的输出，因此需要计算输入图像的大小。
  13.     def conv2d_size_out(size, kernel_size = 5, stride = 2):
  14.         return (size - (kernel_size - 1) - 1) // stride  + 1
  15.     convw = conv2d_size_out(conv2d_size_out(conv2d_size_out(w)))
  16.     convh = conv2d_size_out(conv2d_size_out(conv2d_size_out(h)))
  17.     linear_input_size = convw * convh * 32
  18.     self.head = nn.Linear(linear_input_size, outputs)
  20. # 使用一个元素调用以确定下一个操作，或在优化期间调用批处理。返回张量
  21. def forward(self, x):
  22.     x = F.relu(self.bn1(self.conv1(x)))
  23.     x = F.relu(self.bn2(self.conv2(x)))
  24.     x = F.relu(self.bn3(self.conv3(x)))
  25.     return self.head(x.view(x.size(0), -1))

获取输入

下面的代码是用于从环境中提取和处理渲染图像的实用程序。它使用了torchvision包，这样就可以很容易地组合图像转换。运行单元后，它将显示它提取的示例帧。

  1. resize = T.Compose([T.ToPILImage(),
  2.                     T.Resize(40, interpolation=Image.CUBIC),
  3.                     T.ToTensor()])
  6. def get_cart_location(screen_width):
  7.     world_width = env.x_threshold * 2
  8.     scale = screen_width / world_width
  9.     return int(env.state[0] * scale + screen_width / 2.0)  # MIDDLE OF CART
  11. def get_screen():
  12.     # 返回 gym 需要的400x600x3 图片, 但有时会更大，如800x1200x3. 将其转换为torch (CHW).
  13.     screen = env.render(mode='rgb_array').transpose((2, 0, 1))
  14.     # 车子在下半部分，因此请剥去屏幕的顶部和底部。
  15.     screen = screen[:, int(screen_height*0.4):int(screen_height * 0.8)]
  16.     view_width = int(screen_width * 0.6)
  17.     cart_location = get_cart_location(screen_width)
  18.     if cart_location < view_width // 2:
  19.         slice_range = slice(view_width)
  20.     elif cart_location > (screen_width - view_width // 2):
  21.         slice_range = slice(-view_width, None)
  22.     else:
  23.         slice_range = slice(cart_location - view_width // 2,
  24.                             cart_location + view_width // 2)
  25.     # 去掉边缘，这样我们就可以得到一个以车为中心的正方形图像。
  26.     screen = screen[:, :, slice_range]
  27.     # 转化为 float, 重新裁剪, 转化为 torch 张量(这并不需要拷贝)
  28.     screen = np.ascontiguousarray(screen, dtype=np.float32) / 255
  29.     screen = torch.from_numpy(screen)
  30.     # 重新裁剪,加入批维度 (BCHW)
  31.     return resize(screen).unsqueeze(0).to(device)
  34. env.reset()
  35. plt.figure()
  36. plt.imshow(get_screen().cpu().squeeze(0).permute(1, 2, 0).numpy(),
  37.            interpolation='none')
  38. plt.title('Example extracted screen')
  39. plt.show()

训练

超参数和配置

此单元实例化模型及其优化器，并定义一些实用程序：

  • select_action- 将根据迭代次数贪婪策略选择一个行动。简单地说，我们有时会使用我们的模型来选择动作，有时我们只会对其中一个进行统一的采样。选择随机动作的概率将从 EPS_START 开始并以指数形式向 EPS_END衰减。 EPS_DECAY 控制衰减速率。
  • plot_durations- 一个帮助绘制迭代次数持续时间，以及过去100迭代次数的平均值(官方评估中使用的度量）。迭代次数将在包含主训练循环的单元下方，并在每迭代之后更新。
  2. BATCH_SIZE = 128
  3. GAMMA = 0.999
  4. EPS_START = 0.9
  5. EPS_END = 0.05
  6. EPS_DECAY = 200
  7. TARGET_UPDATE = 10
  9. # 获取屏幕大小，以便我们可以根据从ai-gym返回的形状正确初始化层。
  10. # 这一点上的典型尺寸接近3x40x90，这是在get_screen(）中抑制和缩小的渲染缓冲区的结果。
  11. init_screen = get_screen()
  12. _, _, screen_height, screen_width = init_screen.shape
  14. # Get number of actions from gym action space
  15. n_actions = env.action_space.n
  17. policy_net = DQN(screen_height, screen_width, n_actions).to(device)
  18. target_net = DQN(screen_height, screen_width, n_actions).to(device)
  19. target_net.load_state_dict(policy_net.state_dict())
  20. target_net.eval()
  22. optimizer = optim.RMSprop(policy_net.parameters())
  23. memory = ReplayMemory(10000)
  26. steps_done = 0
  29. def select_action(state):
  30.     global steps_done
  31.     sample = random.random()
  32.     eps_threshold = EPS_END + (EPS_START - EPS_END) * \
  33.         math.exp(-1. * steps_done / EPS_DECAY)
  34.     steps_done += 1
  35.     if sample > eps_threshold:
  36.         with torch.no_grad():
  37.             # t.max(1) will return largest column value of each row.
  38.             # second column on max result is index of where max element was
  39.             # found, so we pick action with the larger expected reward.
  40.             return policy_net(state).max(1)[1].view(1, 1)
  41.     else:
  42.         return torch.tensor([[random.randrange(n_actions)]], device=device, dtype=torch.long)
  45. episode_durations = []
  48. def plot_durations():
  49.     plt.figure(2)
  50.     plt.clf()
  51.     durations_t = torch.tensor(episode_durations, dtype=torch.float)
  52.     plt.title('Training...')
  53.     plt.xlabel('Episode')
  54.     plt.ylabel('Duration')
  55.     plt.plot(durations_t.numpy())
  56.     # Take 100 episode averages and plot them too
  57.     if len(durations_t) >= 100:
  58.         means = durations_t.unfold(0, 100, 1).mean(1).view(-1)
  59.         means = torch.cat((torch.zeros(99), means))
  60.         plt.plot(means.numpy())
  62.     plt.pause(0.001)  # pause a bit so that plots are updated
  63.     if is_ipython:
  64.         display.clear_output(wait=True)
  65.         display.display(plt.gcf())

训练循环

最后，训练我们的模型的代码。

在这里，你可以找到执行最优化的一个步骤的optimize_model功能。它执行优化的一个步骤。它首先对一批数据进行采样，将所有张量连接成一个张量，计算出

强化学习(DQN）教程 - 图22

和

强化学习(DQN）教程 - 图23

，并将它们组合成我们的损失。根据定义，如果

强化学习(DQN）教程 - 图24

是结束状态，我们设置

强化学习(DQN）教程 - 图25

。我们还使用目标网络来计算

强化学习(DQN）教程 - 图26

以增加稳定性。目标网络的权重大部分时间保持不变，但每隔一段时间就会更新一次策略网络的权重。这通常是一组步骤，但为了简单起见，我们将使用迭代次数。

  1. def optimize_model():
  2.     if len(memory) < BATCH_SIZE:
  3.         return
  4.     transitions = memory.sample(BATCH_SIZE)
  5.     # 转置批样本(有关详细说明，请参阅https://stackoverflow.com/a/19343/3343043）。这会将转换的批处理数组转换为批处理数组的转换。
  6.     batch = Transition(*zip(*transitions))
  8.     # 计算非最终状态的掩码并连接批处理元素(最终状态将是模拟结束后的状态）
  9.     non_final_mask = torch.tensor(tuple(map(lambda s: s is not None,
  10.                                           batch.next_state)), device=device, dtype=torch.uint8)
  11.     non_final_next_states = torch.cat([s for s in batch.next_state
  12.                                                 if s is not None])
  13.     state_batch = torch.cat(batch.state)
  14.     action_batch = torch.cat(batch.action)
  15.     reward_batch = torch.cat(batch.reward)
  17.     # 计算Q(s_t, a)-模型计算 Q(s_t)，然后选择所采取行动的列。这些是根据策略网络对每个批处理状态所采取的操作。
  18.     state_action_values = policy_net(state_batch).gather(1, action_batch)
  20.     # 计算下一个状态的V(s_{t+1})。非最终状态下一个状态的预期操作值是基于“旧”目标网络计算的；选择max(1)[0]的最佳奖励。这是基于掩码合并的，这样当状态为最终状态时，我们将获得预期状态值或0。
  21.     next_state_values = torch.zeros(BATCH_SIZE, device=device)
  22.     next_state_values[non_final_mask] = target_net(non_final_next_states).max(1)[0].detach()
  23.     # 计算期望 Q 值
  24.     expected_state_action_values = (next_state_values * GAMMA) + reward_batch
  26.     # 计算 Huber 损失
  27.     loss = F.smooth_l1_loss(state_action_values, expected_state_action_values.unsqueeze(1))
  29.     # 优化模型
  30.     optimizer.zero_grad()
  31.     loss.backward()
  32.     for param in policy_net.parameters():
  33.         param.grad.data.clamp_(-1, 1)
  34.     optimizer.step()

接下来，你可以找到主训练循环。开始时，我们重置环境并初始化state张量。然后，我们对一个操作进行采样，执行它，观察下一个屏幕和奖励(总是1），并对我们的模型进行一次优化。当 episode 结束(我们的模型失败）时，我们重新启动循环。

num_episodes设置得很小。你可以下载并运行更多的epsiodes，比如300+来进行有意义的持续时间改进。

  1. num_episodes = 50
  2. for i_episode in range(num_episodes):
  3.     # 初始化环境和状态
  4.     env.reset()
  5.     last_screen = get_screen()
  6.     current_screen = get_screen()
  7.     state = current_screen - last_screen
  8.     for t in count():
  9.         # 选择并执行动作
  10.         action = select_action(state)
  11.         _, reward, done, _ = env.step(action.item())
  12.         reward = torch.tensor([reward], device=device)
  14.         # 观察新状态
  15.         last_screen = current_screen
  16.         current_screen = get_screen()
  17.         if not done:
  18.             next_state = current_screen - last_screen
  19.         else:
  20.             next_state = None
  22.         # 在内存中储存当前参数
  23.         memory.push(state, action, next_state, reward)
  25.         # 进入下一状态
  26.         state = next_state
  28.         # 记性一步优化 (在目标网络)
  29.         optimize_model()
  30.         if done:
  31.             episode_durations.append(t + 1)
  32.             plot_durations()
  33.             break
  34.     #更新目标网络, 复制在 DQN 中的所有权重偏差
  35.     if i_episode % TARGET_UPDATE == 0:
  36.         target_net.load_state_dict(policy_net.state_dict())
  38. print('Complete')
  39. env.render()
  40. env.close()
  41. plt.ioff()
  42. plt.show()

下面是一个图表，它说明了整个结果数据流。

img/reinforcement_learning_diagram.jpg

动作可以是随机选择的，也可以是基于一个策略，从gym环境中获取下一步的样本。我们将结果记录在回放内存中，并在每次迭代中运行优化步骤。优化从重放内存中随机抽取一批来训练新策略。 “旧的”target_net也用于优化计算预期的Q值；它偶尔会更新以保持其最新。

脚本的总运行时间： (0分钟0.000秒）