强化学习 7. 深度Q网络进阶技巧（实验部分）

发表于 更新于

前言

从零开始学习ai文章系列计划是个人在《动手学深度学习》和《磨菇书》两本书的学习中的个人笔记，文章也会以课本中的章节分开，即每个章节一片笔记。我会尽量的把主要内容以及遇到的难点进行记录与解决，如果哪里有错误的欢迎指正。或者不清晰的可以直接查看原文部分。

《蘑菇书》原文（课本）：https://datawhalechina.github.io/easy-rl/#/

（由于有时候公式太多，可能会直接贴图片）

蘑菇书的文章结构不会跟之前《动手学深度学习》按照原文章节进行，个人会适当调节。

0. 介绍

首先，我们的原式DQN采用上一章节实验的网络。如果不清楚的可以自行去上一章节查看。

我们同样是作用在CartPole-v1中，仅仅只是对DQN网络进行相对应的修改。

原DQN：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
import gym
import torch
import torch.nn as nn
import torch.optim as optim
import random
import numpy as np
if not hasattr(np, "bool8"):
    np.bool8 = np.bool_
from collections import deque



device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

# ====== Q 网络 ======
class QNetwork(nn.Module):
    def __init__(self, state_dim, action_dim):
        super().__init__()
        self.net = nn.Sequential(
            nn.Linear(state_dim, 128),
            nn.ReLU(),
            nn.Linear(128, action_dim)
        )
        self.optimizer = optim.Adam(self.parameters(), lr=1e-3)
        # self.optimizer = optim.Adam(self.parameters(), lr=1e-3,weight_decay=1e-5)
        self.GAMMA = 0.99
        self.grad_norm_ema = None
        self.grad_ema_beta = 0.99

    def forward(self, x):
        return self.net(x)

    def update(self, replay_buffer, target_q_net, BATCH_SIZE=64):

        # 1. 经验池中拿需要的数据
        s, a, r, s_next, done_mask = replay_buffer.sample(BATCH_SIZE)
        s = s.to(device)
        a = a.to(device)
        r = r.to(device)
        s_next = s_next.to(device)
        done_mask = done_mask.to(device)

        # 2.计算Q(s,a)
        # self.forward(s)的结果是每行一组动作的概率[a1,a2,a3].
        # gather(1,0) 第一个参数是维度，1指的是列维度。第二个参数是索引。作用是拿出列方向索引为0的那个数。
        # q_values = [Q(s_0, a_0),Q(s_1, a_1),...,Q(s_batch, a_batch)]
        q_values = self.forward(s).gather(1, a.reshape(-1, 1)).reshape(-1)

        # 3. 计算目标值y
        # 我们不需要更新target Q网络，因此关掉他的梯度计算。
        # y = r_i + max_a Q^(s_{i+1}, a)
        with torch.no_grad():
            next_q_values = target_q_net(s_next).max(1)[0]
            target = r + self.GAMMA * next_q_values * (1 - done_mask)

        # 4.计算Q(s,a)与y的loss
        loss = nn.MSELoss()(q_values, target)

        # 5. 计算梯度，并裁剪梯度
        self.optimizer.zero_grad()
        loss.backward()

        # torch.nn.utils.clip_grad_norm_(self.parameters(), 10) # 整个网络梯度的 L2 范数 ≤ 10
        # 裁剪梯度 （写成函数是为了IDE中能直接折叠）
        def clip():
            # 自适应裁剪梯度
            total_norm = torch.nn.utils.clip_grad_norm_(self.parameters(), float('inf')).item()

            if self.grad_norm_ema is None:
                self.grad_norm_ema = total_norm
            else:
                # 指数加权移动平均
                self.grad_norm_ema = self.grad_ema_beta * self.grad_norm_ema + (1 - self.grad_ema_beta) * total_norm

            # 前几步 EMA 还没收敛，adaptive_max_norm 会偏小，可以加一个保底值1
            adaptive_max_norm = max(2.0 * self.grad_norm_ema, 1.0)  # 留出2倍余量
            torch.nn.utils.clip_grad_norm_(self.parameters(), adaptive_max_norm)

        clip()
        # --------------

        # 6.更新网络
        self.optimizer.step()

# ====== 经验回放缓冲区 ======
class ReplayBuffer:
    def __init__(self, capacity):
        self.buffer = deque(maxlen=capacity)

    def push(self, s, a, r, s_next, done):
        self.buffer.append((s, a, r, s_next, done))

    def sample(self, batch_size):
        batch = random.sample(self.buffer, batch_size)
        s, a, r, s_next, done = zip(*batch)
        return (
            torch.tensor(np.array(s)),  # 警告说非常慢，让我加上np。为了不一直被警告就加上了。
            torch.tensor(a),
            torch.tensor(r),
            torch.tensor(np.array(s_next)),
            torch.tensor(done, dtype=torch.int64), # done是bool值，我们改成数字在后续(1 - done_mask)中使用
        )

    def __len__(self):
        return len(self.buffer)

# ====== ε-greedy 策略 ======
def select_action(state, step, env, q_net):
    EPSILON_START = 1.0
    EPSILON_END = 0.01
    EPSILON_DECAY = 5000

    # 指数衰减的 ε-greedy 策略
    epsilon = EPSILON_END + (EPSILON_START - EPSILON_END) * np.exp(-1.0 * step / EPSILON_DECAY)

    if random.random() < epsilon:
        # 随意返回0或1（左右）
        return env.action_space.sample()
    else:
        # unsqueeze(0)是最前面增加一个维度。理解成展开即可。比如x.shape=(4)，经过unsqueeze(0)后，x.shape=(1,4).
        # 增加state维度主要是为了增加batch维度，然后放入网络中计算q_values = q_net(state)
        state = torch.tensor(state, dtype=torch.float32).unsqueeze(0).to(device)
        with torch.no_grad():
            q_values = q_net(state)

        # 获取最大的Q值。max(dim=1)指在每行上找最大值。你可以理解为dim=1指的列方向，那么列上面有个指针，可以随意动，根据max指令，指针去寻找最大的值。返回的结果是(values, indices)。max(dim=1)[0]指的是我们只要最大值。
        max_q = q_values.max(dim=1)[0]
        # (q_values == max_q)的结果是布尔值列表，相同为True,不同为False，比如有3个动作，后两个都是最大Q值，那么结果为[False, True, True]。
        # .nonzero(as_tuple=True) 作用是返回这些位置的索引，上面的例子来说，返回([0,1],[0,2])。其中第一个数是batch维度索引号，第二个是action维度号（你输入1x4的矩阵，那就有横竖2个维度）。通过.nonzero(as_tuple=True)[1]，我们仅获取action维度的索引号。因此以之前结果为例，我们的输出为[1,2]
        actions = (q_values == max_q).nonzero(as_tuple=True)[1]
        # actions中是最大Q值的列索引号。torch.randint(len(actions), (1,)) 中(1,)指的是生成结果维度，len(actions)则是我们生成数字的范围。准确来说，根据输入的矩阵大小，生成每个[0,x)数字，x就是我们的len(actions)。
        # 这里作用就是随即从最大Q值的动作中选一个。
        action = actions[torch.randint(len(actions), (1,))].item()

        return action




def collect_and_store_data(state, global_step, env, q_net,episode_reward,replay_buffer):

    # 1. 根据 Q 网络 + ε-greedy 选动作
    action = select_action(state, global_step, env, q_net)

    # 2. 与环境交互
    next_state, reward, terminated, truncated, _ = env.step(action)
    done = terminated or truncated
    episode_reward[0] += reward

    # 3. 存入回放缓冲区
    replay_buffer.push(state, action, reward, next_state, done)
    # state = next_state
    return next_state ,done





def train():
    # ====== 初始化环境和网络 ======
    # 后面用到的参数
    targetQ_update_steps = 100  # Q网络每更新100次，目标Q网络就重新加载一次参数
    num_episodes = 500
    Qupdate_batch_size = 64  # Q网络每次更新都从经验池中拿64条数据
    global_step = 0

    # 创建环境
    env = gym.make("CartPole-v1")
    state_dim = env.observation_space.shape[0]
    action_dim = env.action_space.n

    # 创建网络和目标网络
    q_net = QNetwork(state_dim, action_dim).to(device)
    target_q_net = QNetwork(state_dim, action_dim).to(device)
    target_q_net.load_state_dict(q_net.state_dict())

    # 创建经验池
    replay_buffer = ReplayBuffer(10000)

    # 打印用的奖励值
    episode_reward = [0]

    for episode in range(num_episodes):
        state, _ = env.reset()
        # episode_reward = [0]

        while True:
            global_step += 1

            # 1. 收集 (s,a,r,s') 数据到缓存区中
            state ,done = collect_and_store_data(state, global_step, env, q_net, episode_reward, replay_buffer)

            # 2. 从经验回放中采样并训练
            if len(replay_buffer) >= Qupdate_batch_size:
                q_net.update(replay_buffer,target_q_net,Qupdate_batch_size)

            # 3. 每 C 步同步目标网络
            if global_step % targetQ_update_steps == 0:
                target_q_net.load_state_dict(q_net.state_dict())

            if done:
                # print(f"Episode {episode}, Reward: {episode_reward[0]}")
                break

        if (episode+1)%50==0:
            print(f"Episode {episode+1}, Reward: {episode_reward[0]/50}")
            episode_reward = [0]
            torch.save(q_net.state_dict(), "DQN_policy_cartpole.pth")
            print("Policy saved to DQN_policy_cartpole.pth")

    env.close()


    return q_net


def load_policy(model_path):
    state_dim = 4   # CartPole 固定
    action_dim = 2

    policy = QNetwork(state_dim, action_dim)
    # policy = PolicyNet(state_dim, action_dim)
    policy.load_state_dict(torch.load(model_path))
    policy.to(device)
    policy.eval()   # 重要：进入评估模式

    return policy


def render_policy(policy, episodes=10):
    env = gym.make("CartPole-v1", render_mode="human")

    for ep in range(episodes):
        state, _ = env.reset()
        done = False
        total_reward = 0

        while not done:
            state_tensor = torch.as_tensor(state, dtype=torch.float32, device=device)

            with torch.no_grad():
                probs = policy(state_tensor)
                action = torch.argmax(probs).item()

            state, reward, terminated, truncated, _ = env.step(action)
            done = terminated or truncated
            total_reward += reward

        print(f"Episode {ep}, Total Reward: {total_reward}")

    env.close()


if __name__ == "__main__":
    # ① 训练并保存
    train()

    # ② 加载并展示
    policy = load_policy("DQN_policy_cartpole.pth")
    render_policy(policy)

注意：每一种网络都是在单独的DQN代码上修改，并不是重复在同一个DQN代码中。

1. 双深度Q网络（double DQN，DDQN）

作用：避免Q 值过高估计（overestimation）

原始 DQN 的目标公式： \[ y =r+\gamma\max_{a'}Q_{target}(s',a') \] 在DDQN中，我们变成： \[ y =r+\gamma\max_{a'}Q_{target}(s',\arg \max_{a'}Q(s',a')) \] 这样选动作的Q函数与计算值的Q函数 不是同一个。可以减少 Q 值的过高估计。

代码中仅修改一行：

查看执行结果：

2.竞争深度Q网络（dueling DQN）

作用：让模型学“状态价值”和“动作优势”，提高效率。

有些状态根本不需要区分动作，却被迫学习，比如CartPole 中：杆子已经倒了 → 左右都一样烂

原网络是输出Q(s,a)，而竞争DQN输出的是 A(s,a) 和 V(s)：

这里我们变化的是原来的网络结构。

同时，我们的前向传播计算也要进行对应的修改。Dueling DQN 的定义（带零均值）是：

由于我们这里使用了 A.mean(dim=1) ，因此我们输入的维度必须也是符合A的形状的：A.shape=(batch,action_dim)。

因此测试的时候，我们的state必须要添加前面一维batch维

查看修改后的效果

训练好了后，测试代码报错了，因此图片没连在一起。

3. 优先级经验回放（prioritized experience replay，PER）

根据 TD error 的绝对值来给每条经验池中的数据赋予优先级，误差越大的优先级越高，我们越优先使用。

作用：避免重要经验学不到（采样不均）

3.1 初始化

首先初始化添加新的等级队列。以及后续需要的参数：

3.2 push 添加新数据

经验池的push函数修改如下。这里 新样本默认是“最重要的”，因此新样本的等级是直接等于队列中最高那个。

3.3 sample 取出数据

如下图所示。

首先计算概率，我们利用如下公式将等级转换成我们的概率 \[ P(i)=\frac{p^\alpha_i}{\sum_jp^\alpha_j} \] 然后按照我们计算出的概率，对 self.buffer 进行采样。这里我们。

其中 indices = np.random.choice(len(self.buffer), batch_size, p=probs) 指的是：从 [0, 1, 2, ..., len(self.buffer)-1] 这些索引中，按照概率 probs，随机采样 batch_size 个索引。

最后我们计算纠正偏差用的重要性采样权重。公式如下 \[ w_i=[(\frac{1}{N*P(i)})^\beta]/w_{max} \] β 初期比较小是为了稳定训练，随着训练慢慢变成1，主要是为了后期的无偏估计收敛。

然后就是训练中我们利用权重修改loss

我们算出来的权重是针对每条数据的，因此无法单独乘在最外面。因此针对每条数据，loss其实是td_error.pow(2)，使用对应的权重w进行矫正后，将所有结果再进行平均得到最终loss

3.4 更新等级

我们在上一张图中的loss计算完后就需要执行他的等级更新函数。

我们用以下公式，根据TD误差重新分配等级 \[ p_i=|\delta_i|+\epsilon \]

3.5 查看效果

首先是我们固定了 β=0.4 的结果（一开始没加，后来更新代码才加上β趋于1的）

接下来就是 β趋于1的代码

4. n-step DQN

作用：解决奖励传播太慢的问题

我们这里的修改很简单，从原来的一步时序差分，改成多步时序差分。我这里选的是3步。

首先增加 n-step buffer

然后是修改 push 函数。只有存够了 n 条数据，才会开始计算累计奖励。（上面初始化的时候队列大小为3，因此下面append多出来的时候，旧的就会挤出去）

最后修改target的计算。（因为是3步，因此根据公式， self.GAMMA 的指数为3）

查看结果

image-20260409154358528

5. 噪声网络（noisy net）

作用：更“聪明”的探索

我们不再使用 ε-贪心 算法来随机采样动作。而是往网络中添加噪音来达到随机采样动作的目的。

5.1 噪音附加功能函数

首先添加新函数，给输入的网络层添加噪音：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
def add_noise_to_linear(self, layer, sigma_init=0.5):
    in_f = layer.in_features
    out_f = layer.out_features

    # σ 参数（可学习）
    layer.weight_sigma = nn.Parameter(
        torch.full((out_f, in_f), sigma_init / math.sqrt(in_f))
    )
    layer.bias_sigma = nn.Parameter(
        torch.full((out_f,), sigma_init / math.sqrt(out_f))
    )

    # ε 噪声（buffer）
    layer.register_buffer("weight_epsilon", torch.zeros(out_f, in_f))
    layer.register_buffer("bias_epsilon", torch.zeros(out_f))

    return layer

我们给输入的网络层添加了 layer.weight_sigma 、 layer.bias_sigma 和 layer.weight_epsilon、layer.bias_epsilon 共4个参数。其中前两个是决定噪音强度/噪音大小的参数，而后面两个则是与网络层参数同样大小的噪音Tensor，这里初始化为0只是为了占位，在计算前会修改值。

torch.full((out_f, in_f), sigma_init / math.sqrt(in_f)) 的意思是制作一个大小为 (out_f, in_f) 的tensor，其中每个参数值为：sigma_init / math.sqrt(in_f)

  1. 为什么值的内容长这样呢？

我们的目的是让噪声强度随着输入规模缩小，从而保证训练的稳定性，因此我们让 sigma_init 除以 math.sqrt(in_f) 。

因为随着输入的变多，每个输入叠加在一起的噪声也越多，导致最后的输出数值非常不稳定。所以我们对此进行限制。

  1. 为什么tensor的形状是 (out_f, in_f) 而不是(in_f, out_f) 呢？

因为 PyTorch 的计算式这样的： \[ y=x*w^T+b \] 我们可以查看自定义的网络验证这点

  1. 为什么要通过 layer.register_buffer 添加参数，而不能像前面 layer.weight_sigma = 这样直接赋值吗？

nn.Parameter 是 PyTorch 自动识别并注册的模型参数，会自动加入参数列表、参与训练、跟随 device，并被保存。 普通 Tensor 直接赋值不会被 PyTorch 管理：不会跟随 device、不会被保存、也不会出现在模型结构中。 因此，对于“不需要训练但需要参与计算和保存”的变量（如 ε 噪声），必须使用 register_buffer 手动注册，使其能够：

  • 跟随 .to(device)
  • 被 state_dict() 保存
  • 在模型中被正确管理（但不参与梯度更新）
我们这里注意到，强度参数layer.weight_sigma 、 layer.bias_sigma 都是可学习的参数。前期由于TD error很大，根据以下公式 $$ \[\begin{aligned} Q&=f(u+\sigma·\epsilon)\\ L&=(Q-target)^2\\ \end{aligned}\]

$$ Loss对原网络中的参数求导后，原网络参数进行相应调整，朝着梯度为0的方向逼近，即让loss变小的方向接近。

那么对于噪音参数 \(\sigma\) 来说，要想Loss减小，则 \(\sigma\) 的值减小，因为它意味着噪音的强度，噪音越强，loss越大。

训练初期，TD error 大，梯度幅度大，σ 更新幅度大，探索强。随着训练进行，TD error 变小，梯度减小，σ 自然收敛，探索减弱，策略逐渐稳定。

σ 的动态变化是 梯度驱动的自适应探索

然后，我们在初始化中，对网络层添加噪音功能：

5.2 噪音前向传播计算

1
2
3
4
5
6
7
8
def noisy_forward(self,layer, x):
    if layer.training:
        weight = layer.weight + layer.weight_sigma * layer.weight_epsilon
        bias = layer.bias + layer.bias_sigma * layer.bias_epsilon
    else:
        weight = layer.weight
        bias = layer.bias
    return F.linear(x, weight, bias)

我们在5.1中对目标层添加了4个新参数后，我们前向传播就能通过这几个变量，给参数加上噪音。

假设 fc1 = nn.Linear(3, 4) , 那么在计算 fc1(x) 的时候实际等价于 F.linear(x, fc1.weight, fc1.bias)。

因此这里我们修改完 weight 和 bias 之后，就直接利用这两者进行计算 F.linear(x, weight, bias)。在反向传播的时候，因为 weight = layer.weight + layer.weight_sigma * layer.weight_epsilon 与原来层的 layer.weight ，所以是能够获得原网络参数的梯度的。

5.3 前向传播

5.4 重置噪音

对存在噪音参数的层，重置他们的噪音值。

5.5 DQN流程

首先我们去除原来的 ε-贪婪算法。直接选取网络最优动作

然后是训练流程中

课本中我们说是每个episode都需要固定噪声，然后在新一轮episde的时候再更新。这是正确的，但是我们DQN里面是off-policy的，因此我们可以每一步都更新噪声，毕竟我们是经验池中获取旧的数据进行训练，本来也不是同一轮episode的数据。如果是同策略的话，必须遵守每个episode都需要固定噪声这点。

5.6 查看效果

中间层噪声可以理解为特征层探索，类似：探索“策略内部特征”

而全部层加噪音则是有着极强的探索效果，虽然这样意味着不太稳定。

  1. 全部层添加噪音，每个episode固定噪音
  2. 中间层添加噪音，每个episode固定噪音
  3. 中间层添加噪音，每步变噪音
  4. 全部层添加噪音，每步变噪音

我们能看出，2种配置的结果比较符合我们的要求

  1. 每个episode固定噪音，仅中间层添加噪音
  2. 每步都重置噪音，全层添加噪音。

6. 分布式Q函数（distributional Q-function）

课本只是浅显的介绍了输出层从标量变成了分布：

原来DQN： Q(s)-> [action_dim]

分布式DQN： Q(s)->[action_dim, N_atoms]

但实际实现上有许多其他困难的知识点，难以理解实现。暂且没空继续弄（感觉时效比低）。但是根据课本中说的，分布式DQN的提升是我们这几种网络修改里面最大的，因此放这里占个位，以后有需要的时候再继续补充。