基于模块化和快速原型设计的Huskarl深度强化学习框架

• 2019 年 11 月 21 日
• 筆記

深度强化学习报道

来源：Huskarl（Medium）

编辑：DeepRL

前言：Huskarl是一种基于TensorFlow 2.0构建的深度强化学习的框架，其专注于模块化和快速原型设计。设计中尽可能使用了tf.keras API以实现简洁性和可读性。Huskarl可以轻松地跨多个CPU核心并行计算环境动态。这对于加速从多个并发经验源（如A2C或PPO）中受益的策略性学习算法非常有用。并且对于计算密集型环境尤其有用，例如基于物理的环境。其与OpenAI Gym环境无缝协作，并支持多智能体环境和Unity3D环境。

一、简介

深度学习革命一直是从计算机视觉到自然语言处理等领域的许多最新进展和突破的原因。已经看到非凡增长的一个特殊领域是深度强化学习。2013年，DeepMind发布了“使用深度强化学习玩Atari”，他们的模型只是通过观看屏幕上的像素来学习玩Atari游戏。三年后，AlphaGo击败了Go世界冠军，吸引了全球观众。最近，AlphaZero打破了从人类比赛中学习的需要，通过自我发挥将学习推广到任何完美的信息游戏，并有效地成为Go，Chess和Shogi的世界冠军。

1.1

Huskarl在Atari游戏中的应用

Huskarl在#PoweredByTF 2.0挑战赛中获得第一名。其目标是让研究人员轻松实现，测试，调整和比较深度RL算法。类似于TensorFlow如何抽象出计算图的管理，以及Keras创建高级模型，Huskarl抽象出代理 – 环境交互。这使用户可以专注于开发和理解算法，同时还可以防止数据泄漏。目前项目仍处于早期阶段，但它已经包括深度Q学习网络（DQN），Double DQN，AC，DDPG等算法的实现，同时提供了解决离散和连续状态下的方法。并且，Huskarl使用OpenAI Gym工具包开发和比较RL算法，因此与其他开源框架等具有可比性。

下文是创建和可视化DQN代理所需的完整代码，该智能体学习平衡一个cartpole，可以看出整个的代码非常的简洁，后文将会详细讲述过程。

from tensorflow.keras.models import Sequential  from tensorflow.keras.layers import Dense  import huskarl as hk  import gym    # Setup gym environment  create_env = lambda: gym.make('CartPole-v0').unwrapped  dummy_env = create_env()    # Build a simple neural network with 3 fully connected layers as our model  model = Sequential([    Dense(16, activation='relu', input_shape=dummy_env.observation_space.shape),    Dense(16, activation='relu'),    Dense(16, activation='relu'),  ])    # Create Deep Q-Learning Network agent  agent = hk.agent.DQN(model, actions=dummy_env.action_space.n, nsteps=2)    # Create simulation, train and then test  sim = hk.Simulation(create_env, agent)  sim.train(max_steps=3000, visualize=True)  sim.test(max_steps=1000)  

目前，Huskarl实现了包括三个可调智能体的若干算法。DQN智能体实现了深度Q-Learning以及多种增强功能，例如可变步长跟踪，双DQN和可调整的决斗架构。DQN是一种非策略算法，过程实现默认使用优先级经验重放。A2C代理实现Advantage Actor-Critic的同步，多步版本，这是一种on-policy算法。Huskarl允许像A2C这样的on-policy算法轻松地同时从多个环境实例中获取经验。这有助于将数据解相关成一个更加固定的过程，最后，DDPG代理使用变步长跟踪实现深度确定性策略梯度，默认情况下也使用优先级经验重放。DDPG代理处理连续动作空间的问题。

Huskarl可以轻松地跨多个CPU核心并行计算环境动态。这对于加速从多个并发经验源（例如A2C或PPO）中受益的策略性学习算法非常有用。首先，要同时使用多个环境实例，只需为策略上的代理和模拟提供所需数量的实例。然后，将环境实例分布在多个进程上，这些进程在可用的CPU内核上自动并行化，只需在调用sim.train（）时为max_subprocesses参数提供所需的值，如下面的代码段所示。另外，请注意为每个环境实例使用不同的策略是多么简单 – 只需提供策略列表而不是单个策略对象：

# We will be running multiple concurrent environment instances  instances = 16    # Create a policy for each instance with a different distribution for epsilon  policy = [hk.policy.Greedy()] + [hk.policy.GaussianEpsGreedy(eps, 0.1) for eps in np.arange(0, 1, 1/(instances-1))]    # Create Advantage Actor-Critic agent  agent = hk.agent.A2C(model, actions=dummy_env.action_space.n, nsteps=2, instances=instances, policy=policy)    # Create simulation, train and then test  sim = hk.Simulation(create_env, agent)  sim.train(max_steps=5000, instances=instances, max_subprocesses=8)  sim.test(max_steps=1000)  

值得注意的是，某些环境（如cartpole环境）非常简单，因此使用多个进程实际上会因进程间通信开销而减慢培训速度。只有计算上昂贵的环境才能从跨进程传播中受益。在所有实现的智能体中，使用的神经网络由用户提供，因为它们依赖于每个问题规范。它们可以如所希望的那样简单或者复杂。此外，所有算法都充分利用自定义Keras损耗尽可能快速和简洁

下一步是什么

1.2

Huskarl目前支持的算法

二、Huskarl环境安装与使用

源方式

2.1

源代码安装

git clone https://github.com/danaugrs/huskarl.git  cd huskarl  pip install -e .

2.2

PiP安装

pip install huskarl  

2.3

使用方法

相对于其他框架和方法的实现上，Huskarl在代码的简洁性上可以说美妙绝伦。本部以huskarl利用DDPG算法实现Pendulum为例子进行分析。

1、首先引入相关的包（包括tensorflow kera、Gym环境和huskarl）

from tensorflow.keras.models import Sequential, Model  from tensorflow.keras.layers import Dense, Input, Concatenate  import matplotlib.pyplot as plt  import gym  import huskarl as hk  

2、Build环境（同往常一样）

   create_env = lambda: gym.make('Pendulum-v0')     dummy_env = create_env()     action_size = dummy_env.action_space.shape[0]     state_shape = dummy_env.observation_space.shape  

3、建立Actor模型

由于DDPG算法使用了AC框架，因此需要创建对应的网络

   actor = Sequential([          Dense(16, activation='relu', input_shape=state_shape),          Dense(16, activation='relu'),          Dense(16, activation='relu'),          Dense(action_size, activation='linear')      ])  

4、建立Critic环境

   action_input = Input(shape=(action_size,), name='action_input')      state_input = Input(shape=state_shape, name='state_input')      x = Concatenate()([action_input, state_input])      x = Dense(32, activation='relu')(x)      x = Dense(32, activation='relu')(x)      x = Dense(32, activation='relu')(x)      x = Dense(1, activation='linear')(x)      critic = Model(inputs=[action_input, state_input], outputs=x)  

5、创建深层确定性策略梯度智能体

非常关键，也是使用huskarl核心的一步。

  agent = hk.agent.DDPG(actor=actor, critic=critic, nsteps=2)

6、画过程和曲线图

 def plot_rewards(episode_rewards, episode_steps, done=False):          plt.clf()          plt.xlabel('Step')          plt.ylabel('Reward')          for ed, steps in zip(episode_rewards, episode_steps):              plt.plot(steps, ed)          plt.show() if done else plt.pause(0.001) # Pause a bit so that the graph is updated  

7、创建模拟训练和测试的过程

  sim = hk.Simulation(create_env, agent)    sim.train(max_steps=30_000, visualize=True, plot=plot_rewards)    sim.test(max_steps=5_000)  

2.4

完整DDPG代码实现Pendulum

from tensorflow.keras.models import Sequential, Model  from tensorflow.keras.layers import Dense, Input, Concatenate    import matplotlib.pyplot as plt  import gym    import huskarl as hk    if __name__ == "__main__":        # Setup gym environment      create_env = lambda: gym.make('Pendulum-v0')      dummy_env = create_env()      action_size = dummy_env.action_space.shape[0]      state_shape = dummy_env.observation_space.shape        # Build a simple actor model      actor = Sequential([          Dense(16, activation='relu', input_shape=state_shape),          Dense(16, activation='relu'),          Dense(16, activation='relu'),          Dense(action_size, activation='linear')      ])        # Build a simple critic model      action_input = Input(shape=(action_size,), name='action_input')      state_input = Input(shape=state_shape, name='state_input')      x = Concatenate()([action_input, state_input])      x = Dense(32, activation='relu')(x)      x = Dense(32, activation='relu')(x)      x = Dense(32, activation='relu')(x)      x = Dense(1, activation='linear')(x)      critic = Model(inputs=[action_input, state_input], outputs=x)        # Create Deep Deterministic Policy Gradient agent      agent = hk.agent.DDPG(actor=actor, critic=critic, nsteps=2)        def plot_rewards(episode_rewards, episode_steps, done=False):          plt.clf()          plt.xlabel('Step')          plt.ylabel('Reward')          for ed, steps in zip(episode_rewards, episode_steps):              plt.plot(steps, ed)          plt.show() if done else plt.pause(0.001) # Pause a bit so that the graph is updated        # Create simulation, train and then test      sim = hk.Simulation(create_env, agent)      sim.train(max_steps=30_000, visualize=True, plot=plot_rewards)      sim.test(max_steps=5_000)  

参考资料：

1. https://tensorflow.devpost.com/

2. https://github.com/danaugrs/huskarl