强化学习笔记+代码（二）：SARSA算法原理和Agent实现

迈不过友情╰ 2023-07-17 15:59 442阅读 0赞

本文主要整理和参考了[李宏毅的强化学习系列课程][Link 1]和莫烦python的[强化学习教程][Link 2]  
本系列主要分几个部分进行介绍

1.  [强化学习背景介绍][Link 3]
2.  [SARSA算法原理和Agent实现][SARSA_Agent]
3.  [Q-learning算法原理和Agent实现][Q-learning_Agent]
4.  [DQN算法原理和Agent实现(tensorflow)][DQN_Agent_tensorflow]
5.  [Double-DQN、Dueling DQN算法原理和Agent实现(tensorflow)][Double-DQN_Dueling DQN_Agent_tensorflow]
6.  [Policy Gradients算法原理和Agent实现(tensorflow)][Policy Gradients_Agent_tensorflow]
7.  [Actor-Critic、A2C、A3C算法原理和Agent实现(tensorflow)][Actor-Critic_A2C_A3C_Agent_tensorflow]

#### 一、SARSA算法原理 ####

上一篇内容奖励，强化学习的主要功能就是让agent学习尽可能好的动作action，使其后续获得的奖励尽可能的大。  
假设在时刻t时，处于状态长期奖励为：  
![在这里插入图片描述][2020032311212617.png]  
其中 r t + n r\_\{t+n\} rt\+n为t+n时获得的局部奖励， γ γ γ为下一期奖励传递到当期的衰减因子。则状态s的价值为 G t G\_t Gt的条件期望：  
![在这里插入图片描述][20200323112524528.png]  
同样定义状态s时执行动作a的价值也为 G t G\_t Gt的条件期望：  
![在这里插入图片描述][20200323112851810.png]  
可知： E ( Q π ( s , a ) ) = V π ( s ) E(Q\_π(s,a))=V\_π(s) E(Qπ(s,a))=Vπ(s)  
下面直接给出深度学习最核心的Bellman 公式  
![在这里插入图片描述][20200323113200398.png]  
Bellman 公式十分的直观， V π ( s ) V\_π(s) Vπ(s)=根据动作走到下个状态的奖励+下个状态长期价值\*衰减值的期望。价值会不断的传递，因此可以看出 V π ( s ) V\_π(s) Vπ(s)衡量的是状态s的长期价值。  
因为RL需要是Agent不断变强，就可以理解为让状态或让动作的价值不断变大，因此会选择根据如下方式获得新的 V π ( s ) V\_π(s) Vπ(s)和 Q π ( s , a ) Q\_π(s,a) Qπ(s,a)  
![在这里插入图片描述][2020032311400973.png]  
**上面几个公式是强化学习的精髓**

SARSA算法是一种典型性的value-based和on-policy的算法。下面直接给出SARSA的算法  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTE1MTcxMzI_size_16_color_FFFFFF_t_70]  
SARSA的算法中有几个需要注意的地方下面用彩色标记标记出  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTE1MTcxMzI_size_16_color_FFFFFF_t_70 1]  
由于与环境进行了实际的交互，因此会执行动作a后到达状态s’，agent会根据状态s’直接执行动作a’。上面算法图中红框为新的状态s和动作a的价值，即 Q ∗ ( s , a ) Q\_\*(s,a) Q∗(s,a)，蓝色框是旧的 Q ( s , a ) Q(s,a) Q(s,a)，**可以看出 Q ( s , a ) Q(s,a) Q(s,a)的更新方法与梯度下降十分的类似**。还需要注意到，在循环中黄色框，循环末尾将s’估值给s，将动作a’赋值给a,带到下一次循环中进行运算，说明agent真的与环境进行实际交互。这是算法为on-policy的关键。

#### 二、SARSA代码 ####

此处直接参考莫烦python的[强化学习教程][Link 2]进行代码编写，在基础上说明每一行代码的用途  
**1.environment的编写**  
首先RL需要一个环境，因为我们控制不了环境（比如下围棋时我们不不能改变棋盘的大小，何落子方式，只能只能在范围内落在线与线之间的交叉点上），这个环境是不可以改变的，因此后面的Q-learning也将沿用此环境。通常不同的问题有不同环境，我们真正需要关注的是agent即算法逻辑的编写。  
此处以走方格为例编写一个environment  
![在这里插入图片描述][20200323121120175.png]  
其中红色点为当前所在位置，走到黄色点获得奖励1，走到黑色点获得奖励-1

"""
    Reinforcement learning maze example.
    Red rectangle:          explorer.
    Black rectangles:       hells       [reward = -1].
    Yellow bin circle:      paradise    [reward = +1].
    All other states:       ground      [reward = 0].
    This script is the environment part of this example. The RL is in RL_brain.py.
    View more on my tutorial page: https://morvanzhou.github.io/tutorials/
    """
    
    
    import numpy as np
    import time
    import sys
    if sys.version_info.major == 2:
        import Tkinter as tk
    else:
        import tkinter as tk
    
    
    UNIT = 40   # pixels
    MAZE_H = 4  # grid height
    MAZE_W = 4  # grid width
    
    
    class Maze(tk.Tk, object):
        def __init__(self):
            super(Maze, self).__init__()
            self.action_space = ['u', 'd', 'l', 'r']
            self.n_actions = len(self.action_space)
            self.title('maze')
            self.geometry('{0}x{1}'.format(MAZE_H * UNIT, MAZE_H * UNIT))
            self._build_maze()
    
        def _build_maze(self):
            self.canvas = tk.Canvas(self, bg='white',
                               height=MAZE_H * UNIT,
                               width=MAZE_W * UNIT)
    
            # create grids
            for c in range(0, MAZE_W * UNIT, UNIT):
                x0, y0, x1, y1 = c, 0, c, MAZE_H * UNIT
                self.canvas.create_line(x0, y0, x1, y1)
            for r in range(0, MAZE_H * UNIT, UNIT):
                x0, y0, x1, y1 = 0, r, MAZE_W * UNIT, r
                self.canvas.create_line(x0, y0, x1, y1)
    
            # create origin
            origin = np.array([20, 20])
    
            # hell
            hell1_center = origin + np.array([UNIT * 2, UNIT])
            self.hell1 = self.canvas.create_rectangle(
                hell1_center[0] - 15, hell1_center[1] - 15,
                hell1_center[0] + 15, hell1_center[1] + 15,
                fill='black')
            # hell
            hell2_center = origin + np.array([UNIT, UNIT * 2])
            self.hell2 = self.canvas.create_rectangle(
                hell2_center[0] - 15, hell2_center[1] - 15,
                hell2_center[0] + 15, hell2_center[1] + 15,
                fill='black')
    
            # create oval
            oval_center = origin + UNIT * 2
            self.oval = self.canvas.create_oval(
                oval_center[0] - 15, oval_center[1] - 15,
                oval_center[0] + 15, oval_center[1] + 15,
                fill='yellow')
    
            # create red rect
            self.rect = self.canvas.create_rectangle(
                origin[0] - 15, origin[1] - 15,
                origin[0] + 15, origin[1] + 15,
                fill='red')
    
            # pack all
            self.canvas.pack()
    
        def reset(self):
            self.update()
            time.sleep(0.5)
            self.canvas.delete(self.rect)
            origin = np.array([20, 20])
            self.rect = self.canvas.create_rectangle(
                origin[0] - 15, origin[1] - 15,
                origin[0] + 15, origin[1] + 15,
                fill='red')
            # return observation
            return self.canvas.coords(self.rect)
    
        def step(self, action):
            s = self.canvas.coords(self.rect)
            base_action = np.array([0, 0])
            if action == 0:   # up
                if s[1] > UNIT:
                    base_action[1] -= UNIT
            elif action == 1:   # down
                if s[1] < (MAZE_H - 1) * UNIT:
                    base_action[1] += UNIT
            elif action == 2:   # right
                if s[0] < (MAZE_W - 1) * UNIT:
                    base_action[0] += UNIT
            elif action == 3:   # left
                if s[0] > UNIT:
                    base_action[0] -= UNIT
    
            self.canvas.move(self.rect, base_action[0], base_action[1])  # move agent
    
            s_ = self.canvas.coords(self.rect)  # next state
    
            # reward function
            if s_ == self.canvas.coords(self.oval):
                reward = 1
                done = True
                s_ = 'terminal'
            elif s_ in [self.canvas.coords(self.hell1), self.canvas.coords(self.hell2)]:
                reward = -1
                done = True
                s_ = 'terminal'
            else:
                reward = 0
                done = False
    
            return s_, reward, done
    
        def render(self):
            time.sleep(0.1)
            self.update()
            
    def update():
        for t in range(10):
            s = env.reset()
            while True:
                env.render()
                a = 1
                s, r, done = env.step(a)
                if done:
                    break
    
    if __name__ == '__main__':
        env = Maze()
        env.after(100, update)
        env.mainloop()

**2.agent的编写**  
通常不同的问题有不同环境，我们真正需要关注的是agent即算法逻辑的编写。

from maze_env import Maze    #即为上面的environment
    import numpy as np
    import pandas as pd
    
    #RL的父类定义
    class RL(object):
        #初始化
        #actions为可选动作， learning_rate为学习率，reward_decay为传递奖励是的递减系数gamma，1-e_greed为随机选择其他动作的概率
        def __init__(self, actions, learning_rate=0.01, reward_decay=0.9, e_greedy=0.9):
            self.actions = actions
            self.lr = learning_rate
            self.gamma = reward_decay
            self.epsilon = e_greedy
            #初始化qtable，行为observation的state， 列为当前状态可以选择的action（对于所有列，可以选择的action一样）
            self.q_table = pd.DataFrame(columns = self.actions, dtype=np.float64)
        
        def choose_action(self, observation):
            self.check_state_exist(observation)  #检查当前状态是否存在，不存在就添加这个状态
            
            if np.random.uniform() < self.epsilon:
                state_action = self.q_table.loc[observation, :]  #找到当前状态可以选择的动作
                #由于初始化或更新后一个状态下的动作值可能是相同的，为了避免每次都选择相同动作，用random.choice在值最大的action中损及选择一个
                action = np.random.choice(state_action[state_action==np.max(state_action)].index)
            
            else:
                action = np.random.choice(self.actions)   #0.1的几率随机选择动作
            return action
        
        def check_state_exist(self, state):
            if state not in self.q_table.index:
                #若找不到该obversation的转态，则添加该状态到新的qtable
                #新的state的动作的q初始值赋值为0，列名为dataframe的列名，index为state
                self.q_table = self.q_table.append(pd.Series([0]*len(self.actions), index=self.q_table.columns, name=state))
        
        #不同方式的学习方法不同，用可变参数，直接pass
        def learning(self, *args):
            pass
    class SarsaTable(RL):  #继承上面的RL
        #初始化
        #参数自己定义，含义继承父类RL
        #类方法choose_action、check_state_exist自动继承RL，参数不变
        def __init__(self, actions, learning_rate=0.01, reward_decay=0.9, e_greedy=0.9):
            super(SarsaTable, self).__init__(actions, learning_rate, reward_decay, e_greedy)
        
        def learning(self, s, a,r, s_, a_):
            self.check_state_exist(s_)   #检查动作后状态s_是否存在
            
            q_old = self.q_table.loc[s, a]  #旧的q[s,a]值
            
            if s_!='terminal':
                #取下个状态s_和动作a_下q值
                q_predict = self.q_table.loc[s_, a_]
                q_new = r+self.gamma*q_predict   #计算新的值
            else:
                q_new = r
            
            self.q_table.loc[s,a] = q_old - self.lr*(q_new - q_old)    #根据更新公式更新，类似于梯度下降
     
    def update():
        for episode in range(100):
            #初始化环境
            observation = env.reset()
            
            #根据当前状态选行为
            action = RL.choose_action(str(observation))
            
            while True:
                # 刷新环境
                env.render()
                
                # 在环境中采取行为, 获得下一个 state_ (obervation_), reward, 和是否终止
                observation_, reward, done = env.step(action)
                
                #根据observation_选择observation_下应该选择的动作action_
                action_ = RL.choose_action(str(observation_))
                
                #从当前状态state，当前动作action，奖励r，执行动作后state_，state_下的action_,(s,a,r,s,a)
                RL.learning(str(observation), action, reward, str(observation_), action_)
                
                # 将下一个当成下一步的 state (observation) and action。
                #与qlearning的却别是sarsa在observation_下真正执行了动作action_，供下次使用
                #而qlearning中下次状态observation_时还要重新选择action_
                observation = observation_
                action = action_
                
                 # 终止时跳出循环
            	if done:
                	break
            
        # 大循环完毕
        print('game over')
        env.destroy()
    
    
    if __name__ == "__main__":
        env = Maze()
        
        #Sarsa和SarsaLambda的调用方式一模一样
        #RL = SarsaTable(actions=list(range(env.n_actions)))
        RL = SarsaLambdaTable(actions=list(range(env.n_actions)))
    
        env.after(100, update)
        env.mainloop()

[Link 1]: https://www.bilibili.com/video/av24724071/
[Link 2]: https://morvanzhou.github.io/tutorials/machine-learning/reinforcement-learning/1-1-A-RL/
[Link 3]: https://blog.csdn.net/u011517132/article/details/105042828
[SARSA_Agent]: https://blog.csdn.net/u011517132/article/details/105044388
[Q-learning_Agent]: https://blog.csdn.net/u011517132/article/details/105049583
[DQN_Agent_tensorflow]: https://blog.csdn.net/u011517132/article/details/105051463
[Double-DQN_Dueling DQN_Agent_tensorflow]: https://blog.csdn.net/u011517132/article/details/105065672
[Policy Gradients_Agent_tensorflow]: https://blog.csdn.net/u011517132/article/details/105098610
[Actor-Critic_A2C_A3C_Agent_tensorflow]: https://blog.csdn.net/u011517132/article/details/105118510
[2020032311212617.png]: /images/20230528/fd19add44dbe4afe81d5228688e8b009.png
[20200323112524528.png]: /images/20230528/f35e02b1a76d478a9b6eea97d7250350.png
[20200323112851810.png]: /images/20230528/e8d3655e83a64a54804bdc1e816dca17.png
[20200323113200398.png]: /images/20230528/eb09e1db224b4e7e96759bef86ca54be.png
[2020032311400973.png]: /images/20230528/bb4adc3a8b5740cbad52d43518fea13a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTE1MTcxMzI_size_16_color_FFFFFF_t_70]: /images/20230528/74fd15582a29405d80eb688fa0ac0b8f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTE1MTcxMzI_size_16_color_FFFFFF_t_70 1]: /images/20230528/c65fd03871ac424f95d4fad403cb889a.png
[20200323121120175.png]: /images/20230528/59a53dc26679415d8f2e3190b098f6c4.png