强化学习之蒙特卡洛学习,时序差分学习理论与实战(四)

CristianoC

发布于 2020-12-15 11:23:38

9130

发布于 2020-12-15 11:23:38

文章被收录于专栏：计算机视觉漫谈计算机视觉漫谈

前言

上一讲讲解了如果应用动态规划算法对一个已知状态转移概率的MDP进行策略评估或通过策略迭代或直接的价值迭代来寻找最优策略和最优价值函数,同时也指出了动态规划算法的一些缺点.
从本讲开始的连续两讲将讲解如何解决一个可以被认为是MDP,但却不掌握MDP具体细节的问题,也就是讲述个体如何在没有对环境动力学认识的模型的条件下如何直接通过个体与环境的实际交互来评估一个策略的好坏或者寻找到最优价值函数和最优策略.其中这章将聚焦于策略评估,也就是预测的问题.
本章分为三个部分,将分别从理论上阐述基于完整采样的蒙特卡洛强化学习,基于不完整采样的时序差分强化学习以及介于两者之间的

\lambda

时序差分强化学习.本章将会结合一些实例来加深读者的理解.

蒙特卡洛强化学习

蒙特卡洛强化学习(Monte-Carlo Reinforcement Learning,MC学习):指在不清楚MDP状态转移概率的情况下,直接从经历完整的状态序列(episode)来估计状态的真实价值,并认为某状态的价值等于在多个状态序列中以该状态算得到的所有收获的平均值.
完整的状态序列(complete episode):指从某一个状态开始,个体与环境交互直到终止状态的奖励为止.完整的状态序列不要求起始状态一定是某一个特定的状态,但是要求个体最终进入环境认可的某一个终止状态.
蒙特卡洛强化学习有如下的特点:不依赖状态转移概率(即不依赖模型),直接从经历过的完整的状态序列中学习,使用的思想就是用平均收获值代替价值.理论上完整的状态序列越多,结果越准确.
我们可以使用蒙特卡洛强化学习来评估一个给定的策略.基于特定策略

\pi

的一个状态序列信息可以表示为如下的一个序列(即在初始状态执行某动作,获得离开该状态的即时奖励,到达下一个状态

\cdots

其中

时刻状态

S_t

的收获可以表述为:

其中

时刻为终止时刻.该策略下某一状态

的价值:

在蒙特卡洛算法评估策略的时候要针对多个包含同一状态的完整状态序列求收获继而求收获的平均值.在状态转移过程中,某一需要计算的状态可能出现在序列的多个位置,也就是说个体在与环境交互的过程中从某状态出发后又一次或多次返回该状态.根据收获的定义,不同时刻的同一状态其计算得到的收获值是不一样的.
我们有两种方法可以选择,一是仅把状态序列中第一次出现该状态时收获值纳入到收获平均值的计算中;另一种是针对一个状态序列中每次出现的该状态,都计算对应的收获值并纳入收获平均值的计算当中.两种方法对应的蒙特卡洛评估分别称为首次访问(first visit)和每次访问(every visit)蒙特卡洛评估.
在求解状态收获的平均值的过程中,我们介绍一种非常实用的,不需要存储所有历史收获的计算方法:累进更新平均值(incremental mean).而且这种计算平均值的思想也是强化学习的一个核心思想之一.具体公式如下:

累进更新平均值利用前一次的平均值和当前数据以及数据总个数来计算新的平均值:当每产生一个需要平均的新数据

x_k

时,先计算

x_k

与先前平均值

\mu_{k-1}

的差,再将这个差值乘以一定的系数

\frac{1}{k}

后作为误差对旧平均值进行修正.如果该式中平均值和新数据分别看成是状态的价值和该状态的收获,那么该公式就变成了递增式蒙特卡洛法更新状态价值.公式如下:

在一些实时或者无法统计准确状态被访问次数时,可以用一个系数

\alpha

来代替状态计数的倒数,公式变成:

以上就是蒙特卡洛学习方法的思想和描述,下文将介绍另一种强化学习方法:时序差分学习法

时序差分强化学习

时序差分强化学习(Temporal-Difference reinforcement Learning,TD学习):指从采样得到的不完整的状态序列学习,该方法通过引导(bootstrapping),先估计某状态在该状态序列完整后可能获得的收获,并在此基础上利用前文所属的累进更新平均值的方法得到该状态的价值,再通过不断的采样来持续更新这个价值.
具体地说,在TD学习中,算法在估计某一个状态的收获时,用的是离开该状态的即刻奖励

R_{t+1}

与下一时刻状态

S_{t+1}

的预估状态价值乘以衰减系数

\gamma

组成,这符合贝尔曼方程的描述:

其中

R_{t+1} + \gamma V(S_{t+1})

称为TD目标值.

R_{t+1} + \gamma V(S_{t+1}) - V(S_t)

称为TD误差.

引导(bootstrapping):指的是用TD目标值代替收获

G_t

的过程

可以看出,不管是MC学习还是TD学习,它们都不再需要清楚某一状态的所有可能的后续状态以及对应的状态转移概率,因此也不用像动态规划算法那样进行全宽度的回溯来更新状态的价值.这在解决大规模问题或者不清楚环境动力学特征的问题时十分有效.不过MC学习和TD学习两者也是有着很明显的差别的.下文通过一个例子来详细阐述这两种学习方法的特点.

MC学习和TD学习的区别

想象一下作为个体的你如何预测下班后开车回家这个行程所花费的时间.在回家的路上你会依次经过一段高速公路,普通公路和你家附近街区三段路程.由于你经常开车上下班,在下班的路上多次碰到过各种情形,比如取车的时候发现下雨,高速路况的好坏,普通公路是否堵车等等.在每一种状态下时,你对还需要多久才能到家都有一个经验性的估计.下表的既往经验预计的仍需好时列给出了这个经验估计,这个经验估计基本反映了各个状态对应的价值,通过你对下班回家总耗时的预估是30分钟.

假设你现在又下班准备回家了，当花费了5分钟从办公室到车旁时，发现下雨了。此时根据既往经验，估计还需要35分钟才能到家，因此整个行程将耗费40分钟。随后你进入了高速公路，高速公路路况非常好，你一共仅用了20分钟就离开了高速公路，通常根据经验你只再需要15分钟就能到家，加上已经过去的20分钟，你将这次返家预计总耗时修正为35分钟，比先前的估计少了5分钟。但是当你进入普通公路时，发现交通流量较大，你不得不跟在一辆卡车后面龟速行驶，这个时候距离出发已经过去30 分钟了，根据以往你路径此段的经验，你还需要10分钟才能到家，那么现在你对于回家总耗时的预估又回到了40分钟。最后你在出发40分钟后到达了家附近的街区，根据经验，还需要3分钟就能到家，此后没有再出现新的情况，最终你在43分钟的时候到达家中。经历过这一次的下班回家，你对于处在途中各种状态下返家的还需耗时（对应于各状态的价值）有了新的估计，但分别使用MC算法和TD算法得到的对于各状态返家还需耗时的更新结果和更新时机都是不一样的。
如果使用MC算法，在整个驾车返家的过程中，你对于所处的每一个状态，例如“取车时下雨”，“离开高速公路”，“被迫跟在卡车后”、“进入街区”等时，都不会立即更新这些状态对应的返家还需耗时的估计，这些状态的返家仍需耗时仍然分别是先前的35 分钟、15分钟、10分钟和3分钟。但是当你到家发现整个行程耗时43分钟后，通过用实际总耗时减去到达某状态的已耗时，你发现在本次返家过程中在实际到达上述各状态时，仍需时间则分别变成了：38分钟、23分钟、13分钟和3分钟。如果选择修正系数为1，那么这些新的耗时将成为今后你在各状态时的预估返家仍需耗时,相应的整个行程的预估耗时被更新为43分钟
如果使用TD算法，则又是另外一回事，当取车发现下雨时，同样根据经验你会认为还需要35分钟才能返家，此时，你将立刻更新对于返家总耗时的估计，为仍需的35分钟加上你离开办公室到取车现场花费的5分钟，即40分钟。同样道理，当驶离高速公路，根据经验，你对到家还需时间的预计为15分钟，但由于之前你在高速上较为顺利，节省了不少时间，在第20分钟时已经驶离高速，实际从取车到驶离高速只花费了15分钟，则此时你又立刻更新了从取车时下雨到到家所需的时间为30分钟，而整个回家所需时间更新为35分钟。当你在驶离高速在普通公路上又行驶了10 分钟被堵，你预计还需10分钟才能返家时，你对于刚才驶离高速公路返家还需耗时又做了更新，将不再是根据既往经验预估的15 分钟，而是现在的20分钟，加上从出发到驶离高速已花费的20分钟，整个行程耗时预估因此被更新为40分钟。直到你花费了40分钟只到达家附近的街区还预计有3分钟才能到家时，你更新了在普通公路上对于返家还需耗时的预计为13分钟。最终你按预计3 分钟后进入家门，不再更新剩下的仍需耗时。
通过比较可以看出,MC算法只在整个行程结束后才更新各个状态的仍需耗时,而TD算法则每经过一个状态就会根据在这个状态与前一个状态间实际所花时间来更新前一个状态的仍需耗时.下图用折线图直观的显示了分别使用MC和TD算法时预测的驾车回家总耗时的区别.需要注意的是,在这个例子中,与各状态价值相对应的指标并不是图中显示的驾车返家总耗时,而是处于某个状态时驾车返家的仍需耗时.

TD学习能比MC学习更快速灵活的更新状态的价值估计,这在一些情况下是非常重要的.回到驾车回家这个例子中来,我们给驾车回家制定一个新的目标,不再以耗时多少来评估状态价值,而是要求安全平稳的返回家中.假如有一次你在驾车回家的路上突然碰到险情:对面开过来一辆车准备跟你相撞,不过由于双方驾驶员都采取了紧急措施没有让险情实际发生,最后平安到家.如果用MC学习,路上发生的这一险情可能引发的极大负值奖励将不会被考虑,你不会更新在碰到此类险情时状态的价值;但在TD学习时,碰到这样的险情过后,你会大幅调低这个状态的价值,并在今后再次碰到类似情况时采取其他行为,例如降低速度等来让自身处在一个价值较高的状态中,尽可能避免发生意外事件的发生.
通过这个例子,我们认识到:TD学习在知道结果之前就可以学习,也可以在没有结果时学习,还可以在持续进行的环境中学习,而MC学习则要等到最后结果才能学习.
TD学习在更新状态价值时使用的是TD目标值,即基于即时奖励和下一状态的预估价值来替代当前状态在状态序列结束时可能得到的收获,它是当前状态价值的有偏估计,而MC学习则使用实际的收获来更新状态价值,是基于某一策略下状态价值的无偏估计.TD学习存在偏倚(bias)的原因是在于其更新价值时使用的也是后续状态预估的价值,如果能使用后续状态基于某策略的真实TD目标值(true TD target)来更新当前状态价值的话,那么此时的TD学习得到的价值也是实际价值的无偏估计.虽然绝大多数情况下TD学习得到的价值是有偏估计的,但是其方差(Variance)却比MC学习得到的方法要低,且对初始值敏感,通常比MC学下更加高效,这也主要得益于TD学习价值更新灵活,对初始状态价值的依赖较大.
这里的偏倚指的是距离期望的距离，预估的平均值与实际平均值的偏离程度；变异性指的是方差，评估单次采样结果相对于与平均值变动的范围大小。基本就是统计学上均值与方差的概念。
假设在一个强化学习问题中有A,B两个状态,模型未知,不涉及策略和行为,只涉及状态转换和即时奖励,衰减系数为1.现有如下表所示8个完整状态序列的经历,其中除了第1个状态序列发生了状态转移外,其余7个完整的状态序列均只有一个状态构成.现要求根据现有信息计算状态A,B的价值分别是多少?

我们考虑分别使用MC算法和TD算法来计算状态A,B的价值.首先考虑MC算法,在8个完整的状态序列中,只有第一个序列中包含状态A,因此A价值仅能通过第一个序列来计算,也就等同于计算该序列中状态A的收获:

状态B的价值,则需要通过状态B在8个序列中的收获值来平均,其结果是6/8.因此在使用MC算法时,状态A,B的价值分别为6/8和0.
再来考虑应用TD算法,TD算法在计算状态序列中某状态价值时是应用其后续状态的预估价值来计算的,在8个状态序列中,状态B总是出现在终止状态,因而直接使用终止状态时获得的奖励来计算价值再针对状态序列数做平均,这样得到的状态B的价值依然是6/8.状态A由于只存在于第一个状态序列中,因此直接使用包含状态B的TD目标值来得到状态A的价值,由于状态A的即时奖励为0,因而计算得到的状态A的价值与B的价值相同,均为6/8.
TD算法在计算状态价值时利用了状态序列中前后状态之间的关系,由于已知信息仅有8个完整序列,而且状态A的后续状态100%是状态B,而状态B始终作为终止状态,有1/4获得奖励0,3/4获得奖励1.符合这样的状态转移概率的MDP如下图所示.

可以看出,TD算法试图构建一个MDP并使得这个MDP尽可能的符合已经产生的状态序列,也就是说TD算法将首先根据已有经验估计状态间的转移概率:

同时估计一某一个状态的即时奖励:

最后计算MDP的状态函数.
而MC算法则直接依靠完整状态序列的奖励得到的各状态对应的收获来计算状态价值,因而这种算法是以最小化收获与状态价值之间均方差为目标的:

通过上面的示例,我们能体会到TD算法与MC算法之间的另一个差别:TD算法使用了MDP问题的马尔科夫属性,在具有马尔科夫性的环境下更有效;但是MC算法并不利用马尔科夫属性,适用范围不限于具有马尔科夫性的环境.

小结

本章阐述的MC算法,TD算法和上一章讲述的动态规划算法都可以用来计算状态价值.他们的特点也是十分鲜明的,前两种是在不依赖模型的情况下的常用方法,这其中又以MC学习需要完整的状态序列来更新状态价值,TD学习则不需要完整的状态序列;DP算法则是基于模型的计算状态价值的方法,它通过计算一个状态S所有可能的转移状态S'及其转移概率以及对应的即时奖励来计算这个状态S的价值.
在是否使用引导数据上,MC学习并不使用引导数据,它使用实际产生的奖励值来计算状态价值;TD和DP则都是用后续状态的预估价值作为引导数据来计算当前状态的价值.
在是否采样的问题上,MC和TD不依赖模型,使用的都是个体与环境实际交互产生的采样状态序列来计算状态价值的,而DP则依赖状态转移概率矩阵和奖励函数,全宽度计算状态价值,没有采样之说.
MC算法:深层采样算法,使用一次完整的状态序列进行学习,使用实际收获更新状态预估价值,如下图所示:

TD算法:浅层采样算法.经历可以不完整,使用后续状态的预估状态价值预估收获再更新当前状态价值.如下图所示:

DP算法:浅层全宽度学习.依据模型,全宽度地使用后续状态预估价值来更新当前状态价值.如下图所示:

综合以上,可以小结:当使用单个采样,同时不经历完整的状态序列更新价值的算法是TD学习;当使用单个采样,但依赖完整状态序列的算法是MC算法;当考虑全宽度采样,但对每一个采样经历只考虑后续一个状态时的算法是DP学习;如果既考虑所有状态转移的可能性,同时又依赖完整状态序列的,那么这种算法是穷举法(exhausive search).需要说明的是:DP利用的是整个MDP问题的模型,也就是状态转移概率,虽然它并不实际利用采样经历,但它利用了整个模型的规律,因此也被认为是全宽度采样的.

n步时序差分学习

第二节介绍的TD算法实际上都是TD(0)算法,括号内的数字0表示的是在当前状态下往前多看1步,要是往前多看2步更新状态价值会怎么样?这就引入了n-步预测的概念.
n-步预测指从状态序列的当前状态(

S_t

)开始往序列终止状态方向观察至状态

S_{t+n-1}

,使用这n个状态产生的即时奖励(

R_{t+1},R_{t+2},\cdots,R_{t+n}

)以及状态

S_{t+n}

的预估价值来计算当前的状态

S_t

的价值

TD是TD(0)的简写,是基于1-步预测的.根据n-步预测的定义,可以推出n=1,2和

\infty

时对应的预测值如下表所示.从该表可以看出,MC学习是基于

\infty-

步预测的

定义n-步收获为:

由此可以得到n-步TD学习对应的状态价值函数的更新公式为:

我们可以知道,当n=1时等同于TD(0)学习,n取无穷大时等同于MC学习.由于TD学习和MC学习又各有优劣,那么会不会存在一个n值使得预测能够充分利用两种学习的优点或者得到一个更好的预测效果呢?研究认为不同的问题其对应的比较高效的步数不是一成不变的.选择多少步数作为一个较优的计算参数是需要尝试的超参数调优问题.
为了能在不增加计算复杂度的情况下综合考虑所有步数的预测(即将不同n下的n步回报值做加权平均，构成一个有效的回报值),我们引入了一个新的参数

\lambda

,并定义

\lambda-

收获为:

从n=1到

\infty

的所有步收获的权重之和.其中,任意一个n-步收获的权重被设计为

(1-\lambda)\lambda^{n-1}

,如下图所示.

通过这样的权重设计,可以得到

\lambda-

收获的计算公式为:

对应的TD(

\lambda

)被描述为:

下图显示了TD(

\lambda

)对于n-收获的权重分配,左侧阴影部分是3-步收获的权重值,随着n的增大,其n-收获的权重呈几何级数的衰减.当在T时刻到达终止状态时,未分配的权重(右侧阴影部分)全部给予终止状态的实际收获值.如此设计可以使一个完整的状态序列中所有n-步收获的权重加起来为1,离当前状态越远的收获其权重越小.

TD(

\lambda

)的设计使得在一个episode中，后一个状态的状态价值与之前所有状态的状态价值有关，同时也可以说成是一个状态价值参与决定了后续所有状态的状态价值。但是每个状态的价值对于后续状态价值的影响权重是不同的。

前向认识TD

(\lambda)

:TD

(\lambda)

的设计使得在状态序列中,一个状态的价值

V(S_t)

由

G^{(\lambda)}_t

得到,而后者又间接由所有后续状态价值计算得到,因此可以认为更新一个状态的价值需要知道所有后续状态的价值.也就是说,必须要经历完整的状态序列获得包括终止状态的每一个状态的即时奖励才能更新当前状态的价值.这和MC算法的要求一样,因此TD

(\lambda)

算法有着和MC算法一样的劣势.

\lambda

取值区间为

[0,1]

,当

\lambda = 1

对应的就是MC算法.这个给实际计算带来了不便.这里可以用一个例子方便大家理解:前向认识就假设一个人坐在状态流上拿着望远镜看向前方，前方是那些将来的状态。当估计当前状态的值函数时，从TD(λ)的定义中可以看到，它需要用到将来时刻的值函数。

反向认识TD

(\lambda)

为TD

(\lambda)

算法进行在线实时单步更新学习提供了理论依据.为了解释这一点,需要引入效用迹这个概念.我们通过一个之前的例子来解释这个问题,如下图所示:

老鼠在依次连续接受了3次响铃和1次亮灯信号后遭到了电击,那么在分析遭电击的原因时,到底是响铃的因素较重要还是亮灯的因素更重要呢?
如果把老鼠遭到电击的原因认为是之前接受了较多次数的响铃,则称这种归因为频率启发式(frequency heuristic);而把电击归因于最近少数几次状态的影响,则称为就近启发(recncy heuristic)式.如果给每一个状态引入一个数值:效用(eligibility,E)来表示该状态对后续状态的影响,就可以同时利用到上述两个启发.而所有状态的效用值总称为效用迹(eligibility traces,ES).定义:

公式中的1(

S_t = s

)是一个真判断表达式,表示当

S_t = s

时取值为1,其余条件下取值为0.

下图给出了效用E对于时间t的一个可能的曲线:

该图横坐标是时间,横坐标下有竖线的位置代表当前时刻的状态为s,纵坐标是效用的值.可以看出当某一状态连续出现时,E值会在一定衰减的基础上有一个单位数值的提高,此时认为该状态将对后续状态的影响较大,如果该状态很长时间没有经历,那么该状态的E值将逐渐趋向于0,表明该状态对于较远的后续状态价值的影响越来越少.
效用迹的提出是基于一个信度分配（Credit Assignment）问题的，打个比方，最后我们去跟别人下围棋，最后输了，那到底该中间我们下的哪一步负责？或者说，每一步对于最后我们输掉比赛这个结果，分别承担多少责任？这就是一个信度分配问题。对于小鼠问题，小鼠先听到三次铃声，然后看见灯亮，接着就被电击了，小鼠很生气，它仔细想，究竟是铃声导致的它被电击，还是灯亮导致的呢？如果按照事件的发生频率来看，是铃声导致的，如果按照最近发生原则来看，那就是灯亮导致的，但是，更合理的想法是，这二者共同导致小鼠被电击了，于是小鼠为这两个事件分别分配了权重，如果某个事件

发生，那么

对应的效用迹的值就加1，如果在某一段时间

未发生，则按照某个衰减因子进行衰减，这也就是上面的效用迹的计算公式了。

需要指出的是,针对每一个状态存在一个E值,且E值并不需要等到状态序列到达终止状态才能计算出来,它是根据已经经过的状态序列来计算得到,并且在每一个时刻都对每一个状态进行一次更新.E值存在饱和现象,有一个瞬时最高上限:

E值是一个非常符合神经科学相关理论的,非常精巧的设计.可以把它看成是神经元的一个参数,它反映了神经元对某一刺激的敏感性和适应性.神经元在接受刺激时会有反馈,在持续刺激时反馈一般也比较强,当间歇一段时间不刺激的时候,神经元又逐渐趋向静息状态;同时不论如何增加刺激的频率,神经元有一个最大饱和反馈.

后向视角使用了我们刚刚定义的效用迹，每个状态

都保存了一个效用迹。我们可以将效用迹理解为一个权重，状态

被访问的时间离现在越久远，其对于值函数的影响就越小，状态

被访问的次数越少，其对于值函数的影响也越小。同样用一个例子说明:有个人坐在状态流上，手里拿着话筒，面朝着已经经历过的状态获得当前回报并利用下一个状态的值函数得到TD偏差之后，此人会向已经经历过的状态喊话告诉这些已经经历过的状态处的值函数需要利用当前时刻的TD偏差进行更新。此时过往的每个状态值函数更新的大小应该跟距离当前状态的步数有关。

如果我们在更新状态价值时把该状态的效用同时考虑进来,那么价值更新可以表示为:

\lambda = 0

时,

S_t = t

一直成立,此时价值更新等同于TD(0)算法:

\lambda

= 1时,在每完成一个状态序列后更新状态价值时,其完全等同于MC学习;但在引入了效用迹后,可以每经历一个状态就更新状态的价值,这种实时更新的方法并不完全等同于MC.

\lambda \in (0,1)

时,在每完成一个状态序列后更新价值时,基于前向认识的TD(

\lambda

)与基于反向认识的TD(

\lambda

)完全等效;不过在进行在线实时学习时,两者存在一些差别.

编程实践

本章的编程实践将使用MC学习来评估二十一点游戏中一个玩家的策略。为了完成这个任务，我们需要先了解二十一点游戏的规则，并构建一个游戏场景让庄家和玩家在一个给定的策略下进行博弈生成对局数据。这里的对局数据在强化学习看来就是一个个完整的状态序列组成的集合。然后我们使用本章介绍的蒙特卡罗算法来评估其中玩家的策略。本节的难点不在于蒙特卡罗学习算法的实现，而是对游戏场景的实现并生成让蒙特卡罗学算法学习的多个状态序列.

二十一点游戏规则

二十一点游戏是一个比较经典的对弈游戏，其规则也有各种不同的版本，为了简化，本文仅介绍由一个庄家 (dealer) 和一个普通玩家 (player，下文简称玩家) 共2位游戏者参与的一个比较基本的规则版本。游戏使用一副除大小王以外的52张扑克牌，游戏者的目标是使手中的牌的点数之和不超过21点且尽量大。其中2-10的数字牌点数就是牌面的数字，J,Q,K三类牌均记为10 点，A既可以记为1也可以记为11，由游戏者根据目标自己决定。牌的花色对于计算点数没有影响。
开局时，庄家将依次连续发2张牌给玩家和庄家，其中庄家的第一张牌是明牌，其牌面信息对玩家是开放的，庄家从第二张牌开始的其它牌的信息不对玩家开放。玩家可以根据自己手中牌的点数决定是否继续叫牌 (twist) 或停止叫牌 (stick), 玩家可以持续叫牌，但一旦手中牌点数超过 21 点则停止叫牌。当玩家停止叫牌后，庄家可以决定是否继续叫牌。如果庄家停止叫牌，对局结束，双方亮牌计算输赢计算输赢的规则如下：如果双方点数均超过21点或双方点数相同，则和局；一方21点另一方不是21点，则点数为21 点的游戏者赢；如果双方点数均不到21点，则点数离21点近的玩家获胜

将二十一点游戏建模为强化学习问题

为了讲解基于完整状态序列的蒙特卡罗学习算法，我们把二十一点游戏建模成强化学习问题，设定由下面三个参数来集体描述一个状态：庄家的明牌 (第一张牌) 点数；玩家手中所有牌点数之和；玩家手中是否还有“可用 (useable)”的 A(ace)。前两个比较好理解，第三个参数是与玩家策略相关的，玩家是否有A这个比较好理解，可用的A指的是玩家手中的A按照目标最大化原则是否没有被计作1点，如果这个A没有被记为1点而是计为了11点，则成这个A为可用的A，否则认为没有可用的A，当然如果玩家手中没有A，那么也被认为是没有可用的A。例如玩家手中的牌为“A,3,6”，那么此时根据目标最大化原则，A将被计为 11 点，总点数为20点，此时玩家手中的A称为可用的A。加入玩家手中的牌为“A，5,7”，那么此时的A不能被计为11点只能按1计，相应总点数被计为13点，否则总点数将为23点，这时的A就不能称为可用的A
根据我们对状态的设定，我们使用由三个元素组成的元组来描述一个状态。例如使用 (10,15,0) 表示的状态是庄家的明牌是10，玩家手中的牌加起来点数是15，并且玩家手中没有可用的A，(A,17,1) 表述的状态是庄家第一张牌为A，玩家手中牌总点数为 17, 玩家手中有可用的A。这样的状态设定不考虑玩家手中的具体牌面信息，也不记录庄家除第一张牌外的其它牌信息。所有可能的状态构成了状态空间.
该问题的行为空间比较简单，玩家只有两种选择：“继续叫牌”或“停止叫牌”。
该问题中的状态如何转换取决于游戏者的行为以及后续发给游戏者的牌，状态间的转移概率很难计算.
可以设定奖励如下：当棋局未结束时，任何状态对应的奖励为 0；当棋局结束时，如果玩家赢得对局，奖励值为1，玩家输掉对局，奖励值为-1，和局是奖励为0。
本问题中衰减因子

\gamma = 1

游戏者在选择行为时都会遵循一个策略。在本例中，庄家遵循的策略是只要其手中的牌点数达到或超过17点就停止叫牌。我们设定玩家遵循的策略是只要手中的牌点数不到20点就会继续叫牌，点数达到或超过20点就停止叫牌.
我们的任务是评估玩家的这个策略，即计算在该策略下的状态价值函数，也就是计算状态空间中的每一个状态其对应的价值.

游戏场景的搭建

首先来搭建这个游戏场景，实现生成对局数据的功能，我们要实现的功能包括：统计游戏者手中牌的总点数、判断当前牌局信息对应的奖励、实现庄家与玩家的策略、模拟对局的过程生成对局数据等。为了能尽可能生成较符合实际的对局数据，我们将循环使用一副牌，对局过程中发牌、洗牌、收集已使用牌等过程都将得到较为真实的模拟。我们使用面向对象的编程思想，通过构建游戏者类和游戏场景类来实现上述功能.
首先导入我们要使用的库

from random import shuffle
from queue import  Queue
from tqdm import tqdm
import matplotlib.pyplot as plt
import numpy as np
from mpl_toolkits.mplot3d import Axes3D
from utils import str_key,set_dict,get_dict

经过分析,一个单纯的二十一点游戏者应该至少能记住对局过程中手中牌的信息,知道自己的行为空间,还应该能辨认单张牌的点数以及手中牌的总点数,为此游戏者能够接受发给他的牌以及一局结束后将手中的牌扔掉等.为此编写了一个名为Gamer的游戏者类.代码如下:

class Gamer():
    '''游戏者'''
    def __init__(self, name = "", A = None, display = False):
        self.name = name
        self.cards = [] #手中的牌
        self.display = display #是否显示对局文字信息
        self.policy = None #策略
        self.lerning_method = None #学习方法
        self.A = A #行为空间

    def __str__(self):
        return self.name

    def _value_of(self, card):
        '''
        根据牌的字符判断牌的数值大小,A被输出为1,JQK均为10,其余按牌字符对应的数字取值
        :param card: 牌面信息str
        :return: 牌的大小数值int, A返回1
        '''
        try:
            v = int(card)
        except:
            if card == "A":
                v = 1
            elif card in ['J','Q','K']:
                v = 10
            else:
                v = 0
        finally:
            return v

    def get_points(self):
        '''
        统计一手牌分值,如果使用了A的1点,同时返回True
        :return:tuple(返回牌的总点数,是否使用了可复用的Ace)
        例如['A','10','3'] 返回 (14,False)
        ['A','10'] 返回(21, True)
        '''
        num_of_uneable_ace = 0 #默认没有拿到Ace
        total_point = 0 #总值
        cards = self.cards
        if cards is None:
            return 0, False
        for card in cards:
            v = self._value_of(card)
            if v == 1:
                num_of_uneable_ace += 1
                v = 11
            total_point += v
        while total_point > 21 and num_of_uneable_ace > 0:
            total_point -= 10
            num_of_uneable_ace -= 1
        return total_point,bool(num_of_uneable_ace)

    def receive(self, cards = []):
        '''
        玩家获得一张或者多张牌
        :param cards: 玩家获得的牌
        :return: None
        '''
        cards = list(cards)
        for card in cards:
            self.cards.append(card)

    def discharge_cards(self):
        '''
        玩家把手中的牌清空,扔牌
        :return: None
        '''
        self.cards.clear()

    def cards_info(self):
        '''
        展示牌面具体信息
        :return: None
        '''
        self._info("{}{}现在的牌:{}\n".format(self.role,self,self.cards))
    def _info(self,msg):
        if self.display:
            print(msg, end="")

在代码中,构造一个游戏者可以提供三个参数,分别是该游戏者的姓名 (name)，行为空间 (A) 和是否在终端显示具体信息 (display)。其中设置第三个参数主要是由于调试和展示的需要，我们希望一方面游戏在生成大量对局信息时不要输出每一局的细节，另一方面在观察细节时希望能在终端给出某时刻庄家和玩家手中具体牌的信息以及他们的行为等。我们还给游戏者增加了一些辅助属性，比如游戏者姓名、策略、学习方法等，还设置了一个display以及一些显示信息的方法用来在对局中在终端输出对局信息。在计算单张牌面点数的时候，借用了异常处理。在统计一手牌的点数时，要考虑到可能出现多张 A 的情况。读者可以输入一些测试牌的信息观察这两个方法的输出。
在二十一点游戏中，庄家和玩家都是一个游戏者，我们可以从Gamer类继承出Dealer类和Player类分别表示庄家和普通玩家。庄家和普通玩家的区别在于两者的角色不同、使用的策略不同。其中庄家使用固定的策略，他还能显示第一张明牌给其他玩家。在本章编程实践中，玩家则使用最基本的策略，由于我们的玩家还要进行基于蒙特卡罗算法的策略评估，他还需要具备构建一个状态的能力。我们扩展的庄家类如下:

class Dealer(Gamer):
    '''
    庄家
    '''
    def __init__(self, name = "", A = None, display = False):
        super(Dealer,self).__init__(name, A, display)
        self.role = "庄家" #角色
        self.policy = self.dealer_policy #庄家策略

    def first_card_value(self):
        '''
        显示第一张明牌
        :return: 明牌的点数
        '''
        if self.cards is None or len(self.cards) == 0:
            return 0
        return self._value_of(self.cards[0])

    def dealer_policy(self, Dealer = None):
        '''
        庄家策略的细节
        :param Dealer:庄家
        :return:庄家的策略(行为)
        '''
        action = ""
        dealer_points, _ = self.get_points()
        if dealer_points >= 17:
            action = self.A[1] #停止叫牌
        else:
            action = self.A[0] #继续叫牌
        return action

在庄家类的构造方法中声明其基类是游戏者 (Gamer)，这样他就具备了游戏者的所有属性和方法了。我们给庄家贴了个“庄家”的角色标签，同时指定了其策略，在具体的策略方法中，规定庄家的牌只要达到或超过 17 点就不再继续叫牌。玩家类的代码如下：

class Player(Gamer):
    '''
    玩家
    '''
    def __init__(self, name = "", A = None, display = False):
        super(Player, self).__init__(name, A, display)
        self.policy = self.native_policy
        self.role = "玩家"

    def get_state(self, dealer):
        dealer_first_card_value = dealer.first_card_value()
        player_points, useable_ace = self.get_points()
        return dealer_first_card_value, player_points, useable_ace

    def get_state_name(self, dealer):
        return str_key(self.get_state(dealer))

    def native_policy(self, dealer = None):
        player_points, _ = self.get_points()
        if player_points < 20:
            action = self.A[0]
        else:
            action = self.A[1]
        return action

类似的，我们的玩家类也继承子游戏者 (Gamer)，指定其策略为最原始的策略 (naive_policy)，规定玩家只要点数小于 20 点就会继续叫牌。玩家同时还会根据当前局面信息得到当前局面的状态，为策略评估做准备.
至此游戏者这部分的建模工作就完成了，接下来将准备游戏桌、游戏牌、组织游戏对局、判定输赢等功能。我们把所有的这些功能包装在一个名称为 Arena 的类中。Arena 类的构造方法如下:

class Arena():
    '''
    负责游戏管理
    '''
    def __init__(self, display = None, A = None):
        self.cards = ['A','2','3','4','5','6','7','8','9','10','J','Q',"K"] * 4
        self.card_q = Queue(maxsize=52) #发牌器,里面是洗好的牌
        self.cards_in_pool = [] #已经用过的公开的牌
        self.display = display
        self.episodes = [] #产生的对局信息列表
        self.load_cards(self.cards) #把初始状态的52张牌装入发牌器
        self.A = A #获得行为空间

Arena 类接受两个参数，这两个参数与构建游戏者的参数一样。Arena 包含的属性有：一副不包括大小王、花色信息的牌 (cards)、一个装载洗好了的牌的发牌器 (cards_q)，一个负责收集已经使用过的废牌的池子 (cards_in_pool)，一个记录了对局信息的列表 (episodes)，还包括是否显示具体信息以及游戏的行为空间等。在构造一个 Arena 对象时，我们同时把一副新牌洗好并装进了发牌器，这个工作在 load_cards 方法里完成。我们来看看这个方法的细节:

    def load_cards(self, cards):
        '''
        把收集的牌洗一洗,重新装到发牌器中
        :param cards: 要装入发牌器的多张牌list
        :return: None
        '''
        shuffle(cards) #洗牌
        for card in cards: #deque数据结构只能一个一个添加
            self.card_q.put(card)
        cards.clear()#把原来的牌清空
        return

这个方法接受一个参数 (cards)，多数时候我们将 cards_in_pool 传给这个方法，也就是把桌面上已使用的废牌收集起来传给这个方法，该方法将首先把这些牌的次序打乱，模拟洗牌操作。随后将洗好的牌放入发牌器。完成洗牌装牌功能。Arena 应具备根据庄家和玩家手中的牌的信息判断当前谁赢谁输的能力，该能力通过如下的方法 (reward_of) 来实现:

    def reward_of(self, dealer, player):
        '''
        判断玩家奖励值,附带玩家,庄家的牌点信息
        :param dealer: 庄家
        :param player:玩家
        :return: tuple 奖励值 玩家点数 庄家点数 是否使用A
        '''
        dealer_points, _ = dealer.get_points()
        player_points, useable_ace = player.get_points()
        if player_points > 21:
            reward = -1
        else:
            if player_points > dealer_points or dealer_points > 21:
                reward = 1
            elif player_points == dealer_points:
                reward = 0
            else:
                reward = -1
        return reward, player_points, dealer_points, useable_ace

该方法接受庄家和玩家为参数，计算对局过程中以及对局结束时牌局的输赢信息 (reward)后，同时还返回当前玩家、庄家具体的总点数以及玩家是否有可用的 A 等信息。
下面的方法实现了 Arena 对象如何向庄家或玩家发牌的功能：

    def serve_card_to(self, player, n=1):
        '''
        给庄家或玩家发牌,如果牌不够则将公开牌池的牌洗一洗重新发牌
        :param player: 一个庄家或玩家
        :param n: 一次连续发牌的数量
        :return:None
        '''
        cards = [] #将要发出的牌
        for _ in range(n):
            #要考虑发牌器没有牌的情况
            if self.card_q.empty():
                self._info("\n发牌器没牌了,整理废牌,重新发牌;")
                shuffle(self.cards_in_pool)
                self._info("一共整理了{}张已用牌, 重新放入发牌器\n".format(len(self.cards_in_pool)))
                assert (len(self.cards_in_pool) > 20)
                # 确保一次能收集较多的牌
                # 代码编写不合理时,可能会出现即使某一玩家爆点了也还持续地叫牌,会导致玩家手中的牌变多而发牌器和
                # 已使用的牌都很少,需避免这种情况
                self.load_cards(self.cards_in_pool) # 将收集来的用过的牌洗好送入发牌器重新使用
            cards.append(self.card_q.get()) # 从发牌器发出一张牌
        self._info("发了{}张牌({})给{}{}".format(n, cards, player.role,player))
        player.receive(cards) #某玩家接受发出的牌
        player.cards_info()
    def _info(self, message):
        if self.display:
            print(message, end = "")

这个方法 (serve_card_to) 接受一个玩家 (player) 和一个整数 (n) 作为参数，表示向该玩家一次发出一定数量的牌，在发牌时如果遇到发牌器里没有牌的情况时会将已使用的牌收集起来洗好后送入发牌器，随后在把需要数量的牌发给某一玩家。代码中的方法 (_info) 负责根据条件在终端输出对局信息.
当一局结束时，Arena 对象还负责把玩家手中的牌回收至已使用的废牌区，这个功能由下面这个方法来完成:

    def recycle_cards(self, *players):
        '''
        回收玩家手中的牌到公开使用过的牌池中
        :param players: 一个玩家或庄家
        :return:None
        '''
        if len(players) == 0:
            return
        for player in players:
            for card in player.cards:
                self.cards_in_pool.append(card)
            player.discharge_cards()#玩家不再留有这些牌

剩下一个最关键的功能就是，如何让庄家和玩家进行一次对局，编写下面的方法来实现这个功能:

    def play_game(self, dealer, player):
        '''
        玩一局21点,生成一个状态序列以及最终奖励(中间奖励为0)
        :param dealer: 庄家
        :param player: 玩家
        :return: tuple: episode, reward
        '''
        self._info("=============开始新一局=============\n")
        self.serve_card_to(player, n=2) #发两张牌给玩家
        self.serve_card_to(dealer, n=2) #发两张牌给庄家
        episode = [] #记录一个对局信息
        if player.policy is None:
            self._info("玩家需要一个策略")
            return
        if dealer.policy is None:
            self._info("庄家需要一个策略")
            return
        while True:
            action = player.policy(dealer)
            # 玩家的策略产生一个行为
            self._info("{}{}选择:{};".format(player.role, player, action))
            episode.append((player.get_state_name(dealer), action)) #记录一个(s,a)
            if action == self.A[0]: #继续叫牌
                self.serve_card_to(player) # 发一张牌给玩家
            else: #停止叫牌
                break
        #玩家停止叫牌后要计算下玩家手中的点数,玩家如果爆了,庄家就不用继续了
        reward, player_points, dealer_points, useable_ace = self.reward_of(dealer, player)
        if player_points > 21:
            self._info("玩家爆点{}输了,得分:{}\n".format(player_points,reward))
            self.recycle_cards(player, dealer)
            self.episodes.append((episode, reward)) #预测的时候需要行为episode list后集中学习V
            # 在蒙特卡洛控制的时候, 可以不需要episodes list,生成一个episode学习一个,下同
            self._info("===============本局结束===============")
            return episode, reward
        #玩家并没有超过21点
        self._info("\n")
        while True:
            action = dealer.policy() #庄家从其策略中获取一个行为
            self._info("{}{}选择:{};".format(dealer.role,dealer,action))
            #状态只记录庄家第一张牌信息,此时玩家不再叫牌,(s,a)不必重复记录
            if action == self.A[0]: #庄家继续叫牌
                self.serve_card_to(dealer)
            else:
                break;
        #双方均停止叫牌了
        self._info("\n双方均停止叫牌")
        reward, player_points, dealer_points,useable_ace = self.reward_of(dealer,player)
        player.cards_info()
        dealer.cards_info()
        if reward == +1:
            self._info("玩家赢了!")
        elif reward == -1:
            self._info("玩家输了!")
        else:
            self._info("双方和局!")
        self._info("玩家{}点,庄家{}点\n".format(player_points, dealer_points))
        self._info("=================本局结束==================")
        self.recycle_cards(player, dealer) #回收玩家和庄家手中的牌至公开牌池
        self.episodes.append((episode, reward)) #将刚才产生的完整对局添加值状态序列列表,蒙特卡洛控制不需要
        return episode, reward

这段代码虽然比较长，但里面包含许多反映对局过程的信息，使得代码也比较容易理解。该方法接受一个庄家一个玩家为参数，产生一次对局，并返回该对局的详细信息。需要指出的是玩家的策略要做到在玩家手中的牌超过 21 点时强制停止叫牌。其次在玩家停止叫牌后，Arena 对局面进行一次判断，如果玩家超过 21 点则本局结束，否则提示庄家选择行为。当庄家停止叫牌后后，Arena 对局面再次进行以此判断，结束对局并将该对局产生的详细信息记录一个 episode对象，并附加地把包含了该局信息的 episode 对象联合该局的最终输赢 (奖励) 登记至 Arena 的成员属性 episodes中.
有了生成一次对局的方法，我们编写下面的代码来一次性生成多个对局:

    def play_games(self, dealer, player, num = 2, show_statistic = True):
        '''
        一次性玩多局游戏
        :param dealer:
        :param player:
        :param num:
        :param show_statistic:
        :return:
        '''
        results = [0, 0, 0] #玩家负, 和, 胜局数
        self.episodes.clear()
        for i in tqdm(range(num)):
            episode, reward = self.play_game(dealer, player)
            results[1 + reward] += 1
            if player.lerning_method is not None:
                player.lerning_method(episode, reward)
            if show_statistic:
                print("共玩了{}局,玩家赢{}局,和{}局,输{}局,胜率:{:.2f},不输率:{:.2f}"\
                      .format(num, results[2], results[1], results[0], results[2]/num,\
                      (results[2] + results[1]) / num))
        return
    def _info(self, message):
        if self.display:
            print(message, end="")

该方法接受一个庄家、一个玩家、需要产生的对局数量、以及是否显示多个对局的统计信息这四个参数，生成指定数量的对局信息，这些信息都保存在 Arena 的 episodes 对象中。为了兼容具备学习能力的玩家，我们设置了在每一个对局结束后，如果玩家能够从中学习，则提供玩家一次学习的机会，在本章中的玩家不具备从对局中学习改善策略的能力，这部分内容将在下一章详细讲解。如果参数设置为显示统计信息，则会在指定数量的对局结束后显示一共对局多少，玩家的胜率等.

生成对局数据

下面的代码将生成一个庄家,一个玩家,一个Arena对象,并进行20万次的对局:

A = ["继续叫牌","停止叫牌"]
display = False
#创建一个玩家一个庄家,玩家使用原始策略,庄家使用其固定的策略
player = Player(A = A, display = display)
dealer = Dealer(A = A, display = display)
#创建一个场景
arena = Arena(A = A, display = display)
#生成num个完整的对局
arena.play_games(dealer,player,num=200000)
# 共玩了200000局,玩家赢58564局,和11369局,输130067局,胜率:0.29,不输率:0.35
# 100%|██████████| 200000/200000 [00:17<00:00, 11140.95it/s]

策略评估

对局生成的数据均保存在对象 arena.episodes 中，接下来的工作就是使用这些数据来对player的策略进行评估，下面的代码完成这部分功能:

def policy_evaluate(episodes, V, Ns):
    '''
    统计每个状态的价值,衰减因子为1,中间状态的即时奖励为0,递增式蒙特卡洛评估
    :param episodes: 状态序列
    :param V:状态价值字典
    :param Ns:状态被访问的次数节点
    :return:
    '''
    for episode, r in episodes:
        for s, a in episode:
            ns = get_dict(Ns, s)
            v = get_dict(V, s)
            set_dict(Ns, ns+1, s)
            set_dict(V, v+(r-v)/(ns+1), s)

V = {} #状态价值字典
Ns = {}#状态被访问的次数节点
policy_evaluate(arena.episodes, V, Ns) #学习V值

其中，V 和 Ns 保存着蒙特卡罗策略评估进程中的价值和统计次数数据，我们使用的是每次访问计数的方法。我们还可以编写如下的方法将价值函数绘制出来:

def draw_value(value_dict, useable_ace = 0, is_q_dict = False, A = None):
    # 定义figure
    fig = plt.figure()
    # 将figure变为3d
    ax = Axes3D(fig)
    # 定义x, y
    x = np.arange(1, 11, 1) # 庄家第一张牌
    y = np.arange(12, 22, 1) # 玩家总分数
    # 生成网格数据
    X, Y = np.meshgrid(x, y)
    # 从V字典检索Z轴的高度
    row, col = X.shape
    Z = np.zeros((row,col))
    if is_q_dict:
        n = len(A)
    for i in range(row):
        for j in range(col):
            state_name = str(X[i,j])+"_"+str(Y[i,j])+"_"+str(useable_ace)
            if not is_q_dict:
                Z[i,j] = get_dict(value_dict, state_name)
            else:
                assert(A is not None)
                for a in A:
                    new_state_name = state_name + "_" + str(a)
                    q = get_dict(value_dict, new_state_name)
                    if q >= Z[i,j]:
                        Z[i,j] = q
    # 绘制3D曲面
    ax.plot_surface(X, Y, Z, rstride = 1, cstride = 1, cmap = plt.cm.cool)
    plt.show()

draw_value(V, useable_ace=True, A = A)#绘制有可用的A时状态价值图
draw_value(V, useable_ace=False, A = A)#绘制无可用的A时状态价值图

结果如下,第一个图是有可用的ace的价值函数图,第二个是没有可用的ace的价值函数图:

我们可以设置各对象display的值为True,来生成少量对局并输出对局详细信息:

display = True
player.display, dealer.display, arena.display = display,display,display
arena.play_games(dealer,player, num=2)
# =============开始新一局=============
# 发了2张牌(['2', 'Q'])给玩家玩家现在的牌:['2', 'Q']
# 发了2张牌(['J', 'A'])给庄家庄家现在的牌:['J', 'A']
# 玩家选择:继续叫牌;发了1张牌(['10'])给玩家玩家现在的牌:['2', 'Q', '10']
# 玩家选择:停止叫牌;玩家爆点22输了,得分:-1
# ===============本局结束===============共玩了2局,玩家赢0局,和0局,输1局,胜率:0.00,不输率:0.00
# =============开始新一局=============
# 发了2张牌(['5', 'Q'])给玩家玩家现在的牌:['5', 'Q']
# 发了2张牌(['5', '5'])给庄家庄家现在的牌:['5', '5']
# 玩家选择:继续叫牌;发了1张牌(['9'])给玩家玩家现在的牌:['5', 'Q', '9']
# 玩家选择:停止叫牌;玩家爆点24输了,得分:-1
# ===============本局结束===============共玩了2局,玩家赢0局,和0局,输2局,胜率:0.00,不输率:0.00

本节编程实践中，我们构建了游戏者基类并扩展形成了庄家类和玩家类来模拟玩家的行为，同时构建了游戏场景类来负责进行对局管理。在此基础上使用蒙特卡罗算法对游戏中玩家的原始策略进行了评估。在策略评估环节，我们并没有把价值函数 (字典)、计数函数 (字典) 以及策略评估方法设计为玩家类的成员对象和成员方法，这只是为了讲解的方便，读者完全可以将它们设计为玩家的成员变量和方法。下一章的编程实践中，我们将继续通过二十一点游戏介绍如何使用蒙特卡罗控制寻找最优策略，本节建立的 Dealer, Player 和 Arena 类将得到复用和扩展.