File size: 4,691 Bytes
32fb2ef 73526b1 32fb2ef 73526b1 32fb2ef 73526b1 | 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 | #!/usr/bin/env python3
"""Generate DQN notebook."""
import nbformat as nbf
nb = nbf.v4.new_notebook()
nb.metadata = {"kernelspec": {"display_name": "Python 3", "language": "python", "name": "python3"},"language_info": {"name": "python", "version": "3.12.0"}}
cells = []
def md(s): cells.append(nbf.v4.new_markdown_cell(s))
def code(s): cells.append(nbf.v4.new_code_cell(s))
md("# DQN: Deep Q-Network\n\nReinforcement learning with experience replay on CartPole-v1.")
md("""## 背景
DQN（Mnih et al. 2015）将深度学习与 Q-Learning 结合，首次在 Atari 游戏上达到人类水平。
核心创新：
- **经验回放（Experience Replay）**：存储过去经验，随机采样训练，打破数据相关性
- **目标网络（Target Network）**：固定 Q-target，稳定训练
环境：**CartPole-v1** — 控制小车左右移动，保持杆子不倒。
状态：4 维（位置、速度、角度、角速度），动作：2 维（左、右）。
""")
md("""## 数学原理
### Q-Learning
$$Q(s, a) \\leftarrow Q(s, a) + \\alpha \\left(r + \\gamma \\max_{a'} Q(s', a') - Q(s, a)\\right)$$
### DQN Loss
$$\\mathcal{L} = \\mathbb{E}_{(s,a,r,s') \\sim \\mathcal{D}} \\left[\\left(r + \\gamma \\max_{a'} Q_{\\theta^-}(s', a') - Q_\\theta(s, a)\\right)^2\\right]$$
- $\\mathcal{D}$: 经验回放缓冲区
- $\\theta$: 在线网络参数
- $\\theta^-$: 目标网络参数（每隔 $C$ 步复制一次）
### ε-greedy 探索
$$a = \\begin{cases} \\text{random}, & \\text{概率 } \\varepsilon \\\\ \\arg\\max_a Q(s, a), & \\text{概率 } 1-\\varepsilon \\end{cases}$$
ε 随时间指数衰减。
""")
code("""\
import gymnasium as gym
import numpy as np
import torch
import torch.nn as nn
import torch.optim as optim
from rl.dqn.dqn import DQN, ReplayBuffer, train_episode, epsilon_by_episode
from utils.config import load_config
device = torch.device("mps" if torch.backends.mps.is_available() else "cpu")
print(f"Device: {device}")
env = gym.make("CartPole-v1")
state_dim = env.observation_space.shape[0]
action_dim = env.action_space.n
print(f"State: {state_dim} Action: {action_dim}")
""")
code("""\
model = DQN(state_dim, action_dim, hidden_dim=128).to(device)
target = DQN(state_dim, action_dim, hidden_dim=128).to(device)
target.load_state_dict(model.state_dict())
target.eval()
print(f"Parameters: {sum(p.numel() for p in model.parameters()):,}")
""")
md("""## 训练
> ⏱ 预估耗时：**500 episode × ~0.2s ≈ 2 分钟**（M4 Max）
""")
code("""\
NUM_EPISODES = 500
LR = 0.001
GAMMA = 0.99
BATCH_SIZE = 64
BUFFER_SIZE = 50000
TARGET_UPDATE = 100
EPSILON_START = 1.0
EPSILON_END = 0.01
EPSILON_DECAY = 500
optimizer = optim.Adam(model.parameters(), lr=LR)
buffer = ReplayBuffer(BUFFER_SIZE)
rewards = []
for episode in range(1, NUM_EPISODES + 1):
state, _ = env.reset()
episode_reward = 0
eps = epsilon_by_episode(episode, EPSILON_START, EPSILON_END, EPSILON_DECAY)
while True:
if np.random.random() < eps:
action = env.action_space.sample()
else:
with torch.no_grad():
q = model(torch.tensor(state, dtype=torch.float32, device=device).unsqueeze(0))
action = q.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
episode_reward += reward
_ = train_episode(model, target, optimizer, buffer, BATCH_SIZE, GAMMA)
if done:
break
if episode % TARGET_UPDATE == 0:
target.load_state_dict(model.state_dict())
rewards.append(episode_reward)
if episode % 50 == 0:
avg = np.mean(rewards[-50:])
print(f"Episode [{episode:3d}/{NUM_EPISODES}] Reward: {episode_reward:.0f} Avg(50): {avg:.1f} ε: {eps:.3f}")
env.close()
""")
md("""## Reward 曲线""")
code("""\
import matplotlib.pyplot as plt
plt.figure(figsize=(8, 4))
plt.plot(rewards)
plt.xlabel("Episode"); plt.ylabel("Total Reward"); plt.title("DQN Training on CartPole")
plt.grid(True)
plt.axhline(y=195, color='r', linestyle='--', label='Solved (195)')
plt.legend()
plt.show()
""")
md("""\
## 思考题
1. 为什么需要经验回放（Experience Replay）？在线学习会有什么问题？
2. 目标网络（Target Network）解决了什么？如果不固定 target，Q 值会发散吗？
3. ϵ-greedy 中的 ϵ 从 1.0 开始衰减有什么含义？
4. 如果把 `hidden_dim` 从 128 改到 32，训练速度会怎样？收敛难度呢？
""")
nb.cells = cells
with open("rl/dqn/dqn.ipynb", "w") as f:
nbf.write(nb, f)
print("Generated rl/dqn/dqn.ipynb")
|