用飞桨深度强化学习框架PARL来深度强化学习《明日方舟》

　　每个游戏玩家都有一个梦，希望自己在虚拟世界中成为万众瞩目、无所不能的英雄。然后…然后…闹钟响了梦醒了，又到了挤地铁上班的时间。

　　不过，在这个项目中，我将带大家暂时忘却现实的烦恼，用飞桨深度强化学习框架PARL来实现这个“英雄梦”!先放效果图：

　　知识回顾

　　大家是不是迫不及待了呢？且慢，要实现《明日方舟》游戏的深度强化学习，还是先让我带大家回顾一下深度强化学习算法历史。DQN是深度强化学习算法开山之作，在经典街机游戏上取得了非常好的效果。它使用了ReplyMemory来存储和回放经验，这是Off-policy类型算法的常用技巧。但是，DQN在应对手机游戏时，能力就不够看了。于是我把目光投向了更为强大的算法---?A3C。

　　A3C算法与DQN不同，它设计了异步多线程的Actor-Critic，每个Agent在自己的线程中运行，然后全局共享学习到的网络参数。这样，每时每刻都能有大量的交互数据，并且这些多线程采集到的数据没有关联性(关联性问题：请参考DDQN算法原理)。因此，A3C算法通过“异步多线程+共享全局参数”达到了和ReplyMemory类似的效果。而且，它既有大量数据可以解决训练过程不稳定问题，同时又解决了参数关联性的问题。

　　在经典算法PG中，我们的Agent又被称为Actor，Actor对于一个特定的任务，都有自己的一个策略π。策略π通常用一个神经网络表示，其参数为θ。从一个特定的状态State出发，一直到任务的结束，被称为一个完整的Episode。在每一步，我们都能获得一个奖励r，一个完整的任务所获得的最终奖励被称为R。

　　如果我们用Q函数来预估未来的累积奖励，同时创建一个Critic网络来计算Q函数值，那么我们就得到了Actor-Critic方法。

　　Q函数在A3C里的主要作用是增加一个基线，使得反馈有正有负，这里的基线通常用状态价值函数V来表示。但是，当我们应用这样的方法，则需要同时计算Q函数和V函数，这并不容易。Q函数可以用“Step?t+1的V函数”加上“从Step?t到Step?t+1的r”来代替。这样，我们就可以得到用V来表示的Q值计算，我们一般称为Advantage(优势函数)，此时的Critic网络变为计算优势函数A的网络。

　　A3C是Asynchronous?Advantage?Actor-Critic的缩写，中文翻译为异步的优势动作评价算法。其中，Advantage就是指优势函数A。因此，从名字这里我们可以解读出来A3C实质就是求解πθ网络和Aπ(s,?a)网络。

　　在A3C算法论文中，论文作者对比了四种算法——异步Sarsa、异步Q-Learning、DQN和A3C。论文发表后，各路算法大神验证一个问题——是异步更新让算法表现优于其他算法？。结果非常有趣：多线程是A3C算法快的原因，但是”异步更新“反而是它的缺点。于是，科学家提出同步更新算法A2C(Advantage?Actor-Critic)，让它可以更有效利用CPU资源。

　　PS：算法大神照样被打脸，啪啪啪!

　　在下面部分，我会先对PARL库内置的A2C算法进行简单解读，这样大家在看项目实践部分时，就能少阅读一些代码。

　　Learner

　　这个类有意思的地方是，PARL库用了A3C的名字。原因是A2C和A3C是同源算法。它们实现上的主要区别是step函数(后面会讲到)。

　　env?=?gym.make(config[?'env_name'])

　　env?=?wrap_deepmind(env,?dim=config[?'env_dim'],?obs_format=?'NCHW')

　　obs_shape?=?env.observation_space.shape

　　act_dim?=?env.action_space.n

　　self.config[?'obs_shape']?=?obs_shape

　　self.config[?'act_dim']?=?act_dim

　　model?=?AtariModel(act_dim)

　　algorithm?=?parl.algorithms.A3C(

　　model,?vf_loss_coeff=config[?'vf_loss_coeff'])

　　self.agent?=?AtariAgent(algorithm,?config)

　　create_actors

　　这段代码有意思的地方是，它把自己连接到了XPARL集群，然后去执行run_remote_sample。阅读过DQN源码的同学应该很好理解，它的意思就是在独立进程运行“取样”。

　　defcreate_actors(self):

　　#?先把自己连接到XPARL集群上去

　　parl.connect(self.config[?'master_address'])

　　fori?insix.moves.range(self.config[?'actor_num']):

　　...

　　remote_thread?=?threading.Thread(

　　#?在工作线程中运行run_remote_sample函数

　　#?通过params_queue传递模型的参数

　　target=self.run_remote_sample,?args=(params_queue,?))

　　remote_thread.setDaemon(?True)

　　remote_thread.start

　　...

　　step函数

　　step函数是A2C算法中最重要、独特的函数，作用是同步等待更新操作。因为A2C算法会同步等待所有Agent(Actor)完成一轮训练后，把π网络的参数θ同步上来，更新全局的π网络参数。

　　Actor函数

　　注解@parl.remote_class表明Actor类是在独立的本机进程中执行(因为A2C是利用本机多CPU)。通过两行命令部署了PARL分布式集群，Actor实际是在远程server中运行了。

　　注意，Actor的init方法中保存了env数组，用同样的参数实例化了模型，用同样的模型实例化了算法并作为参数传入到了Agent中。

　　@parl.remote_class

　　classActor(object):

　　def__init__(self,?config):

　　...

　　#?Actor保存了env数组

　　self.envs?=?[]

　　for_?inrange(config[?'env_num']):

　　env?=?gym.make(config[?'env_name'])

　　env?=?wrap_deepmind(env,?dim=config[?'env_dim'],?obs_format=?'NCHW')

　　self.envs.append(env)

　　...

　　model?=?AtariModel(act_dim)

　　algorithm?=?parl.algorithms.A3C(

　　model,?vf_loss_coeff=config[?'vf_loss_coeff'])

　　self.agent?=?AtariAgent(algorithm,?config)

　　大家还要关注的点是，每个Actor对应一个Agent。

　　sample函数

　　Actor中的sample函数会调用Agent的sample函数和Agent的value函数来分别更新本地的π网络和v网络，最终返回sample_data给中心节点。

　　...

　　actions_batch,?values_batch?=?self.agent.sample(np.stack(self.obs_batch))

　　...

　　next_value?=?self.agent.value(next_obs)

　　...

　　sample_data的数据结构：

　　sample_data[?'obs'].extend(env_sample_data[env_id][?'obs'])

　　sample_data[?'actions'].extend(env_sample_data[env_id][?'actions'])

　　sample_data[?'advantages'].extend(advantages)

　　sample_data[?'target_values'].extend(target_values)

　　其中，优势函数的的计算如下：

　　#?gae：generalized?advantage?estimator

　　advantages?=?calc_gae(rewards,?values,?next_value,

　　self.config[?'gamma'],

　　self.config[?'lambda'])

　　target_values?=?advantages?+?values

　　VectorEnv函数

　　这个类是PARL对env环境的封装。我们的模拟真机环境，也采用了同样的定义，主要是为了同时跑多个环境，增加并行计算的效率，如下所示：

　　classVectorEnv(object):

　　def__init__(self,?envs):

　　defreset(self):

　　...

　　defstep(self,?actions):

　　#?env需要实现step方法

　　obs,?reward,?done,?info?=?self.envs[env_id].step(actions[env_id])

　　...

　　ifdone:

　　#?env需要实现reset方法

　　obs?=?self.envs[env_id].reset

　　...

　　returnobs_batch,?reward_batch,?done_batch,?info_batch

　　模拟器的源数据是由此类中的step方法批量返回。

　　实战编程

　　1.游戏模拟器编写&训练

　　新建《明日方舟》模拟器项目：

　　ArKnight_A2C_Simulator

　　因为《明日方舟》是手机网络游戏，数据生产速度实在太慢了!!!为了提高训练速度，需要自己开发模拟器。用模拟器后速度可提升50-100倍。

　　修改Learner的初始化方法：

　　#===========?Create?Agent?==========

　　game?=?ArKnights

　　env?=?PMGE(game)

　　obs_shape?=?(?3,?108,?192)

　　act_dim?=?650

　　定义新的env.py：

　　classPMGE(object):

　　def__init__(self,?game):

　　self.game?=?game

　　defstep(self,?action):

　　#?模拟器简化了状态判断

　　#?实际项目应该实时生成：当前屏幕-->?stateCode?的关系

　　s1?=?[?self.game.stateCode?]

　　#?产生状态变化

　　self.game.act(action,?s1)

　　reward?=?self.game.getScore(s1)

　　isOver?=?self.game.gameOver

　　next_obs?=?self.game.render

　　#?为了匹配标准的API

　　returnnext_obs,?reward,?isOver,?0

　　defreset(self):

　　returnself.game.reset

　　修改Actor：

　　classActor(object):

　　def__init__(self,?config):

　　self.config?=?config

　　self.envs?=?[]

　　for_?inrange(config[?'env_num']):

　　game?=?ArKnights

　　env?=?PMGE(game)

　　self.envs.append(env)

　　self.vector_env?=?VectorEnv(self.envs)

　　self.obs_batch?=?self.vector_env.reset

　　model?=?Model(config[?'act_dim'])

　　algorithm?=?parl.algorithms.A3C(

　　model,?vf_loss_coeff=config[?'vf_loss_coeff'])

　　self.agent?=?Agent(algorithm,?config)

　　定义训练用的模拟环境：

　　classArKnights(object):

　　def__init__(self):

　　"""

　　游戏《明日方舟》智能体定义

　　"""

　　self.stateCode?=?990

　　#?1920x1080?-----?1920/80?x?1080/40?=?24x27

　　self.tap_dim?=?24*?27

　　self.swipe_dim?=?4#?上下左右

　　defrender(self):

imgDir?=?IMAGE_DIR?+?str(self.stateCode)?+?'/'

filenames?=?os.listdir(imgDir)

　　#?在stateCode目录下随机取一张图片

　　filename?=?random.choice(filenames)

returnself.transform_img(imgDir?+?filename)

　　defact(self,?action,?stateCode):

　　ifstateCode[?0]?==?990:

　　ifaction?in[?442,?443,?444,?445,?466,?467,?468,?469]:

　　self.stateCode?=?970

　　ifstateCode[?0]?==?970:

　　ifaction?in[?111,?112,?113,?114,?115,

　　135,?136,?137,?138,?139,

　　159,?160,?161,?162,?163,

　　183,?184,?185,?186,?187,

　　207,?208,?209,?210,?211]:

　　self.stateCode?=?965

　　defgetScore(self,?s1):

　　#?状态没变扣一分

　　ifs1[?0]?==?self.stateCode:

　　return-1

　　return1

　　defgameOver(self):

　　code?=?self.stateCode

　　#?if?(code?==?910?or?code?==?1010):

　　#?for?debug?让算法快速收敛

　　if(code?==?965):

　　returnTrue

　　returnFalse

　　defreset(self):

　　self.stateCode?=?990

imgDir?=?IMAGE_DIR?+?str(self.stateCode)?+?'/'

filenames?=?os.listdir(imgDir)

　　#?在990目录下随机取一张图片

　　filename?=?random.choice(filenames)

returnself.transform_img(imgDir?+?filename)

deftransform_img(self,?filepath):

　　#?直接读取?(h,w)

img?=?cv2.imread(filepath,?cv2.IMREAD_COLOR)

　　#?将图片尺寸缩放道?(image,?(w,h))?192x108

img?=?cv2.resize(img,?(?192,?108))

　　#?因为cv2的数组长宽是反的，所以用numpy转置一下?(C,H,W)

img?=?np.transpose(img,?(?2,?0,?1))

obs?=?img.astype(?'float32')

　　returnobs

　　在模拟器中经过大约10万个steps，模型的loss就收敛了。

　　2.编写状态推理引擎

　　新建项目ARKNIGHT_CLASSIFY，使用残差神经网络对《明日方舟》中的主要游戏界面做了预定义。利用这个引擎，在真机部署的时候可以推断出当前游戏的state，用于计算reward和game?over这两个重要参数。

　　3.评估强化学习模型

　　在深度强化学习中，效果评估非常重要，因为我们要知道算法从数据中学到了什么？

　　我们在第一步中得到了模型，在第二步中得到了真机环境下的reward和game?over函数。

　　那么我们就要在真机环境中去测试。

　　deftest:

　　game?=?ArKnights

　　env?=?PMGE(game)

　　obs_shape?=?(?3,?108,?192)

　　act_dim?=?650

　　config[?'obs_shape']?=?obs_shape

　　config[?'act_dim']?=?act_dim

　　model?=?Model(act_dim)

　　algorithm?=?parl.algorithms.A3C(model,?vf_loss_coeff=config[?'vf_loss_coeff'])

　　agent?=?Agent(algorithm,?config)

　　agent.restore(?"./model_dir")

　　#?初始状态

　　obs?=?env.reset

　　MAX_STEP?=?20

　　step?=?0

　　whileTrue:

　　state_code?=?env.game.stateCode

　　action?=?agent.predict(obs)

　　obs,?reward,?isOver,?_?=?env.step(action)

　　next_state_code?=?env.game.stateCode

　　step?+=?1

　　logger.info(?"evaluate?state_code:{},?action:{}?next_state_code:{},?reward:{},?isOver:{}".format(state_code,?action,?next_state_code,?reward,?isOver))

　　ifisOver?orstep?>MAX_STEP:

　　logger.info(?"GameOver,?state:{}".format(next_state_code))

　　break;

　　可以看到，我只用了2步，算法就成功达到了设定的终止状态[965]。新建部署项目ArKnight_A2C，把模型导入，效果如下：

　　4.模型和状态推理引擎部署到真机

　　定义真机环境：

　　importtime

　　importcv2

　　fromPIL?importImage

　　importnumpy?asnp

　　fromadbutil?importAdbUtil

　　fromresnet?importResNet

　　importpaddle

　　importpaddle.fluid?asfluid

　　classArKnights(object):

　　def__init__(self):

　　self.adbutil?=?AdbUtil

　　#?加载推理模型

　　withfluid.dygraph.guard:

　　#?加载状态推断引擎

　　self.model?=?ResNet(?'resnet',?50)

　　#加载模型参数

　　model_state_dict,?_?=?fluid.load_dygraph(?"arknights")

　　self.model.load_dict(model_state_dict)

　　self.model.eval

　　def_restart(self):

　　"""

　　打开游戏进程

　　如果已经打开，先关闭再重新打开

　　"""

　　self.adbutil.stopArKnights

　　self.adbutil.startArKnights

　　#?每隔1秒在屏幕中心点击1下，持续20秒

　　self.adbutil.taptap(?960,?540,?20,?1)

　　def_stop(self):

　　"""

　　关闭游戏进程

　　"""

　　self.adbutil.stopArKnights

　　defact(self,?action):

　　#?点击动作code映射成动作

　　ifaction?<?648:

　　x?=?(action?%?24)?*?80+?40#?取余

　　y?=?(action?//?24)?*?40+?20#?取商

　　self.adbutil.taptap(x,y,?1,?0.01)?#?x,y,count,frequency

　　elifaction?==?648:

　　self.adbutil.rightswipeswipe(?2,?0.5)

　　elifaction?==?649:

　　self.adbutil.leftswipeswipe(?2,?0.5)

　　else:

　　raise(?"No?such?action?error!"+?str(action))

　　time.sleep(?2)?#?等动作执行完

　　defrender(self):

　　#?TODO?check?shape

img?=?self.adbutil.screencap

img?=?img.resize((?192,?108),?Image.ANTIALIAS)

　　#?因为图片的数组长宽是反的，所以用numpy转置一下?(C,H,W)

img?=?np.transpose(img,?(?2,?0,?1))

obs?=?img.astype(?'float32')

　　returnobs

　　defreset(self):

　　self._restart

　　returnself.render

　　defgameOver(self):

　　state?=?self.inferState

　　print(?"state"+str(state))

　　ifstate[?0]?==?965:

　　returnTrue

　　else:

　　returnFalse

　　definferState(self):

　　"""

　　图片推断

　　"""

　　...

　　这里的游戏状态推断引擎，就是ARKNIGHT_CLASSIFY项目输出的推理模型。有了状态的推理值，代码中的reward和game?over就可以和真机环境匹配上。同时，用AdbUtil类来执行真实动作，就可以操作真机执行算法动作。最终真机运行效果如下(手机屏幕的变化请看视频)：

　　在这个文章中，我给大家展示了如何构建明日方舟的交互环境，以及如何通过PARL快速调用A3C算法实现并行训练，整体实现起来简单易懂。

　　看到这儿，大家是不是迫不及待地想要自己动手尝试!

　　“英雄们”，快用飞桨去实现你们的美梦吧，yyds(永远滴神)!