Q-learning是一個(gè)經(jīng)典的強(qiáng)化學(xué)習(xí)算法，是一種基于價(jià)值(Value-based)的算法，通過(guò)維護(hù)和更新一個(gè)價(jià)值表格(Q表格)進(jìn)行學(xué)習(xí)和預(yù)測(cè)。

【資料圖】

Q-learning是一種off-policy的策略，也就是說(shuō)，它的行動(dòng)策略和Q表格的更新策略是不一樣的。

行動(dòng)時(shí)，Q-learning會(huì)采用epsilon-greedy的方式嘗試多種可能動(dòng)作。

更新時(shí)，Q-learning會(huì)采用潛在收益最大化的動(dòng)作進(jìn)行價(jià)值更新。

總體來(lái)說(shuō)，Q-learning是一個(gè)非常勇敢的策略，在有限動(dòng)作和有限狀態(tài)情況下，它能夠收斂找到全局最優(yōu)策略。

公眾號(hào)算法美食屋后臺(tái)回復(fù)關(guān)鍵詞：torchkeras，獲取本文notebook源代碼~

〇，強(qiáng)化學(xué)習(xí)基本概念 1, 環(huán)境(env)和智能體(agent)

在第n步，agent處于狀態(tài) state(n)中，然后采取行動(dòng)action(n)，env給agent獎(jiǎng)勵(lì)reward(n+1)，同時(shí)agent的狀態(tài)變成 state(n+1)

---reward(n+1),state(n+1)-->envagent(state)<------action(n)----------

以我們玩一個(gè)俄羅斯方塊游戲?yàn)槔?/p>

環(huán)境env就是這個(gè)游戲背后的程序，智能體agent就是玩家。

假設(shè)現(xiàn)在是第n步，state(n)就是目前游戲所處的狀態(tài)，可以表示為一個(gè)矩陣，也就是游戲界面每個(gè)格子的明暗狀態(tài)。

我們可以采取某個(gè) action(n) (向左，向右，向下，變形)。

然后我們會(huì)獲得一個(gè)獎(jiǎng)勵(lì)reward(n)，即得分。獎(jiǎng)勵(lì)很多時(shí)候是稀疏的，即大部分時(shí)候?yàn)?，操作很多步才有一個(gè)不為0的獎(jiǎng)勵(lì)。

同時(shí)游戲界面發(fā)生變化，狀態(tài)由 state(n) 變成 state(n+1)。

2, 馬爾科夫交互鏈

env和agent交互作用若干個(gè)步驟，到達(dá)結(jié)束狀態(tài)，通常叫做一個(gè)episode(片段)。

在俄羅斯方塊游戲的例子中，一局完整的游戲構(gòu)成一個(gè)馬爾科夫交互鏈，叫做一個(gè)episode.

之所以叫做馬爾科夫交互鏈，是因?yàn)檫@個(gè)過(guò)程滿足馬爾科夫假設(shè)。

第n+1步驟的狀態(tài)state(n+1)和獎(jiǎng)勵(lì)reward(n+1)只和第n步驟的狀態(tài)stage(n)和action(n)有關(guān)，而與之前的狀態(tài)和action無(wú)關(guān)。

馬爾科夫假設(shè)要求我們?cè)谠O(shè)計(jì)state和action的時(shí)候，要考慮到所有相關(guān)變量。

并且，只要設(shè)計(jì)出合理的state變量和action變量，任何游戲都可以表示為這樣一個(gè)馬爾科夫交互鏈。

3, 獎(jiǎng)勵(lì)折現(xiàn)公式

為了衡量每個(gè)步驟中action的價(jià)值，需要將該步驟之后的獎(jiǎng)勵(lì)，以及未來(lái)的全部獎(jiǎng)勵(lì)按照類似金融學(xué)中的折現(xiàn)算法求和。

在俄羅斯方塊游戲的例子中，一個(gè)操作action的價(jià)值，不僅跟這個(gè)操作完成后立刻獲得的獎(jiǎng)勵(lì)reward有關(guān)，還要考慮到這個(gè)操作的長(zhǎng)遠(yuǎn)影響。

但這種長(zhǎng)遠(yuǎn)影響不太好精確地計(jì)算，因?yàn)楹竺娅@得的獎(jiǎng)勵(lì)，不僅跟當(dāng)前的action有關(guān)，還跟后面的操作有關(guān)，所以跟當(dāng)前操作的相關(guān)性是小于1的。

作為簡(jiǎn)化起見，我們通過(guò)類似金融學(xué)中現(xiàn)金流折現(xiàn)的方式將未來(lái)的獎(jiǎng)勵(lì)全部折算到當(dāng)前步驟。折算因子gamma一般取值在0.9~1.0之間。

4, epsilon-greedy 學(xué)習(xí)策略

訓(xùn)練時(shí)使用epsilon探索，預(yù)測(cè)時(shí)使用greedy貪心。

訓(xùn)練階段: 以一定epsilon概率選擇隨機(jī)動(dòng)作，以（1-epsilon）選擇最大化Q(s,a)的動(dòng)作。

預(yù)測(cè)階段: 貪心策略，直接選擇最大化Q(s,a)的動(dòng)作。

為了讓模型去探索更優(yōu)策略，我們?cè)谟?xùn)練過(guò)程中會(huì)允許模型以一定的概率去實(shí)施隨機(jī)動(dòng)作，以便評(píng)估不同動(dòng)作的價(jià)值。

這樣也能夠讓模型對(duì)狀態(tài)動(dòng)作空間進(jìn)行更分散的采樣，學(xué)到的結(jié)果也會(huì)更加魯棒。

但在測(cè)試過(guò)程，為了獲得更好的結(jié)果，我們應(yīng)該采用預(yù)期價(jià)值最大的動(dòng)作。

5, Q表格軟更新

獎(jiǎng)勵(lì)折現(xiàn)公式對(duì)每個(gè)action的價(jià)值的計(jì)算方法是一種粗糙的估計(jì)算法。

不同的step或者不同的episode中，按照獎(jiǎng)勵(lì)折現(xiàn)公式對(duì)相同state下相同action價(jià)值的評(píng)估的結(jié)果可能差異很大。

為了保持學(xué)習(xí)過(guò)程的穩(wěn)定性，讓Q值不會(huì)過(guò)分受到某次評(píng)估的影響，我們采用一種軟更新的方式。

也就是我們?cè)诟翾表格的時(shí)候，只讓Q值朝著折現(xiàn)公式計(jì)算結(jié)果靠近一點(diǎn)點(diǎn)(縮小差值)，而不是直接調(diào)整為折現(xiàn)公式的計(jì)算結(jié)果。

這樣，我們最終的Q表格中action的價(jià)值結(jié)果相當(dāng)是許多次不同episode不同step下獎(jiǎng)勵(lì)折現(xiàn)公式計(jì)算結(jié)果的某種平均值。

一，準(zhǔn)備環(huán)境

gym是一個(gè)常用的強(qiáng)化學(xué)習(xí)測(cè)試環(huán)境，可以用make創(chuàng)建環(huán)境。

env具有reset,step,render幾個(gè)方法。

懸崖問題

環(huán)境設(shè)計(jì)如下：

環(huán)境一共有48個(gè)state狀態(tài)。

其中T為目標(biāo)位置，到達(dá)目標(biāo)位置游戲結(jié)束。

10個(gè)用C表示的為懸崖，掉入懸崖會(huì)拉回到起始位置。

智能體設(shè)計(jì)如下：

智能體有4種動(dòng)作action，0表示往上，1往右，2往下，3往左。

reward設(shè)計(jì)如下：

智能體每走一步都會(huì)有-1的reward。

這個(gè)問題希望訓(xùn)練一個(gè)能夠盡可能快的從起始位置到達(dá)目標(biāo)位置T的智能體Agent。

importgymimportnumpyasnpimporttimeimportmatplotlibimportmatplotlib.pyplotaspltfromIPythonimportdisplayprint("gym.__version__=",gym.__version__)%matplotlibinline#可視化函數(shù)：defshow_state(env,step,info=""):plt.figure(num=0,dpi=180)plt.clf()plt.imshow(env.render())plt.title("Step:%d%s"%(step,info))plt.axis("off")display.clear_output(wait=True)display.display(plt.gcf())env=gym.make("CliffWalking-v0",render_mode="rgb_array")#0up,1right,2down,3leftenv.reset()forstepinrange(20):time.sleep(0.2)action=np.random.randint(0,4)obs,reward,done,truncated,info=env.step(action)#env.render()show_state(env,step=step)#print("step{}:action{},obs{},reward{},done{},truncated{},info{}".format(\#step,action,obs,reward,done,truncated,info))display.clear_output(wait=True)

我們先來(lái)看看沒有訓(xùn)練模型，按照隨機(jī)的方式會(huì)怎么走。

二，定義Agent

importtorchfromtorchimportnnclassQAgent(nn.Module):def__init__(self,obs_n,act_n,learning_rate=0.01,gamma=0.9,e_greed=0.1):super().__init__()self.act_n=act_n#動(dòng)作維度，有幾個(gè)動(dòng)作可選self.lr=learning_rate#學(xué)習(xí)率self.gamma=gamma#reward的衰減率self.epsilon=e_greed#按一定概率隨機(jī)選動(dòng)作self.Q=nn.Parameter(torch.zeros((obs_n,act_n)),requires_grad=False)#根據(jù)輸入觀察值，采樣輸出的動(dòng)作值，帶探索defsample(self,obs):ifnp.random.uniform(0,1)<(1.0-self.epsilon):#根據(jù)table的Q值選動(dòng)作action=self.predict(obs)else:action=np.random.choice(self.act_n)#有一定概率隨機(jī)探索選取一個(gè)動(dòng)作returnaction#根據(jù)輸入觀察值，預(yù)測(cè)輸出的動(dòng)作值defforward(self,obs):Q_list=self.Q[obs,:]maxQ=Q_list.max()action_list=torch.where(Q_list==maxQ)[0].tolist()#maxQ可能對(duì)應(yīng)多個(gè)actionaction=np.random.choice(action_list)returnaction@torch.no_grad()defpredict(self,obs):self.eval()returnself.forward(obs)#學(xué)習(xí)方法，也就是更新Q-table的方法deflearn(self,obs,action,reward,next_obs,done):"""on-policyobs:交互前的obs,s_taction:本次交互選擇的action,a_treward:本次動(dòng)作獲得的獎(jiǎng)勵(lì)rnext_obs:本次交互后的obs,s_t+1next_action:根據(jù)當(dāng)前Q表格,針對(duì)next_obs會(huì)選擇的動(dòng)作,a_t+1done:episode是否結(jié)束"""predict_Q=self.Q[obs,action]ifdone:target_Q=reward#沒有下一個(gè)狀態(tài)了else:target_Q=reward+self.gamma*self.Q[next_obs,:].max()#Q-learningself.Q[obs,action]+=self.lr*(target_Q-predict_Q)#修正q

我們創(chuàng)建一下env和agent.

#使用gym創(chuàng)建懸崖環(huán)境env=gym.make("CliffWalking-v0")#0up,1right,2down,3left#創(chuàng)建一個(gè)agent實(shí)例，輸入超參數(shù)agent=QAgent(obs_n=env.observation_space.n,act_n=env.action_space.n,learning_rate=0.1,gamma=0.9,e_greed=0.1)

三，訓(xùn)練Agent

下面我們將套用torchkeras的訓(xùn)練模版來(lái)對(duì)Agent進(jìn)行訓(xùn)練。

由于強(qiáng)化學(xué)習(xí)問題與常用的監(jiān)督學(xué)習(xí)范式有很大的差異，所以我們對(duì)torchkeras的訓(xùn)練模版在

StepRunner, EpochRunner這2個(gè)層級(jí)上都有少量的修改。

classDataLoader:def__init__(self,env,agent,stage="train"):self.env=envself.agent=agentself.stage=stagedef__iter__(self):obs,info=self.env.reset()#重置環(huán)境,重新開一局（即開始新的一個(gè)episode）action=self.agent.sample(obs)#根據(jù)算法選擇一個(gè)動(dòng)作whileTrue:next_obs,reward,done,_,_=self.env.step(action)#與環(huán)境進(jìn)行一個(gè)交互ifself.stage=="train":next_action=self.agent.sample(next_obs)#訓(xùn)練階段使用探索-利用策略else:next_action=self.agent.predict(next_obs)#驗(yàn)證階段使用模型預(yù)測(cè)結(jié)果yieldobs,action,reward,next_obs,doneaction=next_actionobs=next_obsifdone:breakdl_train=DataLoader(env,agent,stage="train")dl_train.size=1000dl_val=DataLoader(env,agent,stage="val")dl_val.size=200

importsys,datetimefromtqdmimporttqdmimportnumpyasnpfromaccelerateimportAcceleratorfromtorchkerasimportKerasModelimportpandasaspdfromtorchkeras.utilsimportis_jupyter,colorfulfromcopyimportdeepcopyclassStepRunner:def__init__(self,net,loss_fn,accelerator=None,stage="train",metrics_dict=None,optimizer=None,lr_scheduler=None):self.net,self.loss_fn,self.metrics_dict,self.stage=net,loss_fn,metrics_dict,stageself.optimizer,self.lr_scheduler=optimizer,lr_schedulerself.accelerator=acceleratorifacceleratorisnotNoneelseAccelerator()def__call__(self,batch):obs,action,reward,next_obs,done=batch#backward()ifself.stage=="train":self.net.learn(obs,action,reward,next_obs,done)#losses（orplainmetric）step_losses={self.stage+"_reward":reward,self.stage+"_done":1.0ifdoneelse0.0}#metrics(statefulmetric)step_metrics={}ifself.stage=="train":step_metrics["lr"]=self.net.lrreturnstep_losses,step_metricsclassEpochRunner:def__init__(self,steprunner,quiet=False):self.steprunner=steprunnerself.stage=steprunner.stageself.accelerator=steprunner.acceleratorself.net=steprunner.netself.quiet=quietdef__call__(self,dataloader):dataloader.agent=self.netn=dataloader.sizeifhasattr(dataloader,"size")elselen(dataloader)loop=tqdm(enumerate(dataloader,start=1),total=n,file=sys.stdout,disable=notself.accelerator.is_local_main_processorself.quiet,ncols=100)epoch_losses={}forstep,batchinloop:step_losses,step_metrics=self.steprunner(batch)step_log=dict(step_losses,**step_metrics)fork,vinstep_losses.items():epoch_losses[k]=epoch_losses.get(k,0.0)+vifstep_log[self.stage+"_done"]<1andstep0.5orstep==n:epoch_metrics=step_metricsepoch_metrics.update({self.stage+"_"+name:metric_fn.compute().item()forname,metric_fninself.steprunner.metrics_dict.items()})epoch_losses={k:vfork,vinepoch_losses.items()}epoch_log=dict(epoch_losses,**epoch_metrics)epoch_log[self.stage+"_step"]=steploop.set_postfix(**epoch_log)forname,metric_fninself.steprunner.metrics_dict.items():metric_fn.reset()loop.close()else:breakreturnepoch_logKerasModel.StepRunner=StepRunnerKerasModel.EpochRunner=EpochRunner

keras_model=KerasModel(net=agent,loss_fn=None)dfhistory=keras_model.fit(train_data=dl_train,val_data=dl_val,epochs=600,ckpt_path="checkpoint.pt",patience=500,monitor="val_reward",mode="max",callbacks=None,quiet=True,plot=True,cpu=True)

dfhistory["val_reward"].max()

-13.0

keras_model.load_ckpt("checkpoint.pt")agent=keras_model.net

四，測(cè)試Agent

deftest_agent(env,agent):total_reward=0obs,info=env.reset()step=0whileTrue:action=agent.predict(obs)#greedynext_obs,reward,done,_,_=env.step(action)total_reward+=rewardobs=next_obstime.sleep(0.5)show_state(env,step)step+=1ifdone:breakplt.close()returntotal_reward

#全部訓(xùn)練結(jié)束，查看算法效果env=gym.make("CliffWalking-v0",render_mode="rgb_array")#0up,1right,2down,3lefttest_reward=test_agent(env,agent)print("testreward=%.1f"%(test_reward))

test reward = -13.0

可以看到，訓(xùn)練完成后，這個(gè)agent非常機(jī)智地在懸崖邊上走了一個(gè)最優(yōu)路線，但卻沒有掉到懸崖里去。

五，保存Agent

torch.save(keras_model.net.state_dict(),"best_ckpt.pt")

公眾號(hào)算法美食屋后臺(tái)回復(fù)關(guān)鍵詞：torchkeras，獲取本文notebook源代碼以及更多有趣范例。

關(guān)鍵詞：