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

【深度强化学习】(4) Actor-Critic 模型解析,附Pytorch完整代码

admin 阅读: 2024-03-23
后台-插件-广告管理-内容页头部广告(手机)

大家好,今天和各位分享一下深度强化学习中的 Actor-Critic 演员评论家算法,Actor-Critic 算法是一种综合了策略迭代和价值迭代的集成算法。我将使用该模型结合 OpenAI 中的 Gym 环境完成一个小游戏,完整代码可以从我的 GitHub 中获得:

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


1. 算法原理

根据 agent 选择动作方法的不同,可以把强化学习方法分为三大类:行动者方法(Actor-only),评论家方法(Critic-only),行动者评论家方法(Actor-critic)。

行动者方法中不会对值函数进行估计,直接按照当前策略和环境进行交互。通过交互后得到的立即奖赏值直接优化当前策略。例如:Policy Gradients

评论家方法没有需要维护的策略,评论家方法的策略是直接通过当前的值函数获得的,并通过值函数获得的策略与环境交互。交互得到的立即奖赏值用来优化当前值函数。例如:DQN

行动者评论家方法是由行动者和评论家两个部分构成。行动者用于选择动作评论家评论选择动作的好坏。行动者选择动作的方法不是依据当前的值函数,而是依据存储的策略。评论家的评论一般采用时间差分误差的形式,时间差分误差是根据当前的值函数计算获得的时间差分误差是是评论家的唯一输出,并且驱动了行动者和评论家之间的所有学习。


2. 公式推导

根据策略梯度算法的定义,策略优化目标函数如下:

L_\pi = \sum _{a\in A} log \pi_ \theta (s_t, a_t) (G_{t}^{n}-V(s_t))


3. 代码实现

Actor-Critic 模型部分的实现方式如下:

  1. import torch
  2. from torch import nn
  3. from torch.nn import functional as F
  4. import numpy as np
  5. # ------------------------------------ #
  6. # 策略梯度Actor,动作选择
  7. # ------------------------------------ #
  8. class PolicyNet(nn.Module):
  9. def __init__(self, n_states, n_hiddens, n_actions):
  10. super(PolicyNet, self).__init__()
  11. self.fc1 = nn.Linear(n_states, n_hiddens)
  12. self.fc2 = nn.Linear(n_hiddens, n_actions)
  13. # 前向传播
  14. def forward(self, x):
  15. x = self.fc1(x) # [b,n_states]-->[b,n_hiddens]
  16. x = F.relu(x)
  17. x = self.fc2(x) # [b,n_hiddens]-->[b,n_actions]
  18. # 每个状态对应的动作的概率
  19. x = F.softmax(x, dim=1) # [b,n_actions]-->[b,n_actions]
  20. return x
  21. # ------------------------------------ #
  22. # 值函数Critic,动作评估输出 shape=[b,1]
  23. # ------------------------------------ #
  24. class ValueNet(nn.Module):
  25. def __init__(self, n_states, n_hiddens):
  26. super(ValueNet, self).__init__()
  27. self.fc1 = nn.Linear(n_states, n_hiddens)
  28. self.fc2 = nn.Linear(n_hiddens, 1)
  29. # 前向传播
  30. def forward(self, x):
  31. x = self.fc1(x) # [b,n_states]-->[b,n_hiddens]
  32. x = F.relu(x)
  33. x = self.fc2(x) # [b,n_hiddens]-->[b,1]
  34. return x
  35. # ------------------------------------ #
  36. # Actor-Critic
  37. # ------------------------------------ #
  38. class ActorCritic:
  39. def __init__(self, n_states, n_hiddens, n_actions,
  40. actor_lr, critic_lr, gamma):
  41. # 属性分配
  42. self.gamma = gamma
  43. # 实例化策略网络
  44. self.actor = PolicyNet(n_states, n_hiddens, n_actions)
  45. # 实例化价值网络
  46. self.critic = ValueNet(n_states, n_hiddens)
  47. # 策略网络的优化器
  48. self.actor_optimizer = torch.optim.Adam(self.actor.parameters(), lr=actor_lr)
  49. # 价值网络的优化器
  50. self.critic_optimizer = torch.optim.Adam(self.critic.parameters(), lr=critic_lr)
  51. # 动作选择
  52. def take_action(self, state):
  53. # 维度变换numpy[n_states]-->[1,n_sates]-->tensor
  54. state = torch.tensor(state[np.newaxis, :])
  55. # 动作价值函数,当前状态下各个动作的概率
  56. probs = self.actor(state)
  57. # 创建以probs为标准类型的数据分布
  58. action_dist = torch.distributions.Categorical(probs)
  59. # 随机选择一个动作 tensor-->int
  60. action = action_dist.sample().item()
  61. return action
  62. # 模型更新
  63. def update(self, transition_dict):
  64. # 训练集
  65. states = torch.tensor(transition_dict['states'], dtype=torch.float)
  66. actions = torch.tensor(transition_dict['actions']).view(-1,1)
  67. rewards = torch.tensor(transition_dict['rewards'], dtype=torch.float).view(-1,1)
  68. next_states = torch.tensor(transition_dict['next_states'], dtype=torch.float)
  69. dones = torch.tensor(transition_dict['dones'], dtype=torch.float).view(-1,1)
  70. # 预测的当前时刻的state_value
  71. td_value = self.critic(states)
  72. # 目标的当前时刻的state_value
  73. td_target = rewards + self.gamma * self.critic(next_states) * (1-dones)
  74. # 时序差分的误差计算,目标的state_value与预测的state_value之差
  75. td_delta = td_target - td_value
  76. # 对每个状态对应的动作价值用log函数
  77. log_probs = torch.log(self.actor(states).gather(1, actions))
  78. # 策略梯度损失
  79. actor_loss = torch.mean(-log_probs * td_delta.detach())
  80. # 值函数损失,预测值和目标值之间
  81. critic_loss = torch.mean(F.mse_loss(self.critic(states), td_target.detach()))
  82. # 优化器梯度清0
  83. self.actor_optimizer.zero_grad() # 策略梯度网络的优化器
  84. self.critic_optimizer.zero_grad() # 价值网络的优化器
  85. # 反向传播
  86. actor_loss.backward()
  87. critic_loss.backward()
  88. # 参数更新
  89. self.actor_optimizer.step()
  90. self.critic_optimizer.step()

4. 案例演示

我们使用 OpenAI 的 gym 库中的环境,完成一个小案例。我们的目的是左右移动黑色小车使得黄色的杆子保持竖直。状态 state 的维度为 4,动作 action 有 2 个。

环境交互与训练部分的代码如下:

  1. import numpy as np
  2. import matplotlib.pyplot as plt
  3. import gym
  4. import torch
  5. from RL_brain import ActorCritic
  6. # ----------------------------------------- #
  7. # 参数设置
  8. # ----------------------------------------- #
  9. num_episodes = 100 # 总迭代次数
  10. gamma = 0.9 # 折扣因子
  11. actor_lr = 1e-3 # 策略网络的学习率
  12. critic_lr = 1e-2 # 价值网络的学习率
  13. n_hiddens = 16 # 隐含层神经元个数
  14. env_name = 'CartPole-v1'
  15. return_list = [] # 保存每个回合的return
  16. # ----------------------------------------- #
  17. # 环境加载
  18. # ----------------------------------------- #
  19. env = gym.make(env_name, render_mode="human")
  20. n_states = env.observation_space.shape[0] # 状态数 4
  21. n_actions = env.action_space.n # 动作数 2
  22. # ----------------------------------------- #
  23. # 模型构建
  24. # ----------------------------------------- #
  25. agent = ActorCritic(n_states=n_states, # 状态数
  26. n_hiddens=n_hiddens, # 隐含层数
  27. n_actions=n_actions, # 动作数
  28. actor_lr=actor_lr, # 策略网络学习率
  29. critic_lr=critic_lr, # 价值网络学习率
  30. gamma=gamma) # 折扣因子
  31. # ----------------------------------------- #
  32. # 训练--回合更新
  33. # ----------------------------------------- #
  34. for i in range(num_episodes):
  35. state = env.reset()[0] # 环境重置
  36. done = False # 任务完成的标记
  37. episode_return = 0 # 累计每回合的reward
  38. # 构造数据集,保存每个回合的状态数据
  39. transition_dict = {
  40. 'states': [],
  41. 'actions': [],
  42. 'next_states': [],
  43. 'rewards': [],
  44. 'dones': [],
  45. }
  46. while not done:
  47. action = agent.take_action(state) # 动作选择
  48. next_state, reward, done, _, _ = env.step(action) # 环境更新
  49. # 保存每个时刻的状态\动作\...
  50. transition_dict['states'].append(state)
  51. transition_dict['actions'].append(action)
  52. transition_dict['next_states'].append(next_state)
  53. transition_dict['rewards'].append(reward)
  54. transition_dict['dones'].append(done)
  55. # 更新状态
  56. state = next_state
  57. # 累计回合奖励
  58. episode_return += reward
  59. # 保存每个回合的return
  60. return_list.append(episode_return)
  61. # 模型训练
  62. agent.update(transition_dict)
  63. # 打印回合信息
  64. print(f'iter:{i}, return:{np.mean(return_list[-10:])}')
  65. # -------------------------------------- #
  66. # 绘图
  67. # -------------------------------------- #
  68. plt.plot(return_list)
  69. plt.title('return')
  70. plt.show()

绘制每回合的回报 return

标签:
声明

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

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

后台-插件-广告管理-内容页尾部广告(手机)
关注我们

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

搜索