代碼可以參見(jiàn)，本文我做了一些改動(dòng)
目前，強(qiáng)化學(xué)習(xí)中很火的當(dāng)屬Q(mào)-Learning了，關(guān)于Q-Learning的具體介紹請(qǐng)參加。從上一篇文章中，我們可以看到，Q table可以看做Q-Learning的大腦，Q table對(duì)應(yīng)了一張state-action的表，但在實(shí)際應(yīng)用中，state和action往往很多，內(nèi)存很難裝下Q table，因此需要用神經(jīng)網(wǎng)絡(luò)替代Q table。
訓(xùn)練樣本
首先要解決的問(wèn)題是如何獲取訓(xùn)練樣本。在 DQN 中有 Experience Replay 的概念，就是經(jīng)驗(yàn)回放。即先讓agent去探索環(huán)境，將經(jīng)驗(yàn)（記憶）累積到一定程度，再隨機(jī)抽取出一批樣本進(jìn)行訓(xùn)練。為什么要隨機(jī)抽??？因?yàn)閍gent去探索環(huán)境時(shí)采集到的樣本是一個(gè)時(shí)間序列，樣本之間具有連續(xù)性，如果每次得到樣本就更新Q值，受樣本分布影響，會(huì)對(duì)收斂造成影響。
這里我們聯(lián)想到數(shù)據(jù)庫(kù)領(lǐng)域，我們需要使用benchmark去回放得到不同的action對(duì)應(yīng)的Q值。增強(qiáng)學(xué)習(xí)是試錯(cuò)學(xué)習(xí)(Trail-and-error)，由于沒(méi)有直接的指導(dǎo)信息，agent要以不斷與環(huán)境進(jìn)行交互，通過(guò)試錯(cuò)的方式來(lái)獲得最佳策略。因此一開(kāi)始可以看做是盲目的、隨機(jī)的試驗(yàn)，但是根據(jù)反饋的reward來(lái)優(yōu)化損失函數(shù)可以使得我們想要的Q table慢慢收斂。
損失函數(shù)
上面提到了損失函數(shù)，那么如何選取呢。在DQN中，Q值表中表示的是當(dāng)前已學(xué)習(xí)到的經(jīng)驗(yàn)。而根據(jù)公式計(jì)算出的 Q 值是agent通過(guò)與環(huán)境交互及自身的經(jīng)驗(yàn)總結(jié)得到的一個(gè)分?jǐn)?shù)（即：目標(biāo) Q 值）。最后使用目標(biāo) Q 值(target_q)去更新原來(lái)舊的 Q 值(q)。而目標(biāo) Q 值與舊的 Q 值的對(duì)應(yīng)關(guān)系，正好是監(jiān)督學(xué)習(xí)神經(jīng)網(wǎng)絡(luò)中結(jié)果值與輸出值的對(duì)應(yīng)關(guān)系。
所以，loss = (target_q - q)^2
即：整個(gè)訓(xùn)練過(guò)程其實(shí)就是 Q 值(q)向目標(biāo) Q 值(target_q)逼近的過(guò)程。
代碼實(shí)現(xiàn)
看代碼是最直觀的，我先給出整個(gè)代碼流程，然后再詳細(xì)解釋。下面是全部代碼：
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],
])
# 執(zhí)行步數(shù)。
step_index = 0
# 狀態(tài)數(shù)。
state_num = 6
# 動(dòng)作數(shù)。
action_num = 6
# 訓(xùn)練之前觀察多少步。
OBSERVE = 1000.
# 選取的小批量訓(xùn)練樣本數(shù)。
BATCH = 20
# epsilon 的最小值，當(dāng) epsilon 小于該值時(shí)，將不在隨機(jī)選擇行為。
FINAL_EPSILON = 0.0001
# epsilon 的初始值，epsilon 逐漸減小。
INITIAL_EPSILON = 0.1
# epsilon 衰減的總步數(shù)。
EXPLORE = 3000000.
# 探索模式計(jì)數(shù)。
epsilon = 0
# 訓(xùn)練步數(shù)統(tǒng)計(jì)。
learn_step_counter = 0
# 學(xué)習(xí)率。
learning_rate = 0.001
# γ經(jīng)驗(yàn)折損率。
gamma = 0.9
# 記憶上限。
memory_size = 5000
# 當(dāng)前記憶數(shù)。
memory_counter = 0
# 保存觀察到的執(zhí)行過(guò)的行動(dòng)的存儲(chǔ)器，即：曾經(jīng)經(jīng)歷過(guò)的記憶。
replay_memory_store = deque()
# 生成一個(gè)狀態(tài)矩陣（6 X 6），每一行代表一個(gè)狀態(tài)。
state_list = None
# 生成一個(gè)動(dòng)作矩陣。
action_list = None
# q_eval 網(wǎng)絡(luò)。
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 會(huì)話。
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
# 初始化成一個(gè) 6 X 6 的狀態(tài)矩陣。
self.state_list = np.identity(self.state_num)
# 初始化成一個(gè) 6 X 6 的動(dòng)作矩陣。
self.action_list = np.identity(self.action_num)
# 創(chuàng)建神經(jīng)網(wǎng)絡(luò)。
self.create_network()
# 初始化 tensorflow 會(huì)話。
self.session = tf.InteractiveSession()
# 初始化 tensorflow 參數(shù)。
self.session.run(tf.initialize_all_variables())
# 記錄所有 loss 變化。
self.cost_his = []
def create_network(self):
"""
創(chuàng)建神經(jīng)網(wǎng)絡(luò)。
: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
# 取出當(dāng)前動(dòng)作的得分。
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):
"""
根據(jù)策略選擇動(dòng)作。
:param state_index: 當(dāng)前狀態(tài)。
: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
# 開(kāi)始訓(xùn)練后，在 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: 當(dāng)前狀態(tài) index。
:param current_action_index: 動(dòng)作 index。
:param current_reward: 獎(jiǎng)勵(lì)。
:param next_state_index: 下一個(gè)狀態(tài) index。
:param done: 是否結(jié)束。
: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]
# 記憶動(dòng)作(當(dāng)前狀態(tài)， 當(dāng)前執(zhí)行的動(dòng)作， 當(dāng)前動(dòng)作的得分，下一個(gè)狀態(tài))。
self.replay_memory_store.append((
current_state,
current_action,
current_reward,
next_state,
done))
# 如果超過(guò)記憶的容量，則將最久遠(yuǎn)的記憶移除。
if len(self.replay_memory_store) > self.memory_size:
self.replay_memory_store.popleft()
self.memory_counter += 1
def step(self, state, action):
"""
執(zhí)行動(dòng)作。
:param state: 當(dāng)前狀態(tài)。
:param action: 執(zhí)行的動(dòng)作。
: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:
"""
# 隨機(jī)選擇一小批記憶樣本。
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：下一個(gè)狀態(tài)的 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)):
# 當(dāng)前即時(shí)得分。
current_reward = batch_reward[i][0]
# # 游戲是否結(jié)束。
# current_done = batch_done[i][0]
# 更新 Q 值。
q_value = current_reward + self.gamma * np.max(q_next[0][i])
# 當(dāng)?shù)梅中∮?0 時(shí)，表示走了不可走的位置。
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):
"""
訓(xùn)練。
:return:
"""
# 初始化當(dāng)前狀態(tài)。
current_state = np.random.randint(0, self.action_num - 1)
self.epsilon = self.INITIAL_EPSILON
while True:
# 選擇動(dòng)作。
action = self.select_action(current_state)
# 執(zhí)行動(dòng)作，得到：下一個(gè)狀態(tài)，執(zhí)行動(dòng)作的得分，是否結(jié)束。
next_state, reward, done = self.step(current_state, action)
# 保存記憶。
self.save_store(current_state, action, reward, next_state, done)
# 先觀察一段時(shí)間累積足夠的記憶在進(jìn)行訓(xùn)練。
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):
"""
運(yùn)行并測(cè)試。
:return:
"""
self.train()
# 顯示 R 矩陣。
print(self.r)
for index in range(5):
start_room = index
print("#############################", "Agent 在", start_room, "開(kāi)始行動(dòng)", "#############################")
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, "號(hào)房間移動(dòng)到了", next_state, "號(hào)房間")
current_state = next_state
step += 1
print("Agent 在", start_room, "號(hào)房間開(kāi)始移動(dòng)了", step, "步到達(dá)了目標(biāo)房間 5")
print("#############################", "Agent 在", 5, "結(jié)束行動(dòng)", "#############################")
if __name__ == "__main__":
q_network = DeepQNetwork()
q_network.pay()代碼流程及模塊詳解
1.創(chuàng)建DQN類
全部功能到在這個(gè)類里。最后的main里會(huì)調(diào)用DQN類的pay()來(lái)實(shí)現(xiàn)全部功能。
import tensorflow as tf
import numpy as np
from collections import deque
import random
class DeepQNetwork:
****
2.創(chuàng)建神經(jīng)網(wǎng)絡(luò)
然后，創(chuàng)建一個(gè)神經(jīng)網(wǎng)絡(luò)，并使用該神經(jīng)網(wǎng)絡(luò)來(lái)替換掉 Q 值表（上一篇中的 Q 矩陣）。神經(jīng)網(wǎng)絡(luò)的輸入是 Agent 當(dāng)前的狀態(tài)，輸出是 Agent 當(dāng)前狀態(tài)可以執(zhí)行的動(dòng)作的 Q 值表。由于總共有 6 個(gè)狀態(tài)和 6 種動(dòng)作，所以，這里將創(chuàng)建一個(gè)簡(jiǎn)單 3 層的神經(jīng)網(wǎng)絡(luò)，輸入層的參數(shù)是 6 個(gè)和輸出層輸出 6 個(gè)值，運(yùn)行并調(diào)試好參數(shù)，確認(rèn)能正常運(yùn)行。
單獨(dú)的測(cè)試代碼如下。完整程序中大體和下面的測(cè)試代碼類似，需要注意的是下面的測(cè)試代碼train的是optimizer，可以看做是讓loss function的loss最小化。
import tensorflow as tf
import numpy as np
input_num = 6
output_num = 6
x_data = np.linspace(-1, 1, 300).reshape((-1, input_num)) # 轉(zhuǎn)為列向量
noise = np.random.normal(0, 0.05, x_data.shape)
y_data = np.square(x_data) + 0.5 + noise
xs = tf.placeholder(tf.float32, [None, input_num]) # 樣本數(shù)未知，特征數(shù)為 6，占位符最后要以字典形式在運(yùn)行中填入
ys = tf.placeholder(tf.float32, [None, output_num])
neuro_layer_1 = 3
w1 = tf.Variable(tf.random_normal([input_num, neuro_layer_1]))
b1 = tf.Variable(tf.zeros([1, neuro_layer_1]) + 0.1)
l1 = tf.nn.relu(tf.matmul(xs, w1) + b1)
neuro_layer_2 = output_num
w2 = tf.Variable(tf.random_normal([neuro_layer_1, neuro_layer_2]))
b2 = tf.Variable(tf.zeros([1, neuro_layer_2]) + 0.1)
l2 = tf.matmul(l1, w2) + b2
# reduction_indices=[0] 表示將列數(shù)據(jù)累加到一起。
# reduction_indices=[1] 表示將行數(shù)據(jù)累加到一起。
loss = tf.reduce_mean(tf.reduce_sum(tf.square((ys - l2)), reduction_indices=[1]))
# 選擇梯度下降法
train = tf.train.GradientDescentOptimizer(0.001).minimize(loss)
# train = tf.train.AdamOptimizer(1e-1).minimize(loss)
init = tf.initialize_all_variables()
sess = tf.Session()
sess.run(init)
for i in range(100000):
sess.run(train, feed_dict={xs: x_data, ys: y_data})
if i % 1000 == 0:
print(sess.run(loss, feed_dict={xs: x_data, ys: y_data}))執(zhí)行后 loss 一直持續(xù)減少，確認(rèn)該神經(jīng)網(wǎng)絡(luò)正常運(yùn)行就行了：
確認(rèn)正常后，開(kāi)始實(shí)現(xiàn) DeepQNetwork 類中的 def create_network(self) 函數(shù)：
def create_network(self):
"""
創(chuàng)建神經(jīng)網(wǎng)絡(luò)。
: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
# 取出當(dāng)前動(dòng)作的得分。
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)這里說(shuō)明一下 loss 的計(jì)算，由于狀態(tài)是根據(jù)圖 3.1 的矩陣的方式顯示的，比如，當(dāng)前狀態(tài)如果是在 1 號(hào)房間，則輸入?yún)?shù)（q_eval_input）的值是：[[0, 1, 0, 0, 0, 0]]。由于 Agent 執(zhí)行了動(dòng)作 3，也就是移動(dòng)到了 3 號(hào)房間，所以 Agent 的動(dòng)作參數(shù)（action_input）的值是：[[0, 0, 0, 1, 0, 0]]。因?yàn)樯窠?jīng)網(wǎng)絡(luò)的輸出結(jié)果（q_eval）是 Agent 當(dāng)前狀態(tài)下可執(zhí)行的動(dòng)作的價(jià)值，由于每個(gè)狀態(tài)都有 6 個(gè)動(dòng)作，而狀態(tài)數(shù)也是 6 個(gè)，所以神經(jīng)網(wǎng)絡(luò)的輸出結(jié)果（q_eval）與輸入?yún)?shù)是一樣的，所以輸出的格式也一樣，假設(shè)輸出結(jié)果（q_eval）是：[[0.81, 0.5, 0.24, 0.513, 0.9, 0.71]]
代碼中的
就是矩陣的點(diǎn)積，也就是每個(gè)元素分別相乘。這里表示的就是獲得 Agent 執(zhí)行了 action_input 的價(jià)值（Q 值）。也就是 q = q_eval * action_input = [[0, 0, 0, 0.513, 0, 0]]。所以：
self.loss = tf.reduce_mean(tf.square((self.q_target - self.reward_action)))
就相當(dāng)于：loss = ((1.03 - 0.513)^2) / 1 = 0.2672
3.搜索動(dòng)作
這里是 DQN 需要注意的地方之一，這里的方法將直接影響到 DQN 是否可以收斂，或者是否是陷入局部最小值等情況?，F(xiàn)在在這里選擇了最直接的方法，使用隨機(jī)的方式來(lái)選擇行動(dòng)。使用選擇的方式來(lái)選擇行動(dòng)，可以讓 Agent 能得到更多的探索機(jī)會(huì)，這樣在訓(xùn)練時(shí)才能有效的跳出陷入局部最小值的情況，當(dāng)訓(xùn)練時(shí)，可以減少探索機(jī)會(huì)。
流程如下：
1.初始化 epsilon 變量，并設(shè)置它的最小值（FINAL_EPSILON）與最大值（INITIAL_EPSILON），并將 epsilon 的初始值設(shè)置成 INITIAL_EPSILON。
2.隨機(jī)生成一個(gè)數(shù) n。
3.判斷 n 是否小于 epsilon，如果 n 小于 epsilon 則轉(zhuǎn)到 4，否則轉(zhuǎn)到 5。
4.使用隨機(jī)策略（增加探索機(jī)會(huì)）。
隨機(jī)選擇一個(gè)在 Agent 當(dāng)前狀態(tài)下可以執(zhí)行的動(dòng)作。
5.使用神經(jīng)網(wǎng)絡(luò)直接計(jì)算出結(jié)果（實(shí)際應(yīng)用時(shí)也是應(yīng)用這方法）。
神經(jīng)網(wǎng)絡(luò)會(huì)輸出在當(dāng)前狀態(tài)下所有動(dòng)作的 Q 值，選擇其中最有價(jià)值（Q 值最大）的動(dòng)作返回。
6.判斷是否開(kāi)始訓(xùn)練，如果是，則逐步減少 epsilon 來(lái)減少探索機(jī)會(huì)，否則跳過(guò)。
開(kāi)始實(shí)現(xiàn) DeepQNetwork 類中的 def select_action(self, state_index) 函數(shù)：
def select_action(self, state_index):
"""
根據(jù)策略選擇動(dòng)作。
:param state_index: 當(dāng)前狀態(tài)。
: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
# 開(kāi)始訓(xùn)練后，在 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_index4.執(zhí)行動(dòng)作
這里就是取得游戲是否結(jié)束狀態(tài)，動(dòng)作獎(jiǎng)勵(lì)和下一個(gè)狀態(tài)并返回就可以了。
開(kāi)始實(shí)現(xiàn) DeepQNetwork 類中的 def step(self, state, action) 函數(shù)：
def step(self, state, action):
"""
執(zhí)行動(dòng)作。
:param state: 當(dāng)前狀態(tài)。
:param action: 執(zhí)行的動(dòng)作。
:return:
"""
reward = self.r[state][action]
next_state = action
done = False
if action == 5:
done = True
return next_state, reward, done5.保存記憶
這里使用了一個(gè)先進(jìn)先出的隊(duì)列，設(shè)置好隊(duì)列的 size，直接將“當(dāng)前狀態(tài)”、“執(zhí)行動(dòng)作”、“獎(jiǎng)勵(lì)分?jǐn)?shù)”、“下一個(gè)狀態(tài)”和“游戲是否結(jié)束”保存進(jìn)去就行了。
開(kāi)始實(shí)現(xiàn) DeepQNetwork 類中的 def save_store(self, current_state_index, current_action_index, current_reward, next_state_index, done) 函數(shù)：
def save_store(self, current_state_index, current_action_index, current_reward, next_state_index, done):
"""
保存記憶。
:param current_state_index: 當(dāng)前狀態(tài) index。
:param current_action_index: 動(dòng)作 index。
:param current_reward: 獎(jiǎng)勵(lì)。
:param next_state_index: 下一個(gè)狀態(tài) index。
:param done: 是否結(jié)束。
: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]
# 記憶動(dòng)作(當(dāng)前狀態(tài)， 當(dāng)前執(zhí)行的動(dòng)作， 當(dāng)前動(dòng)作的得分，下一個(gè)狀態(tài))。
self.replay_memory_store.append((
current_state,
current_action,
current_reward,
next_state,
done))
# 如果超過(guò)記憶的容量，則將最久遠(yuǎn)的記憶移除。
if len(self.replay_memory_store) > self.memory_size:
self.replay_memory_store.popleft()
self.memory_counter += 16.記憶回放
這是 DQN 的重點(diǎn)之一，在記憶池里隨機(jī)抽取出一小批的數(shù)據(jù)當(dāng)做訓(xùn)練樣本，并計(jì)算出目標(biāo) Q 值來(lái)訓(xùn)練神經(jīng)網(wǎng)絡(luò)。
流程如下：
1. 初始化時(shí)先設(shè)置抽取的樣本數(shù)。
2. 從記憶池里隨機(jī)抽取出一批樣本。
3. 由于每條樣本中，都保存有當(dāng)時(shí)的數(shù)據(jù)（當(dāng)前狀態(tài)，動(dòng)作，獎(jiǎng)勵(lì)分?jǐn)?shù)，下一個(gè)狀態(tài)，是否結(jié)束），所以，為了計(jì)算出這些樣本數(shù)據(jù)的目標(biāo) Q 值，就必須先取出樣本中“下一個(gè)狀態(tài)（next_state）”（注意：這里取到的是所有這批樣本的“下一個(gè)狀態(tài)”的列表?。?
4. 將 next_state （這是批數(shù)據(jù)?。。┊?dāng)做參數(shù)傳入神經(jīng)網(wǎng)絡(luò)，得到 Agent 在 next_state 狀態(tài)時(shí)所有可執(zhí)行的動(dòng)作的 Q 值表（q_next），q_next 表示這批樣本中所有的 next_state 狀態(tài)的 Q 值表的集合。
5. 現(xiàn)在，已經(jīng)拿到了
Agent 當(dāng)時(shí)的狀態(tài)（state），
當(dāng)時(shí)的動(dòng)作（action），
當(dāng)時(shí)的狀態(tài)（state）下執(zhí)行動(dòng)作（action）得到的獎(jiǎng)勵(lì)R(state, action)，
當(dāng)時(shí)的狀態(tài)（state）下執(zhí)行動(dòng)作（action）后的狀態(tài)（next_state）下所有可執(zhí)行的動(dòng)作的 Q 值表（q_next）。
現(xiàn)在就可以使用上面提到的公式來(lái)計(jì)算出目標(biāo) Q 值Q(state, action)。
Q(state, action) = R(state, action) + Gamma * Max{q_next}
6. 根據(jù)游戲狀態(tài)判斷，當(dāng)前選擇的動(dòng)作是否是違規(guī)（不可執(zhí)行）的動(dòng)作，如果是，則不做經(jīng)驗(yàn)計(jì)算，直接扣除分?jǐn)?shù)，否則使用上面的公式來(lái)計(jì)算出Q(state, action)。
7. 將計(jì)算得到的所有樣本的 Q(state, action) 保存到集合中（q_target）。
8. 將這批樣本的當(dāng)前狀態(tài)的集合，動(dòng)作的集合與 q_target 傳入神經(jīng)網(wǎng)絡(luò)并進(jìn)行訓(xùn)練。
特別注意第 6 條的內(nèi)容，如果這里處理不好，一樣會(huì)得不到結(jié)果的，具體原因可以看上一篇 。
開(kāi)始實(shí)現(xiàn) DeepQNetwork 類中的 def experience_replay(self)函數(shù)：
def experience_replay(self):
"""
記憶回放。
:return:
"""
# 隨機(jī)選擇一小批記憶樣本。
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：下一個(gè)狀態(tài)的 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)):
# 當(dāng)前即時(shí)得分。
current_reward = batch_reward[i][0]
# # 游戲是否結(jié)束。
# current_done = batch_done[i][0]
# 更新 Q 值。
q_value = current_reward + self.gamma * np.max(q_next[0][i])
# 當(dāng)?shù)梅中∮?0 時(shí)，表示走了不可走的位置。
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 += 17.訓(xùn)練
訓(xùn)練就是把上面的過(guò)程串在一起。實(shí)現(xiàn) DeepQNetwork 類中的 def train(self)函數(shù)：
def train(self):
"""
訓(xùn)練。
:return:
"""
# 初始化當(dāng)前狀態(tài)。
current_state = np.random.randint(0, self.action_num - 1)
self.epsilon = self.INITIAL_EPSILON
while True:
# 選擇動(dòng)作。
action = self.select_action(current_state)
# 執(zhí)行動(dòng)作，得到：下一個(gè)狀態(tài)，執(zhí)行動(dòng)作的得分，是否結(jié)束。
next_state, reward, done = self.step(current_state, action)
# 保存記憶。
self.save_store(current_state, action, reward, next_state, done)
# 先觀察一段時(shí)間累積足夠的記憶在進(jìn)行訓(xùn)練。
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 += 18.執(zhí)行并測(cè)試訓(xùn)練結(jié)果
實(shí)現(xiàn) DeepQNetwork 類中的 def pay(self)函數(shù)：
def pay(self):
"""
運(yùn)行并測(cè)試。
:return:
"""
self.train()
# 顯示 R 矩陣。
print(self.r)
for index in range(5):
start_room = index
print("#############################", "Agent 在", start_room, "開(kāi)始行動(dòng)", "#############################")
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, "號(hào)房間移動(dòng)到了", next_state, "號(hào)房間")
current_state = next_state
step += 1
print("Agent 在", start_room, "號(hào)房間開(kāi)始移動(dòng)了", step, "步到達(dá)了目標(biāo)房間 5")
print("#############################", "Agent 在", 5, "結(jié)束行動(dòng)", "#############################")參考文獻(xiàn)：