1.前言

今天利用上篇文章讲解的Policy Gradient理论进行实战，背景仍然是杆子不倒游戏和小车登顶游戏。

image
image

2.算法

理论部分上篇文章已经介绍过了，这里不多赘述，感兴趣的读者可以去看我上一篇文章。

image

2.1算法主循环

我们先定义一下算法的主循环，这里要注意我们采取的是回合更新，而不是Q-Learning等的单步更新

import gym
from RL_brain import PolicyGradient
import matplotlib.pyplot as plt

DISPLAY_REWARD_THRESHOLD = 400  # 当回合总reward大于400时显示模拟窗口
RENDER = False  # 在屏幕上显示模拟窗口会拖慢运行速度，我们等计算机学的差不多了再进行模拟

env = gym.make('CartPole-v0')
env.seed(1)     # 普通的Policy gradient方法，使得回合的方差比较大，所以我们选了一个好点的随机种子
env = env.unwrapped

print(env.action_space)
print(env.observation_space)
print(env.observation_space.high)
print(env.observation_space.low)

RL = PolicyGradient(
    n_actions=env.action_space.n,
    n_features=env.observation_space.shape[0],
    learning_rate=0.02,
    reward_decay=0.99,
    # output_graph=True,
)

for i_episode in range(3000):

    observation = env.reset()

    while True:
        if RENDER: env.render()

        action = RL.choose_action(observation)

        observation_, reward, done, info = env.step(action)

        RL.store_transition(observation, action, reward)#存储这一回合的transition

        if done:
            ep_rs_sum = sum(RL.ep_rs)

            if 'running_reward' not in globals():
                running_reward = ep_rs_sum
            else:
                running_reward = running_reward * 0.99 + ep_rs_sum * 0.01
            if running_reward > DISPLAY_REWARD_THRESHOLD: RENDER = True   # 判断是否显示模拟
            print("episode:", i_episode, "  reward:", int(running_reward))

            vt = RL.learn() #学习，输出vt

            if i_episode == 0:
                plt.plot(vt)    # plot the episode vt
                plt.xlabel('episode steps')
                plt.ylabel('normalized state-action value')
                plt.show()
            break

        observation = observation_

2.2 Policy Gradient部分

我们先初始化我们的神经网络

import numpy as np
import tensorflow as tf

# reproducible
np.random.seed(1)
tf.set_random_seed(1)


class PolicyGradient:
    def __init__(
            self,
            n_actions,
            n_features,
            learning_rate=0.01,
            reward_decay=0.95,
            output_graph=False,
    ):
        self.n_actions = n_actions
        self.n_features = n_features
        self.lr = learning_rate
        self.gamma = reward_decay # reward递减率

        self.ep_obs, self.ep_as, self.ep_rs = [], [], [] #存储回合信息的list

        self._build_net()   #建立policy神经网络

        self.sess = tf.Session()

        if output_graph:
            # $ tensorboard --logdir=logs
            # http://0.0.0.0:6006/
            # tf.train.SummaryWriter soon be deprecated, use following
            tf.summary.FileWriter("logs/", self.sess.graph)

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

然后我们来建立我们的神经网络，我们要建立的神经网络是这样的：

image

    def _build_net(self):
        with tf.name_scope('inputs'):
            #接受observation
            self.tf_obs = tf.placeholder(tf.float32, [None, self.n_features], name="observations")
            #接收我们在这个回合中选过的actions
            self.tf_acts = tf.placeholder(tf.int32, [None, ], name="actions_num")
            #接收每个state-action所对应的value（通过reward计算）
            self.tf_vt = tf.placeholder(tf.float32, [None, ], name="actions_value")
        # fc1
        layer = tf.layers.dense(
            inputs=self.tf_obs,
            units=10,
            activation=tf.nn.tanh,  # tanh activation
            kernel_initializer=tf.random_normal_initializer(mean=0, stddev=0.3),
            bias_initializer=tf.constant_initializer(0.1),
            name='fc1'
        )
        # fc2
        all_act = tf.layers.dense(
            inputs=layer,
            units=self.n_actions,
            activation=None, #后面用softmax
            kernel_initializer=tf.random_normal_initializer(mean=0, stddev=0.3),
            bias_initializer=tf.constant_initializer(0.1),
            name='fc2'
        )

        self.all_act_prob = tf.nn.softmax(all_act, name='act_prob')  # 用softmax出概率

        with tf.name_scope('loss'):
            # 最大化总体reward(log_p*R)就是在最小化-(log_p*R)。而tf的功能里只有最小化loss
            neg_log_prob = tf.nn.sparse_softmax_cross_entropy_with_logits(logits=all_act, labels=self.tf_acts)  # 所选action的概率-log值
            # or in this way:
            # neg_log_prob = tf.reduce_sum(-tf.log(self.all_act_prob)*tf.one_hot(self.tf_acts, self.n_actions), axis=1)
            loss = tf.reduce_mean(neg_log_prob * self.tf_vt)  # （vt=本reward+衰减的未来reward）引导参数的梯度下降

        with tf.name_scope('train'):
            self.train_op = tf.train.AdamOptimizer(self.lr).minimize(loss)

为了确保我们选择的动作真的是“正确的”，我们的loss在原本的cross-entropy形式上乘以vt。用vt来告诉这个cross-entropy算出来的梯度是不是一个值得信任的梯度。如果vt小或者是负的，就说明这个梯度下降是一个错误的方向，我们应该向着另一个方向更新参数。如果这个vt是正的，或很大，vt就会称赞cross-entropy出来的梯度，并朝着这个方向梯度下降。下面有一张从karpathy大神网页下扣下来的图，也正是阐述这个思想。

image
接着我们就可以通过概率而不是Q value来选择我们的行为了。这里即使不采用epsilon-greedy，也具有一定的随机性。

    def choose_action(self, observation):
        prob_weights = self.sess.run(self.all_act_prob, feed_dict={self.tf_obs: observation[np.newaxis, :]})
        action = np.random.choice(range(prob_weights.shape[1]), p=prob_weights.ravel())  # 直接根据概率来选action
        return action

选择完行为就可以存储回合了，就是把这一步的observation，action，reward加到列表中去。因为本回合完毕之后要清空列表，然后存储下一回合的数据，我们会在learn()当中清空列表的动作。

    def store_transition(self, s, a, r):
        self.ep_obs.append(s)
        self.ep_as.append(a)
        self.ep_rs.append(r)

这次的learn()很简单。首先我们要对本回合所有的reward动下手脚，使他变得更适合被学习。第一就是随着时间推进，用gamma衰减未来的reward，然后为了一定程度下减少policy gradient回合方差，我们标准化回合的state-action value，依据在Andrej Karpathy的blog

    def learn(self):
        # discount and normalize episode reward
        discounted_ep_rs_norm = self._discount_and_norm_rewards()

        # train on episode
        self.sess.run(self.train_op, feed_dict={
             self.tf_obs: np.vstack(self.ep_obs),  # shape=[None, n_obs]
             self.tf_acts: np.array(self.ep_as),  # shape=[None, ]
             self.tf_vt: discounted_ep_rs_norm,  # shape=[None, ]
        })

        self.ep_obs, self.ep_as, self.ep_rs = [], [], []    # 清空回合的数据
        return discounted_ep_rs_norm

    def _discount_and_norm_rewards(self):
        # discount episode rewards
        discounted_ep_rs = np.zeros_like(self.ep_rs)
        running_add = 0
        for t in reversed(range(0, len(self.ep_rs))):
            running_add = running_add * self.gamma + self.ep_rs[t]
            discounted_ep_rs[t] = running_add

        # normalize episode rewards
        discounted_ep_rs -= np.mean(discounted_ep_rs)
        discounted_ep_rs /= np.std(discounted_ep_rs)
        return discounted_ep_rs

3. 结果分析

我们来看看vt的输出，看看他是怎么诱导我们的gradient descent。

image
可以看出，左边一段的vt有较高的值，右边较低，这就是vt在说：

“对于回合开始的一些列动作，因为前面一段时间杆子还没有掉下来，所以要重视；对于后面的动作，因为他们让杆子掉下来了，所以要惩罚”或者是

“我每次都想让这个动作在下一次增加被做的可能性（gra(log(Policy))）,但是增加可能性的这种做法是好还是坏呢？这就要由vt告诉我了，所以后段时间的增加可能性做法`并没有被提倡，而前段时间的增加可能性做法是被提倡的”

这样vt就能在这里loss = tf.reduce_mean(log_prob * self.tf_vt)诱导gradient descent朝着正确的方向发展了。

而MountainCar中的vt长这样：

image
这是在说：”请重视我这回合最后的一系列动作，因为这一系列动作让我爬上了山，而且请惩罚我开始的一系列动作，因为这些动作让我没能爬上山“

完整代码：https://github.com/cristianoc20/RL_learning/tree/master/Policy_gradient_softmax
参考：莫烦教程 https://github.com/MorvanZhou