强化学习-4-时序差分

强化学习的笔记、理解、感悟及代码实现,仅按个人思维进行精华总结和记录，使用的教程：动手学强化学习

动态规划里的策略迭代和价值迭代都是需要事先知道环境的奖励函数和状态转移函数，可以直接递归迭代出最优价值或策略。但在大部分场景，无法写出状态转移概率和进行动态规划。在这种情况下，智能体只能和环境进行交互，通过采样到的数据来学习，这类学习方法统称为无模型的强化学习。

时序差分

单步时序差分

时序差分是一种估计策略价值函数的方法，结合了蒙特卡洛(从样本数据中学习，不需要事先知道环境)和动态规划(根据贝尔曼方程利用后续状态的价值估计来更新当前状态的价值估计)的思想。

蒙特卡洛增量更新方法：

将替换为，将a取常数表示对步长的估计，得到时序差分算法：

其中，称为时序差分误差，与的乘积作为状态价值的更新量。

多步时序差分

蒙特卡洛方法利用当前状态之后每一步的奖励而不使用任何价值估计，时序差分算法只利用一步奖励和下一个状态的价值估计，两者结合可以得到多步时序差分：使用n步的奖励，然后使用之后状态的价值估计。

即将替换为：

Sarsa算法

单步Sarsa算法

算法用到了当前状态s、当前动作、获得的奖励r、下一个状态s’和下一个动作，Sarsa算法的更新方式如下：

每一步根据策略选择当前状态s下的动作a，算法流程：
alt text

实验

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
import matplotlib.pyplot as plt
import numpy as np
from tqdm import tqdm  # tqdm是显示循环进度条的库
class CliffWalkingEnv:
    def __init__(self, ncol, nrow):
        self.nrow = nrow
        self.ncol = ncol
        self.x = 0  # 记录当前智能体位置的横坐标
        self.y = self.nrow - 1  # 记录当前智能体位置的纵坐标

    def step(self, action):  # 外部调用这个函数来改变当前位置
        # 4种动作, change[0]:上, change[1]:下, change[2]:左, change[3]:右。坐标系原点(0,0)
        # 定义在左上角
        change = [[0, -1], [0, 1], [-1, 0], [1, 0]]
        self.x = min(self.ncol - 1, max(0, self.x + change[action][0]))
        self.y = min(self.nrow - 1, max(0, self.y + change[action][1]))
        next_state = self.y * self.ncol + self.x
        reward = -1
        done = False
        if self.y == self.nrow - 1 and self.x > 0:  # 下一个位置在悬崖或者目标
            done = True
            if self.x != self.ncol - 1:
                reward = -100
        return reward, next_state, done

    def reset(self):  # 回归初始状态,坐标轴原点在左上角
        self.x = 0
        self.y = self.nrow - 1
        return self.y * self.ncol + self.x
class Sarsa:
    """ Sarsa算法 """
    def __init__(self, env, epsilon, alpha, gamma, n_action=4):
        self.env = env  # 环境
        self.Q_table = np.zeros([self.env.nrow * self.env.ncol, n_action])  # 初始化Q(s,a)表格
        self.n_action = n_action  # 动作个数
        self.alpha = alpha  # 学习率
        self.gamma = gamma  # 折扣因子
        self.epsilon = epsilon  # epsilon-贪婪策略中的参数

    def take_action(self, state):  # 选取下一步的操作,具体实现为epsilon-贪婪
        if np.random.random() < self.epsilon:
            action = np.random.randint(self.n_action)
        else:
            action = np.argmax(self.Q_table[state])
        return action
    def show_result(self):
        """ 打印结果 """
        print("策略：")
        actions = ['^', 'v', '<', '>']
        for i in range(self.env.nrow):
            for j in range(self.env.ncol):
                if i == self.env.nrow - 1 and j == self.env.ncol - 1:#终点
                    print("goal".center(5), end="")
                elif i == self.env.nrow - 1 and self.env.ncol-1>j > 0:#悬崖
                    print("x".center(5), end="")
                else:
                    qsa=self.Q_table[i*self.env.ncol+j]
                    max_qsa=np.max(qsa)
                    a_str=''.join( 'o' if qsa[i]<max_qsa else actions[i] for i in range(len(qsa)) )
                    print(a_str.center(5), end="")
            print('\n')
    """单步时序差分更新"""
    def update_Q(self, s0, a0, r, s1, a1):#s0表示当前状态s，s1表示下一个状态s'
        """计算时序差分误差td_error,更新动作价值函数Q(s,a)"""
        td_error = r + self.gamma * self.Q_table[s1, a1] - self.Q_table[s0, a0]
        self.Q_table[s0, a0] += self.alpha * td_error
    def run(self, episodes_num=1000):
        for episode in tqdm(range(episodes_num)):
            s0 = self.env.reset()  # 初始化
            a0 = self.take_action(s0)
            while True:
                cnt +=1
                r,s1,done = self.env.step(a0)# 环境反馈
                a1 = self.take_action(s1)  #根据下一状态和策略选取动作 
                self.update_Q(s0, a0, r, s1, a1)  # 更新Q
                if done:  # 终止状态
                    break
                s0 = s1  # 更新状态
                a0 = a1
if __name__ == '__main__':
    np.random.seed(0)
    env = CliffWalkingEnv(12, 4)
    agent = Sarsa(env, epsilon=0.1, alpha=0.1, gamma=0.9)
    agent.run(500)
    agent.show_result()
alt text

多步Sarsa算法

单步更新公式为：

相应的有：

实验

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
"""多步时序差分更新"""
class Sarsa:
    ...
    def nstep_update_Q(self,res_list,s1,a1,done):
        """计算多步时序差分误差td_error,更新动作价值函数Q(s,a)"""
        G = self.Q_table[s1, a1]
        for i in range(len(res_list)-1,-1,-1):
            s, a, r = res_list[i]
            G = r + self.gamma * G
            if done and i>0:#res_list最后一个状态是done状态,即使剩余步数不足n_step,也要更新
                self.Q_table[s, a] += self.alpha * (G - self.Q_table[s, a])
        self.Q_table[s, a] += self.alpha * (G - self.Q_table[s, a])
    def nstep_run(self, episodes_num=1000,n_step=5):
        for episode in tqdm(range(episodes_num)):
            res_list = []
            s0 = self.env.reset()  # 初始化
            a0 = self.take_action(s0)
            while True:
                r,s1,done = self.env.step(a0)# 环境反馈
                a1 = self.take_action(s1)  #根据下一状态和策略选取动作 
                res_list.append((s0, a0, r))
                if done:  # 终止状态
                    self.nstep_update_Q(res_list,s1,a1,done)
                    res_list.clear()#终止状态时，res_list所有元素价值都已经更新，所以清空列表
                    break
                if len(res_list) == n_step:
                    self.nstep_update_Q(res_list,s1,a1,done)
                    res_list.pop(0)#res_list首个状态的动作价值已经更新,所以要pop掉
                s0 = s1  # 更新状态
                a0 = a1
if __name__ == '__main__':
    np.random.seed(0)
    env = CliffWalkingEnv(12, 4)
    agent = Sarsa(env, epsilon=0.1, alpha=0.1, gamma=0.9)
    agent.nstep_run(500,n_step=5)
    agent.show_result()
alt text

Q-learning算法

Q-learning 的时序差分更新方式为:

算法流程：
alt text

实验

相比单步sarsa算法，q-learning更新价值函数、计算时序差分误差时直接选取下一状态的动作最大值self.Q_table[s1].max()(贪心/最优略)，而sarsa算法则是依靠当前策略选取动作的价值，多出一个a’的输入。

  • 单步sarsa算法：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
def update_Q(self, s0, a0, r, s1, a1):#s0表示当前状态s，s1表示下一个状态s'
        """计算时序差分误差td_error,更新动作价值函数Q(s,a)"""
        td_error = r + self.gamma * self.Q_table[s1, a1] - self.Q_table[s0, a0]
        self.Q_table[s0, a0] += self.alpha * td_error
def run(self, episodes_num=1000):
        for episode in tqdm(range(episodes_num)):
            s0 = self.env.reset()  ## 初始化
            a0 = self.take_action(s0)
            while True:
                r,s1,done = self.env.step(a0)# 环境反馈
                a1 = self.take_action(s1)  #根据下一状态和策略选取动作 
                self.update_Q(s0, a0, r, s1, a1)  # 更新Q
                if done:  # 终止状态
                    break
                s0 = s1  # 更新状态
                a0 = a1
  • 改动后q-learning算法：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
def update_Q(self, s0, a0, r, s1):#s0表示当前状态s，s1表示下一个状态s'
        """计算时序差分误差td_error,更新动作价值函数Q(s,a)"""
        td_error = r + self.gamma * self.Q_table[s1].max() - self.Q_table[s0, a0]
        self.Q_table[s0, a0] += self.alpha * td_error
def run(self, episodes_num=1000):
        for episode in tqdm(range(episodes_num)):
            s = self.env.reset()  # 初始化
            while True:
                a = self.take_action(s)  # 选取动作
                r,s1,done = self.env.step(a)# 环境反馈
                self.update_Q(s, a, r, s1)  # 更新Q
                if done:  # 终止状态
                    break
                s = s1  # 更新状态
alt text

总结

  • Sarsas算法使用策略来采样数据，也使用策略来更新价值函数，采样数据的策略(行为策略)和利用数据更新的策略(目标策略)必须相同，是在线策略算法。

  • Q-learning算法使用策略(也可以使用其他策略)来采样数据，使用贪心策略即最优策略来更新价值函数，采样数据的策略(行为策略)和利用数据更新的策略(目标策略)可以不同，是离线策略算法。

  • 判断二者的核心在于计算时序差分误差使用的数据是否来自当前的策略，如果是，则是在线策略算法，否则是离线策略算法。

  • Sarsas算法估计的是当前策略的动作价值函数，Q-learning算法则是基于贝尔曼最优方程估计最优动作价值函数。故Sarsas算法比较保守，适合动态、风险环境，更平稳、更新速度慢但稳定(探索和利用的平衡)；Q-learning算法更冒险，适合静态环境和寻找最优解(最优但不一定安全)，更高效、更新速度快但不稳定。

  • 无论是Sarsas算法还是Q-learning算法，最终执行路线都是顺着Q(s,a)最大的序列得到。比较多步sarsa、单步sarsa、q-learning算法的实验结果可知，智能体选择的路线逐渐从远离悬崖到靠近悬崖，路线从保守到冒险，但最终都能到达悬崖(符合算法的预期)。

default
多步sarsa
default
单步sarsa
default
q-learning