强化学习：训练智能出租车学会驾驶

一、强化学习背景知识介绍

强烈建议入门的同学可以先看一下这篇机器之心写的入门文章，以下介绍也会主要参照这篇文章进行介绍。

强化学习的目的

强化学习的目的是希望训练一个智能的实体，使其能够在规定的时间内选择根据不同的情景最优的决策，最终达到一个目标。比如说前几年很火的AlphaGo一代，所用的基本原理就是强化学习。它根据对手下的棋子以及智能体本身棋子的布局构成了智能体所处的环境，它下一步下的棋子就会是当前所身处的环境下所能作出的最大化未来预期收益（也就是最可能赢）的一步动作。

强化学习的组成部分

强化学习的组成部分主要有五个：智能实体（作出策略的主角），奖励（是实体作出动作的动机），环境（智能实体的状态），所能采取的动作（每个不同状态下所能采取的动作都是有限且不同的），目标（要到达的目的地）。
其实强化学习就很像下棋，但是为了更加直观地说明强化学习，用一下文中用巧克力奖励孩子学习的例子。假如说我们想要奖励孩子学习，目标是学习完一本参考书的所有章节并在最后的测试当中取到最高分数。那么孩子很活泼，坐不下来，所以我们要给他一点巧克力当作奖励，激励他采取行动去依次复习每个章节并做测试。但是这样效率会有些低，所以我们要求孩子要在规定时间内完成，如果不能在规定时间内完成就会有惩罚扣减巧克力。那么在这里的话，孩子就是实体，巧克力就是奖励，而环境或者状态就是孩子现在复习的章节，动作就是在孩子在当前章节下接下来从可以复习的章节当中选取一章来复习，那么所有的动作依次序连接起来就构成了一个策略。同时，并不是所有的章节都同样地重要，孩子要自己选择哪些章节更有价值，这就是价值函数所做的（计算出每个章节的价值）。

强化学习的特点

• 延迟奖励。也就是智能实体在决定采取哪个行动的时候并不是只看重下一步所带来的奖励，而是看长远总体的奖励最大化。
• 时间序列。行动具有时序性，下一步行动依赖于上一步行动，智能实体所能采取的所有行动取决于当前的状态。

• 没有大量标注的数据进行监督。在训练的时候每一步都没有办法对系统得到立即的标注，也就是必须要尝试完当前状态下的所有可能的行动，才能得到最终的奖励总和（没有标记的路径的价值是0），这就有点像是不断在摸索道路。

  
  image.png

在每个时间步（t）下，智能体都处于某个状态（一系列观察值组成），看了看周围的道路，决定要采取行动选择一条路，获得一次奖励，这条路走到头之后可以得到最大化的奖励。所以我们要训练智能体不断提升它到达目的地之后能够得到最大奖励。

在这个项目中，目的是构建一个基于Q-learning的智能出租车系统，能够通过对周围环境的训练来适时地自动到达目的地。评判指标有两个：安全性和可靠性。比如说，如果出租车在红灯前面继续形式或者不避让左右来车，那么就会被认为是不安全的。对于可靠性，如果出租车频繁不能按时到达目的地，那么就认为是不可靠的。
安全性和可靠性用字母等级来评估，如下：

等级	安全性	可靠性
A+	代理程序没有任何妨害交通的行为， 并且总是能选择正确的行动。	代理程序在合理时间内到达目的地的次数<br />占行驶次数的100%。
A	代理程序有很少的轻微妨害交通的行为， 如绿灯时未能移动。	代理程序在合理时间内到达目的地的次数<br />占行驶次数的90%。
B	代理程序频繁地有轻微妨害交通行为， 如绿灯时未能移动。	代理程序在合理时间内到达目的地的次数<br />占行驶次数的80%。
C	代理程序有至少一次重大的妨害交通行为， 如闯红灯。	代理程序在合理时间内到达目的地的次数<br />占行驶次数的70%。
D	代理程序造成了至少一次轻微事故， 如绿灯时在对面有车辆情况下左转。	代理程序在合理时间内到达目的地的次数<br />占行驶次数的60%。
F	代理程序造成了至少一次重大事故， 如有交叉车流时闯红灯。	代理程序在合理时间内到达目的地的次数<br />未能达到行驶次数的60%。

二、理解代码

在smartcab/文件夹底下是运行智能实体和模拟世界环境的代码，要构建一个智能实体就要理解它的实现方法。
驾驶代理程序的代码主要放在agent.py里面。

首先是导入要用到的模块，接下来是构建智能车的这个类。

import random
import math
from environment import Agent, Environment
from planner import RoutePlanner
from simulator import Simulator

class LearningAgent(Agent):
    """ An agent that learns to drive in the Smartcab world.
        This is the object you will be modifying. """ 

    def __init__(self, env, learning=True, epsilon=1, alpha=0.5):
        super(LearningAgent, self).__init__(env)     # Set the agent in the evironment 
        self.planner = RoutePlanner(self.env, self)  # Create a route planner
        self.valid_actions = self.env.valid_actions  # The set of valid actions

        # Set parameters of the learning agent
        self.learning = learning # Whether the agent is expected to learn
        self.Q = dict()          # Create a Q-table which will be a dictionary of tuples
        self.epsilon = epsilon   # Random exploration factor
        self.alpha = alpha       # Learning factor

        ###########
        ## TO DO ##
        ###########
        # Set any additional class parameters as needed
        self.t=1

初始化函数 init定义了要用到的变量，定义了包括路线规划的planner，可以操作的有效动作，是否学习，以及Q-table，它是一个由元祖组成的字典，epsilon随机探索系数，alpha是学习系数。

def reset(self, destination=None, testing=True):
        """ The reset function is called at the beginning of each trial.
            'testing' is set to True if testing trials are being used
            once training trials have completed. """

        # Select the destination as the new location to route to
        self.planner.route_to(destination)
        
        ########### 
        ## TO DO ##
        ###########
        # Update epsilon using a decay function of your choice
        # Update additional class parameters as needed
        # If 'testing' is True, set epsilon and alpha to 0
        self.t+=1
        self.a=0.01
        if testing==True:
            self.epsilon=0
            self.alpha=0
        else:
            # self.epsilon-=0.5
            self.epsilon=math.exp(-(self.a*self.t))
        return None

reset函数是在每次开始试验之前就先规定是否要测试，并以此选择不同的epsilon值和alpha值，相当于如果在所有训练次数结束以后，epsilon和alpha的值就会固定是0，而如果是在训练的过程中epsilon就会以一定的我们规定的衰减方式进行变化。

 def build_state(self):
        """ The build_state function is called when the agent requests data from the 
            environment. The next waypoint, the intersection inputs, and the deadline 
            are all features available to the agent. """

        # Collect data about the environment
        waypoint = self.planner.next_waypoint() # The next waypoint 
        inputs = self.env.sense(self)           # Visual input - intersection light and traffic
        deadline = self.env.get_deadline(self)  # Remaining deadline

        ########### 
        ## TO DO ##
        ###########
        # Set 'state' as a tuple of relevant data for the agent        
        state =(waypoint,inputs['light'],inputs['oncoming'],inputs['left'],inputs['right'])

        return state

build_state函数相当于是收集智能体当前的状态信息，包括waypoint（车辆的车头方向），inputs（在十字路口的信息，包括交通灯状况以及来往车辆的行驶方向等），deadline（截止时间），最终的state变量以元组的形式返回。

    def get_maxQ(self, state):
        """ The get_max_Q function is called when the agent is asked to find the
            maximum Q-value of all actions based on the 'state' the smartcab is in. """

        ########### 
        ## TO DO ##
        ###########
        # Calculate the maximum Q-value of all actions for a given state

        maxQ = max(self.Q[state].values())
        return maxQ 

get_maxQ函数是计算给定的一个state（也就是状态）下延伸到终点的所有Q取值的最大值。

    def createQ(self, state):
        """ The createQ function is called when a state is generated by the agent. """

        ########### 
        ## TO DO ##
        ###########
        # When learning, check if the 'state' is not in the Q-table
        # If it is not, create a new dictionary for that state
        #   Then, for each action available, set the initial Q-value to 0.0
        if self.learning:
            if state not in self.Q.keys():
                self.Q[state]={action:0.0 for action in self.valid_actions}
        return

createQ的作用是检查这个状态是否在Q-table当中，如果不在的话，就以当前状态state为键，字典dict为值，其中这个字典dict里面所能采取的有效行动valid_actions为键，对应的值初始化为0。

def choose_action(self, state):
        """ The choose_action function is called when the agent is asked to choose
            which action to take, based on the 'state' the smartcab is in. """

        # Set the agent state and default action
        self.state = state
        self.next_waypoint = self.planner.next_waypoint()
        action = None
        q=self.Q[state]
        ########### 
        ## TO DO ##
        ###########
        # When not learning, choose a random action
        # When learning, choose a random action with '' probability
        #   Otherwise, choose an action with the highest Q-value for the current state
        if not self.learning:
            action = self.valid_actions[random.randint(0,len(self.valid_actions)-1)]
        if self.learning:
            if random.random()<self.epsilon:
                action = self.valid_actions[random.randint(0,len(self.valid_actions)-1)]
            else:
                best=[i for i in self.valid_actions if q[i]==self.get_maxQ(state)]
                action=random.choice(best)
        return action

choose_action这个函数目的是要选择一个动作，里面有两个判断语句，如果是非学习状态下的话，那么我们就随机采取一个有效行动；如果是学习状态下，有两种情况，一种是随机返回的值要小于epsilon，否则的话就从Q-table当中选择一个Q值最大的动作（相当于对号入座）。在这里要说明一下的是，在训练过程中epsilon是会不断减小的，减小的速率是可以通过衰减公式来控制的，但是一定是不断减小的，而随机返回的数字是0到1之间，这样等epsilon衰减到0以后就不可能随机数还有小于epsilon的机会，所以epsilon衰减到0以后，就都会从Q-table中选择一个Q值最大的动作的这种做法。为什么要这么设计呢？是因为Q-table一开始初始化都为0，它的更新是来自于小车不断地去探索未知的状态，有点像是画地图吧，在某个状态下采取某个行动到了一个新的状态就记录下来这个奖励，这样Q-table就会得到更新，而epsilon衰减得越小，探索的区域越多，Q值就越可靠。接下来我们会看到Q-table是如何更新的。

    def learn(self, state, action, reward):
        """ The learn function is called after the agent completes an action and
            receives an award. This function does not consider future rewards 
            when conducting learning. """

        ########### 
        ## TO DO ##
        ###########
        # When learning, implement the value iteration update rule
        #   Use only the learning rate 'alpha' (do not use the discount factor 'gamma')
        if self.learning:
            self.Q[state][action]=self.Q[state][action]+self.alpha*(reward-self.Q[state][action])
        return

learn这个函数当小车在学习的过程中每当完成一个动作都会得到一个reward，对应的Q-table的特定状态下的特定的一个动作的Q值就会得到计算。并且要注意的是，在学习的过程中，并没有考虑到未来的奖励，另外公式也没有涉及折扣因子$\gamma$。这一个留到后面解释。

def update(self):
        """ The update function is called when a time step is completed in the 
            environment for a given trial. This function will build the agent
            state, choose an action, receive a reward, and learn if enabled. """

        state = self.build_state()          # Get current state
        self.createQ(state)                 # Create 'state' in Q-table
        action = self.choose_action(state)  # Choose an action
        reward = self.env.act(self, action) # Receive a reward
        self.learn(state, action, reward)   # Q-learn

        return

update函数相当于是将所有的步骤都整合了起来，先是获得当前状态，然后是创建状态对应的那个条目，然后根据状态选择动作，作出动作之后得到奖励，再根据状态，动作还有奖励来更新Q-table。
最后就是主函数。

def run():
    """ Driving function for running the simulation. 
        Press ESC to close the simulation, or [SPACE] to pause the simulation. """

    ##############
    # Create the environment
    # Flags:
    #   verbose     - set to True to display additional output from the simulation
    #   num_dummies - discrete number of dummy agents in the environment, default is 100
    #   grid_size   - discrete number of intersections (columns, rows), default is (8, 6)
    env = Environment(verbose=True)
    
    ##############
    # Create the driving agent
    # Flags:
    #   learning   - set to True to force the driving agent to use Q-learning
    #    * epsilon - continuous value for the exploration factor, default is 1
    #    * alpha   - continuous value for the learning rate, default is 0.5
    agent = env.create_agent(LearningAgent,alpha=0.5,learning=True)
    
    ##############
    # Follow the driving agent
    # Flags:
    #   enforce_deadline - set to True to enforce a deadline metric
    env.set_primary_agent(agent,enforce_deadline=False)

    ##############
    # Create the simulation
    # Flags:
    #   update_delay - continuous time (in seconds) between actions, default is 2.0 seconds
    #   display      - set to False to disable the GUI if PyGame is enabled
    #   log_metrics  - set to True to log trial and simulation results to /logs
    #   optimized    - set to True to change the default log file name
    sim = Simulator(env,update_delay=0.01,display=True,log_metrics=True,optimized=True)
    
    ##############
    # Run the simulator
    # Flags:
    #   tolerance  - epsilon tolerance before beginning testing, default is 0.05 
    #   n_test     - discrete number of testing trials to perform, default is 0
    sim.run(n_test=10,tolerance=0.01)


if __name__ == '__main__':
    run()

在这里的变量都有注释，值得关注的有几个变量epsilon，alpha，tolerance还有n_test。epsilon是探索系数，正如前面所说，是会不断衰减的，可以理解为我们想让小车探索到什么程度，与tolerance相对应，tolerance是epsilon的节点，小于tolerance之后就会停止训练，开始测试。alpha是学习率，n_test是要测试多少轮。

二、考虑几个问题

奖励情况

当我们先设置learning为False的时候，运行代码，可以在pygame可视化的窗口看到一辆白色小车，并且在对应的log文件可以看到每个动作得到的奖励情况。

image.png image.png image.png

在这里我们可以看到的是每一条记录都是代理程序在每个状态下对应采取的行动以及带来的奖励。在括号里面是代理程序当前的状态，至于分别代表什么，我们可以看一下environment.py里面对应的valid_actions和valid_inputs，environment.py是用来渲染代理程序的运行环境的（也就是给代理程序制定一个规则，包括它可以采取的行动，周围可能遇到的状态）可以发现，代理程序以及路上的其他来车所能采取的行动只有4种，分别是None，forward，left，right；而代理程序状态组成包括“交通灯”、对面来车采取的动作、左边来车所采取的动作和右边来车所采取的动作。

image.png

在模拟过程中出租车没有发生移动 能够正确判明交通灯，比如在绿灯前面不停下，在红灯面前正常停下，都会得到正反馈，除此之外都是负反馈或者是反馈为0。
在绿灯停下也会有正的奖励，比如说类似
('forward', 'green', 'left', None, 'forward')
-- forward : 1.52
-- right : 0.29
-- None : 0.27
-- left : 0.00

('forward', 'green', 'left', 'left', 'forward')
-- forward : 2.25
-- right : 0.00
-- None : 0.01
-- left : 0.00
但是这种正的奖励都是比较小的，基本上是因为左边的车向左转，对面有来车直行，因此在有较多车辆行驶的情况下，为保证不出意外，这个时候也给了绿灯停下较小的奖励。
红灯时，智能车没有移动得到的奖励的范围是0～2.64，在红灯的时候停下没有负值的惩罚。

状态空间

如果我们用过于多的特征来定义小车的状态空间，那么相应地也应该要有足够多的训练才能够让小车真正学到东西，特征过多就有点类似于过拟合。我们计算一下我们定义的特征的所有可能组合情况，也就是状态空间的大小。
如果是以问题4选择的特征来定义一个状态的话，车头有3个朝向，交通灯有2种颜色，'left','right'以及'oncoming'各有4种，分别就是与agent的四种可能'forward', 'None', 'left', 'right'相同，所以一共可能的状态空间是 3×2×4×4×4=384 种，那么应该是384种。 给定合理数量的熟练，代理驾驶可以学到一个较好的策略，原因是代理驾驶可以接触到地图上更多的状态，即使在当下第一次选择的时候随机选择，也可以根据反馈进行下一次重新训练的调整动作获得最大Q值。

合适的epsilon衰减函数

Q-learning当中要选择一个合适的$\epsilon$衰减函数，以合理的速率衰减到tolerance，之后代理程序才可以开始测试。以下是一些衰减函数的例子，t是测试的步数。

image.png

• 我所采用的衰减函数是$\varepsilon =e^{-at}$，并把tolerance降到了0.01，原因是希望尽可能增加训练次数来探索尽可能多的区域，另外一点就是希望这个衰减函数能够让epsilon平缓地衰减。
• 在测试之前驾驶代理大约需要做460次训练。
• 我用的是0.01的epsilon_tolerance和0.5的alpha，一方面希望尽可能地增加训练次数来探索尽可能多的区域，设置比较适中的alpha能够让agent在前期能够积累足够多的经验，但是过大的alpha不利于学习后期（也就是不容易收敛，波动会比较大，学不到东西），因此如果alpha最好是前半段是比较大的，到后半段比较小，如果alpha是一个常数，那么我会选择一个比较适中的。

三、运行结果

最后结果

从最后结果来看，制定了合适的epsilon衰减函数，总共训练了450次，测试10次，这个Q-learning学习器随着训练数量的增加，可靠性在逐步稳定增加，奖励在逐步增加，接收到的惩罚在减少，最终结果安全性和可靠性都达到了A级以上。

四、进一步思考

这个项目中为什么不引入未来奖励？

学习完这个项目，其实我们可以发现Q-value当中并没有涉及到未来的奖励内容，都是短期奖励，也即是都是在当前的Q值基础上再接收到reward再更新。
我认为有两个原因使得未来奖励在这个项目当中不起作用：

1. 关于智能出租车本身它并不知道距离目的地有多远。另外，在实际当中，智能出租车是根据实际情况，车辆，红绿灯变化等情况随机应变的，这些都是随机出现的，不能作为有效的地理信息存储到Q table当中，所以智能出租车只能根据它当前的状态去决定下一步的行动。随机性太大，只能见机行事，根据当前的状态去决定下一步行动，变动性太大使得智能体的状态其实与到达目的地并没有实际上的联系。在走迷宫的例子当中，智能体的状态都是有效的地理信息，具有更稳健的依赖关系。
2. 如果起点和终点固定，未来奖励会随着智能车学习逐渐从目的地“扩散”到起点。我们的训练环境里，每次都随机选择起点和终点，目的地和起点不固定，未来奖励无法从目的地扩散到起点。也即是我们每个状态的未来奖励都是可以不断延伸到终点计算出一个确定的值的，但是在本项目当中每次的起点和终点都是随机的。
   实际上，未来奖励就是与短期奖励相对的，实际在做一个决策的时候，不仅要考虑短期奖励也要考虑未来奖励，而未来奖励是与当前状态有一定关系的，未来奖励能够从目的地扩散到起点。比如走迷宫就是这样的。但在这个项目当中，除了车的waypoint不是随机的，其他的都是随机的，所以未来奖励与当前状态没有关联。所以不会起到作用。

相关资源：

1. 从强化学习基本概念到Q学习的实现，打造自己的迷宫智能体