【環(huán)球時快訊】Q-learning解決懸崖問題

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

【資料圖】

Q-learning是一種off-policy的策略，也就是說，它的行動策略和Q表格的更新策略是不一樣的。

行動時，Q-learning會采用epsilon-greedy的方式嘗試多種可能動作。

更新時，Q-learning會采用潛在收益最大化的動作進(jìn)行價值更新。

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

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

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

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

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

以我們玩一個俄羅斯方塊游戲為例。

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

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

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

然后我們會獲得一個獎勵reward(n)，即得分。獎勵很多時候是稀疏的，即大部分時候為0，操作很多步才有一個不為0的獎勵。

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

2, 馬爾科夫交互鏈

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

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

之所以叫做馬爾科夫交互鏈，是因為這個過程滿足馬爾科夫假設(shè)。

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

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

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

3, 獎勵折現(xiàn)公式

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

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

但這種長遠(yuǎn)影響不太好精確地計算，因為后面獲得的獎勵，不僅跟當(dāng)前的action有關(guān)，還跟后面的操作有關(guān)，所以跟當(dāng)前操作的相關(guān)性是小于1的。

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

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

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

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

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

為了讓模型去探索更優(yōu)策略，我們在訓(xùn)練過程中會允許模型以一定的概率去實施隨機(jī)動作，以便評估不同動作的價值。

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

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

5, Q表格軟更新

獎勵折現(xiàn)公式對每個action的價值的計算方法是一種粗糙的估計算法。

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

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

也就是我們在更新Q表格的時候，只讓Q值朝著折現(xiàn)公式計算結(jié)果靠近一點點(縮小差值)，而不是直接調(diào)整為折現(xiàn)公式的計算結(jié)果。

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

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

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

env具有reset,step,render幾個方法。

懸崖問題

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

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

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

10個用C表示的為懸崖，掉入懸崖會拉回到起始位置。

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

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

reward設(shè)計如下：

智能體每走一步都會有-1的reward。

這個問題希望訓(xùn)練一個能夠盡可能快的從起始位置到達(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)

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

二，定義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#動作維度，有幾個動作可選self.lr=learning_rate#學(xué)習(xí)率self.gamma=gamma#reward的衰減率self.epsilon=e_greed#按一定概率隨機(jī)選動作self.Q=nn.Parameter(torch.zeros((obs_n,act_n)),requires_grad=False)#根據(jù)輸入觀察值，采樣輸出的動作值，帶探索defsample(self,obs):ifnp.random.uniform(0,1)<(1.0-self.epsilon):#根據(jù)table的Q值選動作action=self.predict(obs)else:action=np.random.choice(self.act_n)#有一定概率隨機(jī)探索選取一個動作returnaction#根據(jù)輸入觀察值，預(yù)測輸出的動作值defforward(self,obs):Q_list=self.Q[obs,:]maxQ=Q_list.max()action_list=torch.where(Q_list==maxQ)[0].tolist()#maxQ可能對應(yīng)多個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:本次動作獲得的獎勵rnext_obs:本次交互后的obs,s_t+1next_action:根據(jù)當(dāng)前Q表格,針對next_obs會選擇的動作,a_t+1done:episode是否結(jié)束"""predict_Q=self.Q[obs,action]ifdone:target_Q=reward#沒有下一個狀態(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)建一個agent實例，輸入超參數(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)練模版來對Agent進(jìn)行訓(xùn)練。

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

StepRunner, EpochRunner這2個層級上都有少量的修改。

classDataLoader:def__init__(self,env,agent,stage="train"):self.env=envself.agent=agentself.stage=stagedef__iter__(self):obs,info=self.env.reset()#重置環(huán)境,重新開一局（即開始新的一個episode）action=self.agent.sample(obs)#根據(jù)算法選擇一個動作whileTrue:next_obs,reward,done,_,_=self.env.step(action)#與環(huán)境進(jìn)行一個交互ifself.stage=="train":next_action=self.agent.sample(next_obs)#訓(xùn)練階段使用探索-利用策略else:next_action=self.agent.predict(next_obs)#驗證階段使用模型預(yù)測結(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

四，測試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)練完成后，這個agent非常機(jī)智地在懸崖邊上走了一個最優(yōu)路線，但卻沒有掉到懸崖里去。

五，保存Agent

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

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

關(guān)鍵詞：