强化学习代码实操和讲解（二）_强化学习的step函数

引言

这里是强化学习代码实操第二部分，对应书上第三章有限马尔可夫决策过程的内容。本次实操主要运用了网格世界这一经典的强化学习环境（下面会略作介绍），向我们展示了本章的核心内容——贝尔曼方程的应用以及最优策略的判定方法。看过书的朋友可能会注意到本章其实没有提供具体的算法伪代码，在本次实操中其实运用了下一章中的价值迭代和策略迭代的方法来求解，如果看过下一章可能会更好理解一些。再次贴上代码出处：代码出处

背景介绍

上图就是格子世界的环境展示。每个格子代表一个环境中的状态（state）。在每个格子中有四个可选的动作（action），每个动作都会使智能体在对应的方向上前进一格。如果动作会使智能体出界，那么智能体则会在原地不动，并且获得一个-1的收益（reward）。在状态A中，智能体做任何动作都会转移到A’并且获得+10的收益；在状态B中，智能体做任何动作都会转移到B’并获得+5的收益。在其他情况下智能体的收益都是0。现在我们要来求解每个状态在给定策略$\pi$下的状态价值（$v_{\pi }$），最优价值函数（$v_{\ast }$）以及其对应的最优策略${\pi }_{\ast }$。

重点代码解析

环境设置

import matplotlib
import matplotlib.pyplot as plt
import numpy as np
from matplotlib.table import Table

matplotlib.use('Agg')
#下列网格参数设定的具体意义参见P58
WORLD_SIZE=5   #网格世界的长宽都是5
A_POS=[0,1]   #标记A的位置
A_PRIME_POS=[4,1]   #标记A'的位置
B_POS=[0,3]   #标记B的位置
B_PRIME_POS=[2,3]   #标记B'的位置
DISCOUNT=0.9   #设定折扣系数为0.9

ACTIONS=[np.array([0,-1]),
         np.array([-1,0]),
         np.array([0,1]),
         np.array([1,0])]   #设定动作，由于网格世界是二维平面，所以用一个二元的向量来表示每个动作
ACTION_FIGS=['←','↑','→','↓']   #设定每个动作对应的箭头，方便画图
ACTION_PROB=0.25   #由于在本实验中仅仅估计价值函数，并根据收敛的最优价值采取最优策略，所以探索的过程中并没有对策略（选择方向的概率）进行修改，
                   #所以每个向任意方向的行动都是等概率的，最后根据收敛的最优价值函数获得最优策略
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

首先引入了一些画图和数学计算的包，之后根据问题描述设置了格子世界的参数，比如格子的数量和A、B等点的位置。由于格子世界是一个二维的世界，虽然不能斜着走，但是用一个二维的向量表示动作也是十分巧妙易懂的。值得注意的是，此处的ACTION_PROB指的是向各个位置移动的概率，也就是策略$\pi$，由于本章主要是估计给定策略下的状态价值函数，所以并没有对原策略进行改进，仅仅是在后面通过最优价值函数直接导出最优策略，所以这里的动作概率在估计过程中是一个定值。

step函数：接收环境奖励和状态转移

def step(state,action):
    if state==A_POS:
        return A_PRIME_POS,10   #在A处做任何动作都会转移到A'处并获得+10回报
    if state==B_POS:
        return B_PRIME_POS,5   #在B处做任何动作都会转移到B'处并获得+5回报
    
    next_state=(np.array(state)+action).tolist()
    x,y=next_state   #否则把当前位置加上相应动作向量获得新位置
    if x<0 or x>=WORLD_SIZE or y<0 or y>=WORLD_SIZE:
        reward=-1.0
        next_state=state
    else:
        reward=0   #如果出界，位置不变并获得-1回报，如果没出界且不在A,B处，则获得0回报
    return next_state,reward
#对每一步行动之后的状态和回报（return）进行计算
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

在格子世界中的状态转移与奖励都十分好定义，因为问题描述中已经说清楚所有可能性，因而我们只要用if的组合就能区别各种情况。这里向量化动作的优势就显现出来，我们只要把当前状态和动作这两个向量相加就能很方便地得到下一个状态向量。

figure_3_2：根据给定策略估计状态价值函数

def figure_3_2():
    value = np.zeros((WORLD_SIZE, WORLD_SIZE))
    while True:
        # keep iteration until convergence
        new_value = np.zeros_like(value)
        for i in range(WORLD_SIZE):
            for j in range(WORLD_SIZE):   #对每一个网格进行动作的试探和价值函数迭代
                for action in ACTIONS:
                    (next_i, next_j), reward = step([i, j], action)   #对每个状态下的每个动作记录状态转移和收益（reward）
                    # bellman equation
                    new_value[i, j] += ACTION_PROB * (reward + DISCOUNT * value[next_i, next_j])   #根据贝尔曼方程估计新的状态价值，这里的策略用的是等概率动作，其实还用到了下一章的动态规划
        if np.sum(np.abs(value - new_value)) < 1e-4:   #收敛的时候结束循环，并用最佳函数值画图
            draw_image(np.round(new_value, decimals=2))   #把收敛的价值函数绘制到网格世界中，后面会提到这个函数
            plt.savefig('figure_3_2.png')
            plt.close()
            break
        value = new_value   #更新所有网格的状态价值函数
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

这里用贝尔曼方程估计给定策略下的状态价值的逻辑其实非常简单，就是对每个状态下的每个动作能获得的收益进行收集并通过贝尔曼方程求出新的状态价值，如此反复迭代直至收敛。值得注意的是，这里维护了两张状态价值表格：value和new_value，直到每一循环最后才把new_value表整个覆盖到value表上，实际上这种作法十分类似于线性方程迭代求解法中的雅各比迭代法，在每一循环内部其实都是在用老的状态价值更新，而这种迭代方法其实适合并行运算，像上面代码这样的串行计算其实可以像高斯赛德尔方法那样只维护一张表，实时用新算出来的状态价值来更新后面的状态价值，以达到更快的收敛速度。估计本代码作者的原意是为了展现原汁原味的状态价值估计函数。下面是绘图之后的结果。
可以很明显地看到距离A、B点越近的状态其状态价值函数越大，其实此时顺着最大状态价值函数的方向也可以获得最优策略，但是值得注意的是，（5，4）和（4，5）状态的地位其实是相同的，因为它们靠边的数量是一样的，而且到A、B状态的长度也是一致的，但是它们此时却呈现出不同的状态价值，这是由于它们等概率地走向周围的状态，而周围的状态价值是不同的，所以它们的状态价值不同。而如果像后面那样计算最优价值函数，这两个状态就不会等概率地选择下一个状态，而是会选择使状态价值最高的下一个状态（动作），所以这两个格子的最优状态价值函数是一致的，这也就使得（5，5）格子可以向两个方向选择最优策略，相较于用此时的状态价值函数选择策略多了一个方向。我们在后面会看到计算最优价值函数和选择最优策略的过程。

figure_3_5:计算最优价值函数并绘制最佳策略

def figure_3_5():
    value = np.zeros((WORLD_SIZE, WORLD_SIZE))
    while True:
        # keep iteration until convergence
        new_value = np.zeros_like(value)
        for i in range(WORLD_SIZE):
            for j in range(WORLD_SIZE):
                values = []
                for action in ACTIONS:
                    (next_i, next_j), reward = step([i, j], action)
                    # value iteration
                    values.append(reward + DISCOUNT * value[next_i, next_j])   #用一个list存储每个状态下所有动作对应的价值函数，方便下面选择最大值
                new_value[i, j] = np.max(values)   #从这里可以看出，价值迭代并没有像上面策略迭代一样用原始的随机行动带来的价值更行状态价值，而是用当前采取最优行动后获得的最大价值来更新
        if np.sum(np.abs(new_value - value)) < 1e-4:
            draw_image(np.round(new_value, decimals=2))   #把收敛的价值函数绘制到网格世界中，后面会提到这个函数
            plt.savefig('figure_3_5.png')
            plt.close()
            draw_policy(new_value)   #根据最优价值函数绘制最优策略，后面会提到这个函数
            plt.savefig('figure_3_5_policy.png')
            plt.close()
            break
        value = new_value   #我们会在结果的图表中看出策略迭代和价值迭代的区别
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

在动作选择循环的那一层可以明显看到最优价值函数计算和上面估计给定策略下的状态价值的区别。上面是利用选择动作的概率计算了一个期望值，用期望值来估计一个策略下的状态价值，而最优价值估计则是用当前状态下选择动作后能够获得的最大价值来更新。再根据贝尔曼最优方程阐述的事实：最优策略下各个状态的价值一定等于这个状态下最优动作的期望回报，我们很容易顺着最优价值函数的方向得到最优策略。最优价值函数和最优策略的图像如下图所示。

与上面根据一个平均策略估计的状态价值函数不同，在平等地位上的状态有了相同的状态函数，所有的最佳策略都得到了开发，这体现了在更新状态价值的同时更新策略的重要性，一个不良的策略最后获得的状态价值估计也是不良的，而后面的内容正体现了这种思想（GPI）。

杂项函数解析

draw_image：绘制收敛后的价值函数

def draw_image(image):   #将收敛后的价值函数绘制到网格世界中
    fig,ax=plt.subplots()
    ax.set_axis_off()
    tb=Table(ax,bbox=[0,0,1,1])
    nrows,ncols=image.shape
    width,height=1.0/ncols,1.0/nrows
    for (i,j),val in np.ndenumerate(image):
        if [i,j]==A_POS:
            val=str(val)+'(A)'
        if [i,j]==A_PRIME_POS:
            val=str(val)+"(A')" 
        if [i,j]==B_POS:
            val=str(val)+"(B)"
        if [i,j]==B_PRIME_POS:
            val=str(val)+"(B')"   #在特殊点处的价值函数旁边加上标记
        
        tb.add_cell(i, j, width, height,text=val,loc='center',facecolor='white')
    
    for i in range(len(image)):
        tb.add_cell(i, -1, width, height,text=i+1,loc='right',edgecolor='none',facecolor='none')
        tb.add_cell(-1, i, width, height/2,text=i+1,loc='center',edgecolor='none',facecolor='none')
    
    ax.add_table(tb)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

这个函数接收状态价值函数的矩阵并画到网格世界的表格中，在特殊点（A、B、A’、B’）处的价值函数旁标上标记，其他部分就是规划表格格式等，逻辑比较简单就不再赘述。

draw_policy：选择最佳策略并把对应箭头画在网格世界中

def draw_policy(optimal_values):   #绘制最佳策略，即把箭头画在网格世界中
    fig,ax=plt.subplots()
    ax.set_axis_off()
    tb=Table(ax,bbox=[0,0,1,1])
    nrows,ncols=optimal_values.shape
    width,height=1.0/ncols,1.0/nrows
    
    for (i,j),val in np.ndenumerate(optimal_values):
        next_vals=[]
        for action in ACTIONS:
            next_state,_=step([i,j], action)
            next_vals.append(optimal_values[next_state[0],next_state[1]])   #存储下一个状态的状态价值
        best_actions=np.where(next_vals==np.max(next_vals))[0]   #根据下一个状态的状态价值表选择使状态价值最大的最优策略
        val=''
        for ba in best_actions:   #遍历每个状态下的最佳动作
            val+=ACTION_FIGS[ba]   #从上面每个动作对应的图像中选择图像
        
        if [i,j]==A_POS:
            val=str(val)+"(A)"
        if [i,j]==A_PRIME_POS:
            val=str(val)+"(A')"
        if [i,j]==B_POS:
            val=str(val)+"(B)"
        if [i,j]==B_PRIME_POS:
            val=str(val)+"(B')"   #在特殊点处的最佳动作图像旁边加上标记
        
        tb.add_cell(i, j, width, height,text=val,loc='center',facecolor='white')
        
    for i in range(len(optimal_values)):
        tb.add_cell(i, -1, width, height,text=i+1,loc='right',facecolor='none',edgecolor='none')
        tb.add_cell(-1, i, width, height/2,text=i+1,loc='right',facecolor='none',edgecolor='none')
    ax.add_table(tb)
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

根据最优价值函数选择最优策略的过程其实就是从所在状态向状态价值最大的下一个状态行进，当遇到地位相同的下一个状态时就会随机选择一个，这时候就会在一个格子中出现两个箭头。

总结

本章的代码复杂度没有上一章高，表达的意思也十分明显易懂，基本上就是根据书上的描述来进行编程。实际上，本章的书本内容十分重要，介绍了马尔可夫过程和贝尔曼方程这两个基础性的概念。而在实操部分，我们主要练习了根据贝尔曼方程估计确定策略下的状态价值函数和计算最优价值函数并选择最优策略，其实上面也有提到，相比于固定策略估计状态价值，估计价值和更新策略交替进行能够使两者同时得到提升，这正是下面一章所要提到的广义策略迭代的内容。

完整代码

下面是含中文注释的本章完整代码，当中还有一块使用线性方程的方法求解价值函数估计的部分我没有进行分析，主要是内容和课本结合性不强，有兴趣的可以研究一下。

import matplotlib
import matplotlib.pyplot as plt
import numpy as np
from matplotlib.table import Table

matplotlib.use('Agg')
#下列网格参数设定的具体意义参见P58
WORLD_SIZE=5   #网格世界的长宽都是5
A_POS=[0,1]   #标记A的位置
A_PRIME_POS=[4,1]   #标记A'的位置
B_POS=[0,3]   #标记B的位置
B_PRIME_POS=[2,3]   #标记B'的位置
DISCOUNT=0.9   #设定折扣系数为0.9

ACTIONS=[np.array([0,-1]),
         np.array([-1,0]),
         np.array([0,1]),
         np.array([1,0])]   #设定动作，由于网格世界是二维平面，所以用一个二元的向量来表示每个动作
ACTION_FIGS=['←','↑','→','↓']   #设定每个动作对应的箭头，方便画图
ACTION_PROB=0.25   #由于在本实验中仅仅估计价值函数，并根据收敛的最优价值采取最优策略，所以探索的过程中并没有对策略（选择方向的概率）进行修改，
                   #所以每个向任意方向的行动都是等概率的，最后根据收敛的最优价值函数获得最优策略

def step(state,action):
    if state==A_POS:
        return A_PRIME_POS,10   #在A处做任何动作都会转移到A'处并获得+10回报
    if state==B_POS:
        return B_PRIME_POS,5   #在B处做任何动作都会转移到B'处并获得+5回报
    
    next_state=(np.array(state)+action).tolist()
    x,y=next_state   #否则把当前位置加上相应动作向量获得新位置
    if x<0 or x>=WORLD_SIZE or y<0 or y>=WORLD_SIZE:
        reward=-1.0
        next_state=state
    else:
        reward=0   #如果出界，位置不变并获得-1回报，如果没出界且不在A,B处，则获得0回报
    return next_state,reward
#对每一步行动之后的状态和回报（return）进行计算

def draw_image(image):   #将收敛后的价值函数绘制到网格世界中
    fig,ax=plt.subplots()
    ax.set_axis_off()
    tb=Table(ax,bbox=[0,0,1,1])
    nrows,ncols=image.shape
    width,height=1.0/ncols,1.0/nrows
    for (i,j),val in np.ndenumerate(image):
        if [i,j]==A_POS:
            val=str(val)+'(A)'
        if [i,j]==A_PRIME_POS:
            val=str(val)+"(A')" 
        if [i,j]==B_POS:
            val=str(val)+"(B)"
        if [i,j]==B_PRIME_POS:
            val=str(val)+"(B')"   #在特殊点处的价值函数旁边加上标记
        
        tb.add_cell(i, j, width, height,text=val,loc='center',facecolor='white')
    
    for i in range(len(image)):
        tb.add_cell(i, -1, width, height,text=i+1,loc='right',edgecolor='none',facecolor='none')
        tb.add_cell(-1, i, width, height/2,text=i+1,loc='center',edgecolor='none',facecolor='none')
    
    ax.add_table(tb)
    
def draw_policy(optimal_values):   #绘制最佳策略，即把箭头画在网格世界中
    fig,ax=plt.subplots()
    ax.set_axis_off()
    tb=Table(ax,bbox=[0,0,1,1])
    nrows,ncols=optimal_values.shape
    width,height=1.0/ncols,1.0/nrows
    
    for (i,j),val in np.ndenumerate(optimal_values):
        next_vals=[]
        for action in ACTIONS:
            next_state,_=step([i,j], action)
            next_vals.append(optimal_values[next_state[0],next_state[1]])   #存储下一个状态的状态价值
        best_actions=np.where(next_vals==np.max(next_vals))[0]   #根据下一个状态的状态价值表选择使状态价值最大的最优策略
        val=''
        for ba in best_actions:   #遍历每个状态下的最佳动作
            val+=ACTION_FIGS[ba]   #从上面每个动作对应的图像中选择图像
        
        if [i,j]==A_POS:
            val=str(val)+"(A)"
        if [i,j]==A_PRIME_POS:
            val=str(val)+"(A')"
        if [i,j]==B_POS:
            val=str(val)+"(B)"
        if [i,j]==B_PRIME_POS:
            val=str(val)+"(B')"   #在特殊点处的最佳动作图像旁边加上标记
        
        tb.add_cell(i, j, width, height,text=val,loc='center',facecolor='white')
        
    for i in range(len(optimal_values)):
        tb.add_cell(i, -1, width, height,text=i+1,loc='right',facecolor='none',edgecolor='none')
        tb.add_cell(-1, i, width, height/2,text=i+1,loc='right',facecolor='none',edgecolor='none')
    ax.add_table(tb)
    
def figure_3_2():
    value = np.zeros((WORLD_SIZE, WORLD_SIZE))
    while True:
        # keep iteration until convergence
        new_value = np.zeros_like(value)
        for i in range(WORLD_SIZE):
            for j in range(WORLD_SIZE):   #对每一个网格进行动作的试探和价值函数迭代
                for action in ACTIONS:
                    (next_i, next_j), reward = step([i, j], action)   #对每个状态下的每个动作记录状态转移和收益（reward）
                    # bellman equation
                    new_value[i, j] += ACTION_PROB * (reward + DISCOUNT * value[next_i, next_j])   #根据贝尔曼方程估计新的状态价值，这里的策略用的是等概率动作，其实还用到了下一章的动态规划
        if np.sum(np.abs(value - new_value)) < 1e-4:   #收敛的时候结束循环，并用最佳函数值画图
            draw_image(np.round(new_value, decimals=2))   #把收敛的价值函数绘制到网格世界中，后面会提到这个函数
            plt.savefig('figure_3_2.png')
            plt.close()
            break
        value = new_value   #更新所有网格的状态价值函数

def figure_3_2_linear_system():   #这一部分其实使用线性代数的方法解这个环境
    A = -1 * np.eye(WORLD_SIZE * WORLD_SIZE)
    b = np.zeros(WORLD_SIZE * WORLD_SIZE)
    for i in range(WORLD_SIZE):
        for j in range(WORLD_SIZE):
            s = [i, j]  # current state
            index_s = np.ravel_multi_index(s, (WORLD_SIZE, WORLD_SIZE))
            for a in ACTIONS:
                s_, r = step(s, a)
                index_s_ = np.ravel_multi_index(s_, (WORLD_SIZE, WORLD_SIZE))

                A[index_s, index_s_] += ACTION_PROB * DISCOUNT
                b[index_s] -= ACTION_PROB * r

    x = np.linalg.solve(A, b)
    draw_image(np.round(x.reshape(WORLD_SIZE, WORLD_SIZE), decimals=2))
    plt.savefig('figure_3_2_linear_system.png')
    plt.close()   #其实根据画图的结果来看，线性代数的结果和上面用贝尔曼方程进行动态规划的收敛结果是一致的

def figure_3_5():
    value = np.zeros((WORLD_SIZE, WORLD_SIZE))
    while True:
        # keep iteration until convergence
        new_value = np.zeros_like(value)
        for i in range(WORLD_SIZE):
            for j in range(WORLD_SIZE):
                values = []
                for action in ACTIONS:
                    (next_i, next_j), reward = step([i, j], action)
                    # value iteration
                    values.append(reward + DISCOUNT * value[next_i, next_j])   #用一个list存储每个状态下所有动作对应的价值函数，方便下面选择最大值
                new_value[i, j] = np.max(values)   #从这里可以看出，价值迭代并没有像上面策略迭代一样用原始的随机行动带来的价值更行状态价值，而是用当前采取最优行动后获得的最大价值来更新
        if np.sum(np.abs(new_value - value)) < 1e-4:
            draw_image(np.round(new_value, decimals=2))   #把收敛的价值函数绘制到网格世界中，后面会提到这个函数
            plt.savefig('figure_3_5.png')
            plt.close()
            draw_policy(new_value)   #根据最优价值函数绘制最优策略，后面会提到这个函数
            plt.savefig('figure_3_5_policy.png')
            plt.close()
            break
        value = new_value   #我们会在结果的图表中看出策略迭代和价值迭代的区别


if __name__ == '__main__':
    figure_3_2_linear_system()
    figure_3_2()
    figure_3_5()     
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
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160