深度强化学习开端——DQN算法求解车杆游戏

DQN，即深度Q网络（Deep Q-Network），是一种结合了深度学习和强化学习的算法，其主要用于解决序列决策问题，并且在许多复杂的决策任务中展现出了显著的效果。DQN算法的发明历史可以追溯到2013年，当时DeepMind团队首次提出了一种名为DQN（Deep Q-Network）的新型强化学习算法。这一算法标志着深度学习和强化学习成功结合的开始，为解决高维、连续状态空间的问题提供了一种有效的解决方案。

DQN算法通过深度神经网络对状态的价值函数进行近似，使算法能够自主学习游戏中的策略。然而，早期的DQN算法存在不稳定性问题，为了改进这些问题，DeepMind团队随后又提出了Double DQN、Dueling DQN等改进版本，以提高算法的稳定性和性能。

此外，DQN算法的扩展也在不断发展，研究者们提出了诸如Prioritized DQN、Rainbow等扩展算法，这些算法在复杂的游戏环境中表现出色，并引起了广泛的关注和讨论。值得注意的是，谷歌的AlphaGo中也使用了DQN算法。

接下来，让我们把DQN算法应用到车杆游戏的求解中，讨论并实践该游戏的策略。让我们开始吧！

注意：本文用到了PyTorch库，gym强化学习环境库，需要提前安装。

• gym环境安装：https://github.com/Farama-Foundation/Gymnasium
• gym环境介绍文档：https://gymnasium.farama.org/environments/classic_control/mountain_car/
• pytorch官网：https://pytorch.org/

本文所使用的python版本为3.11.5

step1：车杆（Cart Pole）游戏介绍

官方地址：https://gymnasium.farama.org/environments/classic_control/cart_pole/
车杆游戏中，一个摆杆通过一个非驱动关节连接到一个小车上，小车沿着无摩擦轨道移动。摆杆直立放置在小车上，我们的目标是通过在小车上向左和向右施加力来使摆杆保持平衡。

让我们先看看该环境的状态及动作空间吧！

import gymnasium as gym  # 导入gym包
env = gym.make('CartPole-v1')  # 创建山地车游戏环境
observation, info = env.reset()  # 初始化环境

for _ in range(1000):
    action = env.action_space.sample()  # 随机选择一个动作
    observation, reward, terminated, truncated, info = env.step(action)  # 执行动作

    if terminated or truncated:  # 当达到终点时，重置环境
        observation, info = env.reset()

env.close()  # 关闭环境

从代码输出中，结合官网文档，我们可以知道：
动作空间

动作是一个int64整数值，其可以取值{0, 1}，代表小车被推的方向上的固定力。

• 0: 将小车推向左边
• 1: 将小车推向右边

注意：通过施加的力所减少或增加的速度不是固定的，它取决于摆杆所指的角度。摆杆的重心变化会影响移动其下方小车所需的能量。

观察空间

观察值是一个形状为(4,)的ndarray，其值对应于以下的位置和速度：

编号	观察项目	最小值	最大值
0	小车位置	-4.8	4.8
1	小车速度	-Inf	Inf
2	摆杆角度	~ -0.418 rad (-24°)	~ 0.418 rad (24°)
3	摆杆角速度	-Inf	Inf

注意：尽管上述范围表示了每个元素的观察空间可能的值，但这并不反映在一个未终止的情节中状态空间所允许的值。特别要注意：

• 小车的x位置（索引0）可以在(-4.8, 4.8)之间取值，但如果小车离开(-2.4, 2.4)的范围，则情节将终止。
• 摆杆角度可以在(-.418, .418)弧度（或±24°）之间观察到，但如果摆杆角度不在(-.2095, .2095)（或±12°）的范围内，则情节将终止。

奖励

由于目标是尽可能长时间地保持摆杆直立，因此为每一步（包括终止步骤）分配+1的奖励。对于v1版本，奖励的阈值是500；对于v0版本，奖励的阈值是200。即在v1版本坚持 500 个回合即可获得最高的分数。

情节结束

如果发生以下任一情况，情节将结束：

终止：摆杆角度大于±12°

终止：小车位置大于±2.4（小车中心到达显示屏边缘）

截断：情节长度大于500（对于v0版本为200）

step2:DQN算法架构的搭建

DQN算法原理可参考动手学强化学习——DQN算法，DQN算法的实现具体需要以下几个部件：

• 网络$Q_{\theta }(s,a)$：该网络可以是输入状态$s$，输出每一个动作$a$对应的Q值。也可以是输入状态$s$和动作$a$，输出对应的Q值。
• 目标网络$Q_{\ast }(s,a)$：该网络与网络$Q_{\theta }(s,a)$结构相同，但使用不同的参数。这是为了让训练过程更加稳定。
• 经验回放池：该池用于存储训练过程中产生的经验，包括状态$s$，动作$a$，奖励$r$，下一个状态$s^{\prime }$，游戏是否结束$d$。

在更新网络时，需要从经验回放池中随机抽取一批经验，然后使用这些经验来更新网络。注意：DQN背后的原理为时序差分算法中的Q-learning 算法，其整个训练过程是一个迭代收敛的过程。
下方是DQN中最重要的公式：

$$Q(s,a)\leftarrow Q(s,a)+\alpha \left[r+\gamma \underset{a^{\prime }\in \mathcal{A}}{\max }Q\left(s^{\prime },a^{\prime }\right)-Q(s,a)\right]$$

这表明，在当前状态$s$下，选择动作$a$的Q值应该与当前奖励$r$和下一个状态$s^{\prime }$下，选择某个动作能够获得的最大的Q值$Q\left(s^{\prime },a^{\prime }\right)$ 有关，即$Q(s,a)$应尽可能和$r+\gamma {\text{max}}_{a^{\prime }\in \mathcal{A}}Q\left(s^{\prime },a^{\prime }\right)$ 接近。

接下来让我们参考动手学强化学习——DQN 算法，一步步的构建DQN算法。

import random
import numpy as np
import collections
from tqdm import tqdm
import torch
import torch.nn.functional as F
import matplotlib.pyplot as plt

首先让我们构建一个经验回放池，利用该经验回访池，我们将训练过程中的数据存储起来，以便更快速的更新网络。

class ReplayBuffer:
    ''' 经验回放池 '''
    def __init__(self, capacity):
        '''
        此处，我们使用collections.deque来存储经验，该数据结构可以高效的从两端添加和删除元素。
        '''
        self.buffer = collections.deque(maxlen=capacity)  # 队列,先进先出

    def add(self, state, action, reward, next_state, done):  # 将数据加入buffer
        '''
        我们将状态$s$，动作$a$，奖励$r$，下一个状态$s'$，游戏是否结束$d$存储到经验回放池中。
        将(state, action, reward, next_state, done)打包为元组，放入buffer中。
        '''
        self.buffer.append((state, action, reward, next_state, done))

    def sample(self, batch_size):  # 从buffer中采样数据,数量为batch_size
        transitions = random.sample(self.buffer, batch_size)  # 随机从buffer中采样数据
        state, action, reward, next_state, done = zip(*transitions)  # 将数据解包
        return np.array(state), action, reward, np.array(next_state), done

    def size(self):  # 目前buffer中数据的数量
        return len(self.buffer)

定义一个Q网络

class Qnet(torch.nn.Module):
    ''' 只有一层隐藏层的Q网络 '''
    def __init__(self, state_dim, hidden_dim, action_dim):
        super(Qnet, self).__init__()
        self.fc1 = torch.nn.Linear(state_dim, hidden_dim)
        self.fc2 = torch.nn.Linear(hidden_dim, action_dim)

    def forward(self, x):
        x = F.relu(self.fc1(x))  # 隐藏层使用ReLU激活函数
        return self.fc2(x)

接来下开始实现 DQN 算法

class DQN:
    ''' DQN算法 '''
    def __init__(self, state_dim, hidden_dim, action_dim, learning_rate, gamma,
                 epsilon, target_update, device):
        '''
        初始化：其中state_dim为状态维度，action_dim为动作维度，hidden_dim为隐藏层维度。
        learing_rate为学习率，gamma为折扣因子，epsilon为贪婪策略，target_update为目标网络更新频率。
        device决定在CPU或GPU上运行。
        gamma是折扣因子，它决定了未来奖励的权重。在强化学习中，未来的奖励通常被认为比当前的奖励更重要。
        因此，使用折扣因子可以对未来的奖励进行加权。可参考马尔科夫决策过程。
        epsilon是贪婪策略，它决定了在选择动作时，随机选择动作的概率。这是为了探索新动作，并避免过早收敛到局部最优解。
        '''
        self.action_dim = action_dim
        self.q_net = Qnet(state_dim, hidden_dim, self.action_dim).to(device)  # Q网络
        self.target_q_net = Qnet(state_dim, hidden_dim, self.action_dim).to(device)  # 目标网络
        self.optimizer = torch.optim.Adam(self.q_net.parameters(), lr=learning_rate) # 使用Adam优化器
        self.gamma = gamma  # 折扣因子
        self.epsilon = epsilon  # epsilon-贪婪策略
        self.target_update = target_update  # 目标网络更新频率
        self.count = 0  # 计数器,记录更新次数
        self.device = device

    def take_action(self, state):  # epsilon-贪婪策略采取动作
        if np.random.random() < self.epsilon:  # 当随机数小于epsilon时，这时候我们随机选择一个动作
            action = np.random.randint(self.action_dim)
        else:  # 否则，我们选择当前状态下价值最高的动作
            state = torch.tensor([state], dtype=torch.float).to(self.device)
            action = self.q_net(state).argmax().item()
        return action

    def update(self, transition_dict):
        '''
        更新网络参数，其中传入的transition_dict是一个字典，包含以下字段：
        states: 状态列表
        actions: 动作列表
        rewards: 奖励列表
        next_states: 下一个状态列表
        dones: 是否结束列表
        每个列表中元素都是一一对应的
        '''
        # 将transition_dict中的各量转换为符合神经网络输入的tensor格式
        states = torch.tensor(transition_dict['states'], dtype=torch.float).to(self.device)
        actions = torch.tensor(transition_dict['actions']).view(-1, 1).to(self.device)
        rewards = torch.tensor(transition_dict['rewards'], dtype=torch.float).view(-1, 1).to(self.device)
        next_states = torch.tensor(transition_dict['next_states'], dtype=torch.float).to(self.device)
        dones = torch.tensor(transition_dict['dones'],dtype=torch.float).view(-1, 1).to(self.device)
        
        # 此处我们将结合时序差分算法中Q-learning算法，来更新网络参数
        q_values = self.q_net(states).gather(1, actions)  # 获取对应动作action的Q值
        # 下个状态的最大Q值
        max_next_q_values = self.target_q_net(next_states).max(1)[0].view(-1, 1)
        q_targets = rewards + self.gamma * max_next_q_values * (1 - dones
                                                                )  # TD误差目标
        dqn_loss = torch.mean(F.mse_loss(q_values, q_targets))  # 均方误差损失函数
        self.optimizer.zero_grad()  # PyTorch中默认梯度会累积,这里需要显式将梯度置为0
        dqn_loss.backward()  # 反向传播更新参数
        self.optimizer.step()

        if self.count % self.target_update == 0:
            self.target_q_net.load_state_dict(
                self.q_net.state_dict())  # 更新目标网络
        self.count += 1

接下来让我们开启DQN算法的训练过程

# 初始化各项参数
lr = 2e-3  # 学习率
num_episodes = 500  # 总的训练步数
hidden_dim = 128  # 隐藏层维度
gamma = 0.98  # 折扣因子
epsilon = 0.01  # 贪婪策略的epsilon值
target_update = 10  # 目标网络更新频率
buffer_size = 10000  # 经验回放池的大小，越大越占用内存
minimal_size = 500  # 经验回放池中的最小经验数量，小于这个值不会进行训练
batch_size = 64  # 小批量随机梯度下降中的批量大小
device = torch.device("cuda") if torch.cuda.is_available() else torch.device("cpu")  # 训练设备

# 初始化一些算法部件
env = gym.make('CartPole-v1')  # 创建山地车游戏环境
replay_buffer = ReplayBuffer(buffer_size)  # 创建经验回放池
state_dim = env.observation_space.shape[0]  # 状态维度
action_dim = env.action_space.n  # 动作维度
agent = DQN(state_dim, hidden_dim, action_dim, lr, gamma, epsilon, target_update, device)  # 创建DQN算法智能体

# 开启训练过程
return_list = []  # 记录每一轮的奖励
for i in range(10):
    with tqdm(total=int(num_episodes / 10), desc='Iteration %d' % i) as pbar:
        for i_episode in range(int(num_episodes / 10)):
            episode_return = 0  # 每轮的奖励
            state, info = env.reset()  # 重置环境，返回初始状态
            done = False
            
            # 开启一轮游戏
            while not done:
                # 智能体进行动作，完成一个回合
                action = agent.take_action(state)  # 智能体进行动作
                next_state, reward, terminated, truncated, info = env.step(action)  # 执行动作
                done = terminated or truncated  # 判断游戏是否结束
                replay_buffer.add(state, action, reward, next_state, done)  # 将数据加入经验池
                state = next_state  # 更新状态
                episode_return += reward  # 更新这一轮的奖励
                
                # 当buffer数据的数量超过一定值后,才进行Q网络训练
                if replay_buffer.size() > minimal_size:
                    b_s, b_a, b_r, b_ns, b_d = replay_buffer.sample(batch_size)  # 随机从经验池中获取batch_size个数据
                    transition_dict = {
                        'states': b_s,
                        'actions': b_a,
                        'next_states': b_ns,
                        'rewards': b_r,
                        'dones': b_d
                    }  # 将数据转换为字典格式
                    # 更新智能体参数
                    agent.update(transition_dict)
            return_list.append(episode_return)  # 记录这一轮的奖励
            if (i_episode + 1) % 10 == 0:
                pbar.set_postfix({
                    'episode':
                    '%d' % (num_episodes / 10 * i + i_episode + 1),
                    'return':
                    '%.3f' % np.mean(return_list[-10:])
                })
            pbar.update(1)

Iteration 0:   0%|          | 0/50 [00:00<?, ?it/s]C:\Users\sswun\AppData\Local\Temp\ipykernel_22684\3177917159.py:27: UserWarning: Creating a tensor from a list of numpy.ndarrays is extremely slow. Please consider converting the list to a single numpy.ndarray with numpy.array() before converting to a tensor. (Triggered internally at ..\torch\csrc\utils\tensor_new.cpp:264.)
  state = torch.tensor([state], dtype=torch.float).to(self.device)
Iteration 0: 100%|██████████| 50/50 [00:03<00:00, 13.03it/s, episode=50, return=96.400]
Iteration 1: 100%|██████████| 50/50 [00:30<00:00,  1.63it/s, episode=100, return=223.100]
Iteration 2: 100%|██████████| 50/50 [01:03<00:00,  1.27s/it, episode=150, return=403.800]
Iteration 3: 100%|██████████| 50/50 [01:05<00:00,  1.30s/it, episode=200, return=232.000]
Iteration 4: 100%|██████████| 50/50 [00:44<00:00,  1.12it/s, episode=250, return=338.800]
Iteration 5: 100%|██████████| 50/50 [01:18<00:00,  1.58s/it, episode=300, return=346.500]
Iteration 6: 100%|██████████| 50/50 [01:27<00:00,  1.76s/it, episode=350, return=500.000]
Iteration 7: 100%|██████████| 50/50 [01:28<00:00,  1.76s/it, episode=400, return=360.000]
Iteration 8: 100%|██████████| 50/50 [01:18<00:00,  1.57s/it, episode=450, return=266.300]
Iteration 9: 100%|██████████| 50/50 [01:22<00:00,  1.65s/it, episode=500, return=473.800]

可以看到，对应车杆游戏，DQN算法具有一定的收敛性，一般可以达到较好的效果，但是算法并不是十分稳定。

step3:DQN算法改进——Dueling DQN

Dueling DQN算法是DQN算法的改进，它通过引入一个优势函数来改进Q网络的输出。优势函数的引入使得网络能够更好地学习到状态价值函数。在 Dueling DQN 中，Q 网络被建模为：
$$Q_{\eta ,\alpha ,\beta }(s,a)=V_{\eta ,\alpha }(s)+A_{\eta ,\beta }(s,a)$$

其中, $V_{\eta ,\alpha }(s)$ 表示状态价值函数，$A_{\eta ,\beta }(s,a)$ 表示优势函数。$\eta ,\alpha ,\beta$ 分别是网络的参数。其中$\eta$ 为状态价值函数和优势函数的共享参数，$\alpha$ 和 $\beta$ 分别表示状态价值函数和优势函数的各自独立部分参数。在这样的模型下，我们不再让神经网络直接输出Q值，而是训练神经网络的最后几层的两个分支，分别输出状态价值函数和优势函数，再求和得到Q值。

Dueling DQN 与 DQN 相比的差异只是在网络结构上，大部分代码依然可以继续沿用。我们定义状态价值函数和优势函数的复合神经网络VAnet。

class VAnet(torch.nn.Module):
    ''' 只有一层隐藏层的A网络和V网络 '''
    def __init__(self, state_dim, hidden_dim, action_dim):
        super(VAnet, self).__init__()
        self.fc1 = torch.nn.Linear(state_dim, hidden_dim)  # 共享网络部分
        self.fc_A = torch.nn.Linear(hidden_dim, action_dim)
        self.fc_V = torch.nn.Linear(hidden_dim, 1)

    def forward(self, x):
        A = self.fc_A(F.relu(self.fc1(x)))
        V = self.fc_V(F.relu(self.fc1(x)))
        Q = V + A - A.mean(1).view(-1, 1)  # Q值由V值和A值计算得到
        # 注意：此处我们不是单纯的让Q = V + A，而是让Q = V + A - A.mean(1).view(-1, 1)，
        # 这样做是因为Q值获得不唯一性问题，这样做可以使得Q值计算及网络更新更加稳定
        return Q

接下来让我们利用该网络再次进行训练

# 初始化一些算法部件
lr = 1e-3  # 学习率
env = gym.make('CartPole-v1')  # 创建山地车游戏环境
replay_buffer = ReplayBuffer(buffer_size)  # 创建经验回放池
state_dim = env.observation_space.shape[0]  # 状态维度
action_dim = env.action_space.n  # 动作维度
agent = DQN(state_dim, hidden_dim, action_dim, lr, gamma, epsilon, target_update, device)  # 创建DQN算法智能体
# 将智能体中网络结构更新为VAnet
agent.q_net = VAnet(state_dim, hidden_dim, agent.action_dim).to(device)  # Q网络
agent.target_q_net = VAnet(state_dim, hidden_dim, agent.action_dim).to(device)  # 目标网络
agent.optimizer = torch.optim.Adam(agent.q_net.parameters(), lr=lr) # 使用Adam优化器

# 开启训练过程
return_list = []  # 记录每一轮的奖励
for i in range(10):
    with tqdm(total=int(num_episodes / 10), desc='Iteration %d' % i) as pbar:
        for i_episode in range(int(num_episodes / 10)):
            episode_return = 0  # 每轮的奖励
            state, info = env.reset()  # 重置环境，返回初始状态
            done = False
            
            # 开启一轮游戏
            while not done:
                # 智能体进行动作，完成一个回合
                action = agent.take_action(state)  # 智能体进行动作
                next_state, reward, terminated, truncated, info = env.step(action)  # 执行动作
                done = terminated or truncated  # 判断游戏是否结束
                replay_buffer.add(state, action, reward, next_state, done)  # 将数据加入经验池
                state = next_state  # 更新状态
                episode_return += reward  # 更新这一轮的奖励
                
                # 当buffer数据的数量超过一定值后,才进行Q网络训练
                if replay_buffer.size() > minimal_size:
                    b_s, b_a, b_r, b_ns, b_d = replay_buffer.sample(batch_size)  # 随机从经验池中获取batch_size个数据
                    transition_dict = {
                        'states': b_s,
                        'actions': b_a,
                        'next_states': b_ns,
                        'rewards': b_r,
                        'dones': b_d
                    }  # 将数据转换为字典格式
                    # 更新智能体参数
                    agent.update(transition_dict)
            return_list.append(episode_return)  # 记录这一轮的奖励
            if (i_episode + 1) % 10 == 0:
                pbar.set_postfix({
                    'episode':
                    '%d' % (num_episodes / 10 * i + i_episode + 1),
                    'return':
                    '%.3f' % np.mean(return_list[-10:])
                })
            pbar.update(1)

Iteration 0: 100%|██████████| 50/50 [00:04<00:00, 11.09it/s, episode=50, return=36.400]
Iteration 1: 100%|██████████| 50/50 [00:30<00:00,  1.64it/s, episode=100, return=333.000]
Iteration 2: 100%|██████████| 50/50 [00:51<00:00,  1.03s/it, episode=150, return=413.100]
Iteration 3: 100%|██████████| 50/50 [01:01<00:00,  1.24s/it, episode=200, return=480.200]
Iteration 4: 100%|██████████| 50/50 [01:02<00:00,  1.25s/it, episode=250, return=500.000]
Iteration 5: 100%|██████████| 50/50 [01:03<00:00,  1.28s/it, episode=300, return=459.900]
Iteration 6: 100%|██████████| 50/50 [01:00<00:00,  1.21s/it, episode=350, return=434.400]
Iteration 7: 100%|██████████| 50/50 [00:52<00:00,  1.04s/it, episode=400, return=411.300]
Iteration 8: 100%|██████████| 50/50 [00:57<00:00,  1.15s/it, episode=450, return=433.000]
Iteration 9: 100%|██████████| 50/50 [00:59<00:00,  1.19s/it, episode=500, return=407.300]

对比DQN算法和Dueling DQN算法的训练结果，可以看到有时候Dueling DQN算法可能效果不如DQN，但是Dueling DQN算法的稳定性相对较好，在大部分情况下表现比DQN更好。

让我们看看智能体的表现吧！

import matplotlib.pyplot as plt
import gym
%matplotlib inline
from IPython import display

def show_state(env, step=0, info=""):
    plt.figure(3)
    plt.clf()
    plt.imshow(env.render())
    plt.title("Step: %d %s" % (step, info))
    plt.axis('off')
    display.clear_output(wait=True)
    display.display(plt.gcf())

env = gym.make('CartPole-v1', render_mode='rgb_array')
state, info = env.reset()
for _ in range(1000):
    action = agent.take_action(state)
    state, reward, terminated, truncated, info = env.step(action)
    done = truncated or terminated
    show_state(env, action, info)

    if done:
        state, info = env.reset()
env.close()

可以看到智能体表现已经很好了，但不难发现，智能体并未学习到一个最好的策略，读者可以结合游戏本身，对算法进行一些改进。