PPO

摘要：

我们提出了一种新的强化学习策略梯度方法，它通过与环境的交互在采样数据之间交替，并使用随机梯度上升优化替代目标函数。虽然标准的策略梯度方法对每个数据样本执行一次梯度更新，但我们提出了一个新的目标函数，它可以实现多个时期的小批量更新。这些新方法被称为“最近策略优化”(proximal policy optimization, PPO)，具有信任区域策略优化(TRPO)的一些优点，但它们实现起来更简单、更通用，并且(从经验上)具有更好的样本复杂度。我们的实验在一组基准任务上测试PPO，包括模拟机器人移动和玩Atari游戏，我们表明PPO优于其他在线策略梯度方法，并且在样本复杂性、简单性和实际时间之间取得了良好的平衡。

1简介：

近年来，几种不同的方法被提出用于强化学习的神经网络函数逼近器。领先的竞争者有deep Q-learning[Mni+15]、vanilla政策梯度方法[Mni+16]和信任区域/自然政策梯度方法[Sch+15b]。然而，在开发一种可伸缩(针对大型模型和并行实现)、数据高效和健壮(即，在不进行超参数调优的情况下成功解决各种问题)的方法方面，还有改进的空间。Q-learning(带有函数逼近)在许多简单问题上1都失败了，人们对它的理解也很差，一般的政策梯度方法数据效率和鲁棒性都很差;而信任区域策略优化(TRPO)相对复杂，与包含噪声(如dropout)或参数共享(策略与价值函数之间，或与辅助任务之间)的架构不兼容。

本文试图通过引入一种算法来改善目前的情况，该算法在只使用一阶优化的情况下，能够获得TRPO的数据效率和可靠性能。我们提出了一个新的目标与修剪概率比率，形成一个悲观估计(即，下界)的政策的表现。为了优化策略，我们在从策略中采样数据和对采样数据执行几次优化之间交替进行。

我们的实验比较了代理目标的不同版本的性能，发现具有剪切概率比的版本性能最好。我们也比较PPO与几个以前的算法从文献。在连续控制任务上，它的性能优于我们所比较的算法。在Atari上，它的性能(就样本复杂度而言)明显优于A2C，与ACER相似，但简单得多。

1虽然DQN在具有离散动作空间的街机学习环境[Bel+15]等游戏环境中表现良好，但在OpenAI Gym [Bro+16]和Duan等人[Dua+16]所描述的连续控制基准中，它并没有被证明表现良好

2背景知识：策略优化

2.1 Policy Gradient Methods

策略梯度方法的工作原理是计算策略梯度的估计量，并将其插入随机梯度上升算法中。最常用的梯度估计器有：

 

其中πθ是随机策略，而At是时间步长t处优势函数的估计量。在这里，期望Et[...]表示在抽样和优化之间交替的算法中，有限批次样本的经验平均值。使用自动微分软件的实现方法是构造一个目标函数，其梯度是策略梯度估计量；估计量g是通过对目标进行微分得到

 

虽然使用相同的轨迹对这个损失lpg执行多个优化步骤是有吸引力的，但是这样做是不合理的，而且从经验上讲，这通常会导致破坏性的大的策略更新（见第6.1节；结果未显示，但与“无剪裁”或“惩罚”设置相似或更差）。

2.2Trust Region Methods

在TRPO[Sch+15b]中，一个目标函数（替代目标）在策略更新的大小上受到约束而最大化。明确地，

 

这里，θold是更新前策略参数的向量。在对目标进行线性近似和对约束进行二次逼近后，利用共轭梯度算法可以有效地近似求解该问题。

证明TRPO的理论实际上建议使用惩罚而不是约束，也就是说，解决无约束优化问题：

 

这是基于这样的事实，即某个替代目标（它计算状态上的最大KL而不是均值）对策略π的性能形成了下限（即悲观界限）。 TRPO使用硬约束而不是惩罚，因为很难选择一个在不同问题上甚至在单个问题中表现都良好的β值，因为在学习过程中其特征会发生变化。 因此，为了实现模拟TRPO单调改进的一阶算法的目标，实验表明，仅选择固定罚因子β并使用SGD优化受罚目标方程（5）是不够的。需要其他修改。

3Clipped Surrogate Objective

设rt（θ）表示概率比，因此。TRPO最大化替代目标:

.

上标CPI指的是保守的政策迭代[KL02]，其中提出了这一目标。没有限制，LCPI的最大化将导致过大的政策更新。因此，我们现在考虑如何修改目标，以惩罚将rt（θ）偏离1的策略更改。

我们提出的主要目标如下:

 

其中epsilon是一个超参数，例如= 0.2。 该目标的动机如下。 最小值内的第一项是LCPI。 第二项通过裁剪概率比来修改替代目标，从而消除了将rt移到区间[1，1 +]之外的动机。 最后，我们采用了已修剪和未修剪的目标的最小值，因此最终目标是未修剪的目标的下限（即悲观的边界）。 使用此方案，我们仅在概率比率使目标改善时忽略它的变化，而在使目标恶化时将其包括在内。注意，LCLIP（θ）= LCPI（θ）在θold附近（即，r = 1）处于一阶，但是，当θ远离θold时，它们会变得不同。 图1绘制了LCLIP中的一个术语（即单个t）； 注意，根据优势是正还是负，概率比r固定为1或1 +。

 

图1：表示正向优势（左）和负向优势（右）的替代函数L CLIP的一项（即单个时间步长）作为概率比r的函数的图。 每个图上的红色圆圈显示了优化的起点，即r =1。请注意，L CLIP将这些项中的许多相加。

 

图2提供了关于替代目标L CLIP的另一种直觉来源。它显示了当我们沿着策略更新方向进行插值时，几个目标是如何变化的，这是通过对连续控制问题进行近端策略优化（我们将很快介绍的算法）获得的。 我们可以看到L CLIP是L CPI的下限，这是由于策略更新太大而受到惩罚。

 

图2：替代目标，因为我们在初始策略参数θold和更新的策略参数之间进行插值，我们在PPO的一次迭代之后进行了计算。更新后的策略与初始策略的KL差异约为0.02，这是L CLIP达到最大的点。 该图对应于关于Hopper-v1问题的首次策略更新，使用第6.1节中提供的超参数。

 

4 Adaptive KL Penalty Coefficient

另一种方法，可以作为裁剪替代目标的替代方法，或者除此之外，对KL散度使用惩罚，并调整惩罚系数，以便在每次更新策略时实现KL散度d的目标值。在我们的实验中，我们发现KL惩罚比裁剪的替代目标表现得更差，然而，我们在这里加入它是因为它是一个重要的基线。

在此算法的最简单实例化中，我们在每个策略更新中执行以下步骤。

●利用小批量SGD的几个阶段，优化KL惩罚目标：

 

●计算

 

更新后的β用于下一次策略更新。在这个方案中，我们偶尔会看到政策更新，其中KL差异与dtarg显著不同，但是这些很少，而且β很快就会调整。上述参数1.5和2是启发式选择的，但算法对它们不是很敏感。β的初始值是另一个超参数，但在实际应用中并不重要，因为算法可以快速调整它。

5 算法

可以通过对典型的策略梯度实施方案进行较小的更改来计算和区分前几部分中的代理损失。对于使用自动微分的实现，只需构造损耗L CLIP或L KLPEN而不是LPG，就可以实现此目标的随机梯度上升的多个步骤。

大多数计算方差约化优势函数估计量的技术都使用学习状态值函数V（s）；例如，广义优势估计[Sch+15a]或[Mni+16]中的有限时域估计量。如果使用在策略和值函数之间共享参数的神经网络结构，则必须使用一个结合策略代理项和值函数错误项的损失函数。这一目标可以通过增加熵加值来进一步扩大，以确保足够的探索，正如过去的工作所建议的那样[Wil92；Mni+16]。结合这些条件，我们得到如下目标，即（近似地）使每次迭代最大化

 

其中c1，c2是系数，S表示熵加，L VF t是平方误差损失。

在[Mni+16]中推广的一种策略梯度实现风格非常适合用于递归神经网络，它运行T时间步长的策略(其中T远小于事件长度)，并使用收集的样本进行更新。这种风格需要一个不超越时间步长t的优势估计器。[Mni+16]使用的估计器是

 

其中t在给定的长度T轨迹段内指定[0，T]中的时间索引。概括这个选择，我们可以使用广义优势估计的截断形式，当λ=1时，它可以简化为公式（10）：

 

使用固定长度轨迹段的最近策略优化（PPO）算法如下所示。每次迭代，N个（并行）参与者中的每一个都收集T个时间步长的数据。然后，我们在这些NT时间步长上构造代理损失，并使用minibatch SGD（或通常为获得更好的性能，Adam[KB14]）对K个时间段进行优化。

 

6 实验

6.1 替代目标的比较

首先，我们比较了不同超参数下几种不同的替代目标。在这里，我们比较了几个自然版本和替代版本。

 

对于KL惩罚，可以使用固定的惩罚系数β，也可以使用第4节中使用目标KL值dtarg的自适应系数。请注意，我们还尝试了裁剪日志空间，但发现性能没有提高。

因为我们正在为每个算法变体搜索超参数，所以我们选择了一个计算成本低廉的基准测试算法。也就是说，我们使用了OpenAI Gym[Bro+16]中实现的7个模拟机器人任务2，这些任务使用的是MuJoCo[TET12]物理引擎。我们对每个实验进行100万次的训练。除了用于剪切的超参数和我们搜索的KL惩罚（β，dtarg），其他超参数如表3所示。

为了表示该策略，我们使用了一个完全连接的MLP，它有两个64个单元的隐藏层，并且具有tanh非线性，输出高斯分布的平均值，标准差可变，如下所示[Sch+15b；Dua+16]。我们在策略和值函数之间不共享参数（因此系数c1无关紧要），也不使用熵加成。

每个算法在所有7个环境下运行，每个环境中有3个随机种子。我们通过计算最近100集的平均总回报来对算法的每次运行进行评分。我们对每个环境的分数进行了移位和缩放，这样随机策略的得分为0，最佳结果设置为1，并在21次运行中取平均值，为每个算法设置生成一个标量。

结果见表1。请注意，在没有剪辑或惩罚的情况下，该设置的分数为负数，因为对于一个环境（半猎豹），它会导致非常负的分数，这比初始随机策略更糟糕。

 

表1：连续控制基准测试的结果。 每个算法/超参数设置的平均归一化分数（在7种环境下，运行21次以上的算法）。 β初始化为1。

2   HalfCheetah, Hopper, InvertedDoublePendulum, InvertedPendulum, Reacher, Swimmer, and Walker2d, all “-v1” 

 

6.2与连续域中其他算法的比较

接下来，我们将PPO（与第3节中的裁剪代理目标）和文献中的其他几种方法进行比较，这些方法被认为对连续问题是有效的。我们比较了以下算法的优化实现：信任域策略优化[Sch+15b]、交叉熵方法（CEM）[SL06]、具有自适应步长A2C的香草策略梯度[Mni+16]、具有信任域的A2C[Wan+16]。A2C代表advantage actor critic，是A3C的同步版本，我们发现它的性能与异步版本相同或更好。对于PPO，我们使用上一节中的超参数，m=0.2。我们发现，在几乎所有的连续控制环境中，PPO的性能都优于以前的方法。

 

图3:在几个MuJoCo环境中，训练一百万步时间的几种算法的比较。

6.3连续领域的展示：仿人跑步和转向

为了展示PPO在高维连续控制问题上的表现，我们对一组涉及3D人形机器人的问题进行了培训，在这些问题中，机器人必须运行、转向和离地，可能是在被立方体投掷时。我们测试的三个任务是（1）RoboschoolHumanoid：仅向前移动，（2）RoboschoolHumanoidFlagrun：目标位置每200个时间步或目标到达时随机变化，（3）RoboschoolHumanoidFlagrunHarder，机器人被立方体投掷，需要从地上起来。学习策略的静态框架见图5，三个任务的学习曲线见图4。超参数见表4。在并行工作中，Heess等人使用PPO的自适应KL变体（第4节）来学习3D机器人的运动策略。

 

 

图5：从RoboschoolHumanoidFlagrun学到的策略框架。在前六帧中，机器人朝目标跑去。然后随机改变位置，机器人转向并朝新目标跑去。 

6.4与Atari领域其他算法的比较

我们还在Arcade学习环境[Bel+15]基准测试上运行PPO，并与A2C[Mni+16]和ACER[Wan+16]的优化实现进行了比较。对于这三种算法，我们使用了与[Mni+16]中相同的策略网络架构。PPO的超参数见表5。对于其他两个算法，我们使用了经过优化的超参数，以在该基准测试中最大限度地提高性能。附录B提供了所有49场比赛的成绩和学习曲线表。我们考虑以下两个评分指标：（1）整个训练期间每集的平均奖励（有利于快速学习）；和（2）过去100集训练中每集的平均奖励（有利于最终表现）。表2显示了每种算法赢得的游戏数量，其中我们通过三次测试的平均得分来计算胜利者。

 

我们介绍了近端策略优化，这是一系列策略优化方法，它们使用多个随机梯度上升时间来执行每个策略更新。 这些方法具有信任区方法的稳定性和可靠性，但是实现起来要简单得多，仅需要几行代码就可以更改为原始策略梯度实现，适用于更常规的设置（例如，当对策略使用联合体系结构时） 和价值函数），并具有更好的整体效果。

import os
import gym
import numpy as np
import pandas as pd
import tensorflow as tf


class PPO:
    def __init__(self, ep, batch, t='ppo2'):
        self.t = t
        self.ep = ep
        self.batch = batch
        self.log = 'model/{}_log'.format(t)

        self.env = gym.make('Pendulum-v0')
        self.bound = self.env.action_space.high[0]

        self.gamma = 0.9
        self.A_LR = 0.0001
        self.C_LR = 0.0002
        self.A_UPDATE_STEPS = 10
        self.C_UPDATE_STEPS = 10

        # KL penalty, d_target、β for ppo1
        self.kl_target = 0.01
        self.lam = 0.5
        # ε for ppo2
        self.epsilon = 0.2

        self.sess = tf.Session()
        self.build_model()

    def _build_critic(self):
        """critic model.
        """
        with tf.variable_scope('critic'):
            x = tf.layers.dense(self.states, 100, tf.nn.relu)

            self.v = tf.layers.dense(x, 1)
            self.advantage = self.dr - self.v

    def _build_actor(self, name, trainable):
        """actor model.
        """
        with tf.variable_scope(name):
            x = tf.layers.dense(self.states, 100, tf.nn.relu, trainable=trainable)

            mu = self.bound * tf.layers.dense(x, 1, tf.nn.tanh, trainable=trainable)
            sigma = tf.layers.dense(x, 1, tf.nn.softplus, trainable=trainable)

            norm_dist = tf.distributions.Normal(loc=mu, scale=sigma)

        params = tf.get_collection(tf.GraphKeys.GLOBAL_VARIABLES, scope=name)

        return norm_dist, params

    def build_model(self):
        """build model with ppo loss.
        """
        # inputs
        self.states = tf.placeholder(tf.float32, [None, 3], 'states')
        self.action = tf.placeholder(tf.float32, [None, 1], 'action')
        self.adv = tf.placeholder(tf.float32, [None, 1], 'advantage')
        self.dr = tf.placeholder(tf.float32, [None, 1], 'discounted_r')

        # build model
        self._build_critic()
        nd, pi_params = self._build_actor('actor', trainable=True)
        old_nd, oldpi_params = self._build_actor('old_actor', trainable=False)

        # define ppo loss
        with tf.variable_scope('loss'):
            # critic loss
            self.closs = tf.reduce_mean(tf.square(self.advantage))

            # actor loss
            with tf.variable_scope('surrogate'):
                ratio = tf.exp(nd.log_prob(self.action) - old_nd.log_prob(self.action))
                surr = ratio * self.adv

            if self.t == 'ppo1':
                self.tflam = tf.placeholder(tf.float32, None, 'lambda')
                kl = tf.distributions.kl_divergence(old_nd, nd)
                self.kl_mean = tf.reduce_mean(kl)
                self.aloss = -(tf.reduce_mean(surr - self.tflam * kl))
            else:
                self.aloss = -tf.reduce_mean(tf.minimum(
                    surr,
                    tf.clip_by_value(ratio, 1.- self.epsilon, 1.+ self.epsilon) * self.adv))

        # define Optimizer
        with tf.variable_scope('optimize'):
            self.ctrain_op = tf.train.AdamOptimizer(self.C_LR).minimize(self.closs)
            self.atrain_op = tf.train.AdamOptimizer(self.A_LR).minimize(self.aloss)

        with tf.variable_scope('sample_action'):
            self.sample_op = tf.squeeze(nd.sample(1), axis=0)

        # update old actor
        with tf.variable_scope('update_old_actor'):
            self.update_old_actor = [oldp.assign(p) for p, oldp in zip(pi_params, oldpi_params)]

        tf.summary.FileWriter(self.log, self.sess.graph)

        self.sess.run(tf.global_variables_initializer())

    def choose_action(self, state):
        """choice continuous action from normal distributions.

        Arguments:
            state: state.

        Returns:
           action.
        """
        state = state[np.newaxis, :]
        action = self.sess.run(self.sample_op, {self.states: state})[0]

        return np.clip(action, -self.bound, self.bound)

    def get_value(self, state):
        """get q value.

        Arguments:
            state: state.

        Returns:
           q_value.
        """
        if state.ndim < 2: state = state[np.newaxis, :]

        return self.sess.run(self.v, {self.states: state})

    def discount_reward(self, states, rewards, next_observation):
        """Compute target value.

        Arguments:
            states: state in episode.
            rewards: reward in episode.
            next_observation: state of last action.

        Returns:
            targets: q targets.
        """
        s = np.vstack([states, next_observation.reshape(-1, 3)])
        q_values = self.get_value(s).flatten()

        targets = rewards + self.gamma * q_values[1:]
        targets = targets.reshape(-1, 1)

        return targets

# not work.
#    def neglogp(self, mean, std, x):
#        """Gaussian likelihood
#        """
#        return 0.5 * tf.reduce_sum(tf.square((x - mean) / std), axis=-1) \
#               + 0.5 * np.log(2.0 * np.pi) * tf.to_float(tf.shape(x)[-1]) \
#               + tf.reduce_sum(tf.log(std), axis=-1)

    def update(self, states, action, dr):
        """update model.

        Arguments:
            states: states.
            action: action of states.
            dr: discount reward of action.
        """
        self.sess.run(self.update_old_actor)

        adv = self.sess.run(self.advantage,
                            {self.states: states,
                             self.dr: dr})

        # update actor
        if self.t == 'ppo1':
            # run ppo1 loss
            for _ in range(self.A_UPDATE_STEPS):
                _, kl = self.sess.run(
                    [self.atrain_op, self.kl_mean],
                    {self.states: states,
                     self.action: action,
                     self.adv: adv,
                     self.tflam: self.lam})

            if kl < self.kl_target / 1.5:
                self.lam /= 2
            elif kl > self.kl_target * 1.5:
                self.lam *= 2
        else:
            # run ppo2 loss
            for _ in range(self.A_UPDATE_STEPS):
                self.sess.run(self.atrain_op,
                              {self.states: states,
                               self.action: action,
                               self.adv: adv})

        # update critic
        for _ in range(self.C_UPDATE_STEPS):
            self.sess.run(self.ctrain_op,
                          {self.states: states,
                           self.dr: dr})

    def train(self):
        """train method.
        """
        tf.reset_default_graph()

        history = {'episode': [], 'Episode_reward': []}

        for i in range(self.ep):
            observation = self.env.reset()

            states, actions, rewards = [], [], []
            episode_reward = 0
            j = 0

            while True:
                a = self.choose_action(observation)
                next_observation, reward, done, _ = self.env.step(a)
                states.append(observation)
                actions.append(a)

                episode_reward += reward
                rewards.append((reward + 8) / 8)

                observation = next_observation

                if (j + 1) % self.batch == 0:
                    states = np.array(states)
                    actions = np.array(actions)
                    rewards = np.array(rewards)
                    d_reward = self.discount_reward(states, rewards, next_observation)

                    self.update(states, actions, d_reward)

                    states, actions, rewards = [], [], []

                if done:
                    break
                j += 1

            history['episode'].append(i)
            history['Episode_reward'].append(episode_reward)
            print('Episode: {} | Episode reward: {:.2f}'.format(i, episode_reward))

        return history

    def save_history(self, history, name):
        name = os.path.join('history', name)

        df = pd.DataFrame.from_dict(history)
        df.to_csv(name, index=False, encoding='utf-8')


if __name__ == '__main__':
    model = PPO(1000, 32, 'ppo1')
    history = model.train()
    model.save_history(history, 'ppo1.csv')


参考 深度强化学习–TRPO与PPO实现

哈哈初学，复现龙龙老师的实例！
state：是平衡小车上的杆子，观测状态由 4 个连续的参数组成：推车位置 [-2.4，2.4]，车速 [-∞，∞]，杆子角度 [～-41.8°，～41.8°] 与杆子末端速度 [-∞，∞]。
游戏结束：当极点与垂直方向的夹角超过15度时，或者推车从中心移出2.4个单位以上

向推车施加+1或-1的力来控制系统

杆保持直立的每个时间步长都提供+1的奖励
代码分析
经验池缓存
批训练条件：
Transition = namedtuple('Transition', ['state', 'action', 'a_log_prob', 'reward', 'next_state'])
trans = Transition(state, action, action_prob, reward, next_state)
agent.store_transition(trans)
self.buffer.append(transition)

            if done: # 合适的时间点训练网络（回合结束done=True才训练）
                if len(agent.buffer) >= batch_size:
                    agent.optimize() # 训练网络
                break

数据迭代次数:round()函数不设四舍五入位数，默认计算到整数。
 # 对缓冲池数据大致迭代10遍
        for _ in range(round(10*len(self.buffer)/batch_size)):

奖励计算
 Rs = []
        for r in reward[::-1]:
            R = r + gamma * R
            Rs.insert(0, R)  #在List 列表前添加新的R
        Rs = tf.constant(Rs, dtype=tf.float32) #将List转换为Tensor格式

重要性采样：old_action_log_prob为Actor策略网络得出的已经生成的buffer动作概率（策略采样offline）;pi_a为index对应的（历史）buffer中的状态再进行Actor网络在线得出的动作概率（目标策略online）
ratio = (pi_a / tf.gather(old_action_log_prob, index, axis=0))

PPO2误差：


ratio = (pi_a / tf.gather(old_action_log_prob, index, axis=0))
surr1 = ratio * advantage
surr2 = tf.clip_by_value(ratio, 1 - epsilon, 1 + epsilon) * advantage
# PPO误差函数
policy_loss = -tf.reduce_mean(tf.minimum(surr1, surr2))

函数备注：
tf.reshape(tensor,[-1,1])将张量变为一维列向量

tf.reshape(tensor,[1,-1])将张量变为一维行向量
tf.gather 可以根据索引号收集数据的目的。

如：考虑班级成绩册的例子,共有 4 个班

级,每个班级 35 个学生,8 门科目,保存成绩册的张量 shape 为[4,35,8]。例如在班级的维度上面我们可以像下面这样收集1和2班级的成绩册。
x = tf.random.uniform([4,35,8],maxval=100,dtype=tf.int32)
y=tf.gather(x,[0,1],axis=0)

buffer.append(transition) 在列表后面添加存储数据。
range()返回从0到4的5个数构成的list，而arange()返回一个array对象。不过他们的元素都是一样的。
Transition = namedtuple('Transition', ['state', 'action', 'a_log_prob', 'reward', 'next_state'])

下面的函数的作用为从数组x中取出n个元素，false表示每个元素不一样
**np.random.choice**(x,n,replace=false)

np.arange(len(self.buffer))      ：生成长度的数组，这里为0~39,40个数（buffer的数目），后面会用这些抽取的随机数作为训练样本的索引下标。
中间prob概率截图



a = tf.random.categorical(tf.math.log(prob), 1)[0]
变量监视：











PPO实战代码
import  matplotlib
from matplotlib import pyplot as plt
matplotlib.rcParams['font.size'] = 18
matplotlib.rcParams['figure.titlesize'] = 18
matplotlib.rcParams['figure.figsize'] = [9, 7]
matplotlib.rcParams['font.family'] = ['Microsoft YaHei']
matplotlib.rcParams['axes.unicode_minus']=False

plt.figure()

import  gym,os
import  numpy as np
import  tensorflow as tf
from    tensorflow import keras
from    tensorflow.keras import layers,optimizers,losses
from    collections import namedtuple
#from    torch.utils.data import SubsetRandomSampler,BatchSampler

env = gym.make('CartPole-v1')  # 创建游戏环境
env.seed(2222)
tf.random.set_seed(2222)
np.random.seed(2222)
os.environ['TF_CPP_MIN_LOG_LEVEL'] = '2'
assert tf.__version__.startswith('2.')



gamma = 0.98 # 激励衰减因子
epsilon = 0.2 # PPO误差超参数0.8~1.2
batch_size = 32 # batch size


# 创建游戏环境
env = gym.make('CartPole-v0').unwrapped
Transition = namedtuple('Transition', ['state', 'action', 'a_log_prob', 'reward', 'next_state'])


class Actor(keras.Model):
    def __init__(self):
        super(Actor, self).__init__()
        # 策略网络，也叫Actor网络，输出为概率分布pi(a|s)
        self.fc1 = layers.Dense(100, kernel_initializer='he_normal')
        self.fc2 = layers.Dense(2, kernel_initializer='he_normal')

    def call(self, inputs):
        x = tf.nn.relu(self.fc1(inputs))
        x = self.fc2(x)
        x = tf.nn.softmax(x, axis=1) # 转换成概率
        return x

class Critic(keras.Model):
    def __init__(self):
        super(Critic, self).__init__()
        # 偏置b的估值网络，也叫Critic网络，输出为v(s)
        self.fc1 = layers.Dense(100, kernel_initializer='he_normal')
        self.fc2 = layers.Dense(1, kernel_initializer='he_normal')

    def call(self, inputs):
        x = tf.nn.relu(self.fc1(inputs))
        x = self.fc2(x)
        return x




class PPO():
    # PPO算法主体
    def __init__(self):
        super(Actor, self).__init__()
        self.actor = Actor() # 创建Actor网络
        self.critic = Critic() # 创建Critic网络
        self.buffer = [] # 数据缓冲池
        self.actor_optimizer = optimizers.Adam(1e-3) # Actor优化器
        self.critic_optimizer = optimizers.Adam(3e-3) # Critic优化器

    def select_action(self, s):
        # 送入状态向量，获取策略: [4]
        s = tf.constant(s, dtype=tf.float32)
        # s: [4] => [1,4]
        s = tf.expand_dims(s, axis=0)
        # 获取策略分布: [1, 2]
        prob = self.actor(s)
        # 从类别分布中采样1个动作, shape: [1]
        a = tf.random.categorical(tf.math.log(prob), 1)[0]
        a = int(a)  # Tensor转数字
        return a, float(prob[0][a]) # 返回动作及其概率

    def get_value(self, s):
        # 送入状态向量，获取策略: [4]
        s = tf.constant(s, dtype=tf.float32)
        # s: [4] => [1,4]
        s = tf.expand_dims(s, axis=0)
        # 获取策略分布: [1, 2]
        v = self.critic(s)[0]
        return float(v) # 返回v(s)

    def store_transition(self, transition):
        # 存储采样数据
        self.buffer.append(transition)

    def optimize(self):
        # 优化网络主函数
        # 从缓存中取出样本数据，转换成Tensor
        state = tf.constant([t.state for t in self.buffer], dtype=tf.float32)
        action = tf.constant([t.action for t in self.buffer], dtype=tf.int32)
        action = tf.reshape(action,[-1,1])
        reward = [t.reward for t in self.buffer]
        old_action_log_prob = tf.constant([t.a_log_prob for t in self.buffer], dtype=tf.float32)
        old_action_log_prob = tf.reshape(old_action_log_prob, [-1,1])
        # 通过MC方法循环计算R(st)
        R = 0
        Rs = []
        for r in reward[::-1]:
            R = r + gamma * R
            Rs.insert(0, R)
        Rs = tf.constant(Rs, dtype=tf.float32)
        # 对缓冲池数据大致迭代10遍
        for _ in range(round(10*len(self.buffer)/batch_size)):
            # 随机从缓冲池采样batch size大小样本
            index = np.random.choice(np.arange(len(self.buffer)), batch_size, replace=False)
            # 构建梯度跟踪环境
            with tf.GradientTape() as tape1, tf.GradientTape() as tape2:
                # 取出R(st)，[b,1]
                v_target = tf.expand_dims(tf.gather(Rs, index, axis=0), axis=1)
                # 计算v(s)预测值，也就是偏置b，我们后面会介绍为什么写成v
                v = self.critic(tf.gather(state, index, axis=0))
                delta = v_target - v # 计算优势值
                advantage = tf.stop_gradient(delta) # 断开梯度连接
                # 由于TF的gather_nd与pytorch的gather功能不一样，需要构造
                # gather_nd需要的坐标参数，indices:[b, 2]
                # pi_a = pi.gather(1, a) # pytorch只需要一行即可实现
                a = tf.gather(action, index, axis=0) # 取出batch的动作at
                # batch的动作分布pi(a|st)
                pi = self.actor(tf.gather(state, index, axis=0))
                indices = tf.expand_dims(tf.range(a.shape[0]), axis=1)
                indices = tf.concat([indices, a], axis=1)
                pi_a = tf.gather_nd(pi, indices)  # 动作的概率值pi(at|st), [b]
                pi_a = tf.expand_dims(pi_a, axis=1)  # [b]=> [b,1]
                # 重要性采样
                ratio = (pi_a / tf.gather(old_action_log_prob, index, axis=0))
                surr1 = ratio * advantage
                surr2 = tf.clip_by_value(ratio, 1 - epsilon, 1 + epsilon) * advantage
                # PPO误差函数
                policy_loss = -tf.reduce_mean(tf.minimum(surr1, surr2))
                # 对于偏置v来说，希望与MC估计的R(st)越接近越好
                value_loss = losses.MSE(v_target, v)
            # 优化策略网络
            grads = tape1.gradient(policy_loss, self.actor.trainable_variables)
            self.actor_optimizer.apply_gradients(zip(grads, self.actor.trainable_variables))
            # 优化偏置值网络
            grads = tape2.gradient(value_loss, self.critic.trainable_variables)
            self.critic_optimizer.apply_gradients(zip(grads, self.critic.trainable_variables))

        self.buffer = []  # 清空已训练数据


def main():
    agent = PPO()
    returns = [] # 统计总回报
    total = 0 # 一段时间内平均回报
    for i_epoch in range(1000): # 训练回合数
        state = env.reset() # 复位环境
        for t in range(500): # 最多考虑500步
            # 通过最新策略与环境交互
            action, action_prob = agent.select_action(state)
            next_state, reward, done, _ = env.step(action)
            # 构建样本并存储
            trans = Transition(state, action, action_prob, reward, next_state)
            agent.store_transition(trans)
            state = next_state # 刷新状态
            total += reward # 累积激励
            env.render()
            if done: # 合适的时间点训练网络
                if len(agent.buffer) >= batch_size:
                    agent.optimize() # 训练网络
                break

        if i_epoch % 20 == 0: # 每20个回合统计一次平均回报
            returns.append(total/20)
            total = 0
            print(i_epoch, returns[-1])

    print(np.array(returns))
    plt.figure()
    plt.plot(np.arange(len(returns))*20, np.array(returns))
    plt.plot(np.arange(len(returns))*20, np.array(returns), 's')
    plt.xlabel('回合数')
    plt.ylabel('总回报')
    plt.savefig('ppo-tf-cartpole.svg')


if __name__ == '__main__':
    main()
    print("end")


 List item

近端策略优化（Proximal Policy Optimization,PPO）
算法背景：

策略梯度虽然在有一定难度的问题中取得了一些成效，但是这类方法对于迭代步骤的数量非常敏感；如果选的太小，训练的结果就会令人绝望；如果选的过大，反馈信号就会淹没在噪声中，甚至有可能使训练模型呈现雪崩式的下降。这类方法的采样效率也是非常低，学习简单的任务就需要百万级甚至以上的总迭代次数。
策略梯度方法的缺点是数据效率和鲁棒性不好。对于TRPO算法比较复杂，且不兼容dropout和参数共享（策略和价值网络间），且使用约束的方法参数难以调节，同时算法实现的复杂性。故不建议使用TRPO算法，而使用PPO算法。

PPO算法的提出：旨在借鉴TRPO算法，使用一阶优化，在采样效率、算法表现，以及实现和调试的复杂度之间取得了新的平衡。这是因为PPO会在每一次迭代中尝试计算新的策略，让损失函数最小化，并且保证每一次新计算出的策略能够和原策略相差不大，目前，openai已经把PPO作为自己RL中的首选算法。