快乐的强化学习4——Policy Gradients及其实现方法

今天药忘吃喽~ 2021-10-29 14:24 367阅读 0赞

### 快乐的强化学习4——Policy Gradients及其实现方法 ###

*  学习前言
 *   *  简介
     *  举例应用
     *   *  神经网络的构建
         *  动作的选择
         *  神经网络的学习
     *  具体实现代码

# 学习前言 #

刚刚从大学毕业，近来闲来无事，开始了机器学习的旅程，深度学习是机器学习的重要一环，其可以使得机器自我尝试，并通过结果进行学习。  
![在这里插入图片描述][20190723165901974.jpg_pic_center]  
在机器学习的过程中，我自网上了解到大神morvanzhou，一个从土木工程转向了计算机的“聪明绝顶”的、英语特好的男人。  
[morvanzhou的python个人主页][morvanzhou_python]，请有兴趣的同学关注大神morvanzhou的python教程。

## 简介 ##

Policy Gradients的是一种基于策略的强化学习方法，基于策略的意思就是直接根据状态输出动作或者动作的概率。  
我们使用神经网络输入当前的状态，网络就可以输出我们在这个状态下采取每个动作的概率。  
与之前的DQN和Q-Learmnig这类基于价值的学习方法相比，二者的主要不同为：  
1、基于价值的强化学习方法对应的最优策略通常是确定性策略，一般是从众多行为价值中选择一个最大价值的行为，仅有一定的概率会随机选择，而有些问题的最优策略却是随机策略，无法通过基于价值的学习方法求解。此时Policy Gradients可以起到很好的作用。  
2、DQN之类的方法一般都是只处理离散动作，无法处理连续动作。

## 举例应用 ##

本文使用了OpenAI Gym中的CartPole-v0游戏来验证算法的实用性。  
该环境只有两个离散动作，板子可以向左滑动或者向右滑动，保持杆子不倒即可。  
state状态就是这个板子的位置和速度， pole的角度和角速度，4维的特征。

### 神经网络的构建 ###

在Policy Gradients中，通过输入状态来得到每个动作的概率。  
其建立的神经网络结构如下：  
神经网络的输入量是self.tf\_obs，指的是环境的状态。  
神经网络的输出量是self.all\_act\_prob，指的是每个动作的概率。  
神经网络共两层。

def _build_net(self):
        # 输入量
        # self.tf_obs指的是环境的观测量
        # self.tf_acts指的是可执行的动作
        # self.tf_vt指的是每个动作的奖惩情况
        with tf.name_scope('inputs'):
            self.tf_obs = tf.placeholder(tf.float32, [None, self.n_features], name="observations")
            self.tf_acts = tf.placeholder(tf.int32, [None, ], name="actions_num")
            self.tf_vt = tf.placeholder(tf.float32, [None, ], name="actions_value")
    
        # 利用tf.layers.dense函数生成全连接层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'
        )
        # 利用tf.layers.dense函数生成全连接层fc2
        all_act = tf.layers.dense(
            inputs=layer,
            units=self.n_actions,
            activation=None,
            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')  # use softmax to convert to probability
    
        with tf.name_scope('loss'):
            # 可以通过最小化-(log_p * R)的方式实现最大化(log_p * R)
            neg_log_prob = tf.nn.sparse_softmax_cross_entropy_with_logits(logits=all_act, labels=self.tf_acts)   
            loss = tf.reduce_mean(neg_log_prob * self.tf_vt)  # reward guided loss
    
        with tf.name_scope('train'):
            self.train_op = tf.train.AdamOptimizer(self.lr).minimize(loss)

### 动作的选择 ###

其根据神经网络的输出按照概率选择动作。

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())
        return action

### 神经网络的学习 ###

神经网络的学习过程是这样的，通过记录每一轮开始直到done结束的所有的状态、动作及其得分。  
通过所有的状态得到所有状态下每个动作的概率，利用：

neg_log_prob = tf.nn.sparse_softmax_cross_entropy_with_logits(logits=all_act, labels=self.tf_acts)   
    loss = tf.reduce_mean(neg_log_prob * self.tf_vt)  # reward guided loss

计算每个动作的概率和实际动作的交叉熵，呈上实际得分，得到损失函数值。  
在实际得分用于计算LOSS前，需要进行标准化。

def learn(self):
        # 计算并标准化reward
        discounted_ep_rs_norm = self._discount_and_norm_rewards()
    
        # 训练
        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):
        # 标准化得分
        # 生成与self.ep_rs数组shape相同的全0数组
        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
    
        # 标准化
        discounted_ep_rs -= np.mean(discounted_ep_rs)
        # 计算标准差
        discounted_ep_rs /= np.std(discounted_ep_rs)
        return discounted_ep_rs

在每次学习完成后，需要对历史数据进行清零。

## 具体实现代码 ##

具体的实现代码分为两个部分，这是第一部分，主函数：

import gym
    from RL_brain import PolicyGradient
    import matplotlib.pyplot as plt
    import math
    RENDER = True  # 是否显示图像
    
    env = gym.make('CartPole-v0')
    env.seed(1)     # reproducible, general Policy gradient has high variance
    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)
    
            x, x_dot, theta, theta_dot = observation_   
    
            # r1代表车的 x水平位移 与 x最大边距 的距离差的得分
            r1 = math.exp((env.x_threshold - abs(x))/env.x_threshold) - math.exp(1)/2
            # r1代表棒子的 theta离垂直的角度 与 theta最大角度 的差的得分
            r2 = math.exp((env.theta_threshold_radians - abs(theta))/env.theta_threshold_radians) - math.exp(1)/2
            # 总 reward 是 r1 和 r2 的结合, 既考虑位置, 也考虑角度。
            reward = r1 + r2  
    
            #在未done前进行数据存储
            RL.store_transition(observation, action, reward)
    
            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
                print("episode:", i_episode, "  reward:", int(running_reward))
                vt = RL.learn() # 学习, 输出 vt, 我们下节课讲这个 vt 的作用
                break
    
            observation = observation_

这是第二部分，RL\_brain.py：

import numpy as np
    import tensorflow as tf
    
    np.random.seed(1)
    tf.set_random_seed(1)
    
    class PolicyGradient:
        def __init__(
                self,
                n_actions,
                n_features,
                learning_rate=0.001,
                reward_decay=0.95,
                output_graph=False,
        ):  # 初始化函数，需要输入action的数量，每个环境的features数量，学习率等
            self.n_actions = n_actions
            self.n_features = n_features
            self.lr = learning_rate
            # 衰减率
            self.gamma = reward_decay
            
            # ep_obs用于存储历史观测结果，ep_as用于存储历史动作，ep_rs用于存储历史奖惩
            self.ep_obs, self.ep_as, self.ep_rs = [], [], []
    
            self._build_net()
    
            # 建立会话
            self.sess = tf.Session()
    
            if output_graph:
                # tensorboard --logdir=logs
                tf.summary.FileWriter("logs/", self.sess.graph)
    
            self.sess.run(tf.global_variables_initializer())
    
        def _build_net(self):
            # 输入量
            # self.tf_obs指的是环境的观测量
            # self.tf_acts指的是可执行的动作
            # self.tf_vt指的是每个动作的奖惩情况
            with tf.name_scope('inputs'):
                self.tf_obs = tf.placeholder(tf.float32, [None, self.n_features], name="observations")
                self.tf_acts = tf.placeholder(tf.int32, [None, ], name="actions_num")
                self.tf_vt = tf.placeholder(tf.float32, [None, ], name="actions_value")
    
            # 利用tf.layers.dense函数生成全连接层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'
            )
            # 利用tf.layers.dense函数生成全连接层fc2
            all_act = tf.layers.dense(
                inputs=layer,
                units=self.n_actions,
                activation=None,
                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')  # use softmax to convert to probability
    
            with tf.name_scope('loss'):
                # 可以通过最小化-(log_p * R)的方式实现最大化(log_p * R)
                neg_log_prob = tf.nn.sparse_softmax_cross_entropy_with_logits(logits=all_act, labels=self.tf_acts)   
                loss = tf.reduce_mean(neg_log_prob * self.tf_vt) 
    
            with tf.name_scope('train'):
                self.train_op = tf.train.AdamOptimizer(self.lr).minimize(loss)
    
        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())
            return action
    
        def store_transition(self, s, a, r):
            # 往历史记录中存储数据
            # ep_obs用于存储历史观测结果，ep_as用于存储历史动作，ep_rs用于存储历史奖惩
            self.ep_obs.append(s)
            self.ep_as.append(a)
            self.ep_rs.append(r)
    
        def learn(self):
            # 计算并标准化reward
            discounted_ep_rs_norm = self._discount_and_norm_rewards()
    
            # 训练
            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):
            # 标准化得分
            # 生成与self.ep_rs数组shape相同的全0数组
            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
    
            # 标准化
            discounted_ep_rs -= np.mean(discounted_ep_rs)
            # 计算标准差
            discounted_ep_rs /= np.std(discounted_ep_rs)
            return discounted_ep_rs

主函数和RL\_brain部分我都进行了详细的备注。  
由于代码并不是自己写的，所以就不上传github了，不过还是欢迎大家关注我和我的github。  
https://github.com/bubbliiiing/  
希望得到朋友们的喜欢。

**有不懂的朋友可以评论询问噢。**

[20190723165901974.jpg_pic_center]: /images/20211029/d67fbe1e29e549c5b4277720dad46fdb.png
[morvanzhou_python]: https://morvanzhou.github.io/