每個(gè)游戲玩家都有一個(gè)夢(mèng)，希望自己在虛擬世界中成為萬(wàn)眾矚目、無(wú)所不能的英雄。然后…然后…鬧鐘響了夢(mèng)醒了，又到了擠地鐵上班的時(shí)間。
不過(guò)，在這個(gè)項(xiàng)目中，我將帶大家暫時(shí)忘卻現(xiàn)實(shí)的煩惱，用飛槳深度強(qiáng)化學(xué)習(xí)框架PARL來(lái)實(shí)現(xiàn)這個(gè)“英雄夢(mèng)”！先放效果圖：
大家是不是迫不及待了呢？且慢，要實(shí)現(xiàn)《明日方舟》游戲的深度強(qiáng)化學(xué)習(xí)，還是先讓我?guī)Т蠹一仡櫼幌律疃葟?qiáng)化學(xué)習(xí)算法歷史。DQN是深度強(qiáng)化學(xué)習(xí)算法開(kāi)山之作，在經(jīng)典街機(jī)游戲上取得了非常好的效果。它使用了ReplyMemory來(lái)存儲(chǔ)和回放經(jīng)驗(yàn)，這是Off-policy類(lèi)型算法的常用技巧。但是，DQN在應(yīng)對(duì)手機(jī)游戲時(shí)，能力就不夠看了。于是我把目光投向了更為強(qiáng)大的算法--- A3C。
A3C算法與DQN不同，它設(shè)計(jì)了異步多線程的Actor-Critic，每個(gè)Agent在自己的線程中運(yùn)行，然后全局共享學(xué)習(xí)到的網(wǎng)絡(luò)參數(shù)。這樣，每時(shí)每刻都能有大量的交互數(shù)據(jù)，并且這些多線程采集到的數(shù)據(jù)沒(méi)有關(guān)聯(lián)性（關(guān)聯(lián)性問(wèn)題：請(qǐng)參考DDQN算法原理）。因此，A3C算法通過(guò)“異步多線程+共享全局參數(shù)”達(dá)到了和ReplyMemory類(lèi)似的效果。而且，它既有大量數(shù)據(jù)可以解決訓(xùn)練過(guò)程不穩(wěn)定問(wèn)題，同時(shí)又解決了參數(shù)關(guān)聯(lián)性的問(wèn)題。
在經(jīng)典算法PG中，我們的Agent又被稱(chēng)為Actor，Actor對(duì)于一個(gè)特定的任務(wù)，都有自己的一個(gè)策略π。策略π通常用一個(gè)神經(jīng)網(wǎng)絡(luò)表示，其參數(shù)為θ。從一個(gè)特定的狀態(tài)State出發(fā)，一直到任務(wù)的結(jié)束，被稱(chēng)為一個(gè)完整的Episode。在每一步，我們都能獲得一個(gè)獎(jiǎng)勵(lì)r，一個(gè)完整的任務(wù)所獲得的最終獎(jiǎng)勵(lì)被稱(chēng)為R。
如果我們用Q函數(shù)來(lái)預(yù)估未來(lái)的累積獎(jiǎng)勵(lì)，同時(shí)創(chuàng)建一個(gè)Critic網(wǎng)絡(luò)來(lái)計(jì)算Q函數(shù)值，那么我們就得到了Actor-Critic方法。
Q函數(shù)在A3C里的主要作用是增加一個(gè)基線，使得反饋有正有負(fù)，這里的基線通常用狀態(tài)價(jià)值函數(shù)V來(lái)表示。但是，當(dāng)我們應(yīng)用這樣的方法，則需要同時(shí)計(jì)算Q函數(shù)和V函數(shù)，這并不容易。Q函數(shù)可以用“Step t+1的V函數(shù)”加上“從Step t到Step t+1的r”來(lái)代替。這樣，我們就可以得到用V來(lái)表示的Q值計(jì)算，我們一般稱(chēng)為Advantage（優(yōu)勢(shì)函數(shù)），此時(shí)的Critic網(wǎng)絡(luò)變?yōu)橛?jì)算優(yōu)勢(shì)函數(shù)A的網(wǎng)絡(luò)。
A3C是Asynchronous Advantage Actor-Critic的縮寫(xiě)，中文翻譯為異步的優(yōu)勢(shì)動(dòng)作評(píng)價(jià)算法。其中，Advantage就是指優(yōu)勢(shì)函數(shù)A。因此，從名字這里我們可以解讀出來(lái)A3C實(shí)質(zhì)就是求解πθ網(wǎng)絡(luò)和Aπ(s, a)網(wǎng)絡(luò)。
在A3C算法論文中，論文作者對(duì)比了四種算法——異步Sarsa、異步Q-Learning、DQN和A3C。論文發(fā)表后，各路算法大神驗(yàn)證一個(gè)問(wèn)題——是異步更新讓算法表現(xiàn)優(yōu)于其他算法？。結(jié)果非常有趣：多線程是A3C算法快的原因，但是”異步更新“反而是它的缺點(diǎn)。于是，科學(xué)家提出同步更新算法A2C（Advantage Actor-Critic），讓它可以更有效利用CPU資源。
PS：算法大神照樣被打臉，啪啪啪！
在下面部分，我會(huì)先對(duì)PARL庫(kù)內(nèi)置的A2C算法進(jìn)行簡(jiǎn)單解讀，這樣大家在看項(xiàng)目實(shí)踐部分時(shí)，就能少閱讀一些代碼。
這個(gè)類(lèi)有意思的地方是，PARL庫(kù)用了A3C的名字。原因是A2C和A3C是同源算法。它們實(shí)現(xiàn)上的主要區(qū)別是step函數(shù)（后面會(huì)講到）。
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)
這段代碼有意思的地方是，它把自己連接到了XPARL集群，然后去執(zhí)行run_remote_sample。閱讀過(guò)DQN源碼的同學(xué)應(yīng)該很好理解，它的意思就是在獨(dú)立進(jìn)程運(yùn)行“取樣”。
def create_actors(self):
# 先把自己連接到XPARL集群上去
parl.connect(self.config['master_address'])
for i in six.moves.range(self.config['actor_num']):
...
remote_thread = threading.Thread(
# 在工作線程中運(yùn)行run_remote_sample函數(shù)
# 通過(guò)params_queue傳遞模型的參數(shù)
target=self.run_remote_sample, args=(params_queue, ))
remote_thread.setDaemon(True)
remote_thread.start()
...
step函數(shù)是A2C算法中最重要、獨(dú)特的函數(shù)，作用是同步等待更新操作。因?yàn)锳2C算法會(huì)同步等待所有Agent（Actor)完成一輪訓(xùn)練后，把π網(wǎng)絡(luò)的參數(shù)θ同步上來(lái)，更新全局的π網(wǎng)絡(luò)參數(shù)。
注解@parl.remote_class表明Actor類(lèi)是在獨(dú)立的本機(jī)進(jìn)程中執(zhí)行（因?yàn)锳2C是利用本機(jī)多CPU）。通過(guò)兩行命令部署了PARL分布式集群，Actor實(shí)際是在遠(yuǎn)程server中運(yùn)行了。
注意，Actor的init方法中保存了env數(shù)組，用同樣的參數(shù)實(shí)例化了模型，用同樣的模型實(shí)例化了算法并作為參數(shù)傳入到了Agent中。
@parl.remote_class
class Actor(object):
def __init__(self, config):
...
# Actor保存了env數(shù)組
self.envs = []
for _ in range(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)
大家還要關(guān)注的點(diǎn)是，每個(gè)Actor對(duì)應(yīng)一個(gè)Agent。
Actor中的sample函數(shù)會(huì)調(diào)用Agent的sample函數(shù)和Agent的value函數(shù)來(lái)分別更新本地的π網(wǎng)絡(luò)和v網(wǎng)絡(luò)，最終返回sample_data給中心節(jié)點(diǎn)。
...
actions_batch, values_batch = self.agent.sample(np.stack(self.obs_batch))
...
next_value = self.agent.value(next_obs)
...
sample_data的數(shù)據(jù)結(jié)構(gòu)：
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)
其中，優(yōu)勢(shì)函數(shù)的的計(jì)算如下：
# gae：generalized advantage estimator
advantages = calc_gae(rewards, values, next_value,
self.config['gamma'],
self.config['lambda'])
target_values = advantages + values
這個(gè)類(lèi)是PARL對(duì)env環(huán)境的封裝。我們的模擬真機(jī)環(huán)境，也采用了同樣的定義，主要是為了同時(shí)跑多個(gè)環(huán)境，增加并行計(jì)算的效率，如下所示：
class VectorEnv(object):
def __init__(self, envs):
def reset(self):
...
def step(self, actions):
# env需要實(shí)現(xiàn)step方法
obs, reward, done, info = self.envs[env_id].step(actions[env_id])
...
if done:
# env需要實(shí)現(xiàn)reset方法
obs = self.envs[env_id].reset()
...
return obs_batch, reward_batch, done_batch, info_batch
模擬器的源數(shù)據(jù)是由此類(lèi)中的step方法批量返回。
因?yàn)椤睹魅辗街邸肥鞘謾C(jī)網(wǎng)絡(luò)游戲，數(shù)據(jù)生產(chǎn)速度實(shí)在太慢了！?。榱颂岣哂?xùn)練速度，需要自己開(kāi)發(fā)模擬器。用模擬器后速度可提升50-100倍。
修改Learner的初始化方法：
#=========== Create Agent ==========
game = ArKnights()
env = PMGE(game)
obs_shape = (3, 108, 192)
act_dim = 650
定義新的env.py：
class PMGE(object):
def __init__(self, game):
self.game = game
def step(self, action):
# 模擬器簡(jiǎn)化了狀態(tài)判斷
# 實(shí)際項(xiàng)目應(yīng)該實(shí)時(shí)生成：當(dāng)前屏幕--> stateCode 的關(guān)系
s1 = [ self.game.stateCode ]
# 產(chǎn)生狀態(tài)變化
self.game.act(action, s1)
reward = self.game.getScore(s1)
isOver = self.game.gameOver()
next_obs = self.game.render()
# 為了匹配標(biāo)準(zhǔn)的API
return next_obs, reward, isOver, 0
def reset(self):
return self.game.reset()
修改Actor：
class Actor(object):
def __init__(self, config):
self.config = config
self.envs = []
for _ in range(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)
定義訓(xùn)練用的模擬環(huán)境：
class ArKnights(object):
def __init__(self):
"""
游戲《明日方舟》智能體定義
"""
self.stateCode = 990
# 1920x1080 ----- 1920/80 x 1080/40 = 24x27
self.tap_dim = 24*27
self.swipe_dim = 4 # 上下左右
def render(self):
imgDir = IMAGE_DIR + str(self.stateCode) + '/'
filenames = os.listdir(imgDir)
# 在stateCode目錄下隨機(jī)取一張圖片
filename = random.choice(filenames)
return self.transform_img(imgDir + filename)
def act(self, action, stateCode):
if stateCode[0] == 990:
if action in [442,443,444,445,466,467,468,469]:
self.stateCode = 970
if stateCode[0] == 970:
if action 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
def getScore(self, s1):
# 狀態(tài)沒(méi)變扣一分
if s1[0] == self.stateCode:
return -1
return 1
def gameOver(self):
code = self.stateCode
# if (code == 910 or code == 1010):
# for debug 讓算法快速收斂
if (code == 965):
return True
return False
def reset(self):
self.stateCode = 990
imgDir = IMAGE_DIR + str(self.stateCode) + '/'
filenames = os.listdir(imgDir)
# 在990目錄下隨機(jī)取一張圖片
filename = random.choice(filenames)
return self.transform_img(imgDir + filename)
def transform_img(self, filepath):
# 直接讀取 (h,w)
img = cv2.imread(filepath, cv2.IMREAD_COLOR)
# 將圖片尺寸縮放道 (image, (w,h)) 192x108
img = cv2.resize(img, (192, 108))
# 因?yàn)閏v2的數(shù)組長(zhǎng)寬是反的，所以用numpy轉(zhuǎn)置一下 (C,H,W)
img = np.transpose(img, (2, 0, 1))
obs = img.astype('float32')
return obs
在模擬器中經(jīng)過(guò)大約10萬(wàn)個(gè)steps，模型的loss就收斂了。
新建項(xiàng)目ARKNIGHT_CLASSIFY，使用殘差神經(jīng)網(wǎng)絡(luò)對(duì)《明日方舟》中的主要游戲界面做了預(yù)定義。利用這個(gè)引擎，在真機(jī)部署的時(shí)候可以推斷出當(dāng)前游戲的state，用于計(jì)算reward和game over這兩個(gè)重要參數(shù)。
3.評(píng)估強(qiáng)化學(xué)習(xí)模型
在深度強(qiáng)化學(xué)習(xí)中，效果評(píng)估非常重要，因?yàn)槲覀円浪惴◤臄?shù)據(jù)中學(xué)到了什么？
我們?cè)诘谝徊街械玫搅四Ｐ?，在第二步中得到了真機(jī)環(huán)境下的reward和game over函數(shù)。
那么我們就要在真機(jī)環(huán)境中去測(cè)試。
def test():
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")
# 初始狀態(tài)
obs = env.reset()
MAX_STEP = 20
step = 0
while True:
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))
if isOver or step >MAX_STEP:
logger.info("GameOver, state:{}".format(next_state_code))
break;
可以看到，我只用了2步，算法就成功達(dá)到了設(shè)定的終止?fàn)顟B(tài)[965]。新建部署項(xiàng)目ArKnight_A2C，把模型導(dǎo)入，效果如下：
4.模型和狀態(tài)推理引擎部署到真機(jī)
定義真機(jī)環(huán)境：
import time
import cv2
from PIL import Image
import numpy as np
from adbutil import AdbUtil
from resnet import ResNet
import paddle
import paddle.fluid as fluid
class ArKnights(object):
def __init__(self):
self.adbutil = AdbUtil()
# 加載推理模型
with fluid.dygraph.guard():
# 加載狀態(tài)推斷引擎
self.model = ResNet('resnet', 50)
#加載模型參數(shù)
model_state_dict, _ = fluid.load_dygraph("arknights")
self.model.load_dict(model_state_dict)
self.model.eval()
def _restart(self):
"""
打開(kāi)游戲進(jìn)程
如果已經(jīng)打開(kāi)，先關(guān)閉再重新打開(kāi)
"""
self.adbutil.stopArKnights()
self.adbutil.startArKnights()
# 每隔1秒在屏幕中心點(diǎn)擊1下，持續(xù)20秒
self.adbutil.taptap(960,540,20,1)
def _stop(self):
"""
關(guān)閉游戲進(jìn)程
"""
self.adbutil.stopArKnights()
def act(self, action):
# 點(diǎn)擊動(dòng)作code映射成動(dòng)作
if action < 648:
x = (action % 24) * 80 + 40 # 取余
y = (action // 24) * 40 + 20 # 取商
self.adbutil.taptap(x,y,1,0.01) # x,y,count,frequency
elif action == 648:
self.adbutil.rightswipeswipe(2,0.5)
elif action == 649:
self.adbutil.leftswipeswipe(2,0.5)
else:
raise("No such action error!" + str(action))
time.sleep(2) # 等動(dòng)作執(zhí)行完
def render(self):
# TODO check shape
img = self.adbutil.screencap()
img = img.resize((192, 108), Image.ANTIALIAS)
# 因?yàn)閳D片的數(shù)組長(zhǎng)寬是反的，所以用numpy轉(zhuǎn)置一下 (C,H,W)
img = np.transpose(img, (2, 0, 1))
obs = img.astype('float32')
return obs
def reset(self):
self._restart()
return self.render()
def gameOver(self):
state = self.inferState()
print("state"+str(state))
if state[0] == 965:
return True
else:
return False
def inferState(self):
"""
圖片推斷
"""
...
這里的游戲狀態(tài)推斷引擎，就是ARKNIGHT_CLASSIFY項(xiàng)目輸出的推理模型。有了狀態(tài)的推理值，代碼中的reward和game over就可以和真機(jī)環(huán)境匹配上。同時(shí)，用AdbUtil類(lèi)來(lái)執(zhí)行真實(shí)動(dòng)作，就可以操作真機(jī)執(zhí)行算法動(dòng)作。最終真機(jī)運(yùn)行效果如下(手機(jī)屏幕的變化請(qǐng)看視頻)：
在這個(gè)文章中，我給大家展示了如何構(gòu)建明日方舟的交互環(huán)境，以及如何通過(guò)PARL快速調(diào)用A3C算法實(shí)現(xiàn)并行訓(xùn)練，整體實(shí)現(xiàn)起來(lái)簡(jiǎn)單易懂。
看到這兒，大家是不是迫不及待地想要自己動(dòng)手嘗試！
“英雄們”，快用飛槳去實(shí)現(xiàn)你們的美夢(mèng)吧，永遠(yuǎn)的神（永遠(yuǎn)滴神）！
欲知詳情，請(qǐng)戳PARL開(kāi)源鏈接：
https://github.com/PaddlePaddle/PARL
