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

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

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

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

一、SARSA算法原理

上一篇内容奖励，强化学习的主要功能就是让agent学习尽可能好的动作action，使其后续获得的奖励尽可能的大。
假设在时刻t时，处于状态长期奖励为：
在这里插入图片描述
其中 r t + n r_{t+n} rt+n为t+n时获得的局部奖励， γ γ γ为下一期奖励传递到当期的衰减因子。则状态s的价值为 G t G_t Gt的条件期望：
在这里插入图片描述
同样定义状态s时执行动作a的价值也为 G t G_t Gt的条件期望：
在这里插入图片描述
可知： E ( Q π ( s , a ) ) = V π ( s ) E(Q_π(s,a))=V_π(s) E(Qπ(s,a))=Vπ(s)
下面直接给出深度学习最核心的Bellman 公式
在这里插入图片描述
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)
在这里插入图片描述
上面几个公式是强化学习的精髓

SARSA算法是一种典型性的value-based和on-policy的算法。下面直接给出SARSA的算法
在这里插入图片描述
SARSA的算法中有几个需要注意的地方下面用彩色标记标记出
在这里插入图片描述
由于与环境进行了实际的交互，因此会执行动作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的强化学习教程进行代码编写，在基础上说明每一行代码的用途
1.environment的编写
首先RL需要一个环境，因为我们控制不了环境（比如下围棋时我们不不能改变棋盘的大小，何落子方式，只能只能在范围内落在线与线之间的交叉点上），这个环境是不可以改变的，因此后面的Q-learning也将沿用此环境。通常不同的问题有不同环境，我们真正需要关注的是agent即算法逻辑的编写。
此处以走方格为例编写一个environment
在这里插入图片描述
其中红色点为当前所在位置，走到黄色点获得奖励1，走到黑色点获得奖励-1

  1. """
  2. Reinforcement learning maze example.
  3. Red rectangle:          explorer.
  4. Black rectangles:       hells       [reward = -1].
  5. Yellow bin circle:      paradise    [reward = +1].
  6. All other states:       ground      [reward = 0].
  7. This script is the environment part of this example. The RL is in RL_brain.py.
  8. View more on my tutorial page: https://morvanzhou.github.io/tutorials/
  9. """
  12. import numpy as np
  13. import time
  14. import sys
  15. if sys.version_info.major == 2:
  16.     import Tkinter as tk
  17. else:
  18.     import tkinter as tk
  21. UNIT = 40   # pixels
  22. MAZE_H = 4  # grid height
  23. MAZE_W = 4  # grid width
  26. class Maze(tk.Tk, object):
  27.     def __init__(self):
  28.         super(Maze, self).__init__()
  29.         self.action_space = ['u', 'd', 'l', 'r']
  30.         self.n_actions = len(self.action_space)
  31.         self.title('maze')
  32.         self.geometry('{0}x{1}'.format(MAZE_H * UNIT, MAZE_H * UNIT))
  33.         self._build_maze()
  35.     def _build_maze(self):
  36.         self.canvas = tk.Canvas(self, bg='white',
  37.                            height=MAZE_H * UNIT,
  38.                            width=MAZE_W * UNIT)
  40.         # create grids
  41.         for c in range(0, MAZE_W * UNIT, UNIT):
  42.             x0, y0, x1, y1 = c, 0, c, MAZE_H * UNIT
  43.             self.canvas.create_line(x0, y0, x1, y1)
  44.         for r in range(0, MAZE_H * UNIT, UNIT):
  45.             x0, y0, x1, y1 = 0, r, MAZE_W * UNIT, r
  46.             self.canvas.create_line(x0, y0, x1, y1)
  48.         # create origin
  49.         origin = np.array([20, 20])
  51.         # hell
  52.         hell1_center = origin + np.array([UNIT * 2, UNIT])
  53.         self.hell1 = self.canvas.create_rectangle(
  54.             hell1_center[0] - 15, hell1_center[1] - 15,
  55.             hell1_center[0] + 15, hell1_center[1] + 15,
  56.             fill='black')
  57.         # hell
  58.         hell2_center = origin + np.array([UNIT, UNIT * 2])
  59.         self.hell2 = self.canvas.create_rectangle(
  60.             hell2_center[0] - 15, hell2_center[1] - 15,
  61.             hell2_center[0] + 15, hell2_center[1] + 15,
  62.             fill='black')
  64.         # create oval
  65.         oval_center = origin + UNIT * 2
  66.         self.oval = self.canvas.create_oval(
  67.             oval_center[0] - 15, oval_center[1] - 15,
  68.             oval_center[0] + 15, oval_center[1] + 15,
  69.             fill='yellow')
  71.         # create red rect
  72.         self.rect = self.canvas.create_rectangle(
  73.             origin[0] - 15, origin[1] - 15,
  74.             origin[0] + 15, origin[1] + 15,
  75.             fill='red')
  77.         # pack all
  78.         self.canvas.pack()
  80.     def reset(self):
  81.         self.update()
  82.         time.sleep(0.5)
  83.         self.canvas.delete(self.rect)
  84.         origin = np.array([20, 20])
  85.         self.rect = self.canvas.create_rectangle(
  86.             origin[0] - 15, origin[1] - 15,
  87.             origin[0] + 15, origin[1] + 15,
  88.             fill='red')
  89.         # return observation
  90.         return self.canvas.coords(self.rect)
  92.     def step(self, action):
  93.         s = self.canvas.coords(self.rect)
  94.         base_action = np.array([0, 0])
  95.         if action == 0:   # up
  96.             if s[1] > UNIT:
  97.                 base_action[1] -= UNIT
  98.         elif action == 1:   # down
  99.             if s[1] < (MAZE_H - 1) * UNIT:
  100.                 base_action[1] += UNIT
  101.         elif action == 2:   # right
  102.             if s[0] < (MAZE_W - 1) * UNIT:
  103.                 base_action[0] += UNIT
  104.         elif action == 3:   # left
  105.             if s[0] > UNIT:
  106.                 base_action[0] -= UNIT
  108.         self.canvas.move(self.rect, base_action[0], base_action[1])  # move agent
  110.         s_ = self.canvas.coords(self.rect)  # next state
  112.         # reward function
  113.         if s_ == self.canvas.coords(self.oval):
  114.             reward = 1
  115.             done = True
  116.             s_ = 'terminal'
  117.         elif s_ in [self.canvas.coords(self.hell1), self.canvas.coords(self.hell2)]:
  118.             reward = -1
  119.             done = True
  120.             s_ = 'terminal'
  121.         else:
  122.             reward = 0
  123.             done = False
  125.         return s_, reward, done
  127.     def render(self):
  128.         time.sleep(0.1)
  129.         self.update()
  131. def update():
  132.     for t in range(10):
  133.         s = env.reset()
  134.         while True:
  135.             env.render()
  136.             a = 1
  137.             s, r, done = env.step(a)
  138.             if done:
  139.                 break
  141. if __name__ == '__main__':
  142.     env = Maze()
  143.     env.after(100, update)
  144.     env.mainloop()

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

  1. from maze_env import Maze    #即为上面的environment
  2. import numpy as np
  3. import pandas as pd
  5. #RL的父类定义
  6. class RL(object):
  7.     #初始化
  8.     #actions为可选动作， learning_rate为学习率，reward_decay为传递奖励是的递减系数gamma，1-e_greed为随机选择其他动作的概率
  9.     def __init__(self, actions, learning_rate=0.01, reward_decay=0.9, e_greedy=0.9):
  10.         self.actions = actions
  11.         self.lr = learning_rate
  12.         self.gamma = reward_decay
  13.         self.epsilon = e_greedy
  14.         #初始化qtable，行为observation的state， 列为当前状态可以选择的action（对于所有列，可以选择的action一样）
  15.         self.q_table = pd.DataFrame(columns = self.actions, dtype=np.float64)
  17.     def choose_action(self, observation):
  18.         self.check_state_exist(observation)  #检查当前状态是否存在，不存在就添加这个状态
  20.         if np.random.uniform() < self.epsilon:
  21.             state_action = self.q_table.loc[observation, :]  #找到当前状态可以选择的动作
  22.             #由于初始化或更新后一个状态下的动作值可能是相同的，为了避免每次都选择相同动作，用random.choice在值最大的action中损及选择一个
  23.             action = np.random.choice(state_action[state_action==np.max(state_action)].index)
  25.         else:
  26.             action = np.random.choice(self.actions)   #0.1的几率随机选择动作
  27.         return action
  29.     def check_state_exist(self, state):
  30.         if state not in self.q_table.index:
  31.             #若找不到该obversation的转态，则添加该状态到新的qtable
  32.             #新的state的动作的q初始值赋值为0，列名为dataframe的列名，index为state
  33.             self.q_table = self.q_table.append(pd.Series([0]*len(self.actions), index=self.q_table.columns, name=state))
  35.     #不同方式的学习方法不同，用可变参数，直接pass
  36.     def learning(self, *args):
  37.         pass
  38. class SarsaTable(RL):  #继承上面的RL
  39.     #初始化
  40.     #参数自己定义，含义继承父类RL
  41.     #类方法choose_action、check_state_exist自动继承RL，参数不变
  42.     def __init__(self, actions, learning_rate=0.01, reward_decay=0.9, e_greedy=0.9):
  43.         super(SarsaTable, self).__init__(actions, learning_rate, reward_decay, e_greedy)
  45.     def learning(self, s, a,r, s_, a_):
  46.         self.check_state_exist(s_)   #检查动作后状态s_是否存在
  48.         q_old = self.q_table.loc[s, a]  #旧的q[s,a]值
  50.         if s_!='terminal':
  51.             #取下个状态s_和动作a_下q值
  52.             q_predict = self.q_table.loc[s_, a_]
  53.             q_new = r+self.gamma*q_predict   #计算新的值
  54.         else:
  55.             q_new = r
  57.         self.q_table.loc[s,a] = q_old - self.lr*(q_new - q_old)    #根据更新公式更新，类似于梯度下降
  59. def update():
  60.     for episode in range(100):
  61.         #初始化环境
  62.         observation = env.reset()
  64.         #根据当前状态选行为
  65.         action = RL.choose_action(str(observation))
  67.         while True:
  68.             # 刷新环境
  69.             env.render()
  71.             # 在环境中采取行为, 获得下一个 state_ (obervation_), reward, 和是否终止
  72.             observation_, reward, done = env.step(action)
  74.             #根据observation_选择observation_下应该选择的动作action_
  75.             action_ = RL.choose_action(str(observation_))
  77.             #从当前状态state，当前动作action，奖励r，执行动作后state_，state_下的action_,(s,a,r,s,a)
  78.             RL.learning(str(observation), action, reward, str(observation_), action_)
  80.             # 将下一个当成下一步的 state (observation) and action。
  81.             #与qlearning的却别是sarsa在observation_下真正执行了动作action_，供下次使用
  82.             #而qlearning中下次状态observation_时还要重新选择action_
  83.             observation = observation_
  84.             action = action_
  86.              # 终止时跳出循环
  87.             if done:
  88.                 break
  90.     # 大循环完毕
  91.     print('game over')
  92.     env.destroy()
  95. if __name__ == "__main__":
  96.     env = Maze()
  98.     #Sarsa和SarsaLambda的调用方式一模一样
  99.     #RL = SarsaTable(actions=list(range(env.n_actions)))
  100.     RL = SarsaLambdaTable(actions=list(range(env.n_actions)))
  102.     env.after(100, update)
  103.     env.mainloop()

发表评论

表情：
评论列表 （有 0 条评论，571人围观）

还没有评论，来说两句吧...

相关阅读