上一部分我们介绍了RL的各种算法分类(主要是policy based)

下面针对policy based 方法的基础,也就是policy gradient,spinningup 做了比较详实的阐述。

这部分我们将讨论policy optimization算法的数学表达式,我们将介绍policy gradients的三个重要结果:

  • 最简单的表达式描述了关于策略参数的Policy performance的梯度
  • 允许我们从表达式中删除无用术语的规则,
  • 以及允许我们向该表达式添加有用术语的规则。
看文字很绕嘴,,,大意就是从最基础的公式进行推导,并且把不易计算的部分替换成容易计算的。

Deriving the Simplest Policy Gradient

考虑一个随机参数化的策略 [公式] ,我们希望最大化期望回报 [公式] ,我们将采用 [公式] 来给出有限非折扣回报,但无限范围折扣回报设置的推导几乎相同。

通过梯度上升提升策略:

[公式]

[公式] 称为policy gradient,并且优化策略的算法称为policy gradient 算法, 比如VPG,TRPO,PPO,尽管PPO有一点不太符合,但也是policy gradient方法

要实际使用该算法,我们需要一个可以数值计算的policy gradient 表达式。 这包括两个步骤:

  1. 导出policy performance 的分析梯度,结果证明它具有期望值的形式,
  2. 然后形成该期望值的样本估计,可以用有限数量 agent-env交互步骤的数据进行计算。
虽然这个式子很明确,但是很难直接用于计算,必须把它细化成可以用交互数据进行计算的形式。

我们将首先列出一些对推导分析梯度有用的事实。

1. Probability of a Trajectory.

轨迹 [公式] 的概率,由来自 [公式] 的动作给出:
[公式]

分别是 从初始状态分布中进行采样,然后依靠策略 [公式] 选择动作,再根据转移概率确定下个状态,循环直到轨迹结束。

2. The Log-Derivative Trick.
[公式]

一个简单的对数变换形式,就是源于 [公式] ,把 [公式] 挪过去就是这个形式

3. Log-Probability of a Trajectory.

[公式]

就是把第一条中的轨迹概率加一个log, 把乘法转换成了加法,因[公式]

4. Gradients of Environment Functions.

环境不依赖于 [公式] ,所以 [公式]  [公式]  [公式] 的梯度为零。

这里对应的就是一开始所说的,删除掉无用的部分,因为求解策略梯度的目的是为了提升策略的performance而优化策略的参数,所以和参数 [公式] 无关的项,全部删掉。

5.Grad-Log-Prob of a Trajectory.

这样我们就得到了轨迹概率的对数梯度。

因为我们要优化的是策略performance,也就是使用策略 [公式] 产生的轨迹 [公式] 的奖励,我们需要优化奖励,所以就有了第1行,我们对在策略 [公式] 下采样的 [公式] 的期望奖励 [公式] 求梯度。
接着第2行 我们把期望写成求和的形式,应用连续情况下的期望公式 [公式] , 这里把 [公式] 拿出来乘进去,其实就是对产生轨迹的概率P求和.
第三行把梯度放进了公式内.
第四行应用了第2条的The log-derivative trick.
第五行再把积分反写成期望形式,吸收了P.
第六行将概率P带入,也就是第5条事实的式子.

这是一个期望,这意味着我们可以用样本均值来估计它。 如果我们收集一组轨迹 [公式] 其中每个轨迹是通过让agent使用策略 [公式] ,策略梯度可以估计为

[公式]

用经验代替期望

其中D是轨迹的条数,最后这个表达式是我们想要可计算表达式的最简单版本。 假设允许我们计算 [公式] 的方式表示我们的策略,并且如果我们能够在环境中运行该策略以收集轨迹数据集,我们可以计算策略梯度并采取更新步骤。

Implementing the Simplest Policy Gradient

下面我们使用上式推导出的最简单的policy gradient version。

1. Making the Policy Network.

 # make core of policy network   
 logits_net = mlp(sizes=[obs_dim]+hidden_sizes+[n_acts]) #构建logits mlp网络, 分别是 in hidden  out
 ​
 # make function to compute action distribution
 def get_policy(obs):
     logits = logits_net(obs)  #输入环境得到的obs,输出Logits
     return Categorical(logits=logits)  #通过Categorical得到各个动作的概率
 ​
 # make action selection function (outputs int actions, sampled from policy)
 def get_action(obs):
     return get_policy(obs).sample().item()

此块构建模块和函数以使用前馈神经网络分类策略(feedforward neural network categorical policy)。 logits_net 的输出可用于构建动作的对数-概率和概率值,get_action 函数根据从 logits 计算的概率对动作进行采样。 (注意:这个特定的 get_action 函数假设只提供一个 obs,因此只有一个整数动作输出。这就是它使用 .item() 的原因,它用于获取只有一个元素的 Tensor 的内容。)

其中 Categorical 对象是一个 PyTorch Distribution 对象,它包含了一些与概率分布相关的数学函数。 特别是,它有一种从分布中采样的方法和一种计算给定样本的对数概率的方法(我们稍后会使用)。

当我们谈论具有“logits”的分类分布时,我们的意思是每个结果的概率由 logits 的 Softmax 函数给出。 也就是说,在 logits x_j 的分类分布下,动作 j 的概率为:
[公式]

2. Making the Loss Function.

 # make loss function whose gradient, for the right data, is policy gradient
 def compute_loss(obs, act, weights):#这里的weights实际上就是奖励信号。
     logp = get_policy(obs).log_prob(act) 
     return -(logp * weights).mean() #为什么是负的? 因为调loss的API时,优化方向是下降方向,而我们想要上升方向

在这个模块中,我们为策略梯度算法构建了一个“损失”函数。 当插入正确的数据时,这个损失的梯度等于策略梯度。 正确的数据意味着根据当前策略收集的一组(obs、action、weights)元组

尽管我们将其描述为损失函数,但它并不是监督学习中典型意义上的损失函数。 与标准损失函数有两个主要区别。

  1. 数据分布取决于参数。 损失函数通常定义在一个固定的数据分布上,它与我们要优化的参数无关。 此处并非如此,必须根据最新策略对数据进行采样。
  2. 它不评估性能。 损失函数通常评估我们关心的性能指标。 这里我们关心预期回报 J(\pi_{\theta}),但我们的“损失”函数根本没有近似这个。 这个“损失”函数只对我们有用,因为当在当前参数下评估时,使用当前参数生成的数据,它具有负的性能梯度。

但是在梯度下降的第一步之后,与性能没有更多的联系。 这意味着对于给定的一批数据,最小化这个“损失”函数并不能保证提高预期回报。 您可以将此损失设置到 [公式] 并且策略性能可能会下降; 事实上,它通常会。 有时,深度 RL 研究人员可能会将这种结果描述为对一批数据的策略“过度拟合”。 这是描述性的,但不应从字面上理解,因为它不涉及泛化错误。

我们提出这一点是因为机器学习从业者在训练期间将损失函数解释为有用的信号是很常见的——“如果损失下降,一切都很好。” 在策略梯度中,这种直觉是错误的,你应该只关心平均回报。 损失函数没有任何意义。

这段描述的是,尽管使用了对数概率作为loss来进行优化,但这并不意味着仅仅优化loss,loss下降的很好,策略就有所提升,我们只是通过这种方法,来优化参数,loss并不是参数优化好坏的度量手段,这是与传统机器学习不同的地方。

此处用于生成 logp 张量的方法(调用 PyTorch Categorical 对象的 log_prob 方法)可能需要进行一些修改才能与其他类型的分布对象一起使用。

例如,如果您使用正态分布(用在高斯策略),调用 policy.log_prob(act) 的输出将为您提供一个张量,其中包含每个向量值动作的每个分布的单独对数概率。 也就是说,你输入一个形状为 (batch, act_dim) 的张量,然后取出一个形状为 (batch, act_dim) 的张量,此时你需要做一个 RL 损失是一个形状为 (batch,) 的张量。 在这种情况下,可以计算:

 logp = get_policy(obs).log_prob(act).sum(axis=-1)
接下来,为了更好地理解这段代码的内容,做了如下一个测试。


在CartPole-v0环境中,动作是01离散的,也就是左或者右 ,obs每一step都是4维的观察向量,weights是记录了reward信息
此时需要计算loss


首先调用get_policy中的动作,根据loss的式子:
[公式]

我们首先要拿到策略动作的对数概率。
通过如下所示的logits_net网络


可以看到最终输出节点为2,也就是动作维度(0左1右)的logits,需要注意的是,我们最终输出的实际上是1个动作,这就意味着,如果换做别的环境,假设 需要输出动作维度为4,每个维度又有4种可能,那这种情况下采用这中离散的categorical是不容易实现的,这时候就要使用高斯策略了,假设动作的输出维度为4,那网络的末尾节点就设置为4,每个节点代表动作的mean均值,再依靠动作的方差sigma来进行采样(如前所示,sigma也可以使用与状态相关, 也可以不与状态相关,这样也就意味着也可以直接加入到末尾节点进行训练),就可以解决这个问题了。


现在这个值还没有经过softmax,使用Categorical, 为了方便查看,我们把logp的那一句拆开来看


可以看到logp,也就是categorical返回类的内容


其中probs就是经过了softmax的内容。
这里再通过log_prob方法,让categorical自动计算


最后乘weights再取均值,就是优化的loss。
需要注意的是,我们两次使用了policy 网络,一次就是我们在更新阶段,需要使用policy来计算对数概率,另外一次就是在采样的过程中,对于当前观察值,依据策略给出动作,也就是下面的情况:


环境在采样时,观察到4维,然后经过policy_net得到logits


最后经过categorical的sample,也就是softmax,来确定概率 而选择动作


为了进一步观察,我们把get_policy(obs).sample().item()拆开


可以看到 此时的value,probs已经计算出,显然是动作0的概率大于动作1的概率。


而依据概率选择,系统还是随机选择了动作1,尽管动作0的选择概率更大,这就是随机概率(stochastic policy)和确定性策略(deterministic policy)的区别,就是尽管从数值上看动作0稍好,系统选取较差动作的可能也是存在的吗,其优点就是确实增加了探索性,但在性能表现上可能不会很稳定。而deterministic policy由于不存在这个特性,所以要考虑如何增加探索性。


最后.item()就是对tensor取值


最后 训练了50个epoch的数据情况:

最后我们给出代码

 #import 各种库
 import torch
 import torch.nn as nn
 from torch.distributions.categorical import Categorical
 from torch.optim import Adam
 import numpy as np
 import gym
 from gym.spaces import Discrete, Box
 ​
 def mlp(sizes, activation=nn.Tanh, output_activation=nn.Identity):
     # Build a feedforward neural network. 建立网络
     layers = []
     for j in range(len(sizes)-1):
         act = activation if j < len(sizes)-2 else output_activation
         layers += [nn.Linear(sizes[j], sizes[j+1]), act()]
     return nn.Sequential(*layers)
 ​
 def train(env_name='CartPole-v0', hidden_sizes=[32], lr=1e-2, 
           epochs=50, batch_size=5000, render=False):
 ​
     # make environment, check spaces, get obs / act dims
     env = gym.make(env_name)
     assert isinstance(env.observation_space, Box), \
         "This example only works for envs with continuous state spaces."
     assert isinstance(env.action_space, Discrete), \
         "This example only works for envs with discrete action spaces."
 ​
     obs_dim = env.observation_space.shape[0]
     n_acts = env.action_space.n
 ​
     # make core of policy network
     logits_net = mlp(sizes=[obs_dim]+hidden_sizes+[n_acts])
 ​
     # make function to compute action distribution
     def get_policy(obs):
         logits = logits_net(obs)
         return Categorical(logits=logits)
 ​
     # make action selection function (outputs int actions, sampled from policy)
     def get_action(obs):
         return get_policy(obs).sample().item()
 ​
     # make loss function whose gradient, for the right data, is policy gradient
     def compute_loss(obs, act, weights):
         logp = get_policy(obs).log_prob(act)
         return -(logp * weights).mean()
 ​
     # make optimizer
     optimizer = Adam(logits_net.parameters(), lr=lr)
 ​
     # for training policy
     def train_one_epoch():
         # make some empty lists for logging.
         batch_obs = []          # for observations
         batch_acts = []         # for actions
         batch_weights = []      # for R(tau) weighting in policy gradient
         batch_rets = []         # for measuring episode returns
         batch_lens = []         # for measuring episode lengths
 ​
         # reset episode-specific variables
         obs = env.reset()       # first obs comes from starting distribution
         done = False            # signal from environment that episode is over
         ep_rews = []            # list for rewards accrued throughout ep
 ​
         # render first episode of each epoch
         finished_rendering_this_epoch = False
 ​
         # collect experience by acting in the environment with current policy
         while True:
 ​
             # rendering
             if (not finished_rendering_this_epoch) and render:
                 env.render()
 ​
             # save obs
             batch_obs.append(obs.copy())
 ​
             # act in the environment
             act = get_action(torch.as_tensor(obs, dtype=torch.float32))
             obs, rew, done, _ = env.step(act)
 ​
             # save action, reward
             batch_acts.append(act)
             ep_rews.append(rew)
 ​
             if done:
                 # if episode is over, record info about episode
                 ep_ret, ep_len = sum(ep_rews), len(ep_rews)
                 batch_rets.append(ep_ret)
                 batch_lens.append(ep_len)
 ​
                 # the weight for each logprob(a|s) is R(tau)
                 batch_weights += [ep_ret] * ep_len
 ​
                 # reset episode-specific variables
                 obs, done, ep_rews = env.reset(), False, []
 ​
                 # won't render again this epoch
                 finished_rendering_this_epoch = True
 ​
                 # end experience loop if we have enough of it
                 if len(batch_obs) > batch_size:
                     break
 ​
         # take a single policy gradient update step
         optimizer.zero_grad()
         batch_loss = compute_loss(obs=torch.as_tensor(batch_obs, dtype=torch.float32),
                                   act=torch.as_tensor(batch_acts, dtype=torch.int32),
                                   weights=torch.as_tensor(batch_weights, dtype=torch.float32)
                                   )
         batch_loss.backward()
         optimizer.step()
         return batch_loss, batch_rets, batch_lens
 ​
     # training loop
     for i in range(epochs):
         batch_loss, batch_rets, batch_lens = train_one_epoch()
         print('epoch: %3d \t loss: %.3f \t return: %.3f \t ep_len: %.3f'%
                 (i, batch_loss, np.mean(batch_rets), np.mean(batch_lens)))
 ​
 if __name__ == '__main__':
     import argparse
     parser = argparse.ArgumentParser()
     parser.add_argument('--env_name', '--env', type=str, default='CartPole-v0')
     parser.add_argument('--render', action='store_true')
     parser.add_argument('--lr', type=float, default=1e-2)
     args = parser.parse_args()
     print('\nUsing simplest formulation of policy gradient.\n')
     train(env_name=args.env_name, render=args.render, lr=args.lr)

Expected Grad-Log-Prob Lemma

在本小节中,我们将推导出一个在整个策略梯度理论中广泛使用的中间结果。 我们将其称为预期梯度对数概率 (EGLP) 引理。 [1]

EGLP 引理。 假设 [公式] 是随机变量 x 上的参数化概率分布。 那么:

[公式]

Don’t Let the Past Distract You

检查我们最近的策略梯度表达式:

[公式]

使用这个梯度来进行可以发现,我们使用了整个的轨迹 [公式] 进行了更新,也就是曾经获得的所有奖励的总和。 但这没有多大意义。 agent更新时在采取行动之前获得的奖励与该行动的好坏无关。

[公式]

这里的原始形式是
[公式]

这里做的改变就是把reward的t改成了当前时刻t`,丢掉了当前时刻动作之前的奖励,就如前所说,agent当前的动作好坏与过去奖励无关。

我们称这个形式为“reward-to-go policy gradient”

[公式]

策略梯度的一个关键问题是需要多少样本轨迹才能为它们获得低方差样本估计。 我们开始的公式包括与过去奖励成正比的强化的行动,所有这些都具有零均值,但非零方差:因此,它们只会在策略梯度的样本估计中添加噪声。 通过移除它们,我们减少了所需的样本轨迹数量。

通过上面移除了过去的奖励部分,降低了整体的方差。

Implementing Reward-to-Go Policy Gradient

与上述代码不同的是 我们使用了在loss中的不同的weights(对数概率乘reward的那部分)

 def reward_to_go(rews):
     n = len(rews)
     rtgs = np.zeros_like(rews)
     for i in reversed(range(n)):
         rtgs[i] = rews[i] + (rtgs[i+1] if i+1 < n else 0)
     return rtgs

Baselines in Policy Gradient

EGLP 引理的直接结果是对于任何仅依赖于状态的函数 b

[公式]

这就使得:

[公式]

b被称为baseline

这里把log的部分拆开来,就可以得到关于baseline的部分,这部分期望为0,所以整体期望不变

最常见的baseline选择是 on-policy 价值函数 [公式] 。如果agent从状态 [公式] 开始,然后在之后根据策略 [公式] 行动,则这是平均回报。

根据经验,选择 [公式] 具有减少策略梯度样本估计方差的理想效果。 这导致更快、更稳定的策略学习。 从概念的角度来看,它编码了一种直觉,即如果agent得到了它的奖励预期,它应该“感觉”中立。

在实际情况下, [公式] 无法准确计算,因此必须对其进行近似。 这通常是通过神经网络 [公式] 完成的,它与策略同时更新(以便价值网络始终接近最新策略的价值函数)。

在大多数策略优化算法(包括 VPG、TRPO、PPO 和 A2C)实现中使用的学习 [公式] 的最简单方法是最小化均方误差目标:

[公式]

这部分描述的是如何选择一个合适的b(s),选择状态值函数是一个不错的选择,因为它代表了当前动作的平均水平,而真实的状态值函数是无法获得的,需要使用近似器来逼近,actor-critic的算法正是为了解决这个问题而提出的,AC也成了很多policy based 算法的最基础的component了。

Other Forms of the Policy Gradient

到目前为止我们看到的是策略梯度具有一般形式

[公式]

其中 [公式] 可以是任何轨迹奖励。

[公式]

或是

[公式]

或是

[公式]

尽管有不同的方差,但所有这些选择都会导致策略梯度的期望相同。 事实证明,还有两个更有效的权重选择 [公式] ,这些选择很重要

1. On-Policy Action-Value Function.

可以使用状态动作值函数

[公式]

2. The Advantage Function.

也可以使用优势函数来进行估计

回想一下,由

[公式]

定义的动作优势描述了 它比其他行动平均(相对于当前政策)更好或更坏的程度 。 这个选择 也是可以的,

而如何更好的使用优势函数进行估计,可以参考 Generalized Advantage Estimation GAE 泛化优势估计,它被广泛用于VPG,TRPO 和PPO中。

在下面的文章中我也进行了详细的推导。。。

至此关于Policy gradient 的一些基础就介绍完毕了。