作者：Raymond Yuan，软件工程实习生

在本教程中，我们将学习如何训练一个能够赢得简单游戏 CartPole 的模型，使用深度强化学习。我们将使用 tf.keras 和 OpenAI 的 gym 来训练一个使用称为异步优势 Actor-Critic (A3C) 的技术的代理。强化学习一直受到极大的关注，但它究竟是什么呢？强化学习是机器学习的一个领域，涉及代理，这些代理应该在环境中采取某些行动以最大化或获得某种奖励。

在这个过程中，我们将积累实践经验，并围绕以下概念形成直觉

• Eager Execution - Eager Execution 是一种命令式、定义执行的接口，其中操作是在从 Python 调用时立即执行的。这使得 TensorFlow 更易于上手，并可以使研究和开发更加直观。
  
• 模型子类化 - 模型子类化允许通过子类化 tf.keras.Model 并定义自己的前向传递来创建完全可定制的模型。当启用 Eager Execution 时，模型子类化特别有用，因为前向传递可以以命令式的方式编写。
  
• 自定义训练循环
  

我们将遵循基本工作流程

• 构建我们的主代理监督者
  
• 构建我们的工作代理
  
• 实现 A3C 算法
  
• 训练我们的代理
  
• 可视化我们的性能
  

受众：本教程面向对强化学习感兴趣的任何人。虽然我们不会深入讨论机器学习的基础知识，但我们将从较高层次涵盖策略和价值网络等主题。此外，我建议阅读论文 Volodymyr Mnih 的深度强化学习的异步方法，这是一篇很棒的读物，并提供了更多关于该算法的详细信息。

什么是 CartPole？

CartPole 是一款游戏，其中一根杆子通过一个非驱动接头连接到一辆手推车，该手推车沿着无摩擦轨道移动。起始状态（手推车位置、手推车速度、杆角和杆尖速度）在 +/-0.05 之间随机初始化。该系统通过对手推车施加 +1 或 -1 的力（向左或向右移动）来控制。摆锤从直立状态开始，目标是防止它倒下。如果杆保持直立，则每个时间步长都会提供 +1 的奖励。当杆与垂直方向的夹角超过 15 度，或手推车从中心移动超过 2.4 个单位时，该回合结束。

代码

2021 年 3 月 32 日更新：您可以在此 链接 中找到本文的完整代码。

建立基线

为了正确判断模型的实际性能以及评估模型的指标，建立基线通常非常有用。例如，当您看到模型返回高分时，它似乎表现良好，但实际上，高分可能并不反映好的算法，或者只是随机行为的结果。在分类示例中，我们可以通过简单地分析类分布并预测最常见的类来建立基线性能。但是，我们如何为强化学习建立基线呢？为此，我们将创建一个随机代理，它将在我们的环境中执行随机行为。

class RandomAgent:
  """Random Agent that will play the specified game

    Arguments:
      env_name: Name of the environment to be played
      max_eps: Maximum number of episodes to run agent for.
  """
  def __init__(self, env_name, max_eps):
    self.env = gym.make(env_name)
    self.max_episodes = max_eps
    self.global_moving_average_reward = 0
    self.res_queue = Queue()

  def run(self):
    reward_avg = 0
    for episode in range(self.max_episodes):
      done = False
      self.env.reset()
      reward_sum = 0.0
      steps = 0
      while not done:
        # Sample randomly from the action space and step
        _, reward, done, _ = self.env.step(self.env.action_space.sample())
        steps += 1
        reward_sum += reward
      # Record statistics
      self.global_moving_average_reward = record(episode, 
                                                 reward_sum, 
                                                 0,
                                                 self.global_moving_average_reward,
                                                 self.res_queue, 0, steps)

      reward_avg += reward_sum
    final_avg = reward_avg / float(self.max_episodes)
    print("Average score across {} episodes: {}".format(self.max_episodes, final_avg))
    return final_avg

对于游戏 CartPole，我们在 4000 个回合中获得约 20 的平均值。要运行随机代理，请运行提供的 py 文件：python a3c_cartpole.py — algorithm=random — max-eps=4000。

什么是异步优势 Actor-Critic 算法？

异步优势 Actor-Critic 真是个拗口的名字！让我们从分解这个名字开始，然后介绍算法本身的机制。

• 异步：该算法是一种异步算法，其中多个工作代理并行训练，每个代理都有自己模型和环境的副本。这使得我们的算法不仅可以更快地训练，因为更多工作代理并行训练，而且还可以获得更多样的训练体验，因为每个工作代理的体验都是独立的。
  
• 优势：优势是一个衡量行动好坏的指标，也是衡量行动结果的指标。这使得算法能够关注网络预测不足的地方。直观地说，这使得我们能够衡量在给定时间步长采取行动 a 而不是遵循策略 π 的优势。
  
• Actor-Critic：算法的 Actor-Critic 方面使用了一个架构，该架构在策略和价值函数之间共享层。
  

但它是如何工作的呢？

从高层次来看，A3C 算法使用一种异步更新方案，该方案在固定长度的体验时间步长上运行。它将使用这些段来计算奖励和优势函数的估计量。每个工作代理执行以下工作流程循环

1. 获取全局网络参数
   
2. 通过遵循本地策略与环境交互，最少执行 min(t_max, 到达终止状态的步数) 个步数。
   
3. 计算价值和策略损失
   
4. 从损失中获取梯度
   
5. 使用梯度更新全局网络
   
6. 重复
   

通过这种训练配置，我们预计会看到代理数量的线性加速。但是，机器可以支持的代理数量受可用 CPU 内核数量的限制。此外，A3C 甚至可以扩展到多台机器，而一些较新的研究（例如 IMPALA）支持将其进一步扩展。添加更多代理可能会对速度和性能造成不利影响。查看 论文 以获取更多深入的信息！

策略和价值函数的复习

如果您已经熟悉策略梯度，那么我建议您跳过本节。否则，如果您不知道策略/价值是什么，或者只是想快速复习一下，请继续阅读！

策略的想法是，在给定某些输入状态的情况下，将动作概率分布参数化。我们通过创建一个网络来实现这一点，该网络接收游戏的当前状态并决定我们应该做什么。因此，当代理玩游戏时，无论何时看到某个状态（或类似状态），它都会计算给定输入状态时每个可用动作的概率，然后根据该概率分布采样一个动作。为了更正式地深入研究数学，策略梯度是更通用的评分函数梯度估计量的特例。一般情况用 Ex p(x | ) [f(x)] 的形式表示；换句话说，在我们的案例中，是根据某些策略网络 p 的某些奖励（或优势）函数 f 的期望值。然后，使用对数导数技巧，我们计算出如何更新网络参数，以便动作样本获得更高的奖励，并最终得到 ∇ Ex[f(x)] =Ex[f(x) ∇ log p(x)]。用英文来说，这个等式解释了如何在我们的梯度方向上移动 θ 将根据我们的奖励函数 f 最大化我们的分数。

价值函数本质上是判断处于某个状态有多好。形式上，价值函数定义了从状态 s 开始并遵循策略 p 时奖励的预期总和。这就是名称中“Critic”部分的意义所在。代理使用价值估计（Critic）来更新策略（Actor）。

实现

让我们首先定义我们将使用哪种模型。主代理将拥有全局网络，每个本地工作代理将在自己的进程中拥有该网络的副本。我们将使用模型子类化来实例化模型。模型子类化为我们提供了最大的灵活性，但代价是更高的冗长性。

class ActorCriticModel(keras.Model):
  def __init__(self, state_size, action_size):
    super(ActorCriticModel, self).__init__()
    self.state_size = state_size
    self.action_size = action_size
    self.dense1 = layers.Dense(100, activation='relu')
    self.policy_logits = layers.Dense(action_size)
    self.dense2 = layers.Dense(100, activation='relu')
    self.values = layers.Dense(1)

  def call(self, inputs):
    # Forward pass
    x = self.dense1(inputs)
    logits = self.policy_logits(x)
    v1 = self.dense2(inputs)
    values = self.values(v1)
    return logits, values

从我们的前向传递中可以看出，我们的模型将接收输入并返回策略概率 logits 和值。

主代理 - 主线程

让我们进入操作的核心。主代理拥有一个共享优化器，用于更新其全局网络。该代理实例化每个工作代理也将更新的全局网络以及我们将用来更新它的优化器。A3C 被证明对学习率的分布非常具有弹性，但对于游戏 CartPole，我们将使用 AdamOptimizer，学习率为 5e-4。

class MasterAgent():
  def __init__(self):
    self.game_name = 'CartPole-v0'
    save_dir = args.save_dir
    self.save_dir = save_dir
    if not os.path.exists(save_dir):
      os.makedirs(save_dir)

    env = gym.make(self.game_name)
    self.state_size = env.observation_space.shape[0]
    self.action_size = env.action_space.n
    self.opt = tf.train.AdamOptimizer(args.lr, use_locking=True)
    print(self.state_size, self.action_size)

    self.global_model = ActorCriticModel(self.state_size, self.action_size)  # global network
    self.global_model(tf.convert_to_tensor(np.random.random((1, self.state_size)), dtype=tf.float32))

主代理将运行训练函数来实例化并启动每个代理。主代理负责协调和监督每个代理。这些代理中的每一个都将异步运行。（从技术上讲，这不是真正的异步，因为在 Python 中，由于 GIL（全局解释器锁），单个 Python 进程无法并行运行线程（利用多个内核）。但是，它可以并发运行它们（在 I/O 绑定操作期间进行上下文切换）。为了简单和清晰地说明示例，我们使用线程来实现。）

def train(self):
    if args.algorithm == 'random':
      random_agent = RandomAgent(self.game_name, args.max_eps)
      random_agent.run()
      return

    res_queue = Queue()

    workers = [Worker(self.state_size,
                      self.action_size,
                      self.global_model,
                      self.opt, res_queue,
                      i, game_name=self.game_name,
                      save_dir=self.save_dir) for i in range(multiprocessing.cpu_count())]

    for i, worker in enumerate(workers):
      print("Starting worker {}".format(i))
      worker.start()

    moving_average_rewards = []  # record episode reward to plot
    while True:
      reward = res_queue.get()
      if reward is not None:
        moving_average_rewards.append(reward)
      else:
        break
    [w.join() for w in workers]

    plt.plot(moving_average_rewards)
    plt.ylabel('Moving average ep reward')
    plt.xlabel('Step')
    plt.savefig(os.path.join(self.save_dir,
                             '{} Moving Average.png'.format(self.game_name)))
    plt.show()

内存类 - 保存我们的体验

此外，为了更轻松地跟踪事物，我们还将实现一个 Memory 类。该类将只提供我们跟踪每个步骤发生的行动、奖励和状态的功能。

class Memory:
  def __init__(self):
    self.states = []
    self.actions = []
    self.rewards = []
    
  def store(self, state, action, reward):
    self.states.append(state)
    self.actions.append(action)
    self.rewards.append(reward)
    
  def clear(self):
    self.states = []
    self.actions = []
    self.rewards = []

现在，我们进入算法的核心：工作代理。工作代理继承自线程类，并且我们覆盖了 Thread 的 run 方法。这将使我们能够实现 A3C 中的第一个 A，即异步。首先，我们将通过实例化一个本地模型并设置特定的训练参数来开始。

class Worker(threading.Thread):
  # Set up global variables across different threads
  global_episode = 0
  # Moving average reward
  global_moving_average_reward = 0
  best_score = 0
  save_lock = threading.Lock()

  def __init__(self,
               state_size,
               action_size,
               global_model,
               opt,
               result_queue,
               idx,
               game_name='CartPole-v0',
               save_dir='/tmp'):
    super(Worker, self).__init__()
    self.state_size = state_size
    self.action_size = action_size
    self.result_queue = result_queue
    self.global_model = global_model
    self.opt = opt
    self.local_model = ActorCriticModel(self.state_size, self.action_size)
    self.worker_idx = idx
    self.game_name = game_name
    self.env = gym.make(self.game_name).unwrapped
    self.save_dir = save_dir
    self.ep_loss = 0.0

运行算法

下一步是实现run函数。这将实际运行我们的算法。我们将为给定的全局最大情节数运行所有线程。这就是 A3C 中的第三个 A，即演员，发挥作用的地方。我们的智能体将根据我们的策略函数“行动”，成为演员，而“评论家”——我们的价值函数——会判断行动。虽然代码的这一部分可能看起来很密集，但实际上并没有做太多事情。在每个情节中，代码只是执行以下操作：

1. 根据当前帧获取我们的策略（动作概率分布）
   
2. 根据策略选择的动作进行下一步
   
3. 如果智能体已经执行了设定的步数（args.update_freq）或已达到终止状态（已死亡），则使用从本地模型计算的梯度更新全局模型
   
4. 重复

def run(self):
    total_step = 1
    mem = Memory()
    while Worker.global_episode < args.max_eps:
      current_state = self.env.reset()
      mem.clear()
      ep_reward = 0.
      ep_steps = 0
      self.ep_loss = 0

      time_count = 0
      done = False
      while not done:
        logits, _ = self.local_model(
            tf.convert_to_tensor(current_state[None, :],
                                 dtype=tf.float32))
        probs = tf.nn.softmax(logits)

        action = np.random.choice(self.action_size, p=probs.numpy()[0])
        new_state, reward, done, _ = self.env.step(action)
        if done:
          reward = -1
        ep_reward += reward
        mem.store(current_state, action, reward)

        if time_count == args.update_freq or done:
          # Calculate gradient wrt to local model. We do so by tracking the
          # variables involved in computing the loss by using tf.GradientTape
          with tf.GradientTape() as tape:
            total_loss = self.compute_loss(done,
                                           new_state,
                                           mem,
                                           args.gamma)
          self.ep_loss += total_loss
          # Calculate local gradients
          grads = tape.gradient(total_loss, self.local_model.trainable_weights)
          # Push local gradients to global model
          self.opt.apply_gradients(zip(grads,
                                       self.global_model.trainable_weights))
          # Update local model with new weights
          self.local_model.set_weights(self.global_model.get_weights())

          mem.clear()
          time_count = 0

          if done:  # done and print information
            Worker.global_moving_average_reward = \
              record(Worker.global_episode, ep_reward, self.worker_idx,
                     Worker.global_moving_average_reward, self.result_queue,
                     self.ep_loss, ep_steps)
            # We must use a lock to save our model and to print to prevent data races.
            if ep_reward > Worker.best_score:
              with Worker.save_lock:
                print("Saving best model to {}, "
                      "episode score: {}".format(self.save_dir, ep_reward))
                self.global_model.save_weights(
                    os.path.join(self.save_dir,
                                 'model_{}.h5'.format(self.game_name))
                )
                Worker.best_score = ep_reward
            Worker.global_episode += 1
        ep_steps += 1

        time_count += 1
        current_state = new_state
        total_step += 1
    self.result_queue.put(None)

损失是如何计算的？

工作智能体计算损失以获得关于其所有自身网络参数的梯度。这就是 A3C 中最后一个 A，即优势，发挥作用的地方。这些梯度然后被应用于全局网络。损失的计算方法如下：

1. 价值损失：L=∑(R — V(s))²
   
2. 策略损失：L = -log(𝝅(s)) * A(s)
   

其中 R 是折扣奖励，V 是我们的价值函数（输入状态的函数），𝛑 是我们的策略函数（也是输入状态的函数），A 是我们的优势函数。我们使用折扣奖励来估计我们的 Q 值，因为我们没有直接用 A3C 确定 Q 值。

def compute_loss(self,
                   done,
                   new_state,
                   memory,
                   gamma=0.99):
    if done:
      reward_sum = 0.  # terminal
    else:
      reward_sum = self.local_model(
          tf.convert_to_tensor(new_state[None, :],
                               dtype=tf.float32))[-1].numpy()[0]

    # Get discounted rewards
    discounted_rewards = []
    for reward in memory.rewards[::-1]:  # reverse buffer r
      reward_sum = reward + gamma * reward_sum
      discounted_rewards.append(reward_sum)
    discounted_rewards.reverse()

    logits, values = self.local_model(
        tf.convert_to_tensor(np.vstack(memory.states),
                             dtype=tf.float32))
    # Get our advantages
    advantage = tf.convert_to_tensor(np.array(discounted_rewards)[:, None],
                            dtype=tf.float32) - values
    # Value loss
    value_loss = advantage ** 2

    # Calculate our policy loss
    actions_one_hot = tf.one_hot(memory.actions, self.action_size, dtype=tf.float32)

    policy = tf.nn.softmax(logits)
    entropy = tf.reduce_sum(policy * tf.log(policy + 1e-20), axis=1)

    policy_loss = tf.nn.softmax_cross_entropy_with_logits_v2(labels=actions_one_hot,
                                                             logits=logits)
    policy_loss *= tf.stop_gradient(advantage)
    policy_loss -= 0.01 * entropy
    total_loss = tf.reduce_mean((0.5 * value_loss + policy_loss))
    return total_loss

就是这样！工作智能体将重复将网络参数重置为全局网络中的所有参数的过程，并重复与环境交互、计算损失，然后将梯度应用于全局网络的过程。通过运行以下命令训练您的算法：python a3c_cartpole.py — train。

测试算法

让我们通过启动一个新的环境，并简单地遵循我们训练好的模型的策略输出，来测试该算法。这将呈现我们的环境，并从我们模型的策略分布中进行采样。

 def play(self):
    env = gym.make(self.game_name).unwrapped
    state = env.reset()
    model = self.global_model
    model_path = os.path.join(self.save_dir, 'model_{}.h5'.format(self.game_name))
    print('Loading model from: {}'.format(model_path))
    model.load_weights(model_path)
    done = False
    step_counter = 0
    reward_sum = 0

    try:
      while not done:
        env.render(mode='rgb_array')
        policy, value = model(tf.convert_to_tensor(state[None, :], dtype=tf.float32))
        policy = tf.nn.softmax(policy)
        action = np.argmax(policy)
        state, reward, done, _ = env.step(action)
        reward_sum += reward
        print("{}. Reward: {}, action: {}".format(step_counter, reward_sum, action))
        step_counter += 1
    except KeyboardInterrupt:
      print("Received Keyboard Interrupt. Shutting down.")
    finally:
      env.close()

训练完模型后，您可以使用以下命令运行它：python a3c_cartpole.py。

要检查我们的分数移动平均值： 我们应该看到分数收敛到大于 200 的分数。该游戏被定义为“解决”为在 100 次连续尝试中获得 195.0 的平均奖励。

以及在新的环境上的性能示例：

关键要点

我们涵盖的内容

• 我们通过实现 A3C 来解决了 CartPole 问题！
  
• 我们通过使用 Eager Execution、模型子类化和自定义训练循环来实现这一点。
  
• Eager 是一种开发训练循环的简单方法，它使编码更轻松、更清晰，因为我们可以直接打印和调试张量。
  
• 我们学习了使用策略和价值网络的强化学习基础知识，然后将它们结合在一起以实现 A3C。
  
• 我们使用 tf.gradient，通过应用优化器的更新规则来迭代地更新我们的全局网络。