0
0

深度强化学习-DQN算法原理与代码

DQN算法是DeepMind团队提出的一种深度强化学习算法，在许多电动游戏中达到人类玩家甚至超越人类玩家的水准，本文就带领大家了解一下这个算法，论文和代码的链接见下方。

论文：Human-level control through deep reinforcement learning | Nature

代码：https://github.com/indigoLovee/DQN

喜欢的话可以点个star呢。

1 DQN算法简介

Q-learning算法采用一个Q-tabel来记录每个状态下的动作值，当状态空间或动作空间较大时，需要的存储空间也会较大。如果状态空间或动作空间连续，则该算法无法使用。因此，Q-learning算法只能用于解决离散低维状态空间和动作空间类问题。DQN算法的核心就是用一个人工神经网络q(s,a;\omega ),s\in S,a\in A来代替Q-tabel，即动作价值函数。网络的输入为状态信息，输出为每个动作的价值，因此DQN算法可以用来解决连续状态空间和离散动作空间问题，无法解决连续动作空间类问题。针对连续动作空间类问题，后面blog会慢慢介绍。

2 DQN算法原理

DQN算法是一种off-policy算法，当同时出现异策、自益和函数近似时，无法保证收敛性，容易出现训练不稳定或训练困难等问题。针对这些问题，研究人员主要从以下两个方面进行了改进。

（1）经验回放：将经验（当前状态s_{t}、动作a_{t}、即时奖励r_{t+1}、下个状态s_{t+1}、回合状态done）存放在经验池中，并按照一定的规则采样。

（2）目标网络：修改网络的更新方式，例如不把刚学习到的网络权重马上用于后续的自益过程。

2.1 经验回放

经验回放就是一种让经验概率分布变得稳定的技术，可以提高训练的稳定性。经验回放主要有“存储”和“回放”两大关键步骤：

存储：将经验以(s_{t},a_{t},r_{t+1},s_{t+1},done)形式存储在经验池中。

回放：按照某种规则从经验池中采样一条或多条经验数据。

从存储的角度来看，经验回放可以分为集中式回放和分布式回放：

集中式回放：智能体在一个环境中运行，把经验统一存储在经验池中。

分布式回放：多个智能体同时在多个环境中运行，并将经验统一存储在经验池中。由于多个智能体同时生成经验，所以能够使用更多资源的同时更快地收集经验。

从采样的角度来看，经验回放可以分为均匀回放和优先回放：

均匀回放：等概率从经验池中采样经验。

优先回放：为经验池中每条经验指定一个优先级，在采样经验时更倾向于选择优先级更高的经验。一般的做法是，如果某条经验（例如经验i）的优先级为p_{i}，那么选取该经验的概率为：

p_{i}=\frac{p_{i}}{\sum p_{k}}

优先回放可以具体参照这篇论文：优先经验回放

经验回放的优点：

1.在训练Q网络时，可以打破数据之间的相关性，使得数据满足独立同分布，从而减小参数更新的方差，提高收敛速度。

2.能够重复使用经验，数据利用率高，对于数据获取困难的情况尤其有用。

经验回放的缺点：

无法应用于回合更新和多步学习算法。但是将经验回放应用于Q学习，就规避了这个缺点。

代码中采用集中式均匀回放，具体如下：

  1. import numpy as np
  3. class ReplayBuffer:
  4.     def __init__(self, state_dim, action_dim, max_size, batch_size):
  5.         self.mem_size = max_size
  6.         self.batch_size = batch_size
  7.         self.mem_cnt = 0
  9.         self.state_memory = np.zeros((self.mem_size, state_dim))
  10.         self.action_memory = np.zeros((self.mem_size, ))
  11.         self.reward_memory = np.zeros((self.mem_size, ))
  12.         self.next_state_memory = np.zeros((self.mem_size, state_dim))
  13.         self.terminal_memory = np.zeros((self.mem_size, ), dtype=np.bool)
  15.     def store_transition(self, state, action, reward, state_, done):
  16.         mem_idx = self.mem_cnt % self.mem_size
  18.         self.state_memory[mem_idx] = state
  19.         self.action_memory[mem_idx] = action
  20.         self.reward_memory[mem_idx] = reward
  21.         self.next_state_memory[mem_idx] = state_
  22.         self.terminal_memory[mem_idx] = done
  24.         self.mem_cnt += 1
  26.     def sample_buffer(self):
  27.         mem_len = min(self.mem_size, self.mem_cnt)
  29.         batch = np.random.choice(mem_len, self.batch_size, replace=True)
  31.         states = self.state_memory[batch]
  32.         actions = self.action_memory[batch]
  33.         rewards = self.reward_memory[batch]
  34.         states_ = self.next_state_memory[batch]
  35.         terminals = self.terminal_memory[batch]
  37.         return states, actions, rewards, states_, terminals
  39.     def ready(self):
  40.         return self.mem_cnt > self.batch_size

2.2 目标网络

对于基于自益的Q学习，动作价值估计和权重w有关。当权重变化时，动作价值的估计也会发生变化。在学习的过程中，动作价值试图追逐一个变化的回报，容易出现不稳定的情况。

目标网络是在原有的神经网络之外重新搭建一个结构完全相同的网络。原先的网络称为评估网络，新构建的网络称为目标网络。在学习过程中，使用目标网络进行自益得到回报的评估值，作为学习目标。在更新过程中，只更新评估网络的权重，而不更新目标网络的权重。这样，更新权重时针对的目标不会在每次迭代都发生变化，是一个固定的目标。在更新一定次数后，再将评估网络的权重复制给目标网络，进而进行下一批更新，这样目标网络也能得到更新。由于在目标网络没有变化的一段时间内回报的估计是相对固定的，因此目标网络的引入增加了学习的稳定性。

目标网络的更新方式：

上述在一段时间内固定目标网络，一定次数后将评估网络权重复制给目标网络的更新方式为**硬更新(hard update)**，即

w_{t}\leftarrow w_{e}

其中w_{t}表示目标网络权重，w_{e}表示评估网络权重。

另外一种常用的更新方式为**软更新(soft update)**，即引入一个学习率\tau，将旧的目标网络参数和新的评估网络参数直接做加权平均后的值赋值给目标网络

w_{t}\leftarrow \tau w_{e}+\left ( 1-\tau \right )w_{t}

学习率\tau\in \left ( 0,1 \right )

3 DQN算法伪代码

DQN算法的实现代码为：

  1. import torch as T
  2. import torch.nn as nn
  3. import torch.optim as optim
  4. import torch.nn.functional as F
  5. import numpy as np
  6. from buffer import ReplayBuffer
  8. device = T.device("cuda:0" if T.cuda.is_available() else "cpu")
  10. class DeepQNetwork(nn.Module):
  11.     def __init__(self, alpha, state_dim, action_dim, fc1_dim, fc2_dim):
  12.         super(DeepQNetwork, self).__init__()
  14.         self.fc1 = nn.Linear(state_dim, fc1_dim)
  15.         self.fc2 = nn.Linear(fc1_dim, fc2_dim)
  16.         self.q = nn.Linear(fc2_dim, action_dim)
  18.         self.optimizer = optim.Adam(self.parameters(), lr=alpha)
  19.         self.to(device)
  21.     def forward(self, state):
  22.         x = T.relu(self.fc1(state))
  23.         x = T.relu(self.fc2(x))
  25.         q = self.q(x)
  27.         return q
  29.     def save_checkpoint(self, checkpoint_file):
  30.         T.save(self.state_dict(), checkpoint_file, _use_new_zipfile_serialization=False)
  32.     def load_checkpoint(self, checkpoint_file):
  33.         self.load_state_dict(T.load(checkpoint_file))
  35. class DQN:
  36.     def __init__(self, alpha, state_dim, action_dim, fc1_dim, fc2_dim, ckpt_dir,
  37.                  gamma=0.99, tau=0.005, epsilon=1.0, eps_end=0.01, eps_dec=5e-4,
  38.                  max_size=1000000, batch_size=256):
  39.         self.tau = tau
  40.         self.gamma = gamma
  41.         self.epsilon = epsilon
  42.         self.eps_min = eps_end
  43.         self.eps_dec = eps_dec
  44.         self.batch_size = batch_size
  45.         self.action_space = [i for i in range(action_dim)]
  46.         self.checkpoint_dir = ckpt_dir
  48.         self.q_eval = DeepQNetwork(alpha=alpha, state_dim=state_dim, action_dim=action_dim,
  49.                                    fc1_dim=fc1_dim, fc2_dim=fc2_dim)
  50.         self.q_target = DeepQNetwork(alpha=alpha, state_dim=state_dim, action_dim=action_dim,
  51.                                      fc1_dim=fc1_dim, fc2_dim=fc2_dim)
  53.         self.memory = ReplayBuffer(state_dim=state_dim, action_dim=action_dim,
  54.                                    max_size=max_size, batch_size=batch_size)
  56.         self.update_network_parameters(tau=1.0)
  58.     def update_network_parameters(self, tau=None):
  59.         if tau is None:
  60.             tau = self.tau
  62.         for q_target_params, q_eval_params in zip(self.q_target.parameters(), self.q_eval.parameters()):
  63.             q_target_params.data.copy_(tau * q_eval_params + (1 - tau) * q_target_params)
  65.     def remember(self, state, action, reward, state_, done):
  66.         self.memory.store_transition(state, action, reward, state_, done)
  68.     def choose_action(self, observation, isTrain=True):
  69.         state = T.tensor([observation], dtype=T.float).to(device)
  70.         actions = self.q_eval.forward(state)
  71.         action = T.argmax(actions).item()
  73.         if (np.random.random() < self.epsilon) and isTrain:
  74.             action = np.random.choice(self.action_space)
  76.         return action
  78.     def learn(self):
  79.         if not self.memory.ready():
  80.             return
  82.         states, actions, rewards, next_states, terminals = self.memory.sample_buffer()
  83.         batch_idx = np.arange(self.batch_size)
  85.         states_tensor = T.tensor(states, dtype=T.float).to(device)
  86.         rewards_tensor = T.tensor(rewards, dtype=T.float).to(device)
  87.         next_states_tensor = T.tensor(next_states, dtype=T.float).to(device)
  88.         terminals_tensor = T.tensor(terminals).to(device)
  90.         with T.no_grad():
  91.             q_ = self.q_target.forward(next_states_tensor)
  92.             q_[terminals_tensor] = 0.0
  93.             target = rewards_tensor + self.gamma * T.max(q_, dim=-1)[0]
  94.         q = self.q_eval.forward(states_tensor)[batch_idx, actions]
  96.         loss = F.mse_loss(q, target.detach())
  97.         self.q_eval.optimizer.zero_grad()
  98.         loss.backward()
  99.         self.q_eval.optimizer.step()
  101.         self.update_network_parameters()
  102.         self.epsilon = self.epsilon - self.eps_dec if self.epsilon > self.eps_min else self.eps_min
  104.     def save_models(self, episode):
  105.         self.q_eval.save_checkpoint(self.checkpoint_dir + 'Q_eval/DQN_q_eval_{}.pth'.format(episode))
  106.         print('Saving Q_eval network successfully!')
  107.         self.q_target.save_checkpoint(self.checkpoint_dir + 'Q_target/DQN_Q_target_{}.pth'.format(episode))
  108.         print('Saving Q_target network successfully!')
  110.     def load_models(self, episode):
  111.         self.q_eval.load_checkpoint(self.checkpoint_dir + 'Q_eval/DQN_q_eval_{}.pth'.format(episode))
  112.         print('Loading Q_eval network successfully!')
  113.         self.q_target.load_checkpoint(self.checkpoint_dir + 'Q_target/DQN_Q_target_{}.pth'.format(episode))
  114.         print('Loading Q_target network successfully!')

算法仿真环境是在gym库中的LunarLander-v2环境，因此需要先配置好gym库。进入Aanconda中对应的Python环境中，执行下面的指令

  1. pip install gym

但是，这样安装的gym库只包括少量的内置环境，如算法环境、简单文字游戏环境和经典控制环境，无法使用LunarLander-v2。因此还要安装一些其他依赖项，具体可以参照这篇blog:AttributeError: module ‘gym.envs.box2d‘ has no attribute ‘LunarLander‘ 解决办法

训练脚本如下：

  1. import gym
  2. import numpy as np
  3. import argparse
  4. from DQN import DQN
  5. from utils import plot_learning_curve, create_directory
  7. parser = argparse.ArgumentParser()
  8. parser.add_argument('--max_episodes', type=int, default=500)
  9. parser.add_argument('--ckpt_dir', type=str, default='./checkpoints/DQN/')
  10. parser.add_argument('--reward_path', type=str, default='./output_images/avg_reward.png')
  11. parser.add_argument('--epsilon_path', type=str, default='./output_images/epsilon.png')
  13. args = parser.parse_args()
  15. def main():
  16.     env = gym.make('LunarLander-v2')
  17.     agent = DQN(alpha=0.0003, state_dim=env.observation_space.shape[0], action_dim=env.action_space.n,
  18.                 fc1_dim=256, fc2_dim=256, ckpt_dir=args.ckpt_dir, gamma=0.99, tau=0.005, epsilon=1.0,
  19.                 eps_end=0.05, eps_dec=5e-4, max_size=1000000, batch_size=256)
  20.     create_directory(args.ckpt_dir, sub_dirs=['Q_eval', 'Q_target'])
  21.     total_rewards, avg_rewards, eps_history = [], [], []
  23.     for episode in range(args.max_episodes):
  24.         total_reward = 0
  25.         done = False
  26.         observation = env.reset()
  27.         while not done:
  28.             action = agent.choose_action(observation, isTrain=True)
  29.             observation_, reward, done, info = env.step(action)
  30.             agent.remember(observation, action, reward, observation_, done)
  31.             agent.learn()
  32.             total_reward += reward
  33.             observation = observation_
  35.         total_rewards.append(total_reward)
  36.         avg_reward = np.mean(total_rewards[-100:])
  37.         avg_rewards.append(avg_reward)
  38.         eps_history.append(agent.epsilon)
  39.         print('EP:{} reward:{} avg_reward:{} epsilon:{}'.
  40.               format(episode + 1, total_reward, avg_reward, agent.epsilon))
  42.         if (episode + 1) % 50 == 0:
  43.             agent.save_models(episode + 1)
  45.     episodes = [i for i in range(args.max_episodes)]
  46.     plot_learning_curve(episodes, avg_rewards, 'Reward', 'reward', args.reward_path)
  47.     plot_learning_curve(episodes, eps_history, 'Epsilon', 'epsilon', args.epsilon_path)
  49. if __name__ == '__main__':
  50.     main()

训练时还会用到画图函数和创建文件夹函数，我将他们另外放在一个utils.py脚本中，具体代码如下：

  1. import os
  2. import matplotlib.pyplot as plt
  4. def plot_learning_curve(episodes, records, title, ylabel, figure_file):
  5.     plt.figure()
  6.     plt.plot(episodes, records, linestyle='-', color='r')
  7.     plt.title(title)
  8.     plt.xlabel('episode')
  9.     plt.ylabel(ylabel)
  11.     plt.show()
  12.     plt.savefig(figure_file)
  14. def create_directory(path: str, sub_dirs: list):
  15.     for sub_dir in sub_dirs:
  16.         if os.path.exists(path + sub_dir):
  17.             print(path + sub_dir + ' is already exist!')
  18.         else:
  19.             os.makedirs(path + sub_dir, exist_ok=True)
  20.             print(path + sub_dir + ' create successfully!')

仿真结果如下图所示：

通过平均奖励曲线可以看出，大概迭代到400步左右时算法趋于收敛。

标签： 算法 人工智能 深度学习
本文转载自: https://blog.csdn.net/weixin_46133643/article/details/121845874
版权归原作者 indigo love 所有， 如有侵权，请联系我们删除。
登录后发布评论

“深度强化学习-DQN算法原理与代码”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

网络安全法-网络运行安全

使用selenium/drissionpage时如何阻止chrome自动跳转http到https

docker desktop 里部署的Open WebUI 管理员密码忘记了的处理方法

在ubuntu20.04中搭建onsite比赛运行环境

利用开源的低代码表单设计器FcDesigner高效管理和渲染复杂表单结构

Kafka学习笔记

【前端】浏览器输入url到页面呈现发生了什么？