Q-learning是一个经典的强化学习算法，是一种基于价值(Value-based)的算法，通过维护和更新一个价值表格(Q表格)进行学习和预测。

(资料图片仅供参考)

Q-learning是一种off-policy的策略，也就是说，它的行动策略和Q表格的更新策略是不一样的。

行动时，Q-learning会采用epsilon-greedy的方式尝试多种可能动作。

更新时，Q-learning会采用潜在收益最大化的动作进行价值更新。

总体来说，Q-learning是一个非常勇敢的策略，在有限动作和有限状态情况下，它能够收敛找到全局最优策略。

公众号算法美食屋后台回复关键词：torchkeras，获取本文notebook源代码~

在第n步，agent处于状态 state(n)中，然后采取行动action(n)，env给agent奖励reward(n+1)，同时agent的状态变成 state(n+1)

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

以我们玩一个俄罗斯方块游戏为例。

环境env就是这个游戏背后的程序，智能体agent就是玩家。

假设现在是第n步，state(n)就是目前游戏所处的状态，可以表示为一个矩阵，也就是游戏界面每个格子的明暗状态。

我们可以采取某个 action(n) (向左，向右，向下，变形)。

然后我们会获得一个奖励reward(n)，即得分。奖励很多时候是稀疏的，即大部分时候为0，操作很多步才有一个不为0的奖励。

同时游戏界面发生变化，状态由 state(n) 变成 state(n+1)。

env和agent交互作用若干个步骤，到达结束状态，通常叫做一个episode(片段)。

在俄罗斯方块游戏的例子中，一局完整的游戏构成一个马尔科夫交互链，叫做一个episode.

之所以叫做马尔科夫交互链，是因为这个过程满足马尔科夫假设。

第n+1步骤的状态state(n+1)和奖励reward(n+1)只和第n步骤的状态stage(n)和action(n)有关，而与之前的状态和action无关。

马尔科夫假设要求我们在设计state和action的时候，要考虑到所有相关变量。

并且，只要设计出合理的state变量和action变量，任何游戏都可以表示为这样一个马尔科夫交互链。

为了衡量每个步骤中action的价值，需要将该步骤之后的奖励，以及未来的全部奖励按照类似金融学中的折现算法求和。

在俄罗斯方块游戏的例子中，一个操作action的价值，不仅跟这个操作完成后立刻获得的奖励reward有关，还要考虑到这个操作的长远影响。

但这种长远影响不太好精确地计算，因为后面获得的奖励，不仅跟当前的action有关，还跟后面的操作有关，所以跟当前操作的相关性是小于1的。

作为简化起见，我们通过类似金融学中现金流折现的方式将未来的奖励全部折算到当前步骤。折算因子gamma一般取值在0.9~1.0之间。

训练时使用epsilon探索，预测时使用greedy贪心。

训练阶段: 以一定epsilon概率选择随机动作，以（1-epsilon）选择最大化Q(s,a)的动作。

预测阶段: 贪心策略，直接选择最大化Q(s,a)的动作。

为了让模型去探索更优策略，我们在训练过程中会允许模型以一定的概率去实施随机动作，以便评估不同动作的价值。

这样也能够让模型对状态动作空间进行更分散的采样，学到的结果也会更加鲁棒。

但在测试过程，为了获得更好的结果，我们应该采用预期价值最大的动作。

奖励折现公式对每个action的价值的计算方法是一种粗糙的估计算法。

不同的step或者不同的episode中，按照奖励折现公式对相同state下相同action价值的评估的结果可能差异很大。

为了保持学习过程的稳定性，让Q值不会过分受到某次评估的影响，我们采用一种软更新的方式。

也就是我们在更新Q表格的时候，只让Q值朝着折现公式计算结果靠近一点点(缩小差值)，而不是直接调整为折现公式的计算结果。

这样，我们最终的Q表格中action的价值结果相当是许多次不同episode不同step下奖励折现公式计算结果的某种平均值。

gym是一个常用的强化学习测试环境，可以用make创建环境。

env具有reset,step,render几个方法。

环境设计如下：

环境一共有48个state状态。

其中T为目标位置，到达目标位置游戏结束。

10个用C表示的为悬崖，掉入悬崖会拉回到起始位置。

智能体设计如下：

智能体有4种动作action，0表示往上，1往右，2往下，3往左。

reward设计如下：

智能体每走一步都会有-1的reward。

这个问题希望训练一个能够尽可能快的从起始位置到达目标位置T的智能体Agent。

importgymimportnumpyasnpimporttimeimportmatplotlibimportmatplotlib.pyplotaspltfromIPythonimportdisplayprint("gym.__version__=",gym.__version__)%matplotlibinline#可视化函数：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)

我们先来看看没有训练模型，按照随机的方式会怎么走。

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#学习率self.gamma=gamma#reward的衰减率self.epsilon=e_greed#按一定概率随机选动作self.Q=nn.Parameter(torch.zeros((obs_n,act_n)),requires_grad=False)#根据输入观察值，采样输出的动作值，带探索defsample(self,obs):ifnp.random.uniform(0,1)<(1.0-self.epsilon):#根据table的Q值选动作action=self.predict(obs)else:action=np.random.choice(self.act_n)#有一定概率随机探索选取一个动作returnaction#根据输入观察值，预测输出的动作值defforward(self,obs):Q_list=self.Q[obs,:]maxQ=Q_list.max()action_list=torch.where(Q_list==maxQ)[0].tolist()#maxQ可能对应多个actionaction=np.random.choice(action_list)returnaction@torch.no_grad()defpredict(self,obs):self.eval()returnself.forward(obs)#学习方法，也就是更新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:根据当前Q表格,针对next_obs会选择的动作,a_t+1done:episode是否结束"""predict_Q=self.Q[obs,action]ifdone:target_Q=reward#没有下一个状态了else:target_Q=reward+self.gamma*self.Q[next_obs,:].max()#Q-learningself.Q[obs,action]+=self.lr*(target_Q-predict_Q)#修正q

我们创建一下env和agent.

#使用gym创建悬崖环境env=gym.make("CliffWalking-v0")#0up,1right,2down,3left#创建一个agent实例，输入超参数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)

下面我们将套用torchkeras的训练模版来对Agent进行训练。

由于强化学习问题与常用的监督学习范式有很大的差异，所以我们对torchkeras的训练模版在

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()#重置环境,重新开一局（即开始新的一个episode）action=self.agent.sample(obs)#根据算法选择一个动作whileTrue:next_obs,reward,done,_,_=self.env.step(action)#与环境进行一个交互ifself.stage=="train":next_action=self.agent.sample(next_obs)#训练阶段使用探索-利用策略else:next_action=self.agent.predict(next_obs)#验证阶段使用模型预测结果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

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

#全部训练结束，查看算法效果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

可以看到，训练完成后，这个agent非常机智地在悬崖边上走了一个最优路线，但却没有掉到悬崖里去。

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

公众号算法美食屋后台回复关键词：torchkeras，获取本文notebook源代码以及更多有趣范例。

关键词：