深度强化学习-DDPG算法原理与代码

看完OU噪声后，可能很多小伙伴是懵的，这个也太复杂了。不过我会告诉大家，其实OU噪声是没必要的，因为我们完全可以采用服从正态分布的噪声来取代它，实验结果也证实了这一点。因此Twin Delayed Deep Deterministic policy gradient (TD3)算法舍弃了OU噪声，而是采用服从正态分布的噪声，实现起来更加简单。

另外还需要提醒大家一点：噪声只会加在训练阶段Actor网络输出的动作上，推理阶段不要加上噪声，以及在更新网络参数时也不要加上噪声，前面已经提醒过了。因为我们只需要在训练阶段让智能体具备探索能力，推理时是不需要的。

3 DDPG算法伪代码

 4 代码实现

Actor和Critic网络的代码实现（networks.py）:

1. import torch as T
2. import torch.nn as nn
3. import torch.optim as optim
5. device = T.device("cuda:0" if T.cuda.is_available() else "cpu")
8. def weight_init(m):
9. if isinstance(m, nn.Linear):
10. nn.init.xavier_normal_(m.weight)
11. if m.bias is not None:
12. nn.init.constant_(m.bias, 0.0)
13. elif isinstance(m, nn.BatchNorm1d):
14. nn.init.constant_(m.weight, 1.0)
15. nn.init.constant_(m.bias, 0.0)
18. class ActorNetwork(nn.Module):
19. def __init__(self, alpha, state_dim, action_dim, fc1_dim, fc2_dim):
20. super(ActorNetwork, self).__init__()
21. self.fc1 = nn.Linear(state_dim, fc1_dim)
22. self.ln1 = nn.LayerNorm(fc1_dim)
23. self.fc2 = nn.Linear(fc1_dim, fc2_dim)
24. self.ln2 = nn.LayerNorm(fc2_dim)
25. self.action = nn.Linear(fc2_dim, action_dim)
27. self.optimizer = optim.Adam(self.parameters(), lr=alpha)
28. self.apply(weight_init)
29. self.to(device)
31. def forward(self, state):
32. x = T.relu(self.ln1(self.fc1(state)))
33. x = T.relu(self.ln2(self.fc2(x)))
34. action = T.tanh(self.action(x))
36. return action
38. def save_checkpoint(self, checkpoint_file):
39. T.save(self.state_dict(), checkpoint_file)
41. def load_checkpoint(self, checkpoint_file):
42. self.load_state_dict(T.load(checkpoint_file))
45. class CriticNetwork(nn.Module):
46. def __init__(self, beta, state_dim, action_dim, fc1_dim, fc2_dim):
47. super(CriticNetwork, self).__init__()
48. self.fc1 = nn.Linear(state_dim, fc1_dim)
49. self.ln1 = nn.LayerNorm(fc1_dim)
50. self.fc2 = nn.Linear(fc1_dim, fc2_dim)
51. self.ln2 = nn.LayerNorm(fc2_dim)
52. self.fc3 = nn.Linear(action_dim, fc2_dim)
53. self.q = nn.Linear(fc2_dim, 1)
55. self.optimizer = optim.Adam(self.parameters(), lr=beta, weight_decay=0.001)
56. self.apply(weight_init)
57. self.to(device)
59. def forward(self, state, action):
60. x_s = T.relu(self.ln1(self.fc1(state)))
61. x_s = self.ln2(self.fc2(x_s))
62. x_a = self.fc3(action)
63. x = T.relu(x_s + x_a)
64. q = self.q(x)
66. return q
68. def save_checkpoint(self, checkpoint_file):
69. T.save(self.state_dict(), checkpoint_file)
71. def load_checkpoint(self, checkpoint_file):
72. self.load_state_dict(T.load(checkpoint_file))

注：Actor和Critic网络中前面两个Linear层后面都跟上了Layer Normalization (LN)层。因为我在实验时发现了一个非常有趣的现象，如果不加LN层，或者加入Batch Normalization (BN)层，整个训练过程很容易坍塌或者训练效果很差，具体原因我也不是特别清楚。感兴趣的小伙伴可以把代码git下来跑一遍，如果知道原因的话不妨一起交流。

DDPG算法的代码实现（DDPG.py）:

1. import torch as T
2. import torch.nn.functional as F
3. import numpy as np
4. from networks import ActorNetwork, CriticNetwork
5. from buffer import ReplayBuffer
7. device = T.device("cuda:0" if T.cuda.is_available() else "cpu")
10. class DDPG:
11. def __init__(self, alpha, beta, state_dim, action_dim, actor_fc1_dim,
12. actor_fc2_dim, critic_fc1_dim, critic_fc2_dim, ckpt_dir,
13. gamma=0.99, tau=0.005, action_noise=0.1, max_size=1000000,
14. batch_size=256):
15. self.gamma = gamma
16. self.tau = tau
17. self.action_noise = action_noise
18. self.checkpoint_dir = ckpt_dir
20. self.actor = ActorNetwork(alpha=alpha, state_dim=state_dim, action_dim=action_dim,
21. fc1_dim=actor_fc1_dim, fc2_dim=actor_fc2_dim)
22. self.target_actor = ActorNetwork(alpha=alpha, state_dim=state_dim, action_dim=action_dim,
23. fc1_dim=actor_fc1_dim, fc2_dim=actor_fc2_dim)
24. self.critic = CriticNetwork(beta=beta, state_dim=state_dim, action_dim=action_dim,
25. fc1_dim=critic_fc1_dim, fc2_dim=critic_fc2_dim)
26. self.target_critic = CriticNetwork(beta=beta, state_dim=state_dim, action_dim=action_dim,
27. fc1_dim=critic_fc1_dim, fc2_dim=critic_fc2_dim)
29. self.memory = ReplayBuffer(max_size=max_size, state_dim=state_dim, action_dim=action_dim,
30. batch_size=batch_size)
32. self.update_network_parameters(tau=1.0)
34. def update_network_parameters(self, tau=None):
35. if tau is None:
36. tau = self.tau
38. for actor_params, target_actor_params in zip(self.actor.parameters(),
39. self.target_actor.parameters()):
40. target_actor_params.data.copy_(tau * actor_params + (1 - tau) * target_actor_params)
42. for critic_params, target_critic_params in zip(self.critic.parameters(),
43. self.target_critic.parameters()):
44. target_critic_params.data.copy_(tau * critic_params + (1 - tau) * target_critic_params)
46. def remember(self, state, action, reward, state_, done):
47. self.memory.store_transition(state, action, reward, state_, done)
49. def choose_action(self, observation, train=True):
50. self.actor.eval()
51. state = T.tensor([observation], dtype=T.float).to(device)
52. action = self.actor.forward(state).squeeze()
54. if train:
55. noise = T.tensor(np.random.normal(loc=0.0, scale=self.action_noise),
56. dtype=T.float).to(device)
57. action = T.clamp(action+noise, -1, 1)
58. self.actor.train()
60. return action.detach().cpu().numpy()
62. def learn(self):
63. if not self.memory.ready():
64. return
66. states, actions, reward, states_, terminals = self.memory.sample_buffer()
67. states_tensor = T.tensor(states, dtype=T.float).to(device)
68. actions_tensor = T.tensor(actions, dtype=T.float).to(device)
69. rewards_tensor = T.tensor(reward, dtype=T.float).to(device)
70. next_states_tensor = T.tensor(states_, dtype=T.float).to(device)
71. terminals_tensor = T.tensor(terminals).to(device)
73. with T.no_grad():
74. next_actions_tensor = self.target_actor.forward(next_states_tensor)
75. q_ = self.target_critic.forward(next_states_tensor, next_actions_tensor).view(-1)
76. q_[terminals_tensor] = 0.0
77. target = rewards_tensor + self.gamma * q_
78. q = self.critic.forward(states_tensor, actions_tensor).view(-1)
80. critic_loss = F.mse_loss(q, target.detach())
81. self.critic.optimizer.zero_grad()
82. critic_loss.backward()
83. self.critic.optimizer.step()
85. new_actions_tensor = self.actor.forward(states_tensor)
86. actor_loss = -T.mean(self.critic(states_tensor, new_actions_tensor))
87. self.actor.optimizer.zero_grad()
88. actor_loss.backward()
89. self.actor.optimizer.step()
91. self.update_network_parameters()
93. def save_models(self, episode):
94. self.actor.save_checkpoint(self.checkpoint_dir + 'Actor/DDPG_actor_{}.pth'.format(episode))
95. print('Saving actor network successfully!')
96. self.target_actor.save_checkpoint(self.checkpoint_dir +
97. 'Target_actor/DDPG_target_actor_{}.pth'.format(episode))
98. print('Saving target_actor network successfully!')
99. self.critic.save_checkpoint(self.checkpoint_dir + 'Critic/DDPG_critic_{}'.format(episode))
100. print('Saving critic network successfully!')
101. self.target_critic.save_checkpoint(self.checkpoint_dir +
102. 'Target_critic/DDPG_target_critic_{}'.format(episode))
103. print('Saving target critic network successfully!')
105. def load_models(self, episode):
106. self.actor.load_checkpoint(self.checkpoint_dir + 'Actor/DDPG_actor_{}.pth'.format(episode))
107. print('Loading actor network successfully!')
108. self.target_actor.load_checkpoint(self.checkpoint_dir +
109. 'Target_actor/DDPG_target_actor_{}.pth'.format(episode))
110. print('Loading target_actor network successfully!')
111. self.critic.load_checkpoint(self.checkpoint_dir + 'Critic/DDPG_critic_{}'.format(episode))
112. print('Loading critic network successfully!')
113. self.target_critic.load_checkpoint(self.checkpoint_dir +
114. 'Target_critic/DDPG_target_critic_{}'.format(episode))
115. print('Loading target critic network successfully!')

算法仿真环境是gym库中的LunarLanderContinuous-v2环境，因此需要先配置好gym库。进入Aanconda中对应的Python环境中，执行下面的指令

但是，这样安装的gym库只包括少量的内置环境，如算法环境、简单文字游戏环境和经典控制环境，无法使用LunarLanderContinuous-v2。因此还要安装一些其他依赖项，具体可以参照这篇blog: AttributeError: module ‘gym.envs.box2d‘ has no attribute ‘LunarLander‘ 解决办法。如果已经配置好环境，那请忽略这一段。

训练脚本（train.py）:

1. import gym
2. import numpy as np
3. import argparse
4. from DDPG import DDPG
5. from utils import create_directory, plot_learning_curve, scale_action
7. parser = argparse.ArgumentParser("DDPG parameters")
8. parser.add_argument('--max_episodes', type=int, default=1000)
9. parser.add_argument('--checkpoint_dir', type=str, default='./checkpoints/DDPG/')
10. parser.add_argument('--figure_file', type=str, default='./output_images/reward.png')
12. args = parser.parse_args()
15. def main():
16. env = gym.make('LunarLanderContinuous-v2')
17. agent = DDPG(alpha=0.0003, beta=0.0003, state_dim=env.observation_space.shape[0],
18. action_dim=env.action_space.shape[0], actor_fc1_dim=400, actor_fc2_dim=300,
19. critic_fc1_dim=400, critic_fc2_dim=300, ckpt_dir=args.checkpoint_dir,
20. batch_size=256)
21. create_directory(args.checkpoint_dir,
22. sub_paths=['Actor', 'Target_actor', 'Critic', 'Target_critic'])
24. reward_history = []
25. avg_reward_history = []
26. for episode in range(args.max_episodes):
27. done = False
28. total_reward = 0
29. observation = env.reset()
30. while not done:
31. action = agent.choose_action(observation, train=True)
32. action_ = scale_action(action.copy(), env.action_space.high, env.action_space.low)
33. observation_, reward, done, info = env.step(action_)
34. agent.remember(observation, action, reward, observation_, done)
35. agent.learn()
36. total_reward += reward
37. observation = observation_
39. reward_history.append(total_reward)
40. avg_reward = np.mean(reward_history[-100:])
41. avg_reward_history.append(avg_reward)
42. print('Ep: {} Reward: {:.1f} AvgReward: {:.1f}'.format(episode+1, total_reward, avg_reward))
44. if (episode + 1) % 200 == 0:
45. agent.save_models(episode+1)
47. episodes = [i+1 for i in range(args.max_episodes)]
48. plot_learning_curve(episodes, avg_reward_history, title='AvgReward',
49. ylabel='reward', figure_file=args.figure_file)
52. if __name__ == '__main__':
53. main()

训练脚本中有三个参数，max_episodes表示训练幕数，checkpoint_dir表示训练权重保存路径，figure_file表示训练结果的保存路径（其实是一张累积奖励曲线图），按照默认设置即可。

训练时还会用到画图函数和创建文件夹函数，它们都被放在utils.py脚本中：

1. import os
2. import numpy as np
3. import matplotlib.pyplot as plt
6. class OUActionNoise:
7. def __init__(self, mu, sigma=0.15, theta=0.2, dt=1e-2, x0=None):
8. self.theta = theta
9. self.mu = mu
10. self.sigma = sigma
11. self.dt = dt
12. self.x0 = x0
13. self.reset()
15. def __call__(self):
16. x = self.x_prev + self.theta * (self.mu - self.x_prev) * self.dt + \
17. self.sigma * np.sqrt(self.dt) * np.random.normal(size=self.mu.shape)
18. self.x_prev = x
20. return x
22. def reset(self):
23. self.x_prev = self.x0 if self.x0 is not None else np.zeros_like(self.mu)
26. def create_directory(path: str, sub_paths: list):
27. for sub_path in sub_paths:
28. if not os.path.exists(path + sub_path):
29. os.makedirs(path + sub_path, exist_ok=True)
30. print('Create path: {} successfully'.format(path+sub_path))
31. else:
32. print('Path: {} is already existence'.format(path+sub_path))
35. def plot_learning_curve(episodes, records, title, ylabel, figure_file):
36. plt.figure()
37. plt.plot(episodes, records, color='r', linestyle='-')
38. plt.title(title)
39. plt.xlabel('episode')
40. plt.ylabel(ylabel)
42. plt.show()
43. plt.savefig(figure_file)
46. def scale_action(action, high, low):
47. action = np.clip(action, -1, 1)
48. weight = (high - low) / 2
49. bias = (high + low) / 2
50. action_ = action * weight + bias
52. return action_

另外我们还提供了测试代码，主要用于测试训练效果以及观察环境的动态渲染 (test.py)：

1. import gym
2. import imageio
3. import argparse
4. from DDPG import DDPG
5. from utils import scale_action
7. parser = argparse.ArgumentParser()
8. parser.add_argument('--filename', type=str, default='./output_images/LunarLander.gif')
9. parser.add_argument('--checkpoint_dir', type=str, default='./checkpoints/DDPG/')
10. parser.add_argument('--save_video', type=bool, default=True)
11. parser.add_argument('--fps', type=int, default=30)
12. parser.add_argument('--render', type=bool, default=True)
14. args = parser.parse_args()
17. def main():
18. env = gym.make('LunarLanderContinuous-v2')
19. agent = DDPG(alpha=0.0003, beta=0.0003, state_dim=env.observation_space.shape[0],
20. action_dim=env.action_space.shape[0], actor_fc1_dim=400, actor_fc2_dim=300,
21. critic_fc1_dim=400, critic_fc2_dim=300, ckpt_dir=args.checkpoint_dir,
22. batch_size=256)
23. agent.load_models(1000)
24. video = imageio.get_writer(args.filename, fps=args.fps)
26. done = False
27. observation = env.reset()
28. while not done:
29. if args.render:
30. env.render()
31. action = agent.choose_action(observation, train=True)
32. action_ = scale_action(action.copy(), env.action_space.high, env.action_space.low)
33. observation_, reward, done, info = env.step(action_)
34. observation = observation_
35. if args.save_video:
36. video.append_data(env.render(mode='rgb_array'))
39. if __name__ == '__main__':
40. main()

测试脚本中包括五个参数，filename表示环境动态图的保存路径，checkpoint_dir表示加载的权重路径，save_video表示是否要保存动态图，fps表示动态图的帧率，rander表示是否开启环境渲染。大家只需要调整save_video和rander这两个参数，其余保持默认即可。

5 实验结果

通过平均奖励曲线可以看出，大概迭代到700步左右时算法趋于收敛。 

这是测试效果图，智能体能够很好地完成降落任务！ 

6 结论

本文主要讲解了DDPG算法的原理以及代码实现。尽管它是一个非常优秀的算法，但是仍然存在一些问题需要改进，例如过估计。后面我们会讲解一下TD3算法，它其实就是在DDPG算法的基础做了一些改进工作，克服了DDPG算法中的一些问题，从而让算法的性能得到显著提升。

以上如果有出现错误的地方，欢迎各位怒斥！