强化学习入门DQN详解

详解自己学习过程中DQN算法代码流程和重要原理，帮助新手快速上手入门强化学习，视频和代码资源移步B站和github。

文章共3,741字 · 阅读需要大约13分钟

一键AI生成摘要，助你高效阅读

问答

芝士就是力量蛙

1261人浏览 · 2023-02-02 19:15:10

芝士就是力量蛙 · 2023-02-02 19:15:10 发布

Deep Q Network

参考资料：

B站莫烦：https://www.bilibili.com/video/BV13W411Y75P/?spm_id_from=333.337.search-card.all.click&vd_source=a8e8676617fb04db42af59b530b145fd

github（tensorflow）：https://github.com/MorvanZhou/Reinforcement-learning-with-tensorflow

pytorch版本：https://github.com/ClownW/Reinforcement-learning-with-PyTorch

1、先验知识：

（1）Q-learning

Q-learning使用Q表存储状态-动作-价值对应关系。

用这种办法进行尝试遍历制作Q表再根据损失函数进行训练逼近，虽然非常直观但是当状态与动作种类非常多时，Q表将会非常巨大，不利于计算效率的提升甚至会无法完成。

Deep-Qlearning引入神经网络通过训练得到Q值，将状态和动作组合作为神经网络的输入，无需构造Q表，借助神经网络生成相应的Q值。也可以指输入状态，通过神经网络生成采取相应动作后对应的价值，直接通过强化学习选取价值最大的动作。

（2）Deep Learning

那么在强化学习中我们是如何训练神经网络的呢？

在这里插入图片描述

$新NN=老NN+\alpha*(Q现实-Q估计)$

（3）Experience replay and fixed Q target

经验回放机制，DQN是一种离线学习法，它存在一个记忆库，用于存放之前的经历。训练的时候会随机抽取一些之前的经历，以此打乱经历之间的相关性使神经网络的更新更有效率。

 Fixed Q target，在DQN中使用两个结构相同但是参数不同的神经网络，用一个不断更新的网络来预测Q估计的值，用一个比较老的神经网络来预测Q现实的值，这样可以加快算法的收敛速度，达到更好的效果。

2、 DQN算法更新循环思路

关键思路：DQN采用经验回放机制，所以一开始先不学习，首先建造一个记忆库，这里用step记录，当step大于200以后，再每五步学习一次（RL.learn）。每个回合的更新流程是

①从环境导入智能体此时的状态

②更新环境，根据e-greedy策略选择动作

③得到并存储该动作产生的下一状态以及获得的奖励（存储很重要，在RL-brain中编写）

④更新状态

⑤判断状态是否是最终状态，到达最终状态跳出循环

def run_maze():
	step = 0
	for episode in range(300):
		print("episode: {}".format(episode))
		observation = env.reset()
		while True:
			print("step: {}".format(step))
			env.render()
			action = RL.choose_action(observation)
			observation_, reward, done = env.step(action)
			RL.store_transition(observation, action, reward, observation_)
			if (step>200) and (step%5==0):
				RL.learn()
			observation = observation_
			if done:
				break
			step += 1
	print('game over')
	env.destroy()

3、神经网络搭建

由于采用fixed Q target方法，我们需要搭建两个结构相同而参数不同的神经网络（这里分别记为eval和target，eval不断更新，target保持过去的参数），将两个网络得到的结果比较，加快算法的收敛速度，pytorch版的代码如下。这里莫烦老师教程采用的是tensorflow，可以在github获取。

（1）定义网络结构

class Net(nn.Module):
	def __init__(self, n_feature, n_hidden, n_output):
		super(Net, self).__init__()
		self.el = nn.Linear(n_feature, n_hidden)
		self.q = nn.Linear(n_hidden, n_output)

	def forward(self, x):
		x = self.el(x)
		x = F.relu(x)
		x = self.q(x)
		return x

（2）拿出两个网络中的Q值

def _build_net(self):
   self.q_eval = Net(self.n_features, self.n_hidden, self.n_actions)
   self.q_target = Net(self.n_features, self.n_hidden, self.n_actions)
   self.optimizer = torch.optim.RMSprop(self.q_eval.parameters(), lr=self.lr)

4、基本参数定义（DQN初始化部分）

定义并初始化可选动作、状态特征、神经网络层数、学习率、折扣因子、神经网络更新周期、记忆库大小、抽取样本大小等强化学习需要的基本参数。并设置learn_step_counter来记录目前学习的步数，方便观察结果和测试。
初始化记忆库（pandas），我们首先将记忆库初始化为全零矩阵，高度是记忆库的大小memory_size。其宽度就是每条记忆的大小取决于run-maze中定义的结构，一共有几个变量需要存储。在这里就是RL.store_transition(observation, action, reward, observation_)结构，这里两个状态observation、observation_加上action和reward所以数据长度是n_features*2+2，注意这里action虽然上下左右四个选项，但是每次只选取一个。
这个顺序也比较关键，在后面抽取数据的时候要记住上一步和下一步的状态分别在哪个位置，可以通过切片索引的方法拿到需要的数据。

class DeepQNetwork():
	def __init__(self, n_actions, n_features, n_hidden=20, learning_rate=0.01, reward_decay=0.9, e_greedy=0.9,
				replace_target_iter=200, memory_size=500, batch_size=32, e_greedy_increment=None,
				):
		self.n_actions = n_actions
		self.n_features = n_features
		self.n_hidden = n_hidden
		self.lr = learning_rate
		self.gamma = reward_decay
		self.epsilon_max = e_greedy
		self.replace_target_iter = replace_target_iter
		self.memory_size = memory_size
		self.batch_size = batch_size
		self.epsilon_increment = e_greedy_increment
		self.epsilon = 0 if e_greedy_increment is not None else self.epsilon_max

		# total learning step
		self.learn_step_counter = 0

		# initialize zero memory [s, a, r, s_]
		self.memory = np.zeros((self.memory_size, n_features*2+2))

		self.loss_func = nn.MSELoss()
		self.cost_his = []

		self._build_net()

5、存储记忆store_transition

定义存储记忆函数，这里的输入参数分别是当前状态s，选取动作a，获得奖励r，下一状态s_。

通过self.memory_counter进行索引，找到记忆库的某一行，再插入此时的这条数据。当counter到记忆库大小时，归零，返回上去，从头开始按条替换，保证记忆库是最近的数据，循环往复，不断覆盖更新。

	def store_transition(self, s, a, r, s_):
		if not hasattr(self, 'memory_counter'):
			self.memory_counter = 0
		transition = np.hstack((s, [a, r], s_))
		# replace the old memory with new memory
		index = self.memory_counter % self.memory_size
		self.memory[index, :] = transition 
		self.memory_counter += 1

6、动作选择部分

将当前状态observation输入神经网络，输出在该状态下选取每一个动作的Q值，根据e_greedy策略，在90%（可任意设置）情况下选取Q值最大的动作，剩下10%随机探索，避免局部最优。

	def choose_action(self, observation):
		observation = torch.Tensor(observation[np.newaxis, :])
		if np.random.uniform() < self.epsilon:
			actions_value = self.q_eval(observation)

			action = np.argmax(actions_value.data.numpy())
		else:
			action = np.random.randint(0, self.n_actions)
		return action

7、fixed Q target（神经网络更新）

该部分比较简单，在开始训练之前，我们第一步要判断此时的神经网络是否需要更新。target_net中的各参数是冻结的（Fixed Q target），当到达一个更新周期（replace_target_iter）后，将eval网络中的参数传给target网络，以此来实现网络的更新，这样可以避免神经网络的参数一直变化，不稳定，收敛困难的问题，加速其收敛速度。

		# check to replace target parameters
		if self.learn_step_counter % self.replace_target_iter == 0:
			self.q_target.load_state_dict(self.q_eval.state_dict())
			print("\ntarget params replaced\n")

8、训练过程

神经网络训练时，每一次参数的更新所需要损失函数并不是由一个单独的数据{data：label}获得的，而是由一组数据加权得到的，这一组数据的数量就是[batch size]。我们用随机梯度下降法对数据集进行批量训练，每次随机抽取记忆库中的样本，这里设置当抽取数量大于记忆库大小时，即抽取已经存入的所有记忆。

def learn(self):
   # check to replace target parameters
   if self.learn_step_counter % self.replace_target_iter == 0:
      self.q_target.load_state_dict(self.q_eval.state_dict())
      print("\ntarget params replaced\n")

   # sample batch memory from all memory
   if self.memory_counter > self.memory_size:
      sample_index = np.random.choice(self.memory_size, size=self.batch_size)
   else:
      sample_index = np.random.choice(self.memory_counter, size=self.batch_size)
   batch_memory = self.memory[sample_index, :]

接下来运行两个神经网络，分别得到他们的估计值，冻结网络target的输出作为Q现实，eval网络输出值作为Q估计，从而实现迭代更新。切片索引时注意这里target网络的输入是下一状态，eval网络输入是当前状态。

# q_next is used for getting which action would be choosed by target network in state s_(t+1)
		q_next, q_eval = self.q_target(torch.Tensor(batch_memory[:, -self.n_features:])), self.q_eval(torch.Tensor(batch_memory[:, :self.n_features]))
		# used for calculating y, we need to copy for q_eval because this operation could keep the Q_value that has not been selected unchanged,
		# so when we do q_target - q_eval, these Q_value become zero and wouldn't affect the calculation of the loss 
		q_target = torch.Tensor(q_eval.data.numpy().copy())

最后计算损失、反向传播、更新参数，这里计算损失比较关键，由于两个Q值都是二维数组，如果直接矩阵相减，对应关系不正确，需要首先找到eval中的状态和对应动作，再去找到target中的对应值，才可以进行相减计算损失。

	batch_index = np.arange(self.batch_size, dtype=np.int32)
	eval_act_index = batch_memory[:, self.n_features].astype(int)
	reward = torch.Tensor(batch_memory[:, self.n_features+1])
	q_target[batch_index, eval_act_index] = reward + self.gamma*torch.max(q_next, 1)[0]

		loss = self.loss_func(q_eval, q_target)
		self.optimizer.zero_grad()
		loss.backward()
		self.optimizer.step()