我的Github实现:gym(GitHub)
本篇博客主要是个人实现过程的主观感受,如果想要使用模型可以直接去GitHub仓库,注释完善且规范。
觉得有用请给我点个star!

前言

最近在学习强化学习,大致过了一遍强化学习的数学原理(视频)
视频讲的很好,但是实践的部分总是感觉有点匮乏(毕竟解决 grid world 方格世界(GitHub) 的问题的很难给人特别大的正反馈),所以就找到了openai gymnaisum 想要玩一下里面的几个环境。
CartPole-v1是我的第一个环境,agent分数成功超过threshold(可喜可贺)。
虽然环境很简单,但是也花了我很多的时间(主要是调参!)来实现。
我的项目也已经放到了GitHub上,欢迎大家使用讨论!gym(GitHub)


DQN简介

DQN的核心思想就是用一个神经网络来近似agent的Q-Value-function,神经网络的输入是state,输出是不同动作的Q值,所以网络的形状应该是:(n_state,hidden_size,n_action)。具体的训练公式(因为是训练神经网络,以损失函数的形式给出)如下:

当然,具体使用我们不可能知道期望,所以使用mbgd方法将期望换成采样:

这里的N就是batch_size,
$s_i$表示第i个experience的状态
$a_i$表示第i个experience的动作
$r_i$表示第i个experience的奖励
$s^{‘}_i$表示第i个experience的下个状态
$a$表示第i个experience的下个状态的Q值最大的动作
其中$(s_i,a_i,r_i,s^{‘}_i)$组成了一个experience,也就是sample样本。
我们深入内部去看:

被称作TD-target,可以看作是我们函数优化的Ground Truth
而$\hat{q}(s_i,a_i,w)$则是函数的prediction,是我们要优化的对象,我们的目标就是要函数的prediction不断逼近函数的Ground Truth。这里要注意的是GT是由神经网络$w_T$参与的,这个神经网络与产生pred的神经网络$w$有所差别。在具体实现中我们会固定$w_T$,只更新$w$,每经过固定步数后,将后者赋值给前者。


环境简介

CartPole-v1,也就是车杆,指一种用于平衡控制的装置,通常由一个竖直的杆和一个连接在杆上的水平平台组成(如下图所示),任务的目的是让杆子保持竖直,并且小车不能离开中心(左或右)太远。

这个任务是比较简单的,因为agent只能进行两个动作:左推,右推。但是失败的判定标准中的竖杆偏离竖直方向一定角度(极容易失败)比较严格。
具体的描述(状态空间、动作空间…)大家可以去看gymnasium的文档,讲的还是非常详细的。


任务实现

具体的代码以及如何运行我在GitHub仓库中都讲的很详细了,整个项目主要分为四大部分: argument.py(参数配置) agent.py (智能体DQN实现)train.py(训练主函数) test.py(测试主函数)。

强化学习的特点就是调参难调(这几天深有体会),所以一个方便的调参接口对于我们来说就是十分重要的了。
argument .py的默认参数是有很大概率可以训练出完成任务的DQN的,如果只是想试试效果的话可以直接使用默认参数。但是作者在这里还是强烈建议大家调一调参数感受一下调参的难度(未来打算做一期关于调参的博客)。

关于agent的部分,我一开始只是想要设计一个最简单的DQN算法,后来发现怎么调参都没法收敛,于是去网上搜集资料了解到对于CartPole这类没有显式奖励(该任务每个step的reward固定是1,所以理论上等价reward固定是0),或者是其他一些只有终点有一个reward的任务有一个特殊的划分:sparse reward稀疏奖励问题。这类问题收敛的速度会很慢(实际上我感觉根本没收敛),所以有一些特殊的方法可以实现收敛的加速。
我于是了解并使用了其中一个叫做 prioritized experience replay(arxiv) 的方法,主要的的思想就是在训练神经网络时并不是以随机的方法进行训练样本的采样,而是根据一定权重进行。
为此我重新定义了reply buffer的类(原先为deque),使其可以完成按权重采样、更新权重的功能,但是我后来又将类修改为等概率采样,因为我发现不收敛的原因不是使用的方法太初级。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
class MyDeque():
def __init__(self,maxlength,obs_size,batch_size):
"""
自定义的经验放回区,用于存储和采样经验,是一个循环队列结构,实质上直接使用dequeue也可实现,这里单独实现主要有两个目的:
1.功能封装,代码更为简介
2.想要直接将经验存储为tensor形式且保存在gpu中,减少cpu和gpu之间的通信次数,加速训练速度(事实证明没有用)
:param maxlength:经验池的大小
:param obs_size:状态空间的size
:param batch_size:每一次采样的经验数量
"""

self.maxlength=maxlength
self.batch_size=batch_size

#初始化存储经验的张量
self.memory_obs = torch.zeros((maxlength, obs_size), dtype=torch.float32, device=device)
self.memory_action = torch.zeros((maxlength, 1), dtype=torch.long, device=device)
self.memory_reward = torch.zeros((maxlength, 1), dtype=torch.float32, device=device)
self.memory_next_obs = torch.zeros((maxlength, obs_size), dtype=torch.float32, device=device)
self.memory_terminated = torch.zeros((maxlength, 1), dtype=torch.bool, device=device)
self.memory_weights = torch.ones((maxlength,), dtype=torch.float32, device=device)
self.idx=0 #当前写入位置
self.full=False #存储区是否已满

def __len__(self):
"""
返回当前缓冲区的经验数量,主要用于训练刚开始时经验数小于批量大小时拒绝采样
:return:当前缓冲区的经验数量
"""
if self.full ==True:
return self.maxlength
return self.idx

def get_batch(self):
"""
从缓冲区采样一个批次的数据
:return:状态,动作,奖励,下一个状态,结束标志
"""
index=self.get_index()
self.last_index=index
obs=self.memory_obs[index]
action=self.memory_action[index]
reward=self.memory_reward[index]
next_obs=self.memory_next_obs[index]
terminated=self.memory_terminated[index]
return obs,action,reward,next_obs,terminated

def get_index(self):
"""
根据权重采样索引,该版本权重相等
:return:索引序列,形状为(batch_size,1)
"""
if self.full==True:
return torch.multinomial(self.memory_weights,self.batch_size,replacement=True)
return torch.multinomial(self.memory_weights[:self.idx],self.batch_size,replacement=True)

def update(self,td_error):
"""
修改每个样本的权重,依据是tderror,该版本未使用
:param td_error:理想q与现实q差值的绝对值
"""
for id,idx in enumerate(self.last_index):
self.memory_weights[idx]=td_error[id]+1e-6

def append(self,obs,action,reward,next_obs,terminated):
"""
向缓冲区添加新的经验
:param obs:当前状态
:param action:采取的动作
:param reward:获得的奖励
:param next_obs:下一状态
:param terminated:结束标志
"""
self.memory_obs[self.idx]=torch.tensor(obs, dtype=torch.float32, device=device)
self.memory_action[self.idx]=torch.tensor(action, dtype=torch.long, device=device)
self.memory_reward[self.idx]=torch.tensor(reward, dtype=torch.float32, device=device)
self.memory_next_obs[self.idx]=torch.tensor(next_obs, dtype=torch.float32, device=device)
self.memory_terminated[self.idx] = torch.tensor(terminated, dtype=torch.bool, device=device)
self.memory_weights[self.idx]=1e7 #初始权重
self.idx+=1
#循环队列
if self.idx==self.maxlength:
self.full=True
self.idx=0

关于这个新的reply buffer我还做了一些小设计,那就是在样本录入时直接录入为tensor类,这样就不用每次采样都要临时转化为tensor,导致信息在cpu和gpu之间频繁传输,导致训练速度下降。(后经过测试,并没有什么用,因为与环境的交互是要在cpu上进行的,交互和神经网络计算迭代进行,频繁的cpu、gpu传输根本无法避免,用了cuda反而导致速度下降)

好笑的是,最终我并没有使用新颖的技巧,只是简单的将我的[128,64,64]的mlp中间层尺寸改为 [64,32]就让DQN收敛了。
这是可以理解的,毕竟任务目标是很简单的,用一个复杂的神经网络去拟合,确实有可能会出现过拟合问题。
但是我又有点疑问了,过拟合的原因是样本噪声大、样本数量少,理论上来说哪怕任务再简单,只要你的样本无穷多,就不会过拟合。那么对于强化学习这样一个不断产生利用新样本的范式,为什么还会出现过拟合呢?
我的猜测是神经网络更新的太频繁了,每一次交互都要进行一个批次的更新,相当于每个样本要平均被使用batchsize次数。那么是不是说,我们把样本更新相对于交互的频率降低,就能避免这个问题呢?(打算过段时间出个博客)


说开来去

实事求是的讲,这是一个简单方法解决简单问题的实现,本来应该很快速的完成,如果花了很多时间,那就是走上歪路了。
但是当我尝试了很多方法都没有效果,不断检查代码是否有问题,最后通过最简单的降低神经网络复杂性解决了问题后,我还是感觉到一种巨大的喜悦。这种喜悦不是我解决了什么问题,而是我在尝试了很多方法后,将所有可能的问题都一一排除,最后找到了问题所在。
去年入门深度学习的时候是看李沐的动手学深度学习(b站),里面的提问环节我很喜欢,我迫切的希望通过前辈的经验尽量避免走到弯路上去。
其实我很多次的学习过拟合的概念(学校机器学习课程、网络博客教程…),觉得自己对此已经熟悉到不能熟悉了,但在实践中却还是最后走投无路到时候才考虑它。
人教人确实学不会,事教人一教就会。