前言
从零开始学习ai文章系列计划是个人在《动手学深度学习》和《磨菇书》两本书的学习中的个人笔记，文章也会以课本中的章节分开，即每个章节一片笔记。我会尽量的把主要内容以及遇到的难点进行记录与解决，如果哪里有错误的欢迎指正。或者不清晰的可以直接查看原文部分。
《蘑菇书》原文（课本）：https://datawhalechina.github.io/easy-rl/#/
（由于有时候公式太多，可能会直接贴图片）
蘑菇书的文章结构不会跟之前《动手学深度学习》按照原文章节进行，个人会适当调节。
1. 深度Q网络（DQN）
1.1 环境介绍
我们和之前策略梯度一样，使用的环境是CartPole-v1。也同样是为了方便区分与之前有什么不同。
状态空间：[ 小车位置，小车速度，杆子的角度，杆子的角速度
]，4个连续值
动作空间： [左，右]，2个离散值
接下来我们创建环境并设置需要的超参数：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
|
targetQ_update_steps = 100
num_episodes = 500
Qupdate_batch_size=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]
|
接下来是我们的策略网络。
能看到其实跟之前一样，也是隐藏层为128个神经元的3层感知机。
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
| 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.GAMMA = 0.99
def forward(self, x):
return self.net(x)
def update(self,replay_buffer,target_q_net, BATCH_SIZE = 64):
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)
q_values = self.forward(s).gather(1, a.reshape(-1, 1)).reshape(-1)
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)
loss = nn.MSELoss()(q_values, target)
self.optimizer.zero_grad()
loss.backward()
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
adaptive_max_norm = max(2.0 * self.grad_norm_ema, 1.0)
torch.nn.utils.clip_grad_norm_(self.parameters(), adaptive_max_norm)
clip()
self.optimizer.step()
|
然后是我们的经验池：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
| 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)),
torch.tensor(a),
torch.tensor(r),
torch.tensor(np.array(s_next)),
torch.tensor(done, dtype=torch.int64),
)
def __len__(self):
return len(self.buffer)
|
我们能看出实际上我们的数据存在一个队列deque中，我们为其实现了放入push和取出sample的操作。
1.2 算法流程
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
| for episode in range(num_episodes):
state, _ = env.reset()
while True:
global_step += 1
state ,done = collect_and_store_data(state, global_step, env, q_net, episode_reward, replay_buffer)
if len(replay_buffer) >= Qupdate_batch_size:
q_net.update(replay_buffer,target_q_net,Qupdate_batch_size)
if global_step % targetQ_update_steps == 0:
target_q_net.load_state_dict(q_net.state_dict())
if done:
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()
|
1.3 收集 (s,a,r,s')
数据到缓存区中
首先实现ε-greedy 策略。
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
| def select_action(state, step, env, q_net):
EPSILON_START = 1.0
EPSILON_END = 0.01
EPSILON_DECAY = 5000
epsilon = EPSILON_END + (EPSILON_START - EPSILON_END) * np.exp(-1.0 * step / EPSILON_DECAY)
if random.random() < epsilon:
return env.action_space.sample()
else:
state = torch.tensor(state, dtype=torch.float32).unsqueeze(0).to(device)
with torch.no_grad():
q_values = q_net(state)
max_q = q_values.max(dim=1)[0]
actions = (q_values == max_q).nonzero(as_tuple=True)[1]
action = actions[torch.randint(len(actions), (1,))].item()
return action
|
这里每经过5000步，衰减一次，ε从最开始的1.0，衰减到最小为0.01的值。衰减公式为指数衰减，如下：
\[
ε=ε_{end}+(ε_{start}-ε_{end})*e^{-\frac{step}{5000}}
\]
接下来就是收集数据，很简单，选出动作然后执行，获得(s,a,r,s')
1
2
3
4
5
6
7
8
9
10
11
12
13
14
| def collect_and_store_data(state, global_step, env, q_net,episode_reward,replay_buffer):
action = select_action(state, global_step, env, q_net)
next_state, reward, terminated, truncated, _ = env.step(action)
done = terminated or truncated
episode_reward[0] += reward
replay_buffer.push(state, action, reward, next_state, done)
return next_state ,done
|
1.4 从经验回放中采样并训练
首先我们先判断经验池中是否已经收集到所必须最低数量的数据，然后才开始更新
1
2
| if len(replay_buffer) >= Qupdate_batch_size:
q_net.update(replay_buffer,target_q_net,Qupdate_batch_size)
|
然后我们做如下操作
- 我们从经验池中拿出我们需要的数据
- 计算Q(s,a)的值
- 计算目标值y
- 计算Q(s,a)与y的loss
- 计算梯度，并裁剪梯度
- 更新网络
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
| def update(self,replay_buffer,target_q_net, BATCH_SIZE = 64):
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)
q_values = self.forward(s).gather(1, a.reshape(-1, 1)).reshape(-1)
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)
loss = nn.MSELoss()(q_values, target)
self.optimizer.zero_grad()
loss.backward()
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
adaptive_max_norm = max(2.0 * self.grad_norm_ema, 1.0)
torch.nn.utils.clip_grad_norm_(self.parameters(), adaptive_max_norm)
clip()
self.optimizer.step()
|
优化器使用adam，能够自适应的调整学习率。为了防止梯度爆炸，我们也需要裁剪梯度，通过限制全部参数梯度的L2范数来实现。
权重衰减在每次更新时将参数乘以一个略小于1的系数（往0方向收缩），参数绝对值越大收缩量越大，从而防止参数过大，起到正则化的作用。
权重衰减影响训练结果，因为额外施加了一个持续把参数拉向0的力，当这个力过强或与任务目标冲突时，就会损害训练效果。
RL 中权重衰减普遍比监督学习用更小的值，1e-4 偏大，1e-5
或直接不用是更常见的选择。
我在本次实验中添加了权重衰减，使得训练结果非常差。
1.5 每
targetQ_update_steps步同步目标网络
我们这里targetQ_update_steps=100，即每100步，就更新一次目标网络，将更新好的Q网络参数复制到目标Q网络中。
1
2
| if global_step % targetQ_update_steps == 0:
target_q_net.load_state_dict(q_net.state_dict())
|
1.6 查看训练效果和使用结果
main 函数和展示函数
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
| def load_policy(model_path):
state_dim = 4
action_dim = 2
policy = QNetwork(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)
|
不带裁剪版。由于我们训练的使用使用 ε-greedy
策略，始终会有部分随机，因此训练不到满分是很正常的事情。
另一次
接下来是带裁剪的版本（裁剪是我后来加的）。
