꒰˃͈꒵˂͈꒱ write in front ꒰˃͈꒵˂͈꒱
ʕ̯•͡˔•̯᷅ʔ大家好，我是xiaoxie.希望你看完之后,有不足之处请多多谅解，让我们一起共同进步૮₍❀ᴗ͈ . ᴗ͈ აxiaoxieʕ̯•͡˔•̯᷅ʔ—CSDN博客
本文由xiaoxieʕ̯•͡˔•̯᷅ʔ 原创 CSDN 如需转载还请通知˶⍤⃝˶
个人主页：xiaoxieʕ̯•͡˔•̯᷅ʔ—CSDN博客系列专栏:机器学习系列专栏
"探索未来,掌握人工智能"🚀 点击加入我们的AI学习之旅，让技术变得有趣又易懂,点击跳转
**我的目标:"团团等我💪( ◡̀_◡́ ҂)" **( ⸝⸝⸝›ᴥ‹⸝⸝⸝ )欢迎各位→点赞👍 + 收藏⭐️ + 留言📝+关注（互三必回）!
一.AI序列决策问题
AI序列决策问题是指在人工智能领域中，智能体需要在一个序列的环境中做出一系列决策，以达到某个目标或最大化某种累积奖励的问题。这类问题通常涉及到强化学习，其中智能体通过与环境的交互来学习最优的行为策略。
1.序列决策问题的特点：
- 时间维度：决策不是一次性的，而是需要在一系列时间步骤中进行。每个决策都会影响后续的状态和可能的决策。
- 状态变化：智能体的每个决策都会使环境从一个状态转移到另一个状态。状态可以是环境的描述，如游戏的当前分数、机器人的位置等。
- 奖励反馈：智能体在每个时间步骤做出决策后，环境会提供一个奖励（或惩罚），这是对智能体决策好坏的反馈。
- 长期目标：智能体的目标通常是长期的，比如最大化累积奖励、达到最终的胜利状态或完成任务。
- 不确定性：智能体在做出决策时可能无法完全了解环境的全部特性，因此需要在不确定性中做出最优的选择。
2.解决序列决策问题的AI方法：
- 强化学习：通过智能体与环境的交互来学习最优策略。智能体通过尝试不同的行动并接收环境的奖励或惩罚来学习。
- 动态规划：一种基于模型的优化方法，通过预测未来的状态和奖励来计算当前行动的价值。
- 蒙特卡洛方法：通过随机模拟来估计行动的价值，适用于难以精确建模的环境。
- 时序差分学习：结合了动态规划和蒙特卡洛方法的特点，通过学习状态和行动之间的差异来更新价值估计。
- 深度学习：使用深度神经网络来近似复杂的价值函数或策略函数，尤其在状态空间高维且连续时表现出色。
3.序列决策问题的应用场景：
- 游戏AI：如棋类游戏、电子游戏等，智能体需要通过一系列行动来赢得比赛。
- 机器人控制：机器人需要根据环境的变化做出连续的移动和操作决策。
- 自动驾驶汽车：汽车需要根据路况和交通规则做出连续的驾驶决策。
- 资源管理：如电网管理、网络带宽分配等，需要根据实时数据做出一系列调度决策。
AI序列决策问题是人工智能中一个非常重要且活跃的研究领域，它不仅挑战着智能体在复杂环境中的学习能力，也推动了AI技术在多个领域的应用和发展。
二深度Q网络（Deep Q-Networks, DQN）解决序列决策问题
1.什么是DQN
**DQN属于DRL（深度强化学习）的一种，它是深度学习与Q学习的结合体。使用 Q-learning 因为采用
S-A
表格的局限性，当状态和行为的组合不可穷尽时，就无法通过查表的方式选取最优的Action了。这时候就该想到深度学习了，想通过深度学习找到最优解在很多情况下确实不太靠谱，但是找到一个无限逼近最优解的次优解，倒是没有问题的。因此DQN实际上，总体思路还是用的Q学习的思路，不过对于给定状态选取哪个动作所能得到的Q值，却是由一个深度神经网络来计算的了，其流程图如下：**
2.DNN如何训练
现在我们的选择哪个动作，是由DNN来做决定的，因此我们需要训练DNN以使其能达到令人满意的表现。这显然是一个监督学习的问题，那么训练集是什么，标签是什么，损失函数又是什么？
首先，我们DNN的输出值，自然是在给定状态的情况下，执行各action后能得到的Q值。然而事实上我们在很多情况下并不知道最优的Q值是什么，比如自动驾驶、围棋等情况，所以似乎我们没法给出标签。但是什么是不变的呢？Reward！
对状态s，执行动作a，那么得到的reward是一定的，而且是不变的！因此需要考虑从reward下手，让预测Q值和真实Q值的比较问题转换成让模型实质上在拟合reward的问题。
如果不能很好的理解，请看下面公式，下面公式中target是什么我们会在后面说到，这里先忽略它，那个红色的 θi- 我们也暂且将它当成 θi来看这个公式描述的就是模型的损失函数，大括号外面就是求一个均方差，我们主要看括号里面。前面被target标出来的地方是这一步得到的reward+下一状态所能得到的最大Q值，它们减去这一步的Q值，那么实际上它们的差就是实际reward减去现有模型认为在s下采取a时能得到的reward值。
现在的问题就已经转换为需要一组训练集，它能够提供一批四元组（s, a, r, s’），其中s’为s执行a后的下一个状态。如果能有这样一个四元组，就能够用来训练DNN了，这就是我们要介绍的Experience reply。
Experience Reply
前面提到我们需要一批四元组（s, a, r, s’）来进行训练，因此我们需要缓存一批这样的四元组到经验池中以供训练之用。由于每次执行一个动作后都能转移到下一个状态，并获得一个reward，因此我们每执行一次动作后都可以获得一个这样的四元组，也可以将这个四元组直接放入经验池中。
我们知道这种四元组之间是存在关联性的，因为状态的转移是连续的，如果直接按顺序取一批四元组作为训练集，那么是容易过拟合的，因为训练样本间不是独立的！为解决这个问题，我们可以简单地从经验池中随机抽取少量四元组作为一个batch，这样既保证了训练样本是独立同分布的，也使得每个batch样本量不大，能加快训练速度。 训练的伪代码为:
Target Network
上面的代码似乎已经能够正常运行了，为什么又冒出一个target network呢？回想下前面那个公式，这里重新搬到下面来，是不是之前说target和θi-都先忽略，现在就解释一下为什么
** **
**这个公式里θi-和θi肯定是有区别的，不然也不会使用两个符号了。事实上，我们需要设计两个DNN，它们结构完全一样，但是参数不一样，即神经网络中各层的权重、偏置等，一个的参数是θi-，而另一个是θi。我们每次迭代中，更新的是θi而不更新θi-，且规定每运行C步后让θi-
=
θi。而其θi-所在的网络就被称为target network。
为什么要弄这么奇怪的东西？
这也是为了防止过拟合。试想如果只有一个神经网络，那么它就在会不停地更新，那么它所追求的目标是在一直改变的，即在θ改变的时候，不止Q(s, a)变了，max *Q(s’, a’)*也变了。这样的好处是使得上面公式中target所标注的部分是暂时固定的，我们不断更新
θ
追逐的是一个固定的目标，而不是一直改变的目标。
如果做了这种考虑，那么伪代码就要做一点点修改了，修改后如下，大家可以对比着看： **
三.Python实现DQN算法训练过程
在DQN中，Q值表中表示的是当前已学习到的经验。而根据公式计算出的 Q 值是agent通过与环境交互及自身的经验总结得到的一个分数（即：目标 Q 值）。最后使用目标 Q 值(target_q)去更新原来旧的 Q 值(q)。而目标 Q 值与旧的 Q 值的对应关系，正好是监督学习神经网络中结果值与输出值的对应关系。
所以，loss = (target_q - q)^2
即：整个训练过程其实就是 Q 值(q)向目标 Q 值(target_q)逼近的过程。
用TensorFlow实现的Deep Q-Network（DQN）的例子
import tensorflow as tf
import numpy as np
from collections import deque
import random
class DeepQNetwork:
r = np.array([[-1, -1, -1, -1, 0, -1],
[-1, -1, -1, 0, -1, 100.0],
[-1, -1, -1, 0, -1, -1],
[-1, 0, 0, -1, 0, -1],
[0, -1, -1, 1, -1, 100],
[-1, 0, -1, -1, 0, 100],
])
# 执行步数。
step_index = 0
# 状态数。
state_num = 6
# 动作数。
action_num = 6
# 训练之前观察多少步。
OBSERVE = 1000.
# 选取的小批量训练样本数。
BATCH = 20
# epsilon 的最小值，当 epsilon 小于该值时，将不在随机选择行为。
FINAL_EPSILON = 0.0001
# epsilon 的初始值，epsilon 逐渐减小。
INITIAL_EPSILON = 0.1
# epsilon 衰减的总步数。
EXPLORE = 3000000.
# 探索模式计数。
epsilon = 0
# 训练步数统计。
learn_step_counter = 0
# 学习率。
learning_rate = 0.001
# γ经验折损率。
gamma = 0.9
# 记忆上限。
memory_size = 5000
# 当前记忆数。
memory_counter = 0
# 保存观察到的执行过的行动的存储器，即：曾经经历过的记忆。
replay_memory_store = deque()
# 生成一个状态矩阵（6 X 6），每一行代表一个状态。
state_list = None
# 生成一个动作矩阵。
action_list = None
# q_eval 网络。
q_eval_input = None
action_input = None
q_target = None
q_eval = None
predict = None
loss = None
train_op = None
cost_his = None
reward_action = None
# tensorflow 会话。
session = None
def __init__(self, learning_rate=0.001, gamma=0.9, memory_size=5000):
self.learning_rate = learning_rate
self.gamma = gamma
self.memory_size = memory_size
# 初始化成一个 6 X 6 的状态矩阵。
self.state_list = np.identity(self.state_num)
# 初始化成一个 6 X 6 的动作矩阵。
self.action_list = np.identity(self.action_num)
# 创建神经网络。
self.create_network()
# 初始化 tensorflow 会话。
self.session = tf.InteractiveSession()
# 初始化 tensorflow 参数。
self.session.run(tf.initialize_all_variables())
# 记录所有 loss 变化。
self.cost_his = []
def create_network(self):
"""
创建神经网络。
:return:
"""
self.q_eval_input = tf.placeholder(shape=[None, self.state_num], dtype=tf.float32)
self.action_input = tf.placeholder(shape=[None, self.action_num], dtype=tf.float32)
self.q_target = tf.placeholder(shape=[None], dtype=tf.float32)
neuro_layer_1 = 3
w1 = tf.Variable(tf.random_normal([self.state_num, neuro_layer_1]))
b1 = tf.Variable(tf.zeros([1, neuro_layer_1]) + 0.1)
l1 = tf.nn.relu(tf.matmul(self.q_eval_input, w1) + b1)
w2 = tf.Variable(tf.random_normal([neuro_layer_1, self.action_num]))
b2 = tf.Variable(tf.zeros([1, self.action_num]) + 0.1)
self.q_eval = tf.matmul(l1, w2) + b2
# 取出当前动作的得分。
self.reward_action = tf.reduce_sum(tf.multiply(self.q_eval, self.action_input), reduction_indices=1)
self.loss = tf.reduce_mean(tf.square((self.q_target - self.reward_action)))
self.train_op = tf.train.GradientDescentOptimizer(self.learning_rate).minimize(self.loss)
self.predict = tf.argmax(self.q_eval, 1)
def select_action(self, state_index):
"""
根据策略选择动作。
:param state_index: 当前状态。
:return:
"""
current_state = self.state_list[state_index:state_index + 1]
if np.random.uniform() < self.epsilon:
current_action_index = np.random.randint(0, self.action_num)
else:
actions_value = self.session.run(self.q_eval, feed_dict={self.q_eval_input: current_state})
action = np.argmax(actions_value)
current_action_index = action
# 开始训练后，在 epsilon 小于一定的值之前，将逐步减小 epsilon。
if self.step_index > self.OBSERVE and self.epsilon > self.FINAL_EPSILON:
self.epsilon -= (self.INITIAL_EPSILON - self.FINAL_EPSILON) / self.EXPLORE
return current_action_index
def save_store(self, current_state_index, current_action_index, current_reward, next_state_index, done):
"""
保存记忆。
:param current_state_index: 当前状态 index。
:param current_action_index: 动作 index。
:param current_reward: 奖励。
:param next_state_index: 下一个状态 index。
:param done: 是否结束。
:return:
"""
current_state = self.state_list[current_state_index:current_state_index + 1]
current_action = self.action_list[current_action_index:current_action_index + 1]
next_state = self.state_list[next_state_index:next_state_index + 1]
# 记忆动作(当前状态， 当前执行的动作， 当前动作的得分，下一个状态)。
self.replay_memory_store.append((
current_state,
current_action,
current_reward,
next_state,
done))
# 如果超过记忆的容量，则将最久远的记忆移除。
if len(self.replay_memory_store) > self.memory_size:
self.replay_memory_store.popleft()
self.memory_counter += 1
def step(self, state, action):
"""
执行动作。
:param state: 当前状态。
:param action: 执行的动作。
:return:
"""
reward = self.r[state][action]
next_state = action
done = False
if action == 5:
done = True
return next_state, reward, done
def experience_replay(self):
"""
记忆回放。
:return:
"""
# 随机选择一小批记忆样本。
batch = self.BATCH if self.memory_counter > self.BATCH else self.memory_counter
minibatch = random.sample(self.replay_memory_store, batch)
batch_state = None
batch_action = None
batch_reward = None
batch_next_state = None
batch_done = None
for index in range(len(minibatch)):
if batch_state is None:
batch_state = minibatch[index][0]
elif batch_state is not None:
batch_state = np.vstack((batch_state, minibatch[index][0]))
if batch_action is None:
batch_action = minibatch[index][1]
elif batch_action is not None:
batch_action = np.vstack((batch_action, minibatch[index][1]))
if batch_reward is None:
batch_reward = minibatch[index][2]
elif batch_reward is not None:
batch_reward = np.vstack((batch_reward, minibatch[index][2]))
if batch_next_state is None:
batch_next_state = minibatch[index][3]
elif batch_next_state is not None:
batch_next_state = np.vstack((batch_next_state, minibatch[index][3]))
if batch_done is None:
batch_done = minibatch[index][4]
elif batch_done is not None:
batch_done = np.vstack((batch_done, minibatch[index][4]))
# q_next：下一个状态的 Q 值。
q_next = self.session.run([self.q_eval], feed_dict={self.q_eval_input: batch_next_state})
q_target = []
for i in range(len(minibatch)):
# 当前即时得分。
current_reward = batch_reward[i][0]
# # 游戏是否结束。
# current_done = batch_done[i][0]
# 更新 Q 值。
q_value = current_reward + self.gamma * np.max(q_next[0][i])
# 当得分小于 0 时，表示走了不可走的位置。
if current_reward < 0:
q_target.append(current_reward)
else:
q_target.append(q_value)
_, cost, reward = self.session.run([self.train_op, self.loss, self.reward_action],
feed_dict={self.q_eval_input: batch_state,
self.action_input: batch_action,
self.q_target: q_target})
self.cost_his.append(cost)
# if self.step_index % 1000 == 0:
# print("loss:", cost)
self.learn_step_counter += 1
def train(self):
"""
训练。
:return:
"""
# 初始化当前状态。
current_state = np.random.randint(0, self.action_num - 1)
self.epsilon = self.INITIAL_EPSILON
while True:
# 选择动作。
action = self.select_action(current_state)
# 执行动作，得到：下一个状态，执行动作的得分，是否结束。
next_state, reward, done = self.step(current_state, action)
# 保存记忆。
self.save_store(current_state, action, reward, next_state, done)
# 先观察一段时间累积足够的记忆在进行训练。
if self.step_index > self.OBSERVE:
self.experience_replay()
if self.step_index > 10000:
break
if done:
current_state = np.random.randint(0, self.action_num - 1)
else:
current_state = next_state
self.step_index += 1
def pay(self):
"""
运行并测试。
:return:
"""
self.train()
# 显示 R 矩阵。
print(self.r)
for index in range(5):
start_room = index
print("#############################", "Agent 在", start_room, "开始行动", "#############################")
current_state = start_room
step = 0
target_state = 5
while current_state != target_state:
out_result = self.session.run(self.q_eval, feed_dict={
self.q_eval_input: self.state_list[current_state:current_state + 1]})
next_state = np.argmax(out_result[0])
print("Agent 由", current_state, "号房间移动到了", next_state, "号房间")
current_state = next_state
step += 1
print("Agent 在", start_room, "号房间开始移动了", step, "步到达了目标房间 5")
print("#############################", "Agent 在", 5, "结束行动", "#############################")
if __name__ == "__main__":
q_network = DeepQNetwork()
q_network.pay()
**这个DQN实现是一个简化的版本，简单介绍一下。在实际应用中，可能需要对网络结构、训练过程和超参数进行调整和优化。此外，代码中的
r
矩阵定义了状态转移和奖励，这是一个特定于问题的矩阵，需要根据具体问题进行设计**
以上就算关于DQN的简单介绍.感谢你的阅读.
版权归原作者 小谢在努力 所有， 如有侵权，请联系我们删除。