强化学习（一）：Q-learning，附源码解读

• Q-learning

强化学习与有监督学习和无监督学习为机器学习的三个方向，它主要解决的是决策问题，尤其是连续决策问题。

插入一幅强化学习框图，其中

学习主体（Agent）：强化学习体系中的“学习者”；
环境（Environment）：主体的行为再环境中产生、环境对主体产生影响；
状态（State）：环境反馈给主体的状态的集合；
奖赏（Reward）：环境对主体行为的反馈的集合；
行为（Action）：主体在环境中的行动的集合。

强化学习根据不同条件有不同的分类，这篇文章讲一讲基于价值的离线强化学习：Q-learning。

Q-learning

Q-learning的目的是学习特定State下、特定Action的价值。建立一个Q-table，以State为行、Action为列，通过每个动作带来的奖赏Reward更新Q-table。

Q-learning属于离线学习，它是一种异策略的学习，所谓的异策略的意思是指行动策略和评估策略不是一个策略。

Q即为Q(s,a)就是在某一时刻的s状态下(s∈S)，采取动作a(a∈A)动作能够获得收益的期望，环境会根据agent的动作反馈相应的回报Reward r，所以算法的主要思想就是将State与Action构建成一张Q-table来存储Q值，然后根据Q值来选取能够获得最大收益的动作。

下面附上源码解析，需要公式推导的小伙伴参考这位博主的博客：
https://blog.csdn.net/qq_30615903/article/details/80739243

几个说明：
1.下面所说的窗口也可以叫做画布，个人比较喜欢叫窗口；
2.源码最后会附上百度云链接给大家，因为文件不是很大，就放百度云上了，放github可能会碰到内网连不上等问题，那还不如百度云；
3.如果有解释错误的地方大家见谅，并及时告知博主一下，互相学习，谢谢。

这里按照运行顺序进行剖析：

主文件是q_learning_agent.py，首先运行主函数，会生成一个env的类，其继承了父类Env()，首先进行初始化init：

主函数：

if __name__ == "__main__":
    env = Env()
    agent = QLearningAgent(actions=list(range(env.n_actions)))
    for episode in range(1000):  #循环1000次
        state = env.reset()
        while True:
            env.render()
            # agent产生动作
            action = agent.get_action(str(state))   #str()是把数字转为字符串
            next_state, reward, done = env.step(action)
            # 更新Q表
            agent.learn(str(state), action, reward, str(next_state))
            state = next_state   #状态更新
            env.print_value_all(agent.q_table)
            # 当到达终点就终止游戏开始新一轮训练
            if done:
                break

env = Env()：

class Env(tk.Tk):    #创建一个父类窗口    
    def __init__(self):
        super(Env, self).__init__()    #super(Ecv,self)首先找到Env的父类（tk.Tk)，然后把类Env的对象转换为类(tk.Tk)的对象
        self.action_space = ['u', 'd', 'l', 'r']   #动作空间，上下左右
        self.n_actions = len(self.action_space)   #动作个数
        self.title('Q Learning')   #应该是窗口标题
        self.geometry('{0}x{1}'.format(HEIGHT * UNIT, HEIGHT * UNIT))    #窗口尺寸
        self.shapes = self.load_images()   #把图片加载进入窗口中
        self.canvas = self._build_canvas()   #建立画布的相关属性
        self.texts = []   #建立一个空列表

相关注释都在后面写上了，这里介绍几个比较难懂的点，窗口尺寸是500×500，文件开头赋初值给写了，整个action共有四个动作，分别是上、下、左、右移动，其分别有一个值来对应，为0、1、2、3；

然后是load_images()：

    def load_images(self):
        rectangle = PhotoImage(
            Image.open("../img/rectangle.png").resize((65, 65)))   #把图片加载到load_images，图片尺寸为65×65，窗口每一个网格尺寸为100×100
        triangle = PhotoImage(
            Image.open("../img/triangle.png").resize((65, 65)))
        circle = PhotoImage(
            Image.open("../img/circle.png").resize((65, 65)))

        return rectangle, triangle, circle

窗口每个栅格为什么是100×100后面讲，这里load_images函数的作用可以这么理解，相当于把图片加载进来，也就是预先加载，后面再放进窗口中；

_build_canvas：

    def _build_canvas(self):   #建立画布的相关属性
        canvas = tk.Canvas(self, bg='white',
                           height=HEIGHT * UNIT,
                           width=WIDTH * UNIT)   #窗口属性，背景色为白色，高度和宽度为500
        # create grids
        for c in range(0, WIDTH * UNIT, UNIT):  # 0~400 by 100，长度为400，每100画一条线，从左往右为x，从上往下为y
            x0, y0, x1, y1 = c, 0, c, HEIGHT * UNIT
            canvas.create_line(x0, y0, x1, y1)    #创建线条
        for r in range(0, HEIGHT * UNIT, UNIT):  # 0~400 by 100
            x0, y0, x1, y1 = 0, r, HEIGHT * UNIT, r
            canvas.create_line(x0, y0, x1, y1)

        # add img to canvas，往画布中加入图形
        self.rectangle = canvas.create_image(50, 50, image=self.shapes[0])
        self.triangle1 = canvas.create_image(250, 150, image=self.shapes[1])
        self.triangle2 = canvas.create_image(150, 250, image=self.shapes[1])
        self.circle = canvas.create_image(250, 250, image=self.shapes[2])

        # pack all
        canvas.pack()   #包装，打包

        return canvas

这里比较迷糊的可能就是for循环那里了，for循环的意思是对窗口进行切割，切成5×5大小的类似栅格地图一样的表格，至于为什么是0~400每100循环一次，这个是python的语法特性，不多解释。

放幅图辅助理解：

比如左边的红框（其实是没有框的，只是便于理解）是一个窗口，右边就是分为5×5的表格。

然后再往里面放入刚刚预加载的图形，两个三角形和一个圆形，如果碰到了三角形则Reward<0，表示惩罚；碰到了圆形则Reware>0，表示奖励。

然后是QLearningAgent：

初始化定义了几个参数，比如学习率以及几个策略的系数，还以字典的格式定义了Q-table：

class QLearningAgent:
    def __init__(self, actions):
        # actions = [0, 1, 2, 3]
        self.actions = actions   #共有四个动作，分别用0，1，2，3代替上下左右
        self.learning_rate = 0.01    #学习率
        self.discount_factor = 0.9    #奖励性衰变系数
        self.epsilon = 0.1    #策略系数
        self.q_table = defaultdict(lambda: [0.0, 0.0, 0.0, 0.0])

进入主函数循环for episode in range(1000):

state = env.reset()：

    def reset(self):
        self.update()    #合并两个集合，重复元素进行合并，不重复元素并存
        time.sleep(0.5)   #推迟0.5s调用线程的运行，也就是下面的命令行要过0.5s才能运行，意思应该是让上方的数据合并完了再训练
        x, y = self.canvas.coords(self.rectangle)    #矩形的位置，初始位置为(50,50)
        self.canvas.move(self.rectangle, UNIT / 2 - x, UNIT / 2 - y)   #调回起点，第一格
        self.render()   #是一个更新的函数
        # return observation
        return self.coords_to_state(self.canvas.coords(self.rectangle))   #返回矩形的状态，应该意思是对应的Q-table的坐标

这里reset意思就是当运行完一回合后，将矩形（矩形相当于agent）归为原点，原点是上面5×5表格中左上角那个位置；

这里有一个函数coords_to_state()：

    def coords_to_state(self, coords):
        x = int((coords[0] - 50) / 100)
        y = int((coords[1] - 50) / 100)
        return [x, y]

它的意思就是正常的坐标（比如原点）是(50,50)，那么它对应的Q-table中的位置就是(0,0)，上面的5×5的表格就可以看成是一个Q-table；

进入主函数的while循环：

env.render()就不介绍了，是一个更新窗口的函数：

    def render(self):
        time.sleep(0.03)   #延时0.03秒
        self.update()      #更新

action = agent.get_action(str(state))：

    def get_action(self, state):
        if np.random.rand() < self.epsilon:    #有概率的进入这个判断语句中
            # 贪婪策略随机探索动作
            action = np.random.choice(self.actions)   #从self.action=[0,1,2,3]随机选择
        else:
            # 从q表中选择
            state_action = self.q_table[state]
            action = self.arg_max(state_action)   #这里的选取动作如果value一致，则随机选取
        return action

字面意思就是获得一个动态action，刚开始这个动作是随机获得的，所以第一次循环进入的一定是else后面的语句中，这里着重讲self.q_table[state]，前面已经说了，q_table是我们定义的字典，那么q_table就是获得字典中对应名为state的值，举例：’[0,0]’ : [0.0,0.0,0.0,0.0]，这是字典q_table中的内容，那么它的名就是’[0,0]’，名’[0,0]'对应的值为[0.0,0.0,0.0,0.0]，所以self.q_table[state]返回的就是[0.0,0.0,0.0,0.0]（有的官方名词可能称呼不规范，见谅）；

后面的self.arg_max(state_action)：

    def arg_max(state_action):
        max_index_list = []
        max_value = state_action[0]
        for index, value in enumerate(state_action):
            if value > max_value:
                max_index_list.clear()
                max_value = value
                max_index_list.append(index)
            elif value == max_value:
                max_index_list.append(index)
        return random.choice(max_index_list)

这里就是选取动作了，前面说了，刚开始运行的时候动作是随机的，我们可以代入一个值具体分析，就比如刚刚的[0.0,0.0,0.0,0.0]，进入for循环，第一次循环的index是0，value=0.0，进入elif语句，就是给max_index_list列表加入index索引，循环四次加入四次索引，最后运行到return语句的时候，max_index_list=[0,1,2,3]，意思就是索引值为0,1,2,3（分别对应一个动作action，前面说到了）的Q值是一样的，所以随机选取一个值，比如2，返回；

比如这幅图，会进入if语句，原因是在遍历矩形所在位置的四个Q值时，发现0.0>-1.0，所以清除-1.0这个值，只在列表max_index_list中留三个Q值为0.0的索引值index然后随机选取，这样做就满足了Q-learning的基本思想：根据Q值来选取能够获得最大收益的动作。

next_state, reward, done = env.step(action)：

    def step(self, action):
        state = self.canvas.coords(self.rectangle)   #转回画布中的坐标
        base_action = np.array([0, 0])
        self.render()

        if action == 0:  # up
            if state[1] > UNIT:
                base_action[1] -= UNIT
        elif action == 1:  # down
            if state[1] < (HEIGHT - 1) * UNIT:
                base_action[1] += UNIT
        elif action == 2:  # left
            if state[0] > UNIT:
                base_action[0] -= UNIT
        elif action == 3:  # right
            if state[0] < (WIDTH - 1) * UNIT:
                base_action[0] += UNIT

        # 移动
        self.canvas.move(self.rectangle, base_action[0], base_action[1])   #根据上面的动作选择移动矩形
        self.canvas.tag_raise(self.rectangle)   #相当于置顶
        next_state = self.canvas.coords(self.rectangle)
        # 判断得分条件，Reward
        if next_state == self.canvas.coords(self.circle):
            reward = 100
            done = True
        elif next_state in [self.canvas.coords(self.triangle1),
                            self.canvas.coords(self.triangle2)]:
            reward = -100
            done = True
        else:
            reward = 0
            done = False

        next_state = self.coords_to_state(next_state)  #又变回Q表的坐标
        return next_state, reward, done

前面已经得到了下一步要进行的动作action，然后函数step就是进行矩形位置的更新以及奖赏reward的计算，以及判断矩形是否碰到三角形（障碍物）或者圆形（终点），如果是，则说明这一回合结束了，done会返回true，后面会退出while循环，重新开始新的一回合；

这一部分的代码还是比较好理解的，主要讲里面的self.canvas.tag_raise(self.rectangle)相当于置顶，意思就是如果矩形运行到了比如圆形的位置，如果没有这行代码，则虽然你知道它运行到了，但是你直观上是看不到的，如果有这行代码，则矩形会覆盖圆形。

后面就是更新Q表：

agent.learn(str(state), action, reward, str(next_state))：

def learn(self, state, action, reward, next_state):
    current_q = self.q_table[state][action]   #找到Q表中对应的坐标，分别给其加入选择状态的Reward
    # 贝尔曼方程更新
    new_q = reward + self.discount_factor * max(self.q_table[next_state])   #更新Q值
    self.q_table[state][action] += self.learning_rate * (new_q - current_q)

这里是Q值的计算就不阐述了；

然后是state更新，print_value_all：

    def print_value_all(self, q_table):
        for i in self.texts:
            self.canvas.delete(i)
        self.texts.clear()
        for i in range(HEIGHT):
            for j in range(WIDTH):
                for action in range(0, 4):   #这里应该是遍历窗口的每一个栅格框进行text更新
                    state = [i, j]
                    if str(state) in q_table.keys():
                        temp = q_table[str(state)][action]   #q_table[str(state)]是根据state查找字典q_table中对应的值，然后action你们懂得
                        self.text_value(j, i, round(temp, 2), action)   #round是四舍五入

这个print_all_value是在窗口上重新发布Q值，大家看一张图就知道了。

最后附上百度云链接：
链接：https://pan.baidu.com/s/1Gt8waFzwWGiKg4ubMZQxAg
提取码：8888