📅 最后修改于: 2023-12-03 14:49:47.163000 🧑 作者: Mango
深度 Q 学习（Deep Q-Learning）是一种通过神经网络代替 Q 表来实现 Q 学习的方法，能够有效解决强化学习中状态空间非常大或连续的问题。在该方法中，神经网络的输入是状态，输出是每个行为的 Q 值，通过选择具有最大 Q 值的行为来执行机器人等复杂任务。
在本篇文章中，我们将使用 TensorFlow 实现深度 Q 学习，步骤包括构建模型、训练模型以及测试模型。
首先，我们需要定义模型。在我们的深度 Q 学习中，我们使用一个包含多个隐藏层的神经网络，以状态为输入，输出为每个可能行动的 Q 值。
使用 TensorFlow 构建模型的第一步是导入所需的模块和库。代码如下：
import tensorflow as tf
import numpy as np
import random
# 定义模型
class DQN:
def __init__(
self,
n_actions,
n_features,
learning_rate=0.01,
reward_decay=0.9,
e_greedy=0.9,
replace_target_iter=300,
memory_size=500,
batch_size=32,
e_greedy_increment=None,
output_graph=False,
):
self.n_actions = n_actions
self.n_features = n_features
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
self.learn_step_counter = 0
# 定义记忆库
self.memory = np.zeros((self.memory_size, n_features*2+2))
# 定义神经网络
self.build_net()
# 定义存储参数的操作
t_params = tf.get_collection('target_net_params')
e_params = tf.get_collection('eval_net_params')
self.replace_target_op = [tf.assign(t, e) for t, e in zip(t_params, e_params)]
self.sess = tf.Session()
# 是否输出计算图，便于调试
if output_graph:
tf.summary.FileWriter("logs/", self.sess.graph)
self.sess.run(tf.global_variables_initializer())
self.cost_his = []
以上代码中，我们首先定义了深度 Q 学习的基本参数以及记忆库，随后建立了神经网络模型，包括输入层、多个隐藏层以及输出层，最后定义了存储模型参数的操作。我们还开启了 TensorFlow 的一个会话并初始化全局变量。
在定义好深度 Q 学习模型后，我们需要使用经验回放和目标网络不断训练并优化神经网络参数以得到更好的 Q 值估计。代码如下：
# 定义神经网络
def build_net(self):
# 定义输入层
self.s = tf.placeholder(tf.float32, [None, self.n_features], name='s') # 输入状态
self.q_target = tf.placeholder(tf.float32, [None, self.n_actions], name='Q_target') # 用于备份的Q值
# 定义评估网络
with tf.variable_scope('eval_net'):
c_names, n_l1, w_initializer, b_initializer = \
['eval_net_params', tf.GraphKeys.GLOBAL_VARIABLES], 10, \
tf.random_normal_initializer(0., 0.3), tf.constant_initializer(0.1)
# 神经网络第一层
with tf.variable_scope('l1'):
w1 = tf.get_variable('w1', [self.n_features, n_l1], initializer=w_initializer, collections=c_names)
b1 = tf.get_variable('b1', [1, n_l1], initializer=b_initializer, collections=c_names)
l1 = tf.nn.relu(tf.matmul(self.s, w1) + b1)
# 神经网络第二层
with tf.variable_scope('l2'):
w2 = tf.get_variable('w2', [n_l1, self.n_actions], initializer=w_initializer, collections=c_names)
b2 = tf.get_variable('b2', [1, self.n_actions], initializer=b_initializer, collections=c_names)
self.q_eval = tf.matmul(l1, w2) + b2
# 定义目标网络
with tf.variable_scope('target_net'):
c_names = ['target_net_params', tf.GraphKeys.GLOBAL_VARIABLES]
# 神经网络第一层
with tf.variable_scope('l1'):
w1 = tf.get_variable('w1', [self.n_features, n_l1], initializer=w_initializer, collections=c_names)
b1 = tf.get_variable('b1', [1, n_l1], initializer=b_initializer, collections=c_names)
l1 = tf.nn.relu(tf.matmul(self.s, w1) + b1)
# 神经网络第二层
with tf.variable_scope('l2'):
w2 = tf.get_variable('w2', [n_l1, self.n_actions], initializer=w_initializer, collections=c_names)
b2 = tf.get_variable('b2', [1, self.n_actions], initializer=b_initializer, collections=c_names)
self.q_next = tf.matmul(l1, w2) + b2
# 定义计算loss的操作
with tf.variable_scope('loss'):
self.loss = tf.reduce_mean(tf.squared_difference(self.q_target, self.q_eval))
with tf.variable_scope('train'):
self._train_op = tf.train.AdamOptimizer(self.lr).minimize(self.loss)
# 整除计数器，每替换一次target_net，计数器就加1
self.t_params = tf.get_collection('target_net_params')
self.e_params = tf.get_collection('eval_net_params')
self.replace_target_op = [tf.assign(t, e) for t, e in zip(self.t_params, self.e_params)]
# 记忆库存储
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_))
index = self.memory_counter % self.memory_size
self.memory[index, :] = transition
self.memory_counter += 1
# epsilon贪心算法决策
def choose_action(self, observation):
observation = observation[np.newaxis, :]
if np.random.uniform() < self.epsilon:
actions_value = self.sess.run(self.q_eval, feed_dict={self.s: observation})
action = np.argmax(actions_value)
else:
action = np.random.randint(0, self.n_actions)
return action
# 训练并更新神经网络
def learn(self):
# 随机从记忆库里提取batch_size策略来更新神经网络
batch_index = np.random.choice(self.memory_size, size=self.batch_size)
batch_memory = self.memory[batch_index, :]
q_next, q_eval = self.sess.run(
[self.q_next, self.q_eval],
feed_dict={
self.s_: batch_memory[:, -self.n_features:], # 前n_features为状态
self.s: batch_memory[:, :self.n_features], # 后n_features为下一步状态
})
q_target = q_eval.copy()
batch_index = np.arange(self.batch_size, dtype=np.int32)
eval_act_index = batch_memory[:, self.n_features].astype(int)
reward = batch_memory[:, self.n_features + 1]
q_target[batch_index, eval_act_index] = reward + self.gamma * np.max(q_next, axis=1)
# 训练神经网络
_, self.cost = self.sess.run([self._train_op, self.loss],
feed_dict={self.s: batch_memory[:, :self.n_features],
self.q_target: q_target})
self.cost_his.append(self.cost)
# 按照epsilon_increment的方式逐步增加epsilon，降低随机行动的概率
if self.epsilon_increment is not None:
self.epsilon = self.epsilon + self.epsilon_increment
else:
self.epsilon = self.epsilon_max
self.learn_step_counter += 1
# 每学习300步，就更新目标网络target_net的参数
if self.learn_step_counter % self.replace_target_iter == 0:
self.sess.run(self.replace_target_op)
在以上代码中，我们首先定义了神经网络的输入层、评估网络和目标网络，并计算了损失并使用 Adam 优化器最小化损失。随后，我们定义了存储策略、计算 Q 目标和训练模型的操作。其中，我们还使用经验回放等方法来训练模型，以提高模型的稳定性和训练效果，最后实现了每替换一次 target_net，计数器就加1 的操作，方便下面的操作。
训练模型之后，我们需要对模型进行测试，以评估模型的性能。代码如下：
if __name__ == '__main__':
env = gym.make('CartPole-v0')
env = env.unwrapped
brain = DQN(n_actions=env.action_space.n,
n_features=env.observation_space.shape[0],
learning_rate=0.01,
reward_decay=0.9,
e_greedy=0.9,
replace_target_iter=300,
memory_size=500,
e_greedy_increment=None,
output_graph=True
)
for i_episode in range(1000):
observation = env.reset()
while True:
env.render()
action = brain.choose_action(observation)
observation_, reward, done, info = env.step(action)
x, x_dot, theta, theta_dot = observation_
r1 = (env.x_threshold - abs(x)) / env.x_threshold - 0.8
r2 = (env.theta_threshold_radians - abs(theta)) / env.theta_threshold_radians - 0.5
reward = r1 + r2
brain.store_transition(observation, action, reward, observation_)
if brain.memory_counter > brain.memory_size:
brain.learn()
if done:
print('episode: ', i_episode,
'epsilon: ', round(brain.epsilon, 2),
'reward: ', round(reward, 2))
break
observation = observation_
在训练好神经网络之后，我们对小车的状态进行实时监测，并进行相应的行动。如果观察到小车倾覆（done 为 True），则结束该次试验并统计分数。
本篇文章中，我们探讨了如何使用 TensorFlow 实现深度 Q 学习，并构建了一个基于 Q-Learning 的小车承载平衡杆的任务测例。该实例中的深度 Q 学习模型可用于其他需要强化学习的任务中，如飞机自动驾驶、机器人走动等，具有广泛的应用价值。