深度强化学习(DRL 3) - 从Q-learning到Deep Q Network(DQN)

迷南。 2024-02-23 07:55 197阅读 0赞

目录

    • 一、Q-learning
    • 二、Deep Q Network
    • 三、Double DQN

一、Q-learning

关于Q-learning,网上的资料很多。

在这里插入图片描述

Q-learning最核心的是有一个Q表,它记录了在环境中的 所有状态(s) 以及每个状态可以进行的 所有行为(a) 的Q值,初值设为0。









































状态 \ 行为 a1 a2 a3 a4
s1
s2
s3
……

Q值的更新公式如下:

Q ( s , a ) ← Q ( s , a ) + α [ r + γ m a x a ′ Q ( s ′ , a ′ ) − Q ( s , a ) ] Q(s,a)←Q(s,a)+α[r+\gamma max_{a’}Q(s^′,a^′)−Q(s,a)] Q(s,a)←Q(s,a)+α[r+γmaxa′Q(s′,a′)−Q(s,a)]

  • 其中 α α α表示学习速率,取值小于1
  • r r r是设置的奖励
  • 智能体在状态s经过行为a转移到状态s’, m a x a ′ Q ( s ′ , a ′ ) max_{a’}Q(s^′,a^′) maxa′Q(s′,a′) 是状态 s ′ s’ s′ 那一行最大Q值
  • γ \gamma γ则是衰减速率,取值在0到1之间, γ \gamma γ越大,智能体就会越注重长期利益, γ \gamma γ越小,智能体越短视。

每改变一次状态就会更新一次Q值

Q-learning 是一个 off-policy 的算法,它可以离线学习。它可以随机进行若干次游戏,将游戏过程保存起来,学习的时候就可以避免陷入局部最优。

Q-learning 的代码实现:

  1. # -*- coding: UTF-8 -*-
  2. import numpy as np
  3. import pandas as pd
  4. class QLearningTable:
  5. def __init__(self, actions, learning_rate=0.01, reward_decay=0.9, e_greedy=0.9):
  6. self.actions = actions # a list
  7. self.lr = learning_rate
  8. self.gamma = reward_decay
  9. self.epsilon = e_greedy
  10. self.q_table = pd.DataFrame(columns=self.actions, dtype=np.float64)
  11. def choose_action(self, observation):
  12. self.check_state_exist(observation)
  13. # action selection
  14. if np.random.uniform() < self.epsilon:
  15. # choose best action
  16. state_action = self.q_table.loc[observation, :]
  17. state_action = state_action.reindex(np.random.permutation(state_action.index)) # some actions have same value
  18. action = state_action.idxmax()
  19. else:
  20. # choose random action
  21. action = np.random.choice(self.actions)
  22. return action
  23. def learn(self, s, a, r, s_):
  24. self.check_state_exist(s_)
  25. q_predict = self.q_table.loc[s, a]
  26. if s_ != 'terminal':
  27. q_target = r + self.gamma * self.q_table.loc[s_, :].max() # next state is not terminal
  28. else:
  29. q_target = r # next state is terminal
  30. self.q_table.loc[s, a] += self.lr * (q_target - q_predict) # update
  31. def check_state_exist(self, state):
  32. if state not in self.q_table.index:
  33. # append new state to q table
  34. self.q_table = self.q_table.append(
  35. pd.Series(
  36. [0]*len(self.actions),
  37. index=self.q_table.columns,
  38. name=state,
  39. )
  40. )

二、Deep Q Network

如果想用Q-learning玩之前的CartPole-v0游戏,会发现CartPole-v0的状态值太多了,它的observation有 车的位置,车的速度,杆子的角度,杆子顶端的速度 四个值,并且是连续的数值,四个值再进行量化,Q表实在是太大了。用Q-learning玩游戏和随机玩游戏没区别,学习不到东西。可能Q-learning只适用于走迷宫类的游戏,每一步的状态都可以很简单的描述出来,4*4的迷宫只需要用16个状态就可以全部囊括。

我们用Deep Q Network来解决这个问题, Deep Q Network融合了神经网络和 Q learning。

  • 我们可以将状态和动作当成神经网络的输入, 然后经过神经网络分析后得到动作的 Q 值, 这样我们就没必要在表格中记录 Q 值, 而是直接使用神经网络生成 Q 值.
  • 或者只输入状态值, 输出所有的动作值, 然后按照 Q learning 的原则, 直接选择拥有最大值的动作当做下一步要做的动作.

    在这里插入图片描述

需要搭建两个神经网络,target_net 和 eval_net ,
target_net 用于预测 q_target 值, 他不会及时更新参数.
target_net 用于预测 q_eval, 这个神经网络拥有最新的神经网络参数.

两个神经网络结构是完全一样的, 只是里面的参数不一样. target_net 是 eval_net 的一个历史版本, 拥有 eval_net 很久之前的一组参数, 而且这组参数被固定一段时间, 然后再被 eval_net 的新参数所替换. 而 eval_net 是不断在被提升的。这样做是为了打乱数据间的相关性,避免陷入局部最优。

神经网络的输入数据是[s,a,r,s’],输出数据是状态s下所有行为a的Q值。
target_net 只会输出Q值(q_target ),不会进行训练,而eval_net 输出Q值(q_eval)之后会根据q_target和q_eval进行反向传播训练,更新eval_net 的参数。

Deep Q Network整个算法的运作:

  1. 初始化target_net 和 target_net。
  2. 观察游戏状态observation,选择合适的observation作为输入,一般情况会对observation做数据处理,使其更容易训练,这里不用。
  3. 设置合适的奖励reward。
  4. 先进行若干次游戏,将游戏数据存储到memory中。
  5. 从memory中随机选取训练数据batch_memory用于批量训练。
  6. 训练eval_net 一段时间后,将eval_net 的参数复制给target_net 。
  7. 训练过程中产生的新的游戏数据会替代memory中的旧数据。

回到游戏CartPole-v0

DQN.py:

  1. import numpy as np
  2. import tensorflow as tf
  3. np.random.seed(1)
  4. tf.set_random_seed(1)
  5. # Deep Q Network off-policy
  6. class DeepQNetwork:
  7. def __init__(
  8. self,
  9. n_actions,
  10. n_features,
  11. learning_rate=0.01,
  12. reward_decay=0.9,
  13. e_greedy=0.9,
  14. replace_target_iter=300,
  15. memory_size=500,
  16. batch_size=32,
  17. e_greedy_increment=None,
  18. output_graph=True,
  19. ):
  20. self.n_actions = n_actions
  21. self.n_features = n_features
  22. self.lr = learning_rate
  23. self.gamma = reward_decay
  24. self.epsilon_max = e_greedy
  25. self.replace_target_iter = replace_target_iter
  26. self.memory_size = memory_size
  27. self.batch_size = batch_size
  28. self.epsilon_increment = e_greedy_increment
  29. self.epsilon = 0 if e_greedy_increment is not None else self.epsilon_max
  30. # total learning step
  31. self.learn_step_counter = 0
  32. # initialize zero memory [s, a, r, s_]
  33. # [4, 1, 1, 4]
  34. self.memory = np.zeros((self.memory_size, n_features * 2 + 2))
  35. # consist of [target_net, evaluate_net]
  36. self._build_net()
  37. t_params = tf.get_collection(tf.GraphKeys.GLOBAL_VARIABLES, scope='target_net')
  38. e_params = tf.get_collection(tf.GraphKeys.GLOBAL_VARIABLES, scope='eval_net')
  39. with tf.variable_scope('hard_replacement'):
  40. self.target_replace_op = [tf.assign(t, e) for t, e in zip(t_params, e_params)]
  41. self.sess = tf.Session()
  42. if output_graph:
  43. # $ tensorboard --logdir=logs
  44. tf.summary.FileWriter("logs/", self.sess.graph)
  45. self.sess.run(tf.global_variables_initializer())
  46. self.cost_his = []
  47. def _build_net(self):
  48. # ------------------ all inputs ------------------------
  49. self.s = tf.placeholder(tf.float32, [None, self.n_features], name='s') # input State
  50. self.s_ = tf.placeholder(tf.float32, [None, self.n_features], name='s_') # input Next State
  51. self.r = tf.placeholder(tf.float32, [None, ], name='r') # input Reward
  52. self.a = tf.placeholder(tf.int32, [None, ], name='a') # input Action
  53. w_initializer, b_initializer = tf.random_normal_initializer(0., 0.3), tf.constant_initializer(0.1)
  54. # ------------------ build evaluate_net ------------------
  55. with tf.variable_scope('eval_net'):
  56. e1 = tf.layers.dense(self.s, 20, tf.nn.relu, kernel_initializer=w_initializer,
  57. bias_initializer=b_initializer, name='e1')
  58. self.q_eval = tf.layers.dense(e1, self.n_actions, kernel_initializer=w_initializer,
  59. bias_initializer=b_initializer, name='q')
  60. # ------------------ build target_net ------------------
  61. with tf.variable_scope('target_net'):
  62. t1 = tf.layers.dense(self.s_, 20, tf.nn.relu, kernel_initializer=w_initializer,
  63. bias_initializer=b_initializer, name='t1')
  64. self.q_next = tf.layers.dense(t1, self.n_actions, kernel_initializer=w_initializer,
  65. bias_initializer=b_initializer, name='t2')
  66. with tf.variable_scope('q_target'):
  67. q_target = self.r + self.gamma * tf.reduce_max(self.q_next, axis=1, name='Qmax_s_') # shape=(None, )
  68. self.q_target = tf.stop_gradient(q_target)
  69. with tf.variable_scope('q_eval'):
  70. a_indices = tf.stack([tf.range(tf.shape(self.a)[0], dtype=tf.int32), self.a], axis=1)
  71. self.q_eval_wrt_a = tf.gather_nd(params=self.q_eval, indices=a_indices) # shape=(None, )
  72. with tf.variable_scope('loss'):
  73. self.loss = tf.reduce_mean(tf.squared_difference(self.q_target, self.q_eval_wrt_a, name='TD_error'))
  74. with tf.variable_scope('train'):
  75. self._train_op = tf.train.RMSPropOptimizer(self.lr).minimize(self.loss)
  76. def store_transition(self, s, a, r, s_):
  77. if not hasattr(self, 'memory_counter'):
  78. self.memory_counter = 0
  79. transition = np.hstack((s, [a, r], s_))
  80. # replace the old memory with new memory
  81. index = self.memory_counter % self.memory_size
  82. self.memory[index, :] = transition
  83. self.memory_counter += 1
  84. def choose_action(self, observation):
  85. # to have batch dimension when feed into tf placeholder
  86. observation = observation[np.newaxis, :]
  87. if np.random.uniform() < self.epsilon:
  88. # forward feed the observation and get q value for every actions
  89. actions_value = self.sess.run(self.q_eval, feed_dict={
  90. self.s: observation})
  91. action = np.argmax(actions_value)
  92. else:
  93. action = np.random.randint(0, self.n_actions)
  94. return action
  95. def learn(self):
  96. # check to replace target parameters
  97. if self.learn_step_counter % self.replace_target_iter == 0:
  98. self.sess.run(self.target_replace_op)
  99. print('\ntarget_params_replaced\n')
  100. # sample batch memory from all memory
  101. if self.memory_counter > self.memory_size:
  102. sample_index = np.random.choice(self.memory_size, size=self.batch_size)
  103. else:
  104. sample_index = np.random.choice(self.memory_counter, size=self.batch_size)
  105. batch_memory = self.memory[sample_index, :]
  106. _, cost = self.sess.run(
  107. [self._train_op, self.loss],
  108. feed_dict={
  109. self.s: batch_memory[:, :self.n_features],
  110. self.a: batch_memory[:, self.n_features],
  111. self.r: batch_memory[:, self.n_features + 1],
  112. self.s_: batch_memory[:, -self.n_features:],
  113. })
  114. self.cost_his.append(cost)
  115. # increasing epsilon
  116. self.epsilon = self.epsilon + self.epsilon_increment if self.epsilon < self.epsilon_max else self.epsilon_max
  117. self.learn_step_counter += 1
  118. def plot_cost(self):
  119. import matplotlib.pyplot as plt
  120. plt.plot(np.arange(len(self.cost_his)), self.cost_his)
  121. plt.ylabel('Cost')
  122. plt.xlabel('training steps')
  123. plt.show()
  124. if __name__ == '__main__':
  125. DQN = DeepQNetwork(3, 4, output_graph=True)

DQN_CarPole.py:

  1. import gym
  2. from DQN import DeepQNetwork
  3. env = gym.make('CartPole-v0')
  4. env = env.unwrapped
  5. RL = DeepQNetwork(n_actions=env.action_space.n,
  6. n_features=env.observation_space.shape[0],
  7. learning_rate=0.01, e_greedy=0.9,
  8. replace_target_iter=100, memory_size=2000,
  9. e_greedy_increment=0.0008, )
  10. total_steps = 0 # 记录步数
  11. for i_episode in range(100):
  12. # 获取回合 i_episode 第一个 observation
  13. observation = env.reset()
  14. ep_r = 0
  15. while True:
  16. env.render() # 刷新环境
  17. action = RL.choose_action(observation) # 选行为
  18. observation_, reward, done, info = env.step(action) # 获取下一个 state
  19. x, x_dot, theta, theta_dot = observation_ # 细分开, 为了修改原配的 reward
  20. # x 是车的水平位移, 所以 r1 是车越偏离中心, 分越少
  21. # theta 是棒子离垂直的角度, 角度越大, 越不垂直. 所以 r2 是棒越垂直, 分越高
  22. r1 = (env.x_threshold - abs(x)) / env.x_threshold - 0.8
  23. r2 = (env.theta_threshold_radians - abs(theta)) / env.theta_threshold_radians - 0.5
  24. reward = r1 + r2 # 总 reward 是 r1 和 r2 的结合, 既考虑位置, 也考虑角度, 这样 DQN 学习更有效率
  25. # 保存这一组记忆
  26. RL.store_transition(observation, action, reward, observation_)
  27. if total_steps > 1000:
  28. RL.learn() # 学习
  29. ep_r += reward
  30. if done:
  31. print('episode: ', i_episode,
  32. 'ep_r: ', round(ep_r, 2),
  33. ' epsilon: ', round(RL.epsilon, 2))
  34. break
  35. observation = observation_
  36. total_steps += 1
  37. # 最后输出 cost 曲线
  38. RL.plot_cost()
  39. env.close()

运行DQN_CarPole.py
发现小车确实变得越来越稳定。

最后输出的损失函数曲线

在这里插入图片描述

三、Double DQN

无论是Q-learning还是DQN,更新Q值的的时候都会用到 m a x Q maxQ maxQ。

Q-learning:
Q ( s , a ) ← Q ( s , a ) + α [ r + γ m a x a ′ Q ( s ′ , a ′ ) − Q ( s , a ) ] Q(s,a)←Q(s,a)+α[r+\gamma max_{a’}Q(s^′,a^′)−Q(s,a)] Q(s,a)←Q(s,a)+α[r+γmaxa′Q(s′,a′)−Q(s,a)]

DQN:

在这里插入图片描述

使用max虽然可以快速让Q值向可能的优化目标靠拢,但是很容易过犹不及,导致过度估计(Over Estimation),所谓过度估计就是最终我们得到的算法模型有很大的偏差(bias),可能就会发现Q值都超级大。

DDQN通过解耦目标Q值动作的选择和目标Q值的计算这两步,来达到消除过度估计的问题。

在这里插入图片描述

DDQN更新:

在这里插入图片描述

在DDQN这里,不再是直接在目标Q网络里面找各个动作中最大Q值,而是先在当前Q网络中先找出最大Q值对应的动作。然后利用这个选择出来的动作在目标Q网络里面去计算目标Q值。

DoubleDQN.py

  1. import numpy as np
  2. import tensorflow as tf
  3. np.random.seed(1)
  4. tf.set_random_seed(1)
  5. class DoubleDQN:
  6. def __init__(
  7. self,
  8. n_actions,
  9. n_features,
  10. learning_rate=0.005,
  11. reward_decay=0.9,
  12. e_greedy=0.9,
  13. replace_target_iter=200,
  14. memory_size=3000,
  15. batch_size=32,
  16. e_greedy_increment=None,
  17. output_graph=False,
  18. ):
  19. self.n_actions = n_actions
  20. self.n_features = n_features
  21. self.lr = learning_rate
  22. self.gamma = reward_decay
  23. self.epsilon_max = e_greedy
  24. self.replace_target_iter = replace_target_iter
  25. self.memory_size = memory_size
  26. self.batch_size = batch_size
  27. self.epsilon_increment = e_greedy_increment
  28. self.epsilon = 0 if e_greedy_increment is not None else self.epsilon_max
  29. self.learn_step_counter = 0
  30. self.memory = np.zeros((self.memory_size, n_features * 2 + 2))
  31. self._build_net()
  32. t_params = tf.get_collection('target_net_params')
  33. e_params = tf.get_collection('eval_net_params')
  34. self.replace_target_op = [tf.assign(t, e) for t, e in zip(t_params, e_params)]
  35. self.sess = tf.Session()
  36. if output_graph:
  37. tf.summary.FileWriter("logs/", self.sess.graph)
  38. self.sess.run(tf.global_variables_initializer())
  39. self.cost_his = []
  40. def _build_net(self):
  41. def build_layers(s, c_names, n_l1, w_initializer, b_initializer):
  42. with tf.variable_scope('l1'):
  43. w1 = tf.get_variable('w1', [self.n_features, n_l1], initializer=w_initializer, collections=c_names)
  44. b1 = tf.get_variable('b1', [1, n_l1], initializer=b_initializer, collections=c_names)
  45. l1 = tf.nn.relu(tf.matmul(s, w1) + b1)
  46. with tf.variable_scope('l2'):
  47. w2 = tf.get_variable('w2', [n_l1, self.n_actions], initializer=w_initializer, collections=c_names)
  48. b2 = tf.get_variable('b2', [1, self.n_actions], initializer=b_initializer, collections=c_names)
  49. out = tf.matmul(l1, w2) + b2
  50. return out
  51. # ------------------ build evaluate_net ------------------
  52. self.s = tf.placeholder(tf.float32, [None, self.n_features], name='s') # input
  53. self.q_target = tf.placeholder(tf.float32, [None, self.n_actions], name='Q_target') # for calculating loss
  54. with tf.variable_scope('eval_net'):
  55. c_names, n_l1, w_initializer, b_initializer = \
  56. ['eval_net_params', tf.GraphKeys.GLOBAL_VARIABLES], 20, \
  57. tf.random_normal_initializer(0., 0.3), tf.constant_initializer(0.1) # config of layers
  58. self.q_eval = build_layers(self.s, c_names, n_l1, w_initializer, b_initializer)
  59. with tf.variable_scope('loss'):
  60. self.loss = tf.reduce_mean(tf.squared_difference(self.q_target, self.q_eval))
  61. with tf.variable_scope('train'):
  62. self._train_op = tf.train.RMSPropOptimizer(self.lr).minimize(self.loss)
  63. # ------------------ build target_net ------------------
  64. self.s_ = tf.placeholder(tf.float32, [None, self.n_features], name='s_') # input
  65. with tf.variable_scope('target_net'):
  66. c_names = ['target_net_params', tf.GraphKeys.GLOBAL_VARIABLES]
  67. self.q_next = build_layers(self.s_, c_names, n_l1, w_initializer, b_initializer)
  68. def store_transition(self, s, a, r, s_):
  69. if not hasattr(self, 'memory_counter'):
  70. self.memory_counter = 0
  71. transition = np.hstack((s, [a, r], s_))
  72. index = self.memory_counter % self.memory_size
  73. self.memory[index, :] = transition
  74. self.memory_counter += 1
  75. def choose_action(self, observation):
  76. observation = observation[np.newaxis, :]
  77. actions_value = self.sess.run(self.q_eval, feed_dict={
  78. self.s: observation})
  79. action = np.argmax(actions_value)
  80. if not hasattr(self, 'q'): # record action value it gets
  81. self.q = []
  82. self.running_q = 0
  83. self.running_q = self.running_q * 0.99 + 0.01 * np.max(actions_value)
  84. self.q.append(self.running_q)
  85. if np.random.uniform() > self.epsilon: # choosing action
  86. action = np.random.randint(0, self.n_actions)
  87. return action
  88. def learn(self):
  89. if self.learn_step_counter % self.replace_target_iter == 0:
  90. self.sess.run(self.replace_target_op)
  91. print('\ntarget_params_replaced\n')
  92. if self.memory_counter > self.memory_size:
  93. sample_index = np.random.choice(self.memory_size, size=self.batch_size)
  94. else:
  95. sample_index = np.random.choice(self.memory_counter, size=self.batch_size)
  96. batch_memory = self.memory[sample_index, :]
  97. q_next, q_eval4next = self.sess.run(
  98. [self.q_next, self.q_eval],
  99. feed_dict={
  100. self.s_: batch_memory[:, -self.n_features:], # next observation
  101. self.s: batch_memory[:, -self.n_features:]}) # next observation
  102. q_eval = self.sess.run(self.q_eval, {
  103. self.s: batch_memory[:, :self.n_features]})
  104. q_target = q_eval.copy()
  105. batch_index = np.arange(self.batch_size, dtype=np.int32)
  106. eval_act_index = batch_memory[:, self.n_features].astype(int)
  107. reward = batch_memory[:, self.n_features + 1]
  108. max_act4next = np.argmax(q_eval4next,
  109. axis=1) # the action that brings the highest value is evaluated by q_eval
  110. selected_q_next = q_next[batch_index, max_act4next] # Double DQN, select q_next depending on above actions
  111. q_target[batch_index, eval_act_index] = reward + self.gamma * selected_q_next
  112. _, self.cost = self.sess.run([self._train_op, self.loss],
  113. feed_dict={
  114. self.s: batch_memory[:, :self.n_features],
  115. self.q_target: q_target})
  116. self.cost_his.append(self.cost)
  117. self.epsilon = self.epsilon + self.epsilon_increment if self.epsilon < self.epsilon_max else self.epsilon_max
  118. self.learn_step_counter += 1
  119. def plot_cost(self):
  120. import matplotlib.pyplot as plt
  121. plt.plot(np.arange(len(self.cost_his)), self.cost_his)
  122. plt.ylabel('Cost')
  123. plt.xlabel('training steps')
  124. plt.show()

DoubleDQN_CarPole.py

  1. import tensorflow as tf
  2. import gym
  3. from DoubleDQN import DoubleDQN
  4. env = gym.make('CartPole-v0')
  5. env = env.unwrapped
  6. double_DQN = DoubleDQN(n_actions=env.action_space.n,
  7. n_features=env.observation_space.shape[0],
  8. memory_size=2000,
  9. e_greedy_increment=0.001,
  10. output_graph=True)
  11. total_steps = 0 # 记录步数
  12. for i_episode in range(100):
  13. # 获取回合 i_episode 第一个 observation
  14. observation = env.reset()
  15. ep_r = 0
  16. while True:
  17. env.render() # 刷新环境
  18. action = double_DQN.choose_action(observation) # 选行为
  19. observation_, reward, done, info = env.step(action) # 获取下一个 state
  20. x, x_dot, theta, theta_dot = observation_ # 细分开, 为了修改原配的 reward
  21. # x 是车的水平位移, 所以 r1 是车越偏离中心, 分越少
  22. # theta 是棒子离垂直的角度, 角度越大, 越不垂直. 所以 r2 是棒越垂直, 分越高
  23. r1 = (env.x_threshold - abs(x)) / env.x_threshold - 0.8
  24. r2 = (env.theta_threshold_radians - abs(theta)) / env.theta_threshold_radians - 0.5
  25. reward = r1 + r2 # 总 reward 是 r1 和 r2 的结合, 既考虑位置, 也考虑角度, 这样 DQN 学习更有效率
  26. # 保存这一组记忆
  27. double_DQN.store_transition(observation, action, reward, observation_)
  28. if total_steps > 1000:
  29. double_DQN.learn() # 学习
  30. ep_r += reward
  31. if done:
  32. print('episode: ', i_episode,
  33. 'ep_r: ', round(ep_r, 2),
  34. ' epsilon: ', round(double_DQN.epsilon, 2))
  35. break
  36. observation = observation_
  37. total_steps += 1
  38. # 最后输出 cost 曲线
  39. double_DQN.plot_cost()
  40. env.close()

输出损失函数结果

在这里插入图片描述

对比DQN,可以发现DoubleDQN效果明显好很多。

发表评论

表情:
评论列表 (有 0 条评论,197人围观)

还没有评论,来说两句吧...

相关阅读