一、无模型的强化学习

在上一节中介绍了基于模型的强化学习方法(动态规划)，其中的前提是知道环境的状态转移概率，但在实际问题中，状态转移的信息往往无法获知，由此需要数据驱动的无模型(model-free)的方法。

1.1、蒙特卡罗(Monte Carlo)方法

在无模型时，一种自然的想法是通过随机采样的经验平均来估计期望值，此即蒙特卡罗法。其过程可以总结如下：

1. 智能体与环境交互后得到交互序列
2. 通过序列计算出各个时刻的奖赏值
3. 将奖赏值累积到值函数中进行更新
4. 根据更新的值函数来更新策略

在动态规划中，为保证算法的收敛性，算法会逐个扫描状态空间中的状态，计算值函数时用到了当前状态的所有后继状态的值函数。而蒙特卡罗利用经验平均估计状态的值函数，此处的经验是指一次试验，而一次试验要等到终止状态出现才结束，因此学习速度慢，效率不高。下图较直观展示了二者的不同。

动态规划

蒙特卡罗

在蒙特卡罗中，如果采用确定性策略，每次试验的轨迹都是一样的，因此无法进一步改进策略。为了使更多状态-动作对参与到交互过程中，即平衡探索和利用，常用ε-greedy策略来产生动作 ，以保证每个状态-动作对都有机会作为初始状态，在评估状态-动作值函数时，需要对每次试验中所有状态-动作对进行估计。

1.2、时序差分方法

由于蒙特卡罗需要获得完整轨迹，才能进行策略评估并更新，效率较低。时序差分法结合了动态规划和蒙特卡罗，即模拟一段轨迹(一步或者几步)，然后利用贝尔曼方程进行自迭代更新，如下图所示：

时序差分

比较三种方法估计值函数的异同点。蒙特卡罗使用的是值函数最原始的定义，即利用所有奖赏的累积和来估计值函数；动态规划和时序差分则利用一步预测方法来计算当前状态值函数，不同的是，动态规划利用模型计算后继状态，时序差分利用实验得到后继状态。

1.3、Q-Learning

Q-Learning是一种异策略(off policy)的时序差分方法，即动作策略为ε-greedy策略，目标策略为贪婪策略。在更新值函数时并不完全遵循交互序列，而是选择来自其他策略的交互序列的子部分替换了原来的交互序列。从思想来说，它结合了子部分的最优价值，更像是结合了价值迭代的更新算法，希望每一次都使用前面迭代积累的最优结果进行更新。

Q-Learning 算法

Q-Learning的收敛性分析

为简明起见，笔者在此仅做原理上的证明，更加严格的证明可见参考资料[2] P189-193.
根据Q-Learning的更新公式(此处“=”表达赋值含义)：
第一次迭代：
第二次迭代：
......
第n次迭代：
由于：，当n足够大时，有，则：
仍然是最原始的贝尔曼方程的形式，说明该算法是收敛的。
下面说明收敛效果：
引理：输出在上的随机过程的更新过程定义为
如果下面的条件能够满足，那么它将依概率1 收敛到0:
（1），同时保证。
（2），同时。
（3），其中。

现假定值函数收敛且收敛值为，则： 令,,则上式变为： 可以验证满足以上三个条件（验证过程需要一定矩阵理论、概率统计知识，读者可以尝试证明），在此不再赘述。

二、函数近似与DQN

2.1、函数近似(Function Approximation)

在此之前介绍的强化学习方法(动态规划、蒙特卡罗、时序差分)都有一个共同前提：状态空间和动作空间是离散的且不能太大。通常值函数用表格的形式的表示，故又称之为表格型强化学习。而在很多问题中，状态空间维数很大，或者状态空间是连续的，无法用表格表示，故需要函数近似的方式。
事实上，对于状态和动作，值函数，存在这样一个映射：，由此求解值函数的问题转化为监督学习的问题，而监督学习是一种常见且易于解决的问题。线性回归(Linear Regression) 、支持向量机(Support Vector Machine)、决策树(Decision Tree)， 以及神经网络(Neural Network )等都可以用来解决此类问题。
下面通过在数学上说明以上过程：
假定状态空间为维实数空间，且值函数能表达为状态的线性函数，即： 其中为状态向量，为参数向量。
为使学习得到的值函数尽可能近似于真实值函数 ，用最小二乘来度量误差： 其中 表示由策略采样而得到的状态上的期望。

为最小化误差，采用梯度下降，对误差求负导数：

于是得到以下更新规则： 又，用当前估计值函数代替真实值函数，则： 与Q-Learning的更新方式相同，仅仅是值函数的表达不同。

2.2、Deep Q-Network(DQN)

之前大量叙述了强化学习的基本原理，至此才开始真正的深度强化学习的部分。Deep Q-Network，简称DQN，来自论文 Human-level control through deep reinforcement learning。论文主要介绍了如何使用DQN 网络训练Agent 在Atari游戏平台上尽可能获得更多的分数。
与Q-Learning相比，DQN主要改进在以下三个方面：
（1）DQN利用深度卷积网络(Convolutional Neural Networks,CNN)来逼近值函数；
（2）DQN利用经验回放训练强化学习的学习过程；
（3）DQN独立设置了目标网络来单独处理时序差分中的偏差。

DQN 结构

下面主要说明经验回放和目标网络：
经验回放(Replay Buffer)
Q-Leaning 方法基于当前策略进行交互和改进，每一次模型利用交互生成的数据进行学习，学习后的样本被直接丢弃。但如果使用机器学习模型代替表格式模型后再采用这样的在线学习方法，就有可能遇到两个问题，这两个问题也都和机器学习有关。

1. 交互得到的序列存在一定的相关性。交互序列中的状态行动存在着一定的相关性，而对于基于最大似然法的机器学习模型来说，我们有一个很重要的假设：训练样本是独立且来自相同分布的，一旦这个假设不成立，模型的效果就会大打折扣。而上面提到的相关性恰好打破了独立同分布的假设，那么学习得到的值函数模型可能存在很大的波动。
2. 交互数据的使用效率。采用梯度下降法进行模型更新时，模型训练往往需要经过多轮迭代才能收敛。每一次迭代都需要使用一定数量的样本计算梯度， 如果每次计算的样本在计算一次梯度后就被丢弃，那么我们就需要花费更多的时间与环境交互并收集样本。

Replay Buffer 结构

DQN 算法

四、案例分析

寻宝小游戏 规则:在4×4的网格中，红色方块为寻宝者，黄色圆形为宝藏，找到宝藏奖励为1，黑色方块为陷阱，落入陷阱奖励为-1，其他位置奖励为0，通过训练使寻宝者找到通往宝藏的路径。

寻宝 小游戏

环境由使用Tkinter包绘制，为简明并突出算法，在此不予介绍，感兴趣的读者可以下载源代码加以了解。

4.1、Q-Learning实践

1. Q-Learning算法主体
   定义一个 QLearningTable的类，首先初始化参数，其意义分别如注释所示，其中Q值用q_table来表示。choose_action()函数定义如何决定行为，根据所在的 state, 或者是在这个 state 上的 观测值 (observation) 来决策，采用ε-greedy策略。learn()函数决定学习过程，根据是否是 terminal state (回合终止符) 来判断应该如何更行 q_table，其更新方式与算法描述相同。check_state_exist()函数功能就是检测 q_table 中有没有当前 state 的步骤了, 如果还没有当前 state, 那我我们就插入一组全 0 数据, 当做这个 state 的所有 action 初始 values。

import numpy as np
import pandas as pd

class QLearningTable:
    def __init__(self, actions, learning_rate=0.01, reward_decay=0.9, e_greedy=0.9):
        self.actions = actions   # a list
        self.lr = learning_rate   # 学习率
        self.gamma = reward_decay    # 奖励衰减
        self.epsilon = e_greedy     # 贪婪度
        self.q_table = pd.DataFrame(columns=self.actions, dtype=np.float64)    # 初始 q_table

    def choose_action(self, observation):
        self.check_state_exist(observation)   # 检测本 state 是否在 q_table 中存在
        # action selection
        if np.random.uniform() < self.epsilon:
            # choose best action
            state_action = self.q_table.loc[observation, :]
            # some actions may have the same value, randomly choose on in these actions
            action = np.random.choice(state_action[state_action == np.max(state_action)].index)
        else:
            # choose random action
            action = np.random.choice(self.actions)
        return action

    def learn(self, s, a, r, s_):
        self.check_state_exist(s_)    # 检测 q_table 中是否存在 s_
        q_predict = self.q_table.loc[s, a]
        if s_ != 'terminal':
            q_target = r + self.gamma * self.q_table.loc[s_, :].max()  # next state is not terminal
        else:
            q_target = r  # next state is terminal
        self.q_table.loc[s, a] += self.lr * (q_target - q_predict)  # update

    def check_state_exist(self, state):
        if state not in self.q_table.index:
            # append new state to q table
            self.q_table = self.q_table.append(
                pd.Series(
                    [0]*len(self.actions),
                    index=self.q_table.columns,
                    name=state,
                )
            )

2. 环境交互及更新过程
   该部分是整个 Q-Learning最重要的迭代更新部分，主体循环部分简单易读，体现了智能体与环境的交互和更新过程。

from maze_env import Maze
from RL_brain import QLearningTable

def update():
    for episode in range(100):
        # initial observation
        observation = env.reset()

        while True:
            # fresh env
            env.render()

            # RL choose action based on observation
            action = RL.choose_action(str(observation))

            # RL take action and get next observation and reward
            observation_, reward, done = env.step(action)

            # RL learn from this transition
            RL.learn(str(observation), action, reward, str(observation_))

            # swap observation
            observation = observation_

            # break while loop when end of this episode
            if done:
                break

    # end of game
    print('game over')
    env.destroy()

if __name__ == "__main__":
    env = Maze()
    RL = QLearningTable(actions=list(range(env.n_actions)))

    env.after(100, update)
    env.mainloop()

Q-Learning 算法实现的效果如下图所示，可见寻宝者发现宝藏的动态过程。

Q-Learning 效果

4.2、DQN实践

1. 神经网络的搭建
   为了使用 Tensorflow 来实现 DQN, 比较推荐的方式是搭建两个神经网络, target_net 用于预测 q_target 值, 他不会及时更新参数,eval_net 用于预测 q_eval, 这个神经网络拥有最新的神经网络参数. 不过这两个神经网络结构是完全一样的, 只是里面的参数不一样。两个神经网络是为了固定住一个神经网络 (target_net) 的参数, target_net 是 eval_net 的一个历史版本, 拥有 eval_net 很久之前的一组参数, 而且这组参数被固定一段时间, 然后再被 eval_net 的新参数所替换. 而 eval_net 是不断在被提升的, 所以是一个可以被训练的网络 trainable=True. 而 target_net 的 trainable=False。

class DeepQNetwork:
    def _build_net(self):
        # ------------------ build evaluate_net ------------------
        self.s = tf.placeholder(tf.float32, [None, self.n_features], name='s')  # input
        self.q_target = tf.placeholder(tf.float32, [None, self.n_actions], name='Q_target')  # for calculating loss
        with tf.variable_scope('eval_net'):
            # c_names(collections_names) are the collections to store variables
            c_names, n_l1, w_initializer, b_initializer = \
                ['eval_net_params', tf.GraphKeys.GLOBAL_VARIABLES], 10, \
                tf.random_normal_initializer(0., 0.3), tf.constant_initializer(0.1)  # config of layers

            # first layer. collections is used later when assign to target net
            with tf.variable_scope('l1'):
                w1 = tf.get_variable('w1', [self.n_features, n_l1], initializer=w_initializer, collections=c_names)
                b1 = tf.get_variable('b1', [1, n_l1], initializer=b_initializer, collections=c_names)
                l1 = tf.nn.relu(tf.matmul(self.s, w1) + b1)

            # second layer. collections is used later when assign to target net
            with tf.variable_scope('l2'):
                w2 = tf.get_variable('w2', [n_l1, self.n_actions], initializer=w_initializer, collections=c_names)
                b2 = tf.get_variable('b2', [1, self.n_actions], initializer=b_initializer, collections=c_names)
                self.q_eval = tf.matmul(l1, w2) + b2

        with tf.variable_scope('loss'):
            self.loss = tf.reduce_mean(tf.squared_difference(self.q_target, self.q_eval))
        with tf.variable_scope('train'):
            self._train_op = tf.train.RMSPropOptimizer(self.lr).minimize(self.loss)

        # ------------------ build target_net ------------------
        self.s_ = tf.placeholder(tf.float32, [None, self.n_features], name='s_')    # input
        with tf.variable_scope('target_net'):
            # c_names(collections_names) are the collections to store variables
            c_names = ['target_net_params', tf.GraphKeys.GLOBAL_VARIABLES]

            # first layer. collections is used later when assign to target net
            with tf.variable_scope('l1'):
                w1 = tf.get_variable('w1', [self.n_features, n_l1], initializer=w_initializer, collections=c_names)
                b1 = tf.get_variable('b1', [1, n_l1], initializer=b_initializer, collections=c_names)
                l1 = tf.nn.relu(tf.matmul(self.s_, w1) + b1)

            # second layer. collections is used later when assign to target net
            with tf.variable_scope('l2'):
                w2 = tf.get_variable('w2', [n_l1, self.n_actions], initializer=w_initializer, collections=c_names)
                b2 = tf.get_variable('b2', [1, self.n_actions], initializer=b_initializer, collections=c_names)
                self.q_next = tf.matmul(l1, w2) + b2

通过tensorboard绘制的网络结构图如下图所示。

网络结构图

2. 思维决策的过程
   定义完上次的神经网络部分以后, 这次来定义其他部分，首先是函数值的初始化。

class DeepQNetwork:
    def __init__(
            self,
            n_actions,
            n_features,
            learning_rate=0.01,
            reward_decay=0.9,
            e_greedy=0.9,
            replace_target_iter=300,
            memory_size=500,
            batch_size=32,
            e_greedy_increment=None,
            output_graph=False,
    ):
        self.n_actions = n_actions
        self.n_features = n_features
        self.lr = learning_rate
        self.gamma = reward_decay
        self.epsilon_max = e_greedy  # epsilon 的最大值
        self.replace_target_iter = replace_target_iter    # 更换 target_net 的步数
        self.memory_size = memory_size    # 记忆上限
        self.batch_size = batch_size       # 每次更新时从 memory 里面取多少记忆出来
        self.epsilon_increment = e_greedy_increment   # epsilon 的增量
        self.epsilon = 0 if e_greedy_increment is not None else self.epsilon_max     # 是否开启探索模式, 并逐步减少探索次数

        # total learning step
        self.learn_step_counter = 0

        # initialize zero memory [s, a, r, s_]
        self.memory = np.zeros((self.memory_size, n_features * 2 + 2))

        # consist of [target_net, evaluate_net]
        self._build_net()
        t_params = tf.get_collection('target_net_params')  # 提取 target_net 的参数
        e_params = tf.get_collection('eval_net_params')    # 提取  eval_net 的参数
        self.replace_target_op = [tf.assign(t, e) for t, e in zip(t_params, e_params)]

        self.sess = tf.Session()

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

        self.sess.run(tf.global_variables_initializer())
        self.cost_his = []   # 记录所有 cost 变化, 用于最后 plot 出来观看

记忆存储，DQN 的精髓部分之一: 记录下所有经历过的步, 这些步可以进行反复的学习, 所以是一种 off-policy 方法。

class DeepQNetwork:
    def store_transition(self, s, a, r, s_):
        if not hasattr(self, 'memory_counter'):
            self.memory_counter = 0

        transition = np.hstack((s, [a, r], s_))

        # replace the old memory with new memory
        index = self.memory_counter % self.memory_size
        self.memory[index, :] = transition

        self.memory_counter += 1

行为选择，让 eval_net 神经网络生成所有 action 的值, 并选择值最大的 action；学习过程就是在 DeepQNetwork 中, 是如何学习, 更新参数的. 这里涉及了 target_net 和 eval_net 的交互使用，这是非常重要的一步。

class DeepQNetwork:
    def choose_action(self, observation):
        # to have batch dimension when feed into tf placeholder
        observation = observation[np.newaxis, :]

        if np.random.uniform() < self.epsilon:
            # forward feed the observation and get q value for every actions
            actions_value = self.sess.run(self.q_eval, feed_dict={self.s: observation})
            action = np.argmax(actions_value)
        else:
            action = np.random.randint(0, self.n_actions)
        return action

    def learn(self):
        # check to replace target parameters
        if self.learn_step_counter % self.replace_target_iter == 0:
            self.sess.run(self.replace_target_op)
            print('\ntarget_params_replaced\n')

        # sample batch memory from all memory
        if self.memory_counter > self.memory_size:
            sample_index = np.random.choice(self.memory_size, size=self.batch_size)
        else:
            sample_index = np.random.choice(self.memory_counter, size=self.batch_size)
        batch_memory = self.memory[sample_index, :]

         # 获取 q_next (target_net 产生了 q) 和 q_eval(eval_net 产生的 q)
        q_next, q_eval = self.sess.run(
            [self.q_next, self.q_eval],
            feed_dict={
                self.s_: batch_memory[:, -self.n_features:],  # fixed params
                self.s: batch_memory[:, :self.n_features],  # newest params
            })

        # 下面这几步十分重要. q_next, q_eval 包含所有 action 的值,
        # 而我们需要的只是已经选择好的 action 的值, 其他的并不需要.
        # 所以我们将其他的 action 值全变成 0, 将用到的 action 误差值 反向传递回去, 作为更新凭据.
        # 这是我们最终要达到的样子, 比如 q_target - q_eval = [1, 0, 0] - [-1, 0, 0] = [2, 0, 0]
        # q_eval = [-1, 0, 0] 表示这一个记忆中有我选用过 action 0, 而 action 0 带来的 Q(s, a0) = -1, 所以其他的 Q(s, a1) = Q(s, a2) = 0.
        # q_target = [1, 0, 0] 表示这个记忆中的 r+gamma*maxQ(s_) = 1, 而且不管在 s_ 上我们取了哪个 action,
        # 我们都需要对应上 q_eval 中的 action 位置, 所以就将 1 放在了 action 0 的位置.

        # 下面是为了达到上面说的目的, 不过为了更方面让程序运算, 达到目的的过程有点不同.
        # 是将 q_eval 全部赋值给 q_target, 这时 q_target-q_eval 全为 0,
        # 不过 我们再根据 batch_memory 当中的 action 这个 column 来给 q_target 中的对应的 memory-action 位置来修改赋值.
        # 使新的赋值为 reward + gamma * maxQ(s_), 这样 q_target-q_eval 就可以变成我们所需的样子.
        # change q_target w.r.t q_eval's action
        q_target = q_eval.copy()

        batch_index = np.arange(self.batch_size, dtype=np.int32)
        eval_act_index = batch_memory[:, self.n_features].astype(int)
        reward = batch_memory[:, self.n_features + 1]

        q_target[batch_index, eval_act_index] = reward + self.gamma * np.max(q_next, axis=1)

        # train eval network
        _, self.cost = self.sess.run([self._train_op, self.loss],
                                     feed_dict={self.s: batch_memory[:, :self.n_features],
                                                self.q_target: q_target})
        self.cost_his.append(self.cost)

        # increasing epsilon
        self.epsilon = self.epsilon + self.epsilon_increment if self.epsilon < self.epsilon_max else self.epsilon_max
        self.learn_step_counter += 1

3. 交互过程
   DQN 与环境交互的过程总体与Q-Learning一致，仅仅增加了记忆存储的过程，这与前边提到的 “Q-Leaning 方法基于当前策略进行交互和改进，每一次模型利用交互生成的数据进行学习，学习后的样本被直接丢弃” 是一致的。

from maze_env import Maze
from RL_brain import DeepQNetwork

def run_maze():
    step = 0
    for episode in range(1000):
        # initial observation
        observation = env.reset()

        while True:
            # fresh env
            env.render()

            # RL choose action based on observation
            action = RL.choose_action(observation)

            # RL take action and get next observation and reward
            observation_, reward, done = env.step(action)

            RL.store_transition(observation, action, reward, observation_)

            if (step > 200) and (step % 5 == 0):
                RL.learn()

            # swap observation
            observation = observation_

            # break while loop when end of this episode
            if done:
                break
            step += 1

    # end of game
    print('game over')
    env.destroy()


if __name__ == "__main__":
    # maze game
    env = Maze()
    RL = DeepQNetwork(env.n_actions, env.n_features,
                      learning_rate=0.01,
                      reward_decay=0.9,
                      e_greedy=0.9,
                      replace_target_iter=200,
                      memory_size=20000,
                      output_graph=True
                      )
    env.after(100, run_maze)
    env.mainloop()
    RL.plot_cost()

最后展示一下运行效果及代价函数值的变化情况，可以看出，整体呈下降趋势，但仍存在明显的波动，这是因为 DQN 中的 input 数据是一步步改变的, 而且会根据学习情况, 获取到不同的数据. 所以这并不像一般的监督学习, DQN 的 cost 曲线就有所不同了。

DQN 效果

代价函数值

获取完整代码请点击这里，感谢莫烦的贡献。

参考资料

[1] Barto A G, Sutton R S. Reinforcement Learning: An Introduction.MIT press, 2018.
[2] 冯超著 强化学习精要：核心算法与TensorFlow实现. ----北京：电子工业出版社 2018.
[3] 郭宪，方纯勇编著 深入浅出强化学习：原理入门. ----北京：电子工业出版社 2018.
[4] 邱锡鹏著，神经网络与深度学习. https://nndl.github.io/ 2019.

人生天地间，忽如远行客。----《古诗十九首》