3. Dueling DQN架构实现
任务目标
在CartPole-v1环境中实现Dueling DQN算法。核心是将Q网络拆分为价值流（V(s)）与优势流（A(s,a)），通过聚合层输出Q值，并验证该架构相对传统DQN的性能提升。
环境与依赖
- Python 3.8+
- PyTorch 2.0+
- Gymnasium 0.29+（CartPole-v1）
核心实现
以下代码完整实现了Dueling DQN的训练与评估闭环。网络结构包含共享特征层、价值流与优势流；训练采用Double DQN更新与经验回放；评估使用滑动平均奖励作为性能指标。
import gymnasium as gym
import torch
import torch.nn as nn
import torch.optim as optim
import numpy as np
from collections import deque
import random
# ---------- Dueling 网络 ----------
class DuelingNet(nn.Module):
def __init__(self, state_dim, action_dim):
super().__init__()
self.feature = nn.Sequential(
nn.Linear(state_dim, 128), nn.ReLU(),
nn.Linear(128, 128), nn.ReLU()
)
self.value_stream = nn.Linear(128, 1)
self.advantage_stream = nn.Linear(128, action_dim)
def forward(self, x):
f = self.feature(x)
v = self.value_stream(f)
a = self.advantage_stream(f)
# Dueling 聚合：Q = V + A - mean(A)
return v + (a - a.mean(dim=1, keepdim=True))
# ---------- 经验回放 ----------
class ReplayBuffer:
def __init__(self, capacity=20000):
self.buffer = deque(maxlen=capacity)
def push(self, s, a, r, s2, d):
self.buffer.append((s, a, r, s2, d))
def sample(self, batch_size):
batch = random.sample(self.buffer, batch_size)
s, a, r, s2, d = map(np.stack, zip(*batch))
return (torch.FloatTensor(s), torch.LongTensor(a).unsqueeze(1),
torch.FloatTensor(r).unsqueeze(1), torch.FloatTensor(s2),
torch.FloatTensor(d).unsqueeze(1))
def __len__(self):
return len(self.buffer)
# ---------- 训练参数 ----------
env = gym.make("CartPole-v1")
state_dim = env.observation_space.shape[0]
action_dim = env.action_space.n
online_net = DuelingNet(state_dim, action_dim)
target_net = DuelingNet(state_dim, action_dim)
target_net.load_state_dict(online_net.state_dict())
optimizer = optim.Adam(online_net.parameters(), lr=1e-3)
buffer = ReplayBuffer()
gamma = 0.99
batch_size = 128
epsilon_start, epsilon_end, epsilon_decay = 1.0, 0.01, 0.995
epsilon = epsilon_start
update_target_every = 200
total_steps = 0
episode_rewards = []
# ---------- 训练循环 ----------
for episode in range(400):
state, _ = env.reset()
ep_reward = 0
done = False
while not done:
# epsilon-greedy 动作选择
if random.random() < epsilon:
action = env.action_space.sample()
else:
with torch.no_grad():
q_vals = online_net(torch.FloatTensor(state).unsqueeze(0))
action = q_vals.argmax().item()
next_state, reward, terminated, truncated, _ = env.step(action)
done = terminated or truncated
buffer.push(state, action, reward, next_state, done)
state = next_state
ep_reward += reward
total_steps += 1
# 网络更新
if len(buffer) > batch_size:
s, a, r, s2, d = buffer.sample(batch_size)
q_current = online_net(s).gather(1, a)
with torch.no_grad():
# Double DQN：用在线网络选动作，目标网络给值
a_next = online_net(s2).argmax(dim=1, keepdim=True)
q_target = r + gamma * (1 - d) * target_net(s2).gather(1, a_next)
loss = nn.MSELoss()(q_current, q_target)
optimizer.zero_grad()
loss.backward()
optimizer.step()
# 软更新目标网络
if total_steps % update_target_every == 0:
target_net.load_state_dict(online_net.state_dict())
epsilon = max(epsilon_end, epsilon * epsilon_decay)
episode_rewards.append(ep_reward)
# 每50回合打印一次滑动平均
if (episode + 1) % 50 == 0:
avg = np.mean(episode_rewards[-50:])
print(f"Episode {episode+1}, Avg Reward (last 50): {avg:.1f}, Epsilon: {epsilon:.3f}")
env.close()
运行与验证
- 运行方式：直接执行上述脚本。训练约400回合，在CPU上耗时约2～4分钟。
- 通过条件：CartPole-v1的奖励阈值是475（即平均475分可视为解决）。当最近50回合平均奖励达到475以上时，判定Dueling DQN在该任务上成功收敛。
- 性能对比验证：将上述网络替换为普通DQN（即去掉价值/优势流拆分，直接输出Q值），保持相同超参数训练。对比两者达到475分所需的回合数。Dueling DQN通常比普通DQN提前约20%～30%的回合达到该阈值。
关键说明
- Dueling架构的核心在于聚合公式
Q = V + A - mean(A)，该设计解决了优势流的可辨识性问题，使价值流能独立学习状态价值。 - 本实现采用Double DQN更新，减少过估计；目标网络每200步硬同步一次。
- 若训练不收敛，优先检查epsilon衰减速率与学习率；CartPole对超参数敏感度较低，上述默认值通常可直接工作。