Python DQN (Deep Q-Network) 完整实现指南
核心概念回顾
DQN 是结合深度神经网络与Q-Learning的强化学习算法,关键创新点：
- 经验回放 (Experience Replay)：打破数据相关性
- 目标网络 (Target Network)：稳定训练目标
- ε-greedy 策略：探索与利用平衡
环境准备
pip install gymnasium numpy torch matplotlib
注意：Gym 已升级为
gymnasium，以下代码基于 gymnasium。
完整代码实现
1 构建 DQN 智能体
import numpy as np
import torch
import torch.nn as nn
import torch.optim as optim
import random
from collections import deque
import gymnasium as gym
class DQN(nn.Module):
"""深度Q网络"""
def __init__(self, state_dim, action_dim, hidden_dim=128):
super(DQN, self).__init__()
self.network = nn.Sequential(
nn.Linear(state_dim, hidden_dim),
nn.ReLU(),
nn.Linear(hidden_dim, hidden_dim),
nn.ReLU(),
nn.Linear(hidden_dim, action_dim)
)
def forward(self, x):
return self.network(x)
class DQNAgent:
"""DQN 智能体"""
def __init__(
self,
state_dim: int,
action_dim: int,
lr: float = 1e-3,
gamma: float = 0.99,
epsilon_start: float = 1.0,
epsilon_end: float = 0.01,
epsilon_decay: float = 0.995,
batch_size: int = 64,
memory_size: int = 10000,
target_update_freq: int = 100,
device: str = "cpu"
):
self.state_dim = state_dim
self.action_dim = action_dim
self.gamma = gamma
self.batch_size = batch_size
self.target_update_freq = target_update_freq
self.device = torch.device(device)
# ε-greedy 参数
self.epsilon = epsilon_start
self.epsilon_end = epsilon_end
self.epsilon_decay = epsilon_decay
# 网络
self.q_network = DQN(state_dim, action_dim).to(self.device)
self.target_network = DQN(state_dim, action_dim).to(self.device)
self.target_network.load_state_dict(self.q_network.state_dict())
self.target_network.eval()
# 优化器 & 损失函数
self.optimizer = optim.Adam(self.q_network.parameters(), lr=lr)
self.loss_fn = nn.MSELoss()
# 经验回放缓冲区
self.memory = deque(maxlen=memory_size)
def select_action(self, state: np.ndarray) -> int:
"""ε-greedy 动作选择"""
if random.random() < self.epsilon:
return random.randint(0, self.action_dim - 1)
else:
state_tensor = torch.FloatTensor(state).unsqueeze(0).to(self.device)
with torch.no_grad():
q_values = self.q_network(state_tensor)
return q_values.argmax(dim=1).item()
def store_transition(self, state, action, reward, next_state, done):
"""存储经验"""
self.memory.append((state, action, reward, next_state, done))
def train_step(self):
"""单步训练"""
if len(self.memory) < self.batch_size:
return None
# 随机采样 batch
batch = random.sample(self.memory, self.batch_size)
states, actions, rewards, next_states, dones = zip(batch)
states = torch.FloatTensor(np.array(states)).to(self.device)
actions = torch.LongTensor(actions).unsqueeze(1).to(self.device)
rewards = torch.FloatTensor(rewards).unsqueeze(1).to(self.device)
next_states = torch.FloatTensor(np.array(next_states)).to(self.device)
dones = torch.FloatTensor(dones).unsqueeze(1).to(self.device)
# 当前 Q 值: Q(s, a)
current_q = self.q_network(states).gather(1, actions)
# 目标 Q 值: r + γ max_a' Q_target(s', a')
with torch.no_grad():
next_q = self.target_network(next_states).max(dim=1, keepdim=True)[0]
target_q = rewards + self.gamma next_q (1 - dones)
# 计算损失并反向传播
loss = self.loss_fn(current_q, target_q)
self.optimizer.zero_grad()
loss.backward()
self.optimizer.step()
# 衰减 ε
self.epsilon = max(self.epsilon_end, self.epsilon self.epsilon_decay)
return loss.item()
def update_target_network(self):
"""更新目标网络"""
self.target_network.load_state_dict(self.q_network.state_dict())
def save(self, path: str):
torch.save(self.q_network.state_dict(), path)
def load(self, path: str):
self.q_network.load_state_dict(torch.load(path, map_location=self.device))
2 训练循环
def train_dqn(
env_name: str = "CartPole-v1",
max_episodes: int = 500,
max_steps_per_episode: int = 1000,
render: bool = False,
save_path: str = "dqn_cartpole.pth"
):
"""训练 DQN"""
# 创建环境
env = gym.make(env_name)
state_dim = env.observation_space.shape[0]
action_dim = env.action_space.n
print(f"Environment: {env_name}")
print(f"State dim: {state_dim}, Action dim: {action_dim}")
# 创建智能体
agent = DQNAgent(
state_dim=state_dim,
action_dim=action_dim,
lr=1e-3,
gamma=0.99,
epsilon_start=1.0,
epsilon_end=0.01,
epsilon_decay=0.995,
batch_size=64,
memory_size=10000,
target_update_freq=100
)
# 记录
episode_rewards = []
losses = []
for episode in range(1, max_episodes + 1):
state, _ = env.reset()
total_reward = 0
episode_loss = []
for step in range(max_steps_per_episode):
if render:
env.render()
action = agent.select_action(state)
next_state, reward, terminated, truncated, _ = env.step(action)
done = ter
minated or truncated
agent.store_transition(state, action, reward, next_state, done)
loss = agent.train_step()
if loss is not None:
episode_loss.append(loss)
total_reward += reward
state = next_state
if done:
break
# 定期更新目标网络
if episode % agent.target_update_freq == 0:
agent.update_target_network()
avg_loss = np.mean(episode_loss) if episode_loss else 0
episode_rewards.append(total_reward)
losses.append(avg_loss)
if episode % 10 == 0:
avg_reward = np.mean(episode_rewards[-10:])
print(
f"Episode {episode:4d} | "
f"Reward: {total_reward:6.1f} | "
f"Avg(10): {avg_reward:6.1f} | "
f"Loss: {avg_loss:.4f} | "
f"ε: {agent.epsilon:.4f}"
)
# 保存模型
agent.save(save_path)
print(f"n模型已保存至: {save_path}")
return episode_rewards, losses
# 启动训练
if __name__ == "__main__":
rewards, losses = train_dqn(
env_name="CartPole-v1",
max_episodes=500,
render=False
)
3 可视化训练曲线
import matplotlib.pyplot as plt
def plot_training(rewards, losses):
"""绘制训练曲线"""
fig, axes = plt.subplots(1, 2, figsize=(14, 5))
# 奖励曲线（滑动平均）
window = 10
smooth_rewards = np.convolve(rewards, np.ones(window)/window, mode='valid')
axes[0].plot(smooth_rewards, linewidth=2)
axes[0].set_title("Smoothed Episode Reward")
axes[0].set_xlabel("Episode")
axes[0].set_ylabel("Reward")
axes[0].grid(True)
# 损失曲线
axes[1].plot(losses, linewidth=2, color='red')
axes[1].set_title("Training Loss")
axes[1].set_xlabel("Episode")
axes[1].set_ylabel("Loss")
axes[1].grid(True)
plt.tight_layout()
plt.savefig("dqn_training.png", dpi=150)
plt.show()
# plot_training(rewards, losses)
4 测试/评估
def evaluate_agent(agent: DQNAgent, env_name: str = "CartPole-v1", n_episodes: int = 10, render: bool = False):
"""评估智能体"""
env = gym.make(env_name)
rewards = []
for ep in range(n_episodes):
state, _ = env.reset()
total_reward = 0
done = False
while not done:
if render:
env.render()
action = agent.select_action(state) # ε≈0，几乎总是贪心
next_state, reward, terminated, truncated, _ = env.step(action)
done = terminated or truncated
total_reward += reward
state = next_state
rewards.append(total_reward)
avg_reward = np.mean(rewards)
print(f"评估结果 ({n_episodes} episodes): 平均奖励 = {avg_reward:.2f}")
return rewards
# 使用示例
if __name__ == "__main__":
# 加载已训练模型
agent = DQNAgent(state_dim=4, action_dim=2)
agent.load("dqn_cartpole.pth")
agent.epsilon = 0.0 # 纯贪心策略评估
evaluate_agent(agent, render=True, n_episodes=5)
关键超参数说明
| 参数 | 推荐值 | 说明 |
|---|---|---|
lr |
1e-3 ~ 1e-4 | 学习率 |
gamma |
99 | 折扣因子 |
epsilon_start |
0 | 初始探索率 |
epsilon_end |
01~0.05 | 最小探索率 |
epsilon_decay |
995~0.999 | 探索率衰减 |
batch_size |
32~128 | 批量大小 |
memory_size |
10000~100000 | 回放缓冲区大小 |
target_update_freq |
100~1000 | 目标网络更新频率 |
进阶改进方向
┌─────────────────────────────────────────────┐
│ DQN 改进版本 │
├─────────────────────────────────────────────┤
│ 1. Double DQN → 解决 Q 值过估计 │
│ 2. Dueling DQN → 分离状态价值与优势函数 │
│ 3. Prioritized Replay → 优先采样重要经验 │
│ 4. Noisy Networks → 参数化噪声替代 ε-greedy│
│ 5. Rainbow DQN → 上述所有改进的集合 │
└─────────────────────────────────────────────┘Double DQN 修改示例（替换 train_step 中的目标计算）：
# 当前网络选择最佳动作 next_actions = self.q_network(next_states).max(dim=1, keepdim=True)[1] # 目标网络评估该动作的 Q 值 next_q = self.target_network(next_states).gather(1, next_actions) target_q = rewards + self.gamma next_q (1 - dones)
常见问题排查
| 问题 | 可能原因 | 解决方案 |
|---|---|---|
| 奖励不收敛 | ε 衰减太快/太慢 | 调整 epsilon_decay |
| 损失震荡 | 学习率过大 | 降低 lr，使用 Adam 默认 |
| 训练不稳定 | 目标网络更新太频繁 | 增大 target_update_freq |
| 无法解决复杂环境 | 网络容量不足 | 增加隐藏层/神经元，或改用 DDPG/PPO |
如需针对特定环境（如 Atari、LunarLander）或实现 Double/Dueling DQN,可以告诉我具体需求！
首发原创文章，作者：王坚，如若转载，请注明出处：https://test.idctop.com/article/485072.html
