【翻译】Sklearn与TensorFlow机器学习实用指南 —— 第16章 强化学习(下)

时间差分学习与 Q 学习

具有离散动作的强化学习问题通常可以被建模为马尔可夫决策过程,但是智能体最初不知道转移概率是什么(它不知道T),并且它不知道奖励会是什么(它不知道R)。它必须经历每一个状态和每一次转变并且至少知道一次奖励,并且如果要对转移概率进行合理的估计,就必须经历多次。

时间差分学习(TD 学习)算法与数值迭代算法非常类似,但考虑到智能体仅具有 MDP 的部分知识。一般来说,我们假设智能体最初只知道可能的状态和动作,没有更多了。智能体使用探索策略,例如,纯粹的随机策略来探索 MDP,并且随着它的发展,TD 学习算法基于实际观察到的转换和奖励来更新状态值的估计(见公式 16-4)。

【翻译】Sklearn与TensorFlow机器学习实用指南 —— 第16章 强化学习(下)

其中:

a是学习率(例如 0.01)

TD 学习与随机梯度下降有许多相似之处,特别是它一次处理一个样本的行为。就像 SGD 一样,只有当你逐渐降低学习速率时,它才能真正收敛(否则它将在极值点震荡)。

对于每个状态S,该算法只跟踪智能体离开该状态时立即获得的奖励的平均值,再加上它期望稍后得到的奖励(假设它的行为最佳)。
类似地,此时的Q 学习算法是 Q 值迭代算法的改编版本,其适应转移概率和回报在初始未知的情况(见公式16-5)。

【翻译】Sklearn与TensorFlow机器学习实用指南 —— 第16章 强化学习(下)

对于每一个状态动作对(s,a),该算法跟踪智能体在以动作A离开状态S时获得的即时奖励平均值R,加上它期望稍后得到的奖励。由于目标策略将最优地运行,所以我们取下一状态的 Q 值估计的最大值。

以下是如何实现 Q 学习:

import numpy.random as rnd
learning_rate0 = 0.05 
learning_rate_decay = 0.1 
n_iterations = 20000
s = 0 # 在状态 0开始
Q = np.full((3, 3), -np.inf)  # -inf 对应着不可能的动作 
for state, actions in enumerate(possible_actions):    
    Q[state, actions] = 0.0  # 对于所有可能的动作初始化为 0.0
for iteration in range(n_iterations):    
    a = rnd.choice(possible_actions[s])  # 随机选择动作   
    sp = rnd.choice(range(3), p=T[s, a]) # 使用 T[s, a] 挑选下一状态   
    reward = R[s, a, sp]    
    learning_rate = learning_rate0 / (1 + iteration * learning_rate_decay)    
    Q[s, a] = learning_rate * Q[s, a] + (1 - learning_rate) * (reward + discount_rate * np.max(Q[sp]))    
    s = sp # 移动至下一状态

给定足够的迭代,该算法将收敛到最优 Q 值。这被称为离线策略算法,因为正在训练的策略不是正在执行的策略。令人惊讶的是,该算法能够通过观察智能体行为随机学习(例如学习当你的老师是一个醉猴子时打高尔夫球)最佳策略。我们能做得更好吗?

探索策略

当然,只有在探索策略充分探索 MDP 的情况下,Q 学习才能起作用。尽管一个纯粹的随机策略保证最终访问每一个状态和每个转换多次,但可能需要很长的时间这样做。因此,一个更好的选择是使用 ε 贪婪策略:在每个步骤中,它以概率ε随机地或以概率为1-ε贪婪地(选择具有最高 Q 值的动作)。ε 贪婪策略的优点(与完全随机策略相比)是,它将花费越来越多的时间来探索环境中有趣的部分,因为 Q 值估计越来越好,同时仍花费一些时间访问 MDP 的未知区域。以ε为很高的值(例如,1)开始,然后逐渐减小它(例如,下降到 0.05)是很常见的。

可选择的,相比于依赖于探索的可能性,另一种方法是鼓励探索策略来尝试它以前没有尝试过的行动。这可以被实现为附加于 Q 值估计的奖金,如公式 16-6 所示。

【翻译】Sklearn与TensorFlow机器学习实用指南 —— 第16章 强化学习(下)

其中:

N计算了在状态s时选择动作a的次数
f是一个探索函数,例如f=q+K/(1+n),其中K是一个好奇超参数,它测量智能体被吸引到未知状态的程度。

近似 Q 学习

Q 学习的主要问题是,它不能很好地扩展到具有许多状态和动作的大(甚至中等)的 MDP。试着用 Q 学习来训练一个智能体去玩 Ms. Pac-Man。Ms. Pac-Man 可以吃超过 250 粒粒子,每一粒都可以存在或不存在(即已经吃过)。因此,可能状态的数目大于 2 的 250 次幂,约等于 10 的 75 次幂(并且这是考虑颗粒的可能状态)。这比在可观测的宇宙中的原子要多得多,所以你绝对无法追踪每一个 Q 值的估计值。

解决方案是找到一个函数,使用可管理数量的参数来近似 Q 值。这被称为近似 Q 学习。多年来,人们都是手工在状态中提取并线性组合特征(例如,最近的鬼的距离,它们的方向等)来估计 Q 值,但是 DeepMind 表明使用深度神经网络可以工作得更好,特别是对于复杂的问题。它不需要任何特征工程。用于估计 Q 值的 DNN 被称为深度 Q 网络(DQN),并且使用近似 Q 学习的 DQN 被称为深度 Q 学习。

在本章的剩余部分,我们将使用深度 Q 学习来训练一个智能体去玩 Ms. Pac-Man,就像 DeepMind 在 2013 所做的那样。代码可以很容易地调整,调整后学习去玩大多数 Atari 游戏的效果都相当好。在大多数动作游戏中,它可以达到超人的技能,但它在长时运行的游戏中却不太好。

学习去使用深度 Q 学习来玩 Ms.Pac-Man

由于我们将使用 Atari 环境,我们必须首先安装 OpenAI gym 的 Atari 环境依赖项。当需要玩其他的时候,我们也会为你想玩的其他 OpenAI gym 环境安装依赖项。在 macOS 上,假设你已经安装了 Homebrew 程序,你需要运行:

$ brew install cmake boost boost-python sdl2 swig wget

在 Ubuntu 上,输入以下命令(如果使用 Python 2,用 Python 替换 Python 3):

$ apt-get install -y python3-numpy python3-dev cmake zlib1g-dev libjpeg-dev\    xvfb libav-tools xorg-dev python3-opengl libboost-all-dev libsdl2-dev swig

随后安装额外的 python 包:

$ pip3 install --upgrade 'gym[all]'

如果一切顺利,你应该能够创造一个 Ms.Pac-Man 环境:

>>> env = gym.make("MsPacman-v0") 
>>> obs = env.reset() 
>>> obs.shape  # [长,宽,通道] 
(210, 160, 3) 
>>> env.action_space 
Discrete(9)

正如你所看到的,有九个离散动作可用,它对应于操纵杆的九个可能位置(左、右、上、下、中、左上等),观察结果是 Atari 屏幕的截图(见图 16-9,左),表示为 3D Numpy 矩阵。这些图像有点大,所以我们将创建一个小的预处理函数,将图像裁剪并缩小到88×80像素,将其转换成灰度,并提高 Ms.Pac-Man 的对比度。这将减少 DQN 所需的计算量,并加快培训练。

mspacman_color = np.array([210, 164, 74]).mean()

def preprocess_observation(obs):    
    img = obs[1:176:2, ::2] # 裁剪    
    img = img.mean(axis=2) # 灰度化    
    img[img==mspacman_color] = 0 # 提升对比度    
    img = (img - 128) / 128 - 1 # 正则化为-1到1.   
    return img.reshape(88, 80, 1)

过程的结果如图 16-9 所示(右)。

【翻译】Sklearn与TensorFlow机器学习实用指南 —— 第16章 强化学习(下)

接下来,让我们创建 DQN。它可以只取一个状态动作对(S,A)作为输入,并输出相应的 Q 值Q(s,a)的估计值,但是由于动作是离散的,所以使用只使用状态S作为输入并输出每个动作的一个 Q 值估计的神经网络是更方便的。DQN 将由三个卷积层组成,接着是两个全连接层,其中包括输出层(如图 16-10)。

【翻译】Sklearn与TensorFlow机器学习实用指南 —— 第16章 强化学习(下)

正如我们将看到的,我们将使用的训练算法需要两个具有相同架构(但不同参数)的 DQN:一个将在训练期间用于驱动 Ms.Pac-Man(the actor,行动者),另一个将观看行动者并从其试验和错误中学习(the critic,评判者)。每隔一定时间,我们把评判者网络复制给行动者网络。因为我们需要两个相同的 DQN,所以我们将创建一个q_network()函数来构建它们:

from tensorflow.contrib.layers import convolution2d, fully_connected
input_height = 88 
input_width = 80 
input_channels = 1 
conv_n_maps = [32, 64, 64] 
conv_kernel_sizes = [(8,8), (4,4), (3,3)] 
conv_strides = [4, 2, 1] 
conv_paddings = ["SAME"]*3 
conv_activation = [tf.nn.relu]*3 
n_hidden_in = 64 * 11 * 10  # conv3 有 64 个 11x10 映射
each n_hidden = 512 
hidden_activation = tf.nn.relu 
n_outputs = env.action_space.n  # 9个离散动作
initializer = tf.contrib.layers.variance_scaling_initializer()

def q_network(X_state, scope):    
    prev_layer = X_state    
    conv_layers = []    
    with tf.variable_scope(scope) as scope:        
        for n_maps, kernel_size, stride, padding, activation in zip(conv_n_maps,                                                                  conv_kernel_sizes, 
                                                                    conv_strides,
                                                                    conv_paddings,                                                                  conv_activation):           
            prev_layer = convolution2d(prev_layer, 
                                       num_outputs=n_maps, 
                                       kernel_size=kernel_size,
                                       stride=stride, padding=padding, 
                                       activation_fn=activation,
                                       weights_initializer=initializer)           
            conv_layers.append(prev_layer)       
        last_conv_layer_flat = tf.reshape(prev_layer, shape=[-1, n_hidden_in])     
        hidden = fully_connected(last_conv_layer_flat, n_hidden,
                                 activation_fn=hidden_activation,                                                  weights_initializer=initializer) 
        outputs = fully_connected(hidden, n_outputs, 
                                  activation_fn=None,
                                  weights_initializer=initializer)  

    trainable_vars = tf.get_collection(tf.GraphKeys.TRAINABLE_VARIABLES,
                                       scope=scope.name)                
    trainable_vars_by_name = {var.name[len(scope.name):]: var 
                              for var in trainable_vars}    
    return outputs, trainable_vars_by_name

该代码的第一部分定义了DQN体系结构的超参数。然后q_network()函数创建 DQN,将环境的状态X_state作为输入,以及变量范围的名称。请注意,我们将只使用一个观察来表示环境的状态,因为几乎没有隐藏的状态(除了闪烁的物体和鬼魂的方向)。

trainable_vars_by_name字典收集了所有 DQN 的可训练变量。当我们创建操作以将评论家 DQN 复制到行动者 DQN 时,这将是有用的。字典的键是变量的名称,去掉与范围名称相对应的前缀的一部分。看起来像这样:

>>> trainable_vars_by_name 
{'/Conv/biases:0': <tensorflow.python.ops.variables.Variable at 0x121cf7b50>, '/Conv/weights:0': <tensorflow.python.ops.variables.Variable...>, 
'/Conv_1/biases:0': <tensorflow.python.ops.variables.Variable...>, '/Conv_1/weights:0': <tensorflow.python.ops.variables.Variable...>, '/Conv_2/biases:0': <tensorflow.python.ops.variables.Variable...>, '/Conv_2/weights:0': <tensorflow.python.ops.variables.Variable...>, '/fully_connected/biases:0': <tensorflow.python.ops.variables.Variable...>, '/fully_connected/weights:0': <tensorflow.python.ops.variables.Variable...>, '/fully_connected_1/biases:0': <tensorflow.python.ops.variables.Variable...>, '/fully_connected_1/weights:0': <tensorflow.python.ops.variables.Variable...>}

现在让我们为两个 DQN 创建输入占位符,以及复制评论家 DQN 给行动者 DQN 的操作:

X_state = tf.placeholder(tf.float32, 
                         shape=[None, input_height, input_width,input_channels])          
actor_q_values, actor_vars = q_network(X_state, scope="q_networks/actor") 
critic_q_values, critic_vars = q_network(X_state, scope="q_networks/critic")
copy_ops = [actor_var.assign(critic_vars[var_name])  
            for var_name, actor_var in actor_vars.items()] 
copy_critic_to_actor = tf.group(*copy_ops)

让我们后退一步:我们现在有两个 DQN,它们都能够将环境状态(即预处理观察)作为输入,并输出在该状态下的每一个可能的动作的估计 Q 值。另外,我们有一个名为copy_critic_to_actor的操作,将评论家 DQN 的所有可训练变量复制到行动者 DQN。我们使用 TensorFlow 的tf.group()函数将所有赋值操作分组到一个方便的操作中。

行动者 DQN 可以用来扮演 Ms.Pac-Man(最初非常糟糕)。正如前面所讨论的,你希望它足够深入地探究游戏,所以通常情况下你想将它用 ε 贪婪策略或另一种探索策略相结合。

但是评论家 DQN 呢?它如何去学习玩游戏?简而言之,它将试图使其预测的 Q 值去匹配行动者通过其经验的游戏估计的 Q 值。具体来说,我们将让行动者玩一段时间,把所有的经验保存在回放记忆存储器中。每个记忆将是一个 5 元组(状态、动作、下一状态、奖励、继续),其中“继续”项在游戏结束时等于 0,否则为 1。接下来,我们定期地从回放存储器中采样一批记忆,并且我们将估计这些存储器中的 Q 值。最后,我们将使用监督学习技术训练评论家 DQN 去预测这些 Q 值。每隔几个训练周期,我们会把评论家 DQN 复制到行动者 DQN。就这样!公式 16-7 示出了用于训练评论家 DQN 的损失函数:

【翻译】Sklearn与TensorFlow机器学习实用指南 —— 第16章 强化学习(下)

其中:

【翻译】Sklearn与TensorFlow机器学习实用指南 —— 第16章 强化学习(下)

回放记忆是可选的,但强烈推荐使它存在。没有它,你会训练评论家 DQN 使用连续的经验,这可能是相关的。这将引入大量的偏差并且减慢训练算法的收敛性。通过使用回放记忆,我们确保馈送到训练算法的存储器可以是不相关的。

让我们添加评论家 DQN 的训练操作。首先,我们需要能够计算其在存储器批处理中的每个状态动作的预测 Q 值。由于 DQN 为每一个可能的动作输出一个 Q 值,所以我们只需要保持与在该存储器中实际选择的动作相对应的 Q 值。为此,我们将把动作转换成一个热向量(记住这是一个满是 0 的向量,除了第i个索引中的1),并乘以 Q 值:这将删除所有与记忆动作对应的 Q 值外的 Q 值。然后只对第一轴求和,以获得每个存储器所需的 Q 值预测。

X_action = tf.placeholder(tf.int32, shape=[None]) 
q_value = tf.reduce_sum(critic_q_values * tf.one_hot(X_action, n_outputs), axis=1, keep_dims=True)

接下来,让我们添加训练操作,假设目标Q值将通过占位符馈入。我们还创建了一个不可训练的变量global_step。优化器的minimize()操作将负责增加它。另外,我们创建了init操作和Saver。

y = tf.placeholder(tf.float32, shape=[None, 1]) 
cost = tf.reduce_mean(tf.square(y - q_value)) 
global_step = tf.Variable(0, trainable=False, name='global_step') 
optimizer = tf.train.AdamOptimizer(learning_rate) 
training_op = optimizer.minimize(cost, global_step=global_step)
init = tf.global_variables_initializer() 
saver = tf.train.Saver()

这就是训练阶段的情况。在我们查看执行阶段之前,我们需要一些工具。首先,让我们从回放记忆开始。我们将使用一个deque列表,因为在将数据推送到队列中并在达到最大内存大小时从列表的末尾弹出它们使是非常有效的。我们还将编写一个小函数来随机地从回放记忆中采样一批处理:

from collections import deque

replay_memory_size = 10000 
replay_memory = deque([], maxlen=replay_memory_size)

def sample_memories(batch_size):    
    indices = rnd.permutation(len(replay_memory))[:batch_size]    
    cols = [[], [], [], [], []] # state, action, reward, next_state, continue    
    for idx in indices:        
        memory = replay_memory[idx]        
        for col, value in zip(cols, memory):            
            col.append(value)    
    cols = [np.array(col) for col in cols]    
    return (cols[0], cols[1], cols[2].reshape(-1, 1), cols[3],cols[4].reshape(-1, 1))

接下来,我们需要行动者来探索游戏。我们使用 ε 贪婪策略,并在 50000 个训练步骤中逐步将ε从 1 降低到 0.05。

eps_min = 0.05 
eps_max = 1.0 
eps_decay_steps = 50000
def epsilon_greedy(q_values, step):    
    epsilon = max(eps_min, eps_max - (eps_max-eps_min) * step/eps_decay_steps)   
     if rnd.rand() < epsilon:        
        return rnd.randint(n_outputs) # 随机动作   
    else:        
        return np.argmax(q_values) # 最优动作

就是这样!我们准备好开始训练了。执行阶段不包含太复杂的东西,但它有点长,所以深呼吸。准备好了吗?来次够!首先,让我们初始化几个变量:

n_steps = 100000  # 总的训练步长 
training_start = 1000  # 在游戏1000次迭代后开始训练
training_interval = 3  # 每3次迭代训练一次
save_steps = 50  # 每50训练步长保存模型
copy_steps = 25  # 每25训练步长后复制评论家Q值到行动者
discount_rate = 0.95 
skip_start = 90  # 跳过游戏开始(只是等待时间) 
batch_size = 50 
iteration = 0  # 游戏迭代
checkpoint_path = "./my_dqn.ckpt" 
done = True # env 需要被重置

接下来,让我们打开会话并开始训练:

with tf.Session() as sess:    
    if os.path.isfile(checkpoint_path):        
        saver.restore(sess, checkpoint_path)    
    else:        
        init.run()    
    while True:        
        step = global_step.eval()        
        if step >= n_steps:            
            break        
        iteration += 1        
        if done: # 游戏结束,重来            
            obs = env.reset()            
            for skip in range(skip_start): # 跳过游戏开头              
                obs, reward, done, info = env.step(0)            
                state = preprocess_observation(obs)

        # 行动者评估要干什么        
        q_values = actor_q_values.eval(feed_dict={X_state: [state]})        
        action = epsilon_greedy(q_values, step)

        # 行动者开始玩游戏       
        obs, reward, done, info = env.step(action)        
        next_state = preprocess_observation(obs)

        # 让我们记下来刚才发生了啥        
        replay_memory.append((state, action, reward, next_state, 1.0 - done))        state = next_state

        if iteration < training_start or iteration % training_interval != 0:                continue

        # 评论家学习        
        X_state_val, X_action_val, rewards, X_next_state_val, continues = (            sample_memories(batch_size))        
        next_q_values = actor_q_values.eval( feed_dict={X_state: X_next_state_val})    
        max_next_q_values = np.max(next_q_values, axis=1, keepdims=True)        
        y_val = rewards + continues * discount_rate * max_next_q_values        
        training_op.run(feed_dict={X_state: X_state_val,X_action: X_action_val, y: y_val})

        # 复制评论家Q值到行动者        
        if step % copy_steps == 0:            
            copy_critic_to_actor.run()

        # 保存模型        
        if step % save_steps == 0:            
            saver.save(sess, checkpoint_path)

如果检查点文件存在,我们就开始恢复模型,否则我们只需初始化变量。然后,主循环开始,其中iteration计算从程序开始以来游戏步骤的总数,同时step计算从训练开始的训练步骤的总数(如果恢复了检查点,也恢复全局步骤)。然后代码重置游戏(跳过第一个无聊的等待游戏的步骤,这步骤啥都没有)。接下来,行动者评估该做什么,并且玩游戏,并且它的经验被存储在回放记忆中。然后,每隔一段时间(热身期后),评论家开始一个训练步骤。它采样一批回放记忆,并要求行动者估计下一状态的所有动作的Q值,并应用公式 16-7 来计算目标 Q 值y_val.这里唯一棘手的部分是,我们必须将下一个状态的 Q 值乘以continues向量,以将对应于游戏结束的记忆 Q 值清零。接下来,我们进行训练操作,以提高评论家预测 Q 值的能力。最后,我们定期将评论家的 Q 值复制给行动者,然后保存模型。

不幸的是,训练过程是非常缓慢的:如果你使用你的破笔记本电脑进行训练的话,想让 Ms. Pac-Man 变好一点点你得花好几天,如果你看看学习曲线,计算一下每次的平均奖励,你会发现到它是非常嘈杂的。在某些情况下,很长一段时间内可能没有明显的进展,直到智能体学会在合理的时间内生存。如前所述,一种解决方案是将尽可能多的先验知识注入到模型中(例如,通过预处理、奖励等),也可以尝试通过首先训练它来模仿基本策略来引导模型。在任何情况下,RL仍然需要相当多的耐心和调整,但最终结果是非常令人兴奋的。

原文发布时间为:2018-07-09
本文作者:ApacheCN【翻译】
本文来自云栖社区合作伙伴“Python爱好者社区”,了解相关信息可以关注“Python爱好者社区

上一篇:VS输出窗口(output view)的小技巧--文件行号字符定位


下一篇:后台(27)——文件上传