快乐的强化学习5——Actor Critic及其实现方法

灰太狼 2021-10-29 23:48 503阅读 0赞

### 快乐的强化学习5——Actor Critic及其实现方法 ###

*  学习前言
 *   *  一、简介
     *  二、实现过程拆解
     *   *  1、神经网络部分
         *   *  a、Actor部分
             *  b、Critic部分
         *  2、动作的选择
         *  3、Actor神经网络的学习
     *  三、具体实现代码

# 学习前言 #

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

## 一、简介 ##

Actor Critic的中文名字是演员&评论家，它的主体可以分为两个部分，分别是演员和评论家。  
actor（演员）的作用与其policy gradient类似，其可以通过输入环境的观测量利用神经网络得到每个动作的概率，并通过每个观测状态的得分更新当前的神经网络。**它就好像一个被指定动作的演员一般。**  
Critic（评论家）的作用与其前身DQN类似，其可以通过输入环境的观测量预测每个环境的价值，并通过实际价值与预测价值更新当前的神经网络，**它的价值输出可以对actor的动作选择进行动作指导，就好像评论演员的评论家，有助于演员做更好的动作。**  
actor的前身是policy gradient，policy gradient可以轻松地在完成连续空间的动作，Actor Critic也有同样的特性，但是由于policy gradien是基于回合更新的，所以学习效率比较慢，这时候我们使用前身为DQN的value-based算法为Actor提供算法的参数支持，有助于算法的快速更新。

## 二、实现过程拆解 ##

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

### 1、神经网络部分 ###

#### a、Actor部分 ####

在Actor部分，通过输入状态来得到每个动作的概率。  
其建立的神经网络结构如下：  
**神经网络的输入量是self.s，指的是环境的状态。  
神经网络的输出量是self.acts\_prob，指的是每个动作的概率。**  
其学习过程需要结合Critic部分输出的self.td\_error进行学习，self.td\_error在训练过程中的含义类似于该环境的评分。

class Actor(object):
        def __init__(self, sess, n_features, n_actions, lr=0.001):
    
            self.sess = sess
            # 状态
            self.s = tf.placeholder(tf.float32, [1, n_features], "state")
            # 动作
            self.a = tf.placeholder(tf.int32, None, "act")
            self.td_error = tf.placeholder(tf.float32, None, "td_error")  # TD_error
    
            with tf.variable_scope('Actor'):
    
                l1 = tf.layers.dense(
                    inputs=self.s,
                    units=20,    # 二十个神经元
                    activation=tf.nn.relu,
                    kernel_initializer=tf.random_normal_initializer(0., .1),
                    bias_initializer=tf.constant_initializer(0.1),
                    name='l1'
                )
    
                l2 = tf.layers.dense(
                    inputs=l1,
                    units=n_actions,    # 状态的维度
                    activation=None,   # 获得执行每个动作的可能性
                    kernel_initializer=tf.random_normal_initializer(0., .1), 
                    bias_initializer=tf.constant_initializer(0.1),
                    name='acts_prob'
                )
            self.acts_prob = tf.nn.softmax(l2)
            label = [self.a]
            with tf.variable_scope('exp_v'):
                # 可以通过最小化-(log_p * R)的方式实现最大化(log_p * R)
                neg_log_prob = tf.nn.sparse_softmax_cross_entropy_with_logits(logits=l2, labels=label)   
                self.exp_v = tf.reduce_mean(neg_log_prob * self.td_error) 
    
            with tf.variable_scope('train'):
                self.train_op = tf.train.AdamOptimizer(lr).minimize(self.exp_v)  # minimize(-exp_v) = maximize(exp_v)
    
        def learn(self, s, a, td):
            s = s[np.newaxis, :]
            feed_dict = {
        self.s: s, self.a: a, self.td_error: td}
            _, exp_v = self.sess.run([self.train_op, self.exp_v], feed_dict)
            return exp_v
    
        def choose_action(self, s):
            s = s[np.newaxis, :]
            # 获得所有动作的概率
            probs = self.sess.run(self.acts_prob, {
        self.s: s})   
            # 返回按照一定概率选择的动作
            return np.random.choice(np.arange(probs.shape[1]), p=probs.ravel())

#### b、Critic部分 ####

在Critic部分，通过输入状态来得到每个环境的评分  
**神经网络的输入量是self.s，指的是环境的状态。  
神经网络的输出量是self.v，该环境的得分。**  
其学习过程与DQN类似，通过所处的当前环境的估计得分、下一步的环境的现实得分进行得分更新的。  
当前环境的估计得分与DQN中的Q估计类似  
下一步的环境的现实得分与DQN中的Q现实类似。

class Critic(object):
        def __init__(self, sess, n_features, lr=0.01):
            self.sess = sess
            # s是状态
            self.s = tf.placeholder(tf.float32, [1, n_features], "state")
            # 下一个状态的预计value
            self.v_ = tf.placeholder(tf.float32, [1, 1], "v_next")
            # r是得分
            self.r = tf.placeholder(tf.float32, None, 'r')
    
            with tf.variable_scope('Critic'):
                l1 = tf.layers.dense(
                    inputs=self.s,
                    units=20, 
                    activation=tf.nn.relu,  
                    kernel_initializer=tf.random_normal_initializer(0., .1),
                    bias_initializer=tf.constant_initializer(0.1),
                    name='l1'
                )
    
                self.v = tf.layers.dense(
                    inputs=l1,
                    units=1,  # output units
                    activation=None,
                    kernel_initializer=tf.random_normal_initializer(0., .1),
                    bias_initializer=tf.constant_initializer(0.1),
                    name='V'
                )
    
            with tf.variable_scope('squared_TD_error'):
                # 获取下一个环境得分与该环境的差
                # self.r + GAMMA * self.v_为现实的价值，V_eval为预计的价值
                self.td_error = self.r + GAMMA * self.v_ - self.v
                self.loss = tf.square(self.td_error)    # TD_error = (r+gamma*V_next) - V_eval
            with tf.variable_scope('train'):
                self.train_op = tf.train.AdamOptimizer(lr).minimize(self.loss)
    
        def learn(self, s, r, s_):
            s, s_ = s[np.newaxis, :], s_[np.newaxis, :]
            
            # 获得下一个环境的预测得分
            v_ = self.sess.run(self.v, {
        self.s: s_})
            
            td_error, _ = self.sess.run([self.td_error, self.train_op],
                                              {
        self.s: s, self.v_: v_, self.r: r})
            return td_error

### 2、动作的选择 ###

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

def choose_action(self, s):
        s = s[np.newaxis, :]
        # 获得所有动作的概率
        probs = self.sess.run(self.acts_prob, {
        self.s: s})   
        # 返回按照一定概率选择的动作
        return np.random.choice(np.arange(probs.shape[1]), p=probs.ravel())

### 3、Actor神经网络的学习 ###

Actor神经网络的学习过程是这样的，得到Critic评出的得分，并通过该得分进行训练。  
以下代码为Critic评出的得分：

self.td_error = self.r + GAMMA * self.v_ - self.v

通过所有的状态得到所有状态下每个动作的概率，利用：

neg_log_prob = tf.nn.sparse_softmax_cross_entropy_with_logits(logits=l2, labels=label)   
    self.exp_v = tf.reduce_mean(neg_log_prob * self.td_error)

计算每个动作的概率和实际动作的交叉熵，乘上实际得分，得到损失函数值。

## 三、具体实现代码 ##

具体的实现代码仅一部分：

import numpy as np
    import tensorflow as tf
    import gym
    import math
    
    np.random.seed(2)
    tf.set_random_seed(2) 
    
    # 在200分前不显示图像
    DISPLAY_REWARD_THRESHOLD = 200
    RENDER = False 
    
    # 是否输出图像
    OUTPUT_GRAPH = False
    
    # 最大世代
    MAX_EPISODE = 3000
    MAX_EP_STEPS = 1000   # maximum time step in one episode
    
    # 衰减率
    GAMMA = 0.9     
    
    # 两个神经网络的学习率
    LR_A = 0.001    # actor学习率
    LR_C = 0.01     # critic学习率
    
    env = gym.make('CartPole-v0')
    env.seed(1)
    env = env.unwrapped
    
    # 环境的维度
    N_F = env.observation_space.shape[0]
    # 动作的数量
    N_A = env.action_space.n
    
    
    class Actor(object):
        def __init__(self, sess, n_features, n_actions, lr=0.001):
    
            self.sess = sess
            # 状态
            self.s = tf.placeholder(tf.float32, [1, n_features], "state")
            # 动作
            self.a = tf.placeholder(tf.int32, None, "act")
            self.td_error = tf.placeholder(tf.float32, None, "td_error")  # TD_error
    
            with tf.variable_scope('Actor'):
    
                l1 = tf.layers.dense(
                    inputs=self.s,
                    units=20,    # 二十个神经元
                    activation=tf.nn.relu,
                    kernel_initializer=tf.random_normal_initializer(0., .1),
                    bias_initializer=tf.constant_initializer(0.1),
                    name='l1'
                )
    
                l2 = tf.layers.dense(
                    inputs=l1,
                    units=n_actions,    # 状态的维度
                    activation=None,   # 获得执行每个动作的可能性
                    kernel_initializer=tf.random_normal_initializer(0., .1), 
                    bias_initializer=tf.constant_initializer(0.1),
                    name='acts_prob'
                )
            self.acts_prob = tf.nn.softmax(l2)
            label = [self.a]
            with tf.variable_scope('exp_v'):
                # 可以通过最小化-(log_p * R)的方式实现最大化(log_p * R)
                neg_log_prob = tf.nn.sparse_softmax_cross_entropy_with_logits(logits=l2, labels=label)   
                self.exp_v = tf.reduce_mean(neg_log_prob * self.td_error) 
    
            with tf.variable_scope('train'):
                self.train_op = tf.train.AdamOptimizer(lr).minimize(self.exp_v)  # minimize(-exp_v) = maximize(exp_v)
    
        def learn(self, s, a, td):
            s = s[np.newaxis, :]
            feed_dict = {
        self.s: s, self.a: a, self.td_error: td}
            _, exp_v = self.sess.run([self.train_op, self.exp_v], feed_dict)
            return exp_v
    
        def choose_action(self, s):
            s = s[np.newaxis, :]
            # 获得所有动作的概率
            probs = self.sess.run(self.acts_prob, {
        self.s: s})   
            # 返回按照一定概率选择的动作
            return np.random.choice(np.arange(probs.shape[1]), p=probs.ravel())   
    
    
    class Critic(object):
        def __init__(self, sess, n_features, lr=0.01):
            self.sess = sess
            # s是状态
            self.s = tf.placeholder(tf.float32, [1, n_features], "state")
            # 下一个状态的预计value
            self.v_ = tf.placeholder(tf.float32, [1, 1], "v_next")
            # r是得分
            self.r = tf.placeholder(tf.float32, None, 'r')
    
            with tf.variable_scope('Critic'):
                l1 = tf.layers.dense(
                    inputs=self.s,
                    units=20, 
                    activation=tf.nn.relu,  
                    kernel_initializer=tf.random_normal_initializer(0., .1),
                    bias_initializer=tf.constant_initializer(0.1),
                    name='l1'
                )
    
                self.v = tf.layers.dense(
                    inputs=l1,
                    units=1,  # output units
                    activation=None,
                    kernel_initializer=tf.random_normal_initializer(0., .1),
                    bias_initializer=tf.constant_initializer(0.1),
                    name='V'
                )
    
            with tf.variable_scope('squared_TD_error'):
                # 获取下一个环境得分与该环境的差
                # self.r + GAMMA * self.v_为现实的价值，V_eval为预计的价值
                self.td_error = self.r + GAMMA * self.v_ - self.v
                self.loss = tf.square(self.td_error)    # TD_error = (r+gamma*V_next) - V_eval
            with tf.variable_scope('train'):
                self.train_op = tf.train.AdamOptimizer(lr).minimize(self.loss)
    
        def learn(self, s, r, s_):
            s, s_ = s[np.newaxis, :], s_[np.newaxis, :]
            
            # 获得下一个环境的预测得分
            v_ = self.sess.run(self.v, {
        self.s: s_})
            
            td_error, _ = self.sess.run([self.td_error, self.train_op],
                                              {
        self.s: s, self.v_: v_, self.r: r})
            return td_error
    
    
    sess = tf.Session()
    
    # critic输出的数据会为actor进行指导，所以critic需要训练的更快一些
    actor = Actor(sess, n_features=N_F, n_actions=N_A, lr=LR_A)
    critic = Critic(sess, n_features=N_F, lr=LR_C)     
    
    sess.run(tf.global_variables_initializer())
    
    if OUTPUT_GRAPH:
        tf.summary.FileWriter("logs/", sess.graph)
    
    for i_episode in range(MAX_EPISODE):
        s = env.reset()
        t = 0
        # 用于记录每一轮的得分
        track_r = []
        while True:
            if RENDER: env.render()
    
            a = actor.choose_action(s)
            
            s_, r, done, info = env.step(a)
    
            x, x_dot, theta, theta_dot = s_   
    
            # 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 的结合, 既考虑位置, 也考虑角度。
            r = r1 + r2  
    
            # 如果没有撑过1000回合，则得分为-20一次。
            if done: r = -20
    
            track_r.append(r)
    
            td_error = critic.learn(s, r, s_)  # gradient = grad[r + gamma * V(s_) - V(s)]
            
            actor.learn(s, a, td_error)     # true_gradient = grad[logPi(s,a) * td_error]
    
            # 状态更新
            s = s_
    
            # t++
            t += 1
    
            if done or t >= MAX_EP_STEPS:
                ep_rs_sum = sum(track_r)
    
                if 'running_reward' not in globals():
                    running_reward = ep_rs_sum
                else:
                    # 保留95%过去的得分，并加上5%现在的得分
                    running_reward = running_reward * 0.95 + ep_rs_sum * 0.05
                if running_reward > DISPLAY_REWARD_THRESHOLD: RENDER = True
                print("episode:", i_episode, "  reward:", int(running_reward))
                break

代码的各部分我都进行了详细的备注。  
相比于原来的代码，我对代码进行了一定的更改，分别是得分部分和Actor神经网络的训练部分。  
首先是得分部分：

x, x_dot, theta, theta_dot = s_   
    
            # 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 的结合, 既考虑位置, 也考虑角度。
            r = r1 + r2

得分部分不仅考虑了棒子的**theta离垂直的角度与 theta最大角度的差的得分**还**考虑了x水平位移与x最大边距的距离差的得分**  
Actor神经网络的训练部分：

self.acts_prob = tf.nn.softmax(l2)
    label = [self.a]
    with tf.variable_scope('exp_v'):
        # 可以通过最小化-(log_p * R)的方式实现最大化(log_p * R)
        neg_log_prob = tf.nn.sparse_softmax_cross_entropy_with_logits(logits=l2, labels=label)   
        self.exp_v = tf.reduce_mean(neg_log_prob * self.td_error)

**利用交叉熵形式**，我认为更加清晰。  
由于代码并不是自己写的，所以就不上传github了，不过还是欢迎大家关注我和我的github。  
https://github.com/bubbliiiing/  
希望得到朋友们的喜欢。

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

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