强化学习第四讲

• FrozenLake—v1(冰湖)环境解释
• 基于策略迭代算法(policy-iteration)
• 基于值迭代算法(value-iteration)
• 一些统计数据和思考
• 总结

FrozenLake-v1环境

先放环境代码和官网简介，恕我英文确实不咋地，理解了好久

综述

简单来说，此模型描述的问题是人物从起点要走到终点，然后地图中存在一些位置是洞，掉进洞或走到终点即游戏结束，但是由于地很滑(冰湖)，运动方向仅部分取决于选取的动作(重点，我就是被这个坑到了，不仔细读文档的屑😭😭😭)

状态空间(observation_space)

以最简单的"4x4"地图为例，起点默认为第一行第一个，终点默认为第四行第四个，中间的洞是随机的，但是起点和终点必然是联通的(环境代码中对于随机生成图是通过宽搜来判定的)，状态空间大小为$4*4=16$，依次按照顺序状态如下：

0	1	2	3
4	5	6	7
8	9	10	11
12	13	14	15

(Markdown表头自动加粗，懒得改HTML了见谅💦)
那么observation_space.n即是16(不知道为什么用Pycharm的时候一开始它会报错，说这个n是未定义的特性引用，大概是因为这是这个模型特有的)

动作空间(action_space)

对于这个问题，动作只有四种，即：

LEFT: 0  
DOWN: 1
RIGHT: 2
UP: 3

状态转移概率P

P[state][action]: 指在状态$s$采取动作$a$的状态转移概率，在本题中，由于规则要求，比如选择动作0(即向左)，那么有三分之一的概率向上，三分之一概率向下，和三分之一概率向左。

那么在这里是不是就可以想到，如果我一直让它背对着洞走，是不是就一定能不掉进去呢，事实上策略就是这样，但是会有例如两个洞包夹的情况，那么肯定会有概率滑进洞里，这时候，就需要计算期望值，选择最不容易掉进去的路线行进。

这里的P[state][action]不止包含了状态转移概率，还包含了下一个状态、下一个状态的reward，以及下一个状态是否结束。(列表形式存储)
注意：

• 在边界上的情况，比如状态1，如果向上运动了，那么相当于停在该点不动
• 在终点(状态15)，此处不会再转移到其他状态，奖励为1，其余点奖励均为0

下图是状态1的所有可能action对应的状态转移P[1][action]：

特别解释一下最后的bool值，因为掉进洞里或者到达终点都视为结束，可以看出状态5是个洞。
下图是状态15对应的状态转移：

可以看出在终点无论采取什么动作都是不动的，且视为结束

reset()函数

参数：无
返回值：起点的state值

step()函数

参数：动作action
返回值：执行完action后得到的state(即s’)值、从s到s’获得的reward奖励、是否结束(done,是一个布尔值)

render()函数

参数：mode(默认为'human')
无返回值，有两种render方法

1. 一种是text文本方式，环境代码中采用了StringIO读取和输出(至于怎样动画化，像官方文档的视频一样还没弄懂)，对应mode='ansi'。
2. 第二种是gui形式，利用pygame包将整个过程动画化并演示出来，对应mode='human'

策略迭代算法

总体思路请参照前述策略迭代部分，下面是各步骤源码和一些解释。
整体为两部分，第一部分是策略迭代，其中包括策略评估(迭代计算状态值函数value_table)和策略改善(通过动作状态值函数q选择最佳策略)，第二部分是对该模型执行策略(run policy)，每轮执行是一个episode。

import gym
import numpy as np


def evaluate_policy(env, value_table, policy, gamma=1.0, theta=1e-4):
    delta = 2 * theta
    while delta > theta:
        # 初始化一个新的v表，注意下面迭代计算v表时使用的+=，所以这里应该将其初始化为0
        new_value_table = np.zeros(env.observation_space.n)
        for state in range(env.observation_space.n):
            # 从待评估策略中提取动作
            action = policy[state]
            # 对每种可能的状态转移过程进行求和得到新v表
            for prob, next_state, reward, done in env.P[state][action]:
                new_value_table[state] += prob * (reward + gamma * value_table[next_state])
        # 计算两表误差，当误差小于限定精度时退出迭代
        delta = sum(np.fabs(new_value_table - value_table))
        value_table = np.copy(new_value_table)
    # 返回待评估策略得到的状态值函数
    return value_table


def improve_policy(env, value_table, policy, gamma=1.0):
    while True:
        # 复制一份旧策略
        old_policy = np.copy(policy)
        for state in range(env.observation_space.n):
            # 初始化一个q函数，注意大小
            q_value = np.zeros(env.action_space.n)
            # 对每个动作求状态动作值函数
            for action in range(env.action_space.n):
                for prob, next_state, reward, done in env.P[state][action]:
                    q_value[action] += prob * (reward + gamma * value_table[next_state])
            # 利用argmax返回q函数的最大值索引，即在state对应的价值最大的action，作为该state的策略
            policy[state] = np.argmax(q_value)
        # 比较新旧策略的每一项，若都相等即退出迭代
        if np.all(policy == old_policy):  break
    # 返回改善后的策略
    return policy


def policy_iterate(env, gamma=1.0, iterations=10000):
    # 初始化原始策略，这里选择利用随机的方式
    policy = np.random.randint(low=0, high=env.action_space.n, size=env.observation_space.n)
    # 初始化原始状态值函数，每个状态均为0
    value_table = np.zeros(env.observation_space.n)
    # 计算迭代次数
    iters = 0
    for i in range(iterations):
        iters += 1
        old_policy = np.copy(policy)
        # 策略评估
        value_table = evaluate_policy(env, value_table, policy, gamma)
        # 策略改善
        policy = improve_policy(env, value_table, policy, gamma)
        if np.all(policy == old_policy):
            break
    # 返回状态值函数，最优策略，策略迭代次数
    return value_table, policy, iters


def run_policy(env, policy, gamma=1.0):
    # 初始化状态
    state = env.reset()
    # 第1帧
    env.render()
    # 初始化这一轮episode的总价值和帧数
    tot_reward, num_step = 0, 1
    while True:
        # 采用policy中当前state对应的action，得到next_state，回报和终止判断
        next_state, reward, done, _ = env.step(policy[state])
        # 可以选择env.render(mode='ansi')
        env.render()
        # 计算总价值，注意该步的价值等于衰减系数的当前帧数次方乘上这一次状态转移的回报
        tot_reward += gamma ** num_step * reward
        num_step += 1
        if done:
            break
        # 更新状态
        state = next_state
    # 返回总价值和结束时的帧数
    return tot_reward, num_step


env = gym.make('FrozenLake-v1', desc=None, map_name="4x4") # 使用默认的4x4地图
# 衰减系数
gamma = 1.0
# 策略迭代
value, policy, iters = policy_iterate(env, iterations=1000)
print('policy:',end='')
print(policy)
print('iterations:',end='')
print(iters)
for episode in range(1):
    cum_reward, nsteps = run_policy(env, policy)
    print('Episode {}:\nreward:{}\tsteps:{}'.format(episode + 1, cum_reward, nsteps))

对于numpy包，有时间会记录一些学习笔记，这里放一个官方教程,这是一个很好的处理多维数据(譬如对于复杂的多维状态空间，动作空间，或者说数据处理统计分类时起很大的作用)的包，在安装anaconda时会同时附加安装。

另外需要注意一点，初始化策略时，这里使用的是np.random.randint，所以策略中的动作均为整数符合条件，如果用np.zeros初始化，注意要将默认参数改为dtype='int'(默认为float64型)

值函数迭代

import gym
import numpy as np


def value_iterate(env, gamma=1.0, theta=1e-4):
    # 初始化策略
    policy = np.random.randint(low=0, high=env.action_space.n, size=env.observation_space.n)
    # 初始化值函数，这里可以在循环外只初始化依次用于迭代的表是因为下面使用的是max进行赋值，而区别于策略迭代的+=
    value_table = np.zeros(env.observation_space.n)
    new_value_table = np.zeros(env.observation_space.n)
    # 迭代次数和迭代误差
    iters = 0
    delta = 2 * theta
    while delta > theta:
        iters += 1
        for state in range(env.observation_space.n):
            # 初始化q函数
            q_value = np.zeros(env.action_space.n)
            for action in range(env.action_space.n):
                for prob, next_state, reward, done in env.P[state][action]:
                    q_value[action] += prob * (reward + gamma * value_table[next_state])
            # 选取q函数中价值最大的action，用于更新v表，即相当于只迭代评估计算一次v表立即进行策略改进
            new_value_table[state] = max(q_value)
            # 找q函数的最大值索引，更新策略
            policy[state] = np.argmax(q_value)
        # 计算迭代前后v表误差
        delta = sum(np.fabs(new_value_table - value_table))
        # 更新v表,注意要用copy进行拷贝
        value_table = np.copy(new_value_table)
    # 返回策略，值函数和值迭代次数
    return policy, value_table, iters


def run_policy(env, policy, gamma=1.0):
# 同策略迭代

需要注意的是更新值函数时都需要用copy，而不是直接赋值(srds我也不知道为什么策略迭代直接赋值不会出问题)，原因是python的一些特性：

Python 赋值过程中不明确区分拷贝和引用，一般对静态变量的传递为拷贝，对动态变量的传递为引用。（注，对静态变量首次传递时也是引用，当需要修改静态变量时，因为静态变量不能改变，所以需要生成一个新的空间存储数据）。

而这里使用copy即是将ndarray数组进行了深拷贝，完全的创立了副本。
参考文章：runoob、numpy

对于4x4地图的结果及分析

默认4x4图

S	F	F	F
F	H	F	H
F	F	F	H
H	F	F	G

• S: starting point, safe(起点)
• F: frozen surface, safe(冰面)
• H: hole, fall to your doom(洞)
• G: goal(终点)

最佳策略和GUI演示

policy:[0 3 3 3 0 0 0 0 3 1 0 0 0 2 1 0]

视频演示
别骂了别骂了，什么dplayer插件是真的用不来啊😤😤，捏麻麻的折腾了一下午都没弄懂，有会的大佬就帮助一下吧。。。
另外可以看到pycharm是会警告的，原因是这个环境中调用的distutils.spawn即将在python3.12移除(现已废弃)，但不妨碍正常运行。

一些定性分析

定量一点的分析咱就不做了，前前后后也倒腾了四天，那些什么用csv格式绘表啊，或者用matplotlib来实现可视化咱也就不做了，属于是啥都不知道的fw了😅
在这里也放一份大佬的博客(注意他的项目文件啥的在文章里有链接，我就不贴了，有兴趣可以看一看,他是采用的SARSA算法进行学习的)
我这里就"盗用"一张这位大佬用matplotlib绘制的一个在FrozenLake-v1中的图(在他的analysis.py文件中可以看到绘制方法，顺带着他还绘制了value_table的图)，在实验中也可以发现，如果单个episode走了100步，step函数也会返回终止为True，至于为什么我也不知道，环境代码中没有给出相应的解释。

总结

• 不打不知道，我是真的蠢，一切模型算法实验都要亲自上手，才能够逐步理解，我在自己照猫画虎写了策略迭代算法后，发现了各种我不认识的错误，包括numpy包的一些不会的地方，包括环境代码的解读，也包括怎样跑策略，将其可视化，或者整理数据，等等
• 虽然花了很多心思，而且这只是一个OpenAI gym里的“easy”的模型😢，但是万事开头难嘛，总得要有开头，生命不止，折腾不息😋😋
• 强推Sutton和Barto的书，就是有点小贵()
• 我的ddl我对不起你！这就来伺候您老人家😍😍

参考文章

• Machine Learning Journey
• CSDN