您现在的位置是：首页 > 技术教程 正文

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

admin 阅读： 2024-03-20
后台-插件-广告管理-内容页头部广告（手机）

大家好，今天和各位分享一下深度确定性策略梯度算法 (Deterministic Policy Gradient，DDPG)。并基于 OpenAI 的 gym 环境完成一个小游戏。完整代码在我的 GitHub 中获得：

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

1. 基本原理

深度确定性策略梯度算法是结合确定性策略梯度算法的思想，对 DQN 的一种改进，是一种无模型的深度强化学习算法。

DDPG 算法使用演员-评论家（Actor-Critic）算法作为其基本框架，采用深度神经网络作为策略网络和动作值函数的近似，使用随机梯度法训练策略网络和价值网络模型中的参数。DDPG 算法的原理如下图所示。

DDPG 算法架构中使用双重神经网络架构，对于策略函数和价值函数均使用双重神经网络模型架构（即 Online 网络和 Target 网络），使得算法的学习过程更加稳定，收敛的速度加快。同时该算法引入经验回放机制，Actor 与环境交互生产生的经验数据样本存储到经验池中，抽取批量数据样本进行训练，即类似于 DQN 的经验回放机制，去除样本的相关性和依赖性，使得算法更加容易收敛。 

2. 公式推导

为了便于大家理解 DDPG 的推导过程，算法框架如下图所示：

DDPG 共包含 4 个神经网络，用于对 Q 值函数和策略的近似表示。Critic 目标网络用于近似估计下一时刻的状态-动作的 Q 值函数 Q_{w'}(S_{t+1},\pi _{\theta '}(S_{t+1}))

3. 代码实现

DDPG 的伪代码如下：

模型代码如下：

  1. import torch
  2. from torch import nn
  3. from torch.nn import functional as F
  4. import numpy as np
  5. import collections
  6. import random
  8. # ------------------------------------- #
  9. # 经验回放池
  10. # ------------------------------------- #
  12. class ReplayBuffer:
  13. def __init__(self, capacity): # 经验池的最大容量
  14. # 创建一个队列，先进先出
  15. self.buffer = collections.deque(maxlen=capacity)
  16. # 在队列中添加数据
  17. def add(self, state, action, reward, next_state, done):
  18. # 以list类型保存
  19. self.buffer.append((state, action, reward, next_state, done))
  20. # 在队列中随机取样batch_size组数据
  21. def sample(self, batch_size):
  22. transitions = random.sample(self.buffer, batch_size)
  23. # 将数据集拆分开来
  24. state, action, reward, next_state, done = zip(*transitions)
  25. return np.array(state), action, reward, np.array(next_state), done
  26. # 测量当前时刻的队列长度
  27. def size(self):
  28. return len(self.buffer)
  30. # ------------------------------------- #
  31. # 策略网络
  32. # ------------------------------------- #
  34. class PolicyNet(nn.Module):
  35. def __init__(self, n_states, n_hiddens, n_actions, action_bound):
  36. super(PolicyNet, self).__init__()
  37. # 环境可以接受的动作最大值
  38. self.action_bound = action_bound
  39. # 只包含一个隐含层
  40. self.fc1 = nn.Linear(n_states, n_hiddens)
  41. self.fc2 = nn.Linear(n_hiddens, n_actions)
  42. # 前向传播
  43. def forward(self, x):
  44. x = self.fc1(x) # [b,n_states]-->[b,n_hiddens]
  45. x = F.relu(x)
  46. x = self.fc2(x) # [b,n_hiddens]-->[b,n_actions]
  47. x= torch.tanh(x) # 将数值调整到 [-1,1]
  48. x = x * self.action_bound # 缩放到 [-action_bound, action_bound]
  49. return x
  51. # ------------------------------------- #
  52. # 价值网络
  53. # ------------------------------------- #
  55. class QValueNet(nn.Module):
  56. def __init__(self, n_states, n_hiddens, n_actions):
  57. super(QValueNet, self).__init__()
  58. #
  59. self.fc1 = nn.Linear(n_states + n_actions, n_hiddens)
  60. self.fc2 = nn.Linear(n_hiddens, n_hiddens)
  61. self.fc3 = nn.Linear(n_hiddens, 1)
  62. # 前向传播
  63. def forward(self, x, a):
  64. # 拼接状态和动作
  65. cat = torch.cat([x, a], dim=1) # [b, n_states + n_actions]
  66. x = self.fc1(cat) # -->[b, n_hiddens]
  67. x = F.relu(x)
  68. x = self.fc2(x) # -->[b, n_hiddens]
  69. x = F.relu(x)
  70. x = self.fc3(x) # -->[b, 1]
  71. return x
  73. # ------------------------------------- #
  74. # 算法主体
  75. # ------------------------------------- #
  77. class DDPG:
  78. def __init__(self, n_states, n_hiddens, n_actions, action_bound,
  79. sigma, actor_lr, critic_lr, tau, gamma, device):
  81. # 策略网络--训练
  82. self.actor = PolicyNet(n_states, n_hiddens, n_actions, action_bound).to(device)
  83. # 价值网络--训练
  84. self.critic = QValueNet(n_states, n_hiddens, n_actions).to(device)
  85. # 策略网络--目标
  86. self.target_actor = PolicyNet(n_states, n_hiddens, n_actions, action_bound).to(device)
  87. # 价值网络--目标
  88. self.target_critic = QValueNet(n_states, n_hiddens, n_actions).to(device
  89. )
  90. # 初始化价值网络的参数，两个价值网络的参数相同
  91. self.target_critic.load_state_dict(self.critic.state_dict())
  92. # 初始化策略网络的参数，两个策略网络的参数相同
  93. self.target_actor.load_state_dict(self.actor.state_dict())
  95. # 策略网络的优化器
  96. self.actor_optimizer = torch.optim.Adam(self.actor.parameters(), lr=actor_lr)
  97. # 价值网络的优化器
  98. self.critic_optimizer = torch.optim.Adam(self.critic.parameters(), lr=critic_lr)
  100. # 属性分配
  101. self.gamma = gamma # 折扣因子
  102. self.sigma = sigma # 高斯噪声的标准差，均值设为0
  103. self.tau = tau # 目标网络的软更新参数
  104. self.n_actions = n_actions
  105. self.device = device
  107. # 动作选择
  108. def take_action(self, state):
  109. # 维度变换 list[n_states]-->tensor[1,n_states]-->gpu
  110. state = torch.tensor(state, dtype=torch.float).view(1,-1).to(self.device)
  111. # 策略网络计算出当前状态下的动作价值 [1,n_states]-->[1,1]-->int
  112. action = self.actor(state).item()
  113. # 给动作添加噪声，增加搜索
  114. action = action + self.sigma * np.random.randn(self.n_actions)
  115. return action
  117. # 软更新, 意思是每次learn的时候更新部分参数
  118. def soft_update(self, net, target_net):
  119. # 获取训练网络和目标网络需要更新的参数
  120. for param_target, param in zip(target_net.parameters(), net.parameters()):
  121. # 训练网络的参数更新要综合考虑目标网络和训练网络
  122. param_target.data.copy_(param_target.data*(1-self.tau) + param.data*self.tau)
  124. # 训练
  125. def update(self, transition_dict):
  126. # 从训练集中取出数据
  127. states = torch.tensor(transition_dict['states'], dtype=torch.float).to(self.device) # [b,n_states]
  128. actions = torch.tensor(transition_dict['actions'], dtype=torch.float).view(-1,1).to(self.device) # [b,1]
  129. rewards = torch.tensor(transition_dict['rewards'], dtype=torch.float).view(-1,1).to(self.device) # [b,1]
  130. next_states = torch.tensor(transition_dict['next_states'], dtype=torch.float).to(self.device) # [b,next_states]
  131. dones = torch.tensor(transition_dict['dones'], dtype=torch.float).view(-1,1).to(self.device) # [b,1]
  133. # 价值目标网络获取下一时刻的动作[b,n_states]-->[b,n_actors]
  134. next_q_values = self.target_actor(next_states)
  135. # 策略目标网络获取下一时刻状态选出的动作价值 [b,n_states+n_actions]-->[b,1]
  136. next_q_values = self.target_critic(next_states, next_q_values)
  137. # 当前时刻的动作价值的目标值 [b,1]
  138. q_targets = rewards + self.gamma * next_q_values * (1-dones)
  140. # 当前时刻动作价值的预测值 [b,n_states+n_actions]-->[b,1]
  141. q_values = self.critic(states, actions)
  143. # 预测值和目标值之间的均方差损失
  144. critic_loss = torch.mean(F.mse_loss(q_values, q_targets))
  145. # 价值网络梯度
  146. self.critic_optimizer.zero_grad()
  147. critic_loss.backward()
  148. self.critic_optimizer.step()
  150. # 当前状态的每个动作的价值 [b, n_actions]
  151. actor_q_values = self.actor(states)
  152. # 当前状态选出的动作价值 [b,1]
  153. score = self.critic(states, actor_q_values)
  154. # 计算损失
  155. actor_loss = -torch.mean(score)
  156. # 策略网络梯度
  157. self.actor_optimizer.zero_grad()
  158. actor_loss.backward()
  159. self.actor_optimizer.step()
  161. # 软更新策略网络的参数
  162. self.soft_update(self.actor, self.target_actor)
  163. # 软更新价值网络的参数
  164. self.soft_update(self.critic, self.target_critic)

4. 案例演示

基于 OpenAI 的 gym 环境完成一个推车游戏，目标是将小车推到山顶旗子处。动作维度为1，属于连续值；状态维度为 2，分别是 x 坐标和小车速度。

代码如下：

  1. import numpy as np
  2. import torch
  3. import matplotlib.pyplot as plt
  4. import gym
  5. from parsers import args
  6. from RL_brain import ReplayBuffer, DDPG
  7. device = torch.device('cuda') if torch.cuda.is_available() else torch.device('cpu')
  9. # -------------------------------------- #
  10. # 环境加载
  11. # -------------------------------------- #
  13. env_name = "MountainCarContinuous-v0" # 连续型动作
  14. env = gym.make(env_name, render_mode="human")
  15. n_states = env.observation_space.shape[0] # 状态数 2
  16. n_actions = env.action_space.shape[0] # 动作数 1
  17. action_bound = env.action_space.high[0] # 动作的最大值 1.0
  20. # -------------------------------------- #
  21. # 模型构建
  22. # -------------------------------------- #
  24. # 经验回放池实例化
  25. replay_buffer = ReplayBuffer(capacity=args.buffer_size)
  26. # 模型实例化
  27. agent = DDPG(n_states = n_states, # 状态数
  28. n_hiddens = args.n_hiddens, # 隐含层数
  29. n_actions = n_actions, # 动作数
  30. action_bound = action_bound, # 动作最大值
  31. sigma = args.sigma, # 高斯噪声
  32. actor_lr = args.actor_lr, # 策略网络学习率
  33. critic_lr = args.critic_lr, # 价值网络学习率
  34. tau = args.tau, # 软更新系数
  35. gamma = args.gamma, # 折扣因子
  36. device = device
  37. )
  39. # -------------------------------------- #
  40. # 模型训练
  41. # -------------------------------------- #
  43. return_list = [] # 记录每个回合的return
  44. mean_return_list = [] # 记录每个回合的return均值
  46. for i in range(10): # 迭代10回合
  47. episode_return = 0 # 累计每条链上的reward
  48. state = env.reset()[0] # 初始时的状态
  49. done = False # 回合结束标记
  51. while not done:
  52. # 获取当前状态对应的动作
  53. action = agent.take_action(state)
  54. # 环境更新
  55. next_state, reward, done, _, _ = env.step(action)
  56. # 更新经验回放池
  57. replay_buffer.add(state, action, reward, next_state, done)
  58. # 状态更新
  59. state = next_state
  60. # 累计每一步的reward
  61. episode_return += reward
  63. # 如果经验池超过容量，开始训练
  64. if replay_buffer.size() > args.min_size:
  65. # 经验池随机采样batch_size组
  66. s, a, r, ns, d = replay_buffer.sample(args.batch_size)
  67. # 构造数据集
  68. transition_dict = {
  69. 'states': s,
  70. 'actions': a,
  71. 'rewards': r,
  72. 'next_states': ns,
  73. 'dones': d,
  74. }
  75. # 模型训练
  76. agent.update(transition_dict)
  78. # 保存每一个回合的回报
  79. return_list.append(episode_return)
  80. mean_return_list.append(np.mean(return_list[-10:])) # 平滑
  82. # 打印回合信息
  83. print(f'iter:{i}, return:{episode_return}, mean_return:{np.mean(return_list[-10:])}')
  85. # 关闭动画窗格
  86. env.close()
  88. # -------------------------------------- #
  89. # 绘图
  90. # -------------------------------------- #
  92. x_range = list(range(len(return_list)))
  94. plt.subplot(121)
  95. plt.plot(x_range, return_list) # 每个回合return
  96. plt.xlabel('episode')
  97. plt.ylabel('return')
  98. plt.subplot(122)
  99. plt.plot(x_range, mean_return_list) # 每回合return均值
  100. plt.xlabel('episode')
  101. plt.ylabel('mean_return')
标签:
声明

1.本站遵循行业规范，任何转载的稿件都会明确标注作者和来源；2.本站的原创文章，请转载时务必注明文章作者和来源，不尊重原创的行为我们将追究责任；3.作者投稿可能会经我们编辑修改或补充。

在线投稿：投稿 站长QQ：1888636

后台-插件-广告管理-内容页尾部广告（手机）
相关文章
关注我们

扫一扫关注我们，了解最新精彩内容

搜索
排行榜
最近发表
标签列表