【深度强化学习】(6) PPO 模型解析，附Pytorch完整代码

大家好，今天和各位分享一下深度强化学习中的近端策略优化算法（proximal policy optimization，PPO），并借助 OpenAI 的 gym 环境完成一个小案例，完整代码可以从我的 GitHub 中获得：

https://github.com/LiSir-HIT/Reinforcement-Learning/tree/main/Model

1. 算法原理

PPO 算法之所以被提出，根本原因在于 Policy Gradient 在处理连续动作空间时 Learning rate 取值抉择困难。Learning rate 取值过小，就会导致深度强化学习收敛性较差，陷入完不成训练的局面，取值过大则导致新旧策略迭代时数据不一致，造成学习波动较大或局部震荡。除此之外，Policy Gradient 因为在线学习的性质，进行迭代策略时原先的采样数据无法被重复利用，每次迭代都需要重新采样；

同样地置信域策略梯度算法（Trust Region Policy Optimization，TRPO）虽然利用重要性采样（Important-sampling）、共轭梯度法求解提升了样本效率、训练速率等，但在处理函数的二阶近似时会面临计算量过大，以及实现过程复杂、兼容性差等缺陷。 

PPO 算法具备 Policy Gradient、TRPO 的部分优点，采样数据和使用随机梯度上升方法优化代替目标函数之间交替进行，虽然标准的策略梯度方法对每个数据样本执行一次梯度更新，但 PPO 提出新目标函数，可以实现小批量更新。

鉴于上述问题，该算法在迭代更新时，观察当前策略在 t 时刻智能体处于状态 s 所采取的行为概率

如果优势函数为正数，需要增大新旧策略比值 

PPO 算法主要由 Actor 和 Critic 两部分构成，Critic 部分更新方式与其他Actor-Critic 类型相似，通常采用计算 TD  error（时序差分误差）形式。对于 Actor 的更新方式，PPO 可在KLPENL 、CLIPL 之间选择对于当前实验环境稳定性适用性更强的目标函数，经过 OpenAI 研究团队实验论证，PPO- Clip 比 PPO- Penalty有更好的数据效率和可行性。 

2. 代码实现

下面我就采用 Clip 形式的 PPO。模型构建代码如下。下面的模型适用于 action 是离散的情况，连续情况的代码可以从我的 GitHub 中获取。

1. # 代码用于离散环境的模型
2. import numpy as np
3. import torch
4. from torch import nn
5. from torch.nn import functional as F
7. # ----------------------------------- #
8. # 构建策略网络--actor
9. # ----------------------------------- #
11. class PolicyNet(nn.Module):
12. def __init__(self, n_states, n_hiddens, n_actions):
13. super(PolicyNet, self).__init__()
14. self.fc1 = nn.Linear(n_states, n_hiddens)
15. self.fc2 = nn.Linear(n_hiddens, n_actions)
16. def forward(self, x):
17. x = self.fc1(x) # [b,n_states]-->[b,n_hiddens]
18. x = F.relu(x)
19. x = self.fc2(x) # [b, n_actions]
20. x = F.softmax(x, dim=1) # [b, n_actions] 计算每个动作的概率
21. return x
23. # ----------------------------------- #
24. # 构建价值网络--critic
25. # ----------------------------------- #
27. class ValueNet(nn.Module):
28. def __init__(self, n_states, n_hiddens):
29. super(ValueNet, self).__init__()
30. self.fc1 = nn.Linear(n_states, n_hiddens)
31. self.fc2 = nn.Linear(n_hiddens, 1)
32. def forward(self, x):
33. x = self.fc1(x) # [b,n_states]-->[b,n_hiddens]
34. x = F.relu(x)
35. x = self.fc2(x) # [b,n_hiddens]-->[b,1] 评价当前的状态价值state_value
36. return x
38. # ----------------------------------- #
39. # 构建模型
40. # ----------------------------------- #
42. class PPO:
43. def __init__(self, n_states, n_hiddens, n_actions,
44. actor_lr, critic_lr, lmbda, epochs, eps, gamma, device):
45. # 实例化策略网络
46. self.actor = PolicyNet(n_states, n_hiddens, n_actions).to(device)
47. # 实例化价值网络
48. self.critic = ValueNet(n_states, n_hiddens).to(device)
49. # 策略网络的优化器
50. self.actor_optimizer = torch.optim.Adam(self.actor.parameters(), lr=actor_lr)
51. # 价值网络的优化器
52. self.critic_optimizer = torch.optim.Adam(self.critic.parameters(), lr = critic_lr)
54. self.gamma = gamma # 折扣因子
55. self.lmbda = lmbda # GAE优势函数的缩放系数
56. self.epochs = epochs # 一条序列的数据用来训练轮数
57. self.eps = eps # PPO中截断范围的参数
58. self.device = device
60. # 动作选择
61. def take_action(self, state):
62. # 维度变换 [n_state]-->tensor[1,n_states]
63. state = torch.tensor(state[np.newaxis, :]).to(self.device)
64. # 当前状态下，每个动作的概率分布 [1,n_states]
65. probs = self.actor(state)
66. # 创建以probs为标准的概率分布
67. action_list = torch.distributions.Categorical(probs)
68. # 依据其概率随机挑选一个动作
69. action = action_list.sample().item()
70. return action
72. # 训练
73. def learn(self, transition_dict):
74. # 提取数据集
75. states = torch.tensor(transition_dict['states'], dtype=torch.float).to(self.device)
76. actions = torch.tensor(transition_dict['actions']).to(self.device).view(-1,1)
77. rewards = torch.tensor(transition_dict['rewards'], dtype=torch.float).to(self.device).view(-1,1)
78. next_states = torch.tensor(transition_dict['next_states'], dtype=torch.float).to(self.device)
79. dones = torch.tensor(transition_dict['dones'], dtype=torch.float).to(self.device).view(-1,1)
81. # 目标，下一个状态的state_value [b,1]
82. next_q_target = self.critic(next_states)
83. # 目标，当前状态的state_value [b,1]
84. td_target = rewards + self.gamma * next_q_target * (1-dones)
85. # 预测，当前状态的state_value [b,1]
86. td_value = self.critic(states)
87. # 目标值和预测值state_value之差 [b,1]
88. td_delta = td_target - td_value
90. # 时序差分值 tensor-->numpy [b,1]
91. td_delta = td_delta.cpu().detach().numpy()
92. advantage = 0 # 优势函数初始化
93. advantage_list = []
95. # 计算优势函数
96. for delta in td_delta[::-1]: # 逆序时序差分值 axis=1轴上倒着取 [], [], []
97. # 优势函数GAE的公式
98. advantage = self.gamma * self.lmbda * advantage + delta
99. advantage_list.append(advantage)
100. # 正序
101. advantage_list.reverse()
102. # numpy --> tensor [b,1]
103. advantage = torch.tensor(advantage_list, dtype=torch.float).to(self.device)
105. # 策略网络给出每个动作的概率，根据action得到当前时刻下该动作的概率
106. old_log_probs = torch.log(self.actor(states).gather(1, actions)).detach()
108. # 一组数据训练 epochs 轮
109. for _ in range(self.epochs):
110. # 每一轮更新一次策略网络预测的状态
111. log_probs = torch.log(self.actor(states).gather(1, actions))
112. # 新旧策略之间的比例
113. ratio = torch.exp(log_probs - old_log_probs)
114. # 近端策略优化裁剪目标函数公式的左侧项
115. surr1 = ratio * advantage
116. # 公式的右侧项，ratio小于1-eps就输出1-eps，大于1+eps就输出1+eps
117. surr2 = torch.clamp(ratio, 1-self.eps, 1+self.eps) * advantage
119. # 策略网络的损失函数
120. actor_loss = torch.mean(-torch.min(surr1, surr2))
121. # 价值网络的损失函数，当前时刻的state_value - 下一时刻的state_value
122. critic_loss = torch.mean(F.mse_loss(self.critic(states), td_target.detach()))
124. # 梯度清0
125. self.actor_optimizer.zero_grad()
126. self.critic_optimizer.zero_grad()
127. # 反向传播
128. actor_loss.backward()
129. critic_loss.backward()
130. # 梯度更新
131. self.actor_optimizer.step()
132. self.critic_optimizer.step()

3. 案例演示

基于 OpenAI 的 gym 环境完成一个推车游戏，一个离散的环境，目标是左右移动小车将黄色的杆子保持竖直。动作维度为2，属于离散值；状态维度为 4，分别是坐标、速度、角度、角速度。

1. import numpy as np
2. import matplotlib.pyplot as plt
3. import gym
4. import torch
5. from RL_brain import PPO
7. device = torch.device('cuda') if torch.cuda.is_available() \
8. else torch.device('cpu')
10. # ----------------------------------------- #
11. # 参数设置
12. # ----------------------------------------- #
14. num_episodes = 100 # 总迭代次数
15. gamma = 0.9 # 折扣因子
16. actor_lr = 1e-3 # 策略网络的学习率
17. critic_lr = 1e-2 # 价值网络的学习率
18. n_hiddens = 16 # 隐含层神经元个数
19. env_name = 'CartPole-v1'
20. return_list = [] # 保存每个回合的return
22. # ----------------------------------------- #
23. # 环境加载
24. # ----------------------------------------- #
26. env = gym.make(env_name, render_mode="human")
27. n_states = env.observation_space.shape[0] # 状态数 4
28. n_actions = env.action_space.n # 动作数 2
30. # ----------------------------------------- #
31. # 模型构建
32. # ----------------------------------------- #
34. agent = PPO(n_states=n_states, # 状态数
35. n_hiddens=n_hiddens, # 隐含层数
36. n_actions=n_actions, # 动作数
37. actor_lr=actor_lr, # 策略网络学习率
38. critic_lr=critic_lr, # 价值网络学习率
39. lmbda = 0.95, # 优势函数的缩放因子
40. epochs = 10, # 一组序列训练的轮次
41. eps = 0.2, # PPO中截断范围的参数
42. gamma=gamma, # 折扣因子
43. device = device
44. )
46. # ----------------------------------------- #
47. # 训练--回合更新 on_policy
48. # ----------------------------------------- #
50. for i in range(num_episodes):
52. state = env.reset()[0] # 环境重置
53. done = False # 任务完成的标记
54. episode_return = 0 # 累计每回合的reward
56. # 构造数据集，保存每个回合的状态数据
57. transition_dict = {
58. 'states': [],
59. 'actions': [],
60. 'next_states': [],
61. 'rewards': [],
62. 'dones': [],
63. }
65. while not done:
66. action = agent.take_action(state) # 动作选择
67. next_state, reward, done, _, _ = env.step(action) # 环境更新
68. # 保存每个时刻的状态\动作\...
69. transition_dict['states'].append(state)
70. transition_dict['actions'].append(action)
71. transition_dict['next_states'].append(next_state)
72. transition_dict['rewards'].append(reward)
73. transition_dict['dones'].append(done)
74. # 更新状态
75. state = next_state
76. # 累计回合奖励
77. episode_return += reward
79. # 保存每个回合的return
80. return_list.append(episode_return)
81. # 模型训练
82. agent.learn(transition_dict)
84. # 打印回合信息
85. print(f'iter:{i}, return:{np.mean(return_list[-10:])}')
87. # -------------------------------------- #
88. # 绘图
89. # -------------------------------------- #
91. plt.plot(return_list)
92. plt.title('return')
93. plt.show()

训练100回合，绘制每回合的 return