策略梯度模型 (Policy Gradient)原理与实现

作者：陆平

1. 智能体与环境

策略梯度（Policy Gradient）模型是强化学习中的一个经典基础模型。它用来在某种环境下训练智能体。对于一些简单场景，我们可以把智能体与环境及关系抽象为：

智能体是在环境中实施行为的实体。它与环境的交互下图所示。首先一个episode开始，环境初始化后将释放出一个状态信息 s 1 s1 s1。智能体观测到状态信息 s 1 s1 s1，依据该信息做思考和决策，得到行为 a 1 a1 a1。环境再根据智能体的行为 a 1 a1 a1,释放出状态信息 s 2 s2 s2。智能体观测到状态信息 s 2 s2 s2，经过自身决策得到行为 a 2 a2 a2。该过程一直持续下去，直到该episode结束。

理解智能体与环境的关系，需要理解episode这个词。它的中文直译是“一段经历”，其实不太好理解。为了更通俗，举以下几个例子；

第一个例子是打游戏。大家应该都玩过超级玛丽，超级玛丽在前进过程中会遇到各种不同的怪兽和陷阱，一不小心就会被吃掉或掉进陷阱，厉害的玩家可以持续玩很长一段时间而不会挂掉，直到打通关。从开始到挂掉或通关这段时间便是一段episode。

第二个例子是历史中朝代更迭。历史中的很多朝代都想尽可能长的存续下来。然而，社会环境、自然环境都处于不断变化，很难通过某种单一策略保持永固。历史车轮滚滚向前，时代潮流浩浩荡荡，只有不断调整策略适应环境，才能上时代步伐，与时俱进。每轮更迭，便可视为一段episode。

第三个例子是探寻宝藏。对于一些特定寻宝或解谜任务，智能体并不追求在更长时间内存活下来，而是找到某个宝藏或解决某个问题，这种便要求在更短时间内完成越好。每次从开始到找到宝藏或是失败，都可以视为一段episode。

《黑客帝国》里有个名句“Everything that has a begin has an end.”翻译成中文便是“世间万物有始皆有终”，一段episode即有始有终。

2. 策略梯度模型推导

采用数学方式描述上述过程：

假设在一个episode中，智能体与环境交互所产生的状态信息和行为形成一个集合，称为迹。

τ = { s 1 , a 1 , s 2 , a 2 , . . . , s T n , a T n } \tau = \{ s_{1},a_{1},s_{2},a_{2},...,s_{T_{n}},a_{T_{n}}\} τ={s1,a1,s2,a2,...,sTn,aTn}

对于某个迹而言，其发生的可能性可以表示为一系列条件概率的乘积。

P θ ( τ ) = P ( s 1 ) P θ ( a 1 ∣ s 1 ) P ( s 2 ∣ s 1 , a 1 ) P θ ( a 2 ∣ s 2 ) P ( s 3 ∣ s 2 , a 2 ) . . . P_{\theta}(\tau) = P(s_{1})P_{\theta}(a_{1}|s_{1})P(s_{2}|s_{1},a_{1})P_{\theta}(a_{2}|s_{2})P(s_{3}|s_{2},a_{2})... Pθ(τ)=P(s1)Pθ(a1∣s1)P(s2∣s1,a1)Pθ(a2∣s2)P(s3∣s2,a2)...

其中，θ是智能体决策参数，可以当做智能体自己的神经网络中的权重。

P ( s 1 ) P(s_{1}) P(s1)表示环境产生第一个状态的概率。 P θ ( a 1 ∣ s 1 ) P_{\theta}(a_{1}|s_{1}) Pθ(a1∣s1)表示智能体根据环境的第一个状态做出自己的行动，它可以看成是一个条件概率，其条件为环境的第一个状态。 P ( s 2 ∣ s 1 , a 1 ) P(s_{2}|s_{1},a_{1}) P(s2∣s1,a1)表示环境产生第二个状态的条件概率，条件是环境的第一个状态和智能体的做出的行动。

用连乘符号，上面这个式子可以表示为：

P θ ( τ ) = P ( s 1 ) ∏ t = 1 T n P θ ( a t ∣ s t ) P ( s t + 1 ∣ s t , a t ) P_{\theta}(\tau) = P(s_{1})\prod_{t=1}^{T_{n}}P_{\theta}(a_{t}|s_{t})P(s_{t+1}|s_{t},a_{t}) Pθ(τ)=P(s1)∏t=1TnPθ(at∣st)P(st+1∣st,at)

对于某智能体而言，在多个episode中，得到的迹可能是不同的。因此，训练智能体的最终目标是要最大化在各个迹上的期望奖励。

R ˉ θ = ∑ τ R ( τ ) P θ ( τ ) \bar{R}_{\theta} = \sum_{\tau}R(\tau)P_{\theta}(\tau) Rˉθ=∑τR(τ)Pθ(τ)

又可以表示为：

R ˉ θ = E τ ∼ P θ ( τ ) [ R ( τ ) ] \bar{R}_{\theta} = E_{\tau\sim P_{\theta}(\tau)}[R(\tau)] Rˉθ=Eτ∼Pθ(τ)[R(τ)]

为了获得最大化的期望奖励，我们可以对上式进行求梯度。在实操的时候，损失函数取最大化函数的负函数：

▽ R ˉ θ = ∑ τ R ( τ ) ▽ P θ ( τ ) = ∑ τ R ( τ ) P θ ( τ ) ▽ l o g P θ ( τ ) \triangledown \bar{R}_{\theta}=\sum_{\tau}R(\tau)\triangledown P_{\theta}(\tau)=\sum_{\tau}R(\tau) P_{\theta}(\tau) \triangledown log P_{\theta}(\tau) ▽Rˉθ=∑τR(τ)▽Pθ(τ)=∑τR(τ)Pθ(τ)▽logPθ(τ)

上式也可以表示为期望形态：

R ˉ θ = E τ ∼ P θ ( τ ) [ R ( τ ) ▽ l o g P θ ( τ ) ] \bar{R}_{\theta} = E_{\tau\sim P_{\theta}(\tau)}[R(\tau) \triangledown log P_{\theta}(\tau)] Rˉθ=Eτ∼Pθ(τ)[R(τ)▽logPθ(τ)]

表示成为期望形态后，就意味着可以采用抽样的方法，从分布 P θ ( τ ) P_{\theta}(\tau) Pθ(τ)中抽取N个样本，然后求平均，便可以近似地表示期望。

R ˉ θ ≈ 1 N ∑ n = 1 N [ R ( τ n ) ▽ l o g P θ ( τ n ) ] \bar{R}_{\theta} \approx \frac{1}{N} \sum_{n=1}^{N} [R(\tau^{n}) \triangledown log P_{\theta}(\tau^{n})] Rˉθ≈N1∑n=1N[R(τn)▽logPθ(τn)]

再展开 ▽ l o g P θ ( τ ) \triangledown log P_{\theta}(\tau) ▽logPθ(τ),得到：

▽ l o g P θ ( τ ) = ∑ τ = 1 T n ▽ l o g P θ ( a t n ∣ s t n ) \triangledown log P_{\theta}(\tau)=\sum_{\tau =1}^{T_{n}} \triangledown log P_{\theta}(a_{t}^{n}|s_{t}^{n}) ▽logPθ(τ)=∑τ=1Tn▽logPθ(atn∣stn)

求得智能体的策略梯度为：

R ˉ θ ≈ 1 N ∑ n = 1 N ∑ t = 1 T n [ R ( τ n ) ▽ l o g P θ ( a t n ∣ s t n ) ] \bar{R}_{\theta} \approx \frac{1}{N} \sum_{n=1}^{N} \sum_{t=1}^{T_{n}} [R(\tau^{n}) \triangledown log P_{\theta}(a_{t}^{n}|s_{t}^{n})] Rˉθ≈N1∑n=1N∑t=1Tn[R(τn)▽logPθ(atn∣stn)]

每次计算完梯度后，采用下式便可以更新智能体的决策参数 θ \theta θ（即神经网络中的权重）。

θ ← θ + η ▽ R ˉ θ \theta \leftarrow \theta + \eta \triangledown \bar{R}_{\theta} θ←θ+η▽Rˉθ

其中， η \eta η为学习率。

以上即为基础的策略梯度算法。为提升学习效果和效率，可以选择做以下改进。

一是增加基线baseline。基线可以有很多种，比如可以是奖励的期望值等。

R ˉ θ ≈ 1 N ∑ n = 1 N ∑ t = 1 T n [ ( R ( τ n ) − b a s e l i n e ) ▽ l o g P θ ( a t n ∣ s t n ) ] \bar{R}_{\theta} \approx \frac{1}{N} \sum_{n=1}^{N} \sum_{t=1}^{T_{n}} [(R(\tau^{n})-baseline) \triangledown log P_{\theta}(a_{t}^{n}|s_{t}^{n})] Rˉθ≈N1∑n=1N∑t=1Tn[(R(τn)−baseline)▽logPθ(atn∣stn)]

二是用当前时刻至episode结束的奖励之和，作为当前智能体行为的奖励。举例说明，假设[2,3,4,6]是智能体在第n个episode中迹的片段（到episode结束），该片段是按产生时刻先后进行排序，分别对应于时刻[t1, t2, t3, t4]，智能体的行为[a1, a2, a3, a4]。那么t1时刻的智能体产生行为a1的奖励为2+3+4+6=15，t2时刻的智能体产生行为a2的奖励为3+4+6=13，t3时间的智能体产生行为a3的奖励为4+6=10，t4时间的智能体产生行为a4的奖励为6。

R ˉ θ ≈ 1 N ∑ n = 1 N ∑ t = 1 T n [ ( ∑ t ′ = t T n r t ′ n − b a s e l i n e ) ▽ l o g P θ ( a t n ∣ s t n ) ] \bar{R}_{\theta} \approx \frac{1}{N} \sum_{n=1}^{N} \sum_{t=1}^{T_{n}} [(\sum_{t'=t}^{T_{n}}r_{t'}^{n}-baseline) \triangledown log P_{\theta}(a_{t}^{n}|s_{t}^{n})] Rˉθ≈N1∑n=1N∑t=1Tn[(∑t′=tTnrt′n−baseline)▽logPθ(atn∣stn)]

三是对奖励按时间进行折扣。还是举上段的例子，假设折扣率为0.9，t1时刻的智能体产生行为a1的奖励为2+3×0.9+4×0.9×0.9+6×0.9×0.9×0.9=9.4。

R ˉ θ ≈ 1 N ∑ n = 1 N ∑ t = 1 T n [ ( ∑ t ′ = t T n g a m m a t ′ − t r t ′ n − b a s e l i n e ) ▽ l o g P θ ( a t n ∣ s t n ) ] \bar{R}_{\theta} \approx \frac{1}{N} \sum_{n=1}^{N} \sum_{t=1}^{T_{n}} [(\sum_{t'=t}^{T_{n}}gamma^{t'-t}r_{t'}^{n}-baseline) \triangledown log P_{\theta}(a_{t}^{n}|s_{t}^{n})] Rˉθ≈N1∑n=1N∑t=1Tn[(∑t′=tTngammat′−trt′n−baseline)▽logPθ(atn∣stn)]

其中，gamma是折扣因子（取值范围0至1之间）。

3. 平衡杆环境（gym）

平衡杆环境中，杆子的一头连接在小车上（连接处可以转动），小车可沿着无摩擦的轨道移动，随机给定杆子的初始状态，智能体可以通过向左或向右移动小车，来防止杆子倒下。小车的移动速度方向和大小，取决于杆子的角度和杆子顶端的速度。杆子重心的偏离程度越大，恢复平衡所需要的能量也就越高。举个生活中类似的例子，手上竖立一根棍子，要保证棍子不倒下，人需要根据棍子倾斜的角度和速度，来控制手的移动方向和力度大小。

平衡杆环境的观测空间（连续值）

编号	物理含义	取值范围
1	小车所处的位置	-4.8至4.8
2	小车的速度	-inf至inf
3	杆子的角度	-24度至24度
4	杆子顶端的速度	-inf至inf

智能体的行动空间（离散值）

值	含义
0	向左移动小车
1	向右移动小车

4. 应用策略梯度模型训练平衡杆智能体

应用PaddlePaddle2.0，构建一个应用策略

import os
from itertools import count
import gym
import paddle
import paddle.nn.functional as F
from paddle.distribution import Categoricalclass PolicyAgent(paddle.nn.Layer):def __init__(self, obs_space, action_space):#初始化super(PolicyAgent, self).__init__()#设样本批量数为batch，输入形状大小为[batch, obs_space]，输出形状大小为[batch, 16]self.linear1 = paddle.nn.Linear(obs_space, 16)#输入形状大小为[batch, 16]，输出形状大小为[batch, action_space]self.linear2 = paddle.nn.Linear(16, action_space)#log_probs列表用来保存历次行为的概率self.log_probs = []#rewards列表用来保存智能体每次与环境交互所产生的回报self.rewards = []def forward(self, x):#前向传播过程#把输入x转化为tensor格式数据，并转成浮点数。x = paddle.to_tensor(x, dtype="float32")#输入x经过linear1网络层运算，再接上relu激活函数x = F.relu(self.linear1(x))#再经过linear2网络层运算，然后在最后一个维度上接softmax函数action_probs = F.softmax(self.linear2(x), axis=-1)#把上述的行动概率，转成分类分布action_distribution = Categorical(action_probs)#从分类分布中进行抽样action = action_distribution.sample([1])#根据上述抽样的结果，求解该抽样结果的概率，对概率进行log运算#每次得到的结果，追加到log_probs列表之中self.log_probs.append(action_distribution.log_prob(action))#返回行动值return action.numpy().item()def loss_calculate(self, gamma):#损失函数rewards = []dis_reward = 0log_probs = self.log_probsfor reward in self.rewards[::-1]: #把rewards倒序排列#计算折扣回报率dis_reward = reward + gamma * dis_reward#每次从0位置插入，后进来的数排在前面。rewards.insert(0, dis_reward)#把rewards转化为tensor，并对rewards进行标准化rewards = paddle.to_tensor(rewards)rewards = (rewards - rewards.mean()) / (rewards.std())loss = 0for logprob, reward in zip(log_probs, rewards):#参考算法推导部分action_loss = -logprob * rewardloss += action_lossreturn lossdef clear_memory(self):#清除智能体记忆的函数del self.rewards[:]del self.log_probs[:]def train():#创建平衡杆cartpole环境env = gym.make('CartPole-v1')env = env.unwrappedenv.seed(1)paddle.seed(1)obs_space = env.observation_space.shape[0] #状态空间，数值为4action_space = env.action_space.n #行动空间，数值为2#设置超参数gamma = 0.9 #折扣率episodes = 500 #设定episodes轮数，即让智能体一共玩多少回游戏#模型初始化model = PolicyAgent(obs_space, action_space)#优化器初始化optimizer = paddle.optimizer.Adam(learning_rate=0.02, parameters=model.parameters())#开始训练duration = [] #记录每轮eposide所持续的时间for i in range(episodes):state = env.reset() #state的类型是ndarrayfor t in count():#一直不断地循环action = model(state)#输入状态，输出智能体选择的行动#环境会根据智能体的动作，输出下一个状态、回报、完成等信息state, reward, done, info = env.step(action)model.rewards.append(reward)if done or t >= 100: #episode终止或玩游戏持续次数超过100break #跳出本局循环duration.append(t) #把本局的持续时间t,追加到持久期里面if i % 250 == 0:#每过250次保存模型paddle.save(model.state_dict(), './lup/'+str(i)+'.pdparams')#清零梯度optimizer.clear_grad()#生成损失函数loss = model.loss_calculate(gamma)#对损失函数求梯度，反向传播loss.backward()#优化器用反向传播计算出来的梯度，对参数进行更新optimizer.step()#清空智能体记忆model.clear_memory()return duration#训练智能体
if __name__ == '__main__':duration = train()

#打印出结果，看智能体在玩平衡杆游戏中所能持续的时间
import matplotlib.pyplot as pltplt.plot(duration)
plt.show()

想运行代码，可以点击进入AI studio项目：

基于PaddlePaddle2.0-构建策略梯度模型