Keras+OpenAI強化學習實踐：深度Q網絡
在之前的 Keras/OpenAI 教程中，我們討論了一個將深度學習應用于強化學習環境的基礎案例，它的效果非常顯著。想象作為訓練數據的完全隨機序列(series)。任何兩個序列都不可能高度彼此重復，因為這些都是隨機產生的。然而，成功的試驗之間存在相同的關鍵特征，例如在 CartPole 游戲中，當桿往右靠時需要將車向右推，反之亦然。因此，通過在所有這些試驗數據上訓練我們的神經網絡，我們提取了有助于成功的共同模式(pattern)，并能夠平滑導致其產生獨立故障的細節。
話雖如此，我們認為這次的環境比上次要困難得多，即游戲：MountainCar。
一、更復雜的環境
即使看上去我們應該能夠應用與上周相同的技術，但是有一個關鍵特征使它變得不可能：我們無法生成訓練數據。與簡單的 CartPole 例子不同，采取隨機移動通常只會導致實驗的結果很差(谷底)。也就是說，我們的實驗結果***都是相同的-200。這用作訓練數據幾乎沒有用。想象一下，如果你無論在考試中做出什么答案，你都會得到 0%，那么你將如何從這些經驗中學習?
「MountainCar-v0」環境的隨機輸入不會產生任何對于訓練有用的輸出。
由于這些問題，我們必須找出一種能逐步改進以前實驗的方法。為此，我們使用強化學習最基本的方法：Q-learning!
二、DQN 的理論背景
Q-learning 的本質是創建一個「虛擬表格」，這個表格包含了當前環境狀態下每個可能的動作能得到多少獎勵。下面來詳細說明：
網絡可以想象為內生有電子表格的網絡，該表格含有當前環境狀態下可能采取的每個可能的動作的值。
「虛擬表格」是什么意思?想像一下，對于輸入空間的每個可能的動作，你都可以為每個可能采取的動作賦予一個分數。如果這可行，那么你可以很容易地「打敗」環境：只需選擇具有***分數的動作!但是需要注意 2 點：首先，這個分數通常被稱為「Q-分數」，此算法也由此命名。第二，與任何其它得分一樣，這些 Q-分數在其模擬的情境外沒有任何意義。也就是說，它們沒有確定的意義，但這沒關系，因為我們只需要做比較。
為什么對于每個輸入我們都需要一個虛擬表格?難道沒有統一的表格嗎?原因是這樣做不和邏輯：這與在谷底談采取什么動作是***的，及在向左傾斜時的***點討論采取什么動作是***的是一樣的道理。
現在，我們的主要問題(為每個輸入建立虛擬表格)是不可能的：我們有一個連續的(***)輸入空間!我們可以通過離散化輸入空間來解決這個問題，但是對于本問題來說，這似乎是一個非常棘手的解決方案，并且在將來我們會一再遇到。那么，我們如何解決呢?那就是通過將神經網絡應用于這種情況：這就是 DQN 中 D 的來歷!
三、DQN agent
現在，我們現在已經將問題聚焦到：找到一種在給定當前狀態下為不同動作賦值 Q-分數的方法。這是使用任何神經網絡時遇到的非常自然的***個問題的答案：我們模型的輸入和輸出是什么?本模型中你需要了解的數學方程是以下等式(不用擔心，我們會在下面講解)：
如上所述，Q 代表了給定當前狀態(s)和采取的動作(a)時我們模型估計的價值。然而，目標是確定一個狀態價值的總和。那是什么意思?即從該位置獲得的即時獎勵和將來會獲得的預期獎勵之和。也就是說，我們要考慮一個事實，即一個狀態的價值往往不僅反映了它的直接收益，而且還反映了它的未來收益。在任何情況下，我們會將未來的獎勵折現，因為對于同樣是收到$100 的兩種情況(一種為將來，一種為現在)，我會永遠選擇現在的交易，因為未來是會變化的。γ因子反映了此狀態預期未來收益的貶值。
這就是我們需要的所有數學!下面是實際代碼的演示!
四、DQN agent 實現
深度 Q 網絡為持續學習(continuous learning)，這意味著不是簡單地累積一批實驗/訓練數據并將其傳入模型。相反，我們通過之前運行的實驗創建訓練數據，并且直接將運行后創建的數據饋送如模型。如果現在感到好像有些模糊，別擔心，該看看代碼了。代碼主要在定義一個 DQN 類，其中將實現所有的算法邏輯，并且我們將定義一組簡單的函數來進行實際的訓練。
1. DQN 超參數
首先，我們將討論一些與 DQN 相關的參數。它們大多數是實現主流神經網絡的標準參數：
- class DQN:
- def __init__(self, env):
- self.env = env
- self.memory = deque(maxlen=2000)
- self.gamma = 0.95
- self.epsilon = 1.0
- self.epsilon_min = 0.01
- self.epsilon_decay = 0.995
- self.learning_rate = 0.01
讓我們來一步一步地講解這些超參數。***個是環境(env)，這僅僅是為了在建立模型時便于引用矩陣的形狀。「記憶(memory)」是 DQN 的關鍵組成部分：如前所述，我們不斷通過實驗訓練模型。然而與直接訓練實驗的數據不同，我們將它們先添加到內存中并隨機抽樣。為什么這樣做呢，難道僅僅將*** x 個實驗數據作為樣本進行訓練不好嗎?原因有點微妙。設想我們只使用最近的實驗數據進行訓練：在這種情況下，我們的結果只會學習其最近的動作，這可能與未來的預測沒有直接的關系。特別地，在本環境下，如果我們在斜坡右側向下移動，使用最近的實驗數據進行訓練將需要在斜坡右側向上移動的數據上進行訓練。但是，這與在斜坡左側的情景需決定采取的動作無關。所以，通過抽取隨機樣本，將保證不會偏離訓練集，而是理想地學習我們將遇到的所有環境。
我們現在來討論模型的超參數：gamma、epsilon 以及 epsilon 衰減和學習速率。***個是前面方程中討論的未來獎勵的折現因子(<1)，***一個是標準學習速率參數，我們不在這里討論。第二個是 RL 的一個有趣方面，值得一談。在任何一種學習經驗中，我們總是在探索與利用之間做出選擇。這不僅限于計算機科學或學術界：我們每天都在做這件事!
考慮你家附近的飯店。你***一次嘗試新飯店是什么時候?可能很久以前。這對應于你從探索到利用的轉變：與嘗試找到新的更好的機會不同，你根據自己以往的經驗找到***的解決方案，從而***化效用。對比當你剛搬家時：當時你不知道什么飯店是好的，所以被誘惑去探索新選擇。換句話說，這時存在明確的學習趨勢：當你不了解它們時，探索所有的選擇，一旦你對其中的一些建立了意見，就逐漸轉向利用。以同樣的方式，我們希望我們的模型能夠捕捉這種自然的學習模型，而 epsilon 扮演著這個角色。
Epsilon 表示我們將致力于探索的時間的一小部分。也就是說，實驗的分數 self.epsilon，我們將僅僅采取隨機動作，而不是我們預測在這種情況下***的動作。如上所述，我們希望在開始時形成穩定評估之前更經常地采取隨機動作：因此開始時初始化ε接近 1.0，并在每一個連續的時間步長中以小于 1 的速率衰減它。
2. DQN 模型
在上面的 DQN 的初始化中排除了一個關鍵環節：用于預測的實際模型!在原來的 Keras RL 教程中，我們直接給出數字向量形式的輸入和輸出。因此，除了全連接層之外，不需要在網絡中使用更復雜的層。具體來說，我們將模型定義為：
- def create_model(self):
- model = Sequential()
- state_shape = self.env.observation_space.shape
- model.add(Dense(24, input_dim=state_shape[0],
- activation="relu"))
- model.add(Dense(48, activation="relu"))
- model.add(Dense(24, activation="relu"))
- model.add(Dense(self.env.action_space.n))
- model.compile(loss="mean_squared_error",
- optimizer=Adam(lr=self.learning_rate))
- return model
并用它來定義模型和目標模型(如下所述)：
- def __init__(self, env):
- self.env = env
- self.memory = deque(maxlen=2000)
- self.gamma = 0.95
- self.epsilon = 1.0
- self.epsilon_min = 0.01
- self.epsilon_decay = 0.995
- self.learning_rate = 0.01
- self.tau = .05
- selfself.model = self.create_model()
- # "hack" implemented by DeepMind to improve convergence
- selfself.target_model = self.create_model()
事實上，有兩個單獨的模型，一個用于做預測，一個用于跟蹤「目標值」，這是反直覺的。明確地說，模型(self.model)的作用是對要采取的動作進行實際預測，目標模型(self.target_model)的作用是跟蹤我們想要模型采取的動作。
為什么不用一個模型做這兩件事呢?畢竟，如果預測要采取的動作，那不會間接地確定我們想要模型采取的模式嗎?這實際上是 DeepMind 發明的深度學習的「不可思議的技巧」之一，它用于在 DQN 算法中獲得收斂。如果使用單個模型，它可以(通常會)在簡單的環境(如 CartPole)中收斂。但是，在這些更為復雜的環境中并不收斂的原因在于我們如何對模型進行訓練：如前所述，我們正在對模型進行「即時」訓練。
因此，在每個時間步長進行訓練模型，如果我們使用單個網絡，實際上也將在每個時間步長時改變「目標」。想想這將多么混亂!那就如同，開始老師告訴你要完成教科書中的第 6 頁，當你完成了一半時，她把它改成了第 9 頁，當你完成一半的時候，她告訴你做第 21 頁!因此，由于缺乏明確方向以利用優化器，即梯度變化太快難以穩定收斂，將導致收斂不足。所以，作為代償，我們有一個變化更慢的網絡以跟蹤我們的最終目標，和一個最終實現這些目標的網絡。
3. DQN 訓練
訓練涉及三個主要步驟：記憶、學習和重新定位目標。***步基本上只是隨著實驗的進行向記憶添加數據：
- def remember(self, state, action, reward, new_state, done):
- self.memory.append([state, action, reward, new_state, done])
這里沒有太多的注意事項，除了我們必須存儲「done」階段，以了解我們以后如何更新獎勵函數。轉到 DQN 主體的訓練函數。這是使用存儲記憶的地方，并積極從我們過去看到的內容中學習。首先，從整個存儲記憶中抽出一個樣本。我們認為每個樣本是不同的。正如我們在前面的等式中看到的，我們要將 Q-函數更新為當前獎勵之和與預期未來獎勵的總和(貶值為 gamma)。在實驗結束時，將不再有未來的獎勵，所以該狀態的價值為此時我們收到的獎勵之和。然而，在非終止狀態，如果我們能夠采取任何可能的動作，將會得到的***的獎勵是什么?我們得到：
- def replay(self):
- batch_size = 32
- if len(self.memory) < batch_size:
- return
- samples = random.sample(self.memory, batch_size)
- for sample in samples:
- state, action, reward, new_state, done = sample
- target = self.target_model.predict(state)
- if done:
- target[0][action] = reward
- else:
- Q_future = max(
- self.target_model.predict(new_state)[0])
- target[0][action] = reward + Q_future * self.gamma
- self.model.fit(state, target, epochs=1, verbose=0)
***，我們必須重新定位目標，我們只需將主模型的權重復制到目標模型中。然而，與主模型訓練的方法不同，目標模型更新較慢：
- def target_train(self):
- weights = self.model.get_weights()
- target_weights = self.target_model.get_weights()
- for i in range(len(target_weights)):
- target_weights[i] = weights[i]
- self.target_model.set_weights(target_weights)
4. DQN 動作
***一步是讓 DQN 實際執行希望的動作，在給定的 epsilon 參數基礎上，執行的動作在隨機動作與基于過去訓練的預測動作之間選擇，如下所示：
- def act(self, state):
- self.epsilon *= self.epsilon_decay
- self.epsilon = max(self.epsilon_min, self.epsilon)
- if np.random.random() < self.epsilon:
- return self.env.action_space.sample()
- return np.argmax(self.model.predict(state)[0])
5. 訓練 agent
現在訓練我們開發的復雜的 agent。將其實例化，傳入經驗數據，訓練 agent，并更新目標網絡：
- def main():
- env = gym.make("MountainCar-v0")
- gamma = 0.9
- epsilon = .95
- trials = 100
- trial_len = 500
- updateTargetNetwork = 1000
- dqn_agent = DQN(envenv=env)
- steps = []
- for trial in range(trials):
- cur_state = env.reset().reshape(1,2)
- for step in range(trial_len):
- action = dqn_agent.act(cur_state)
- env.render()
- new_state, reward, done, _ = env.step(action)
- rewardreward = reward if not done else -20
- print(reward)
- new_statenew_state = new_state.reshape(1,2)
- dqn_agent.remember(cur_state, action,
- reward, new_state, done)
- dqn_agent.replay()
- dqn_agent.target_train()
- cur_state = new_state
- if done:
- break
- if step >= 199:
- print("Failed to complete trial")
- else:
- print("Completed in {} trials".format(trial))
- break
五、完整的代碼
這就是使用 DQN 的「MountainCar-v0」環境的完整代碼!
- import gym
- import numpy as np
- import random
- from keras.models import Sequential
- from keras.layers import Dense, Dropout
- from keras.optimizers import Adam
- from collections import deque
- class DQN:
- def __init__(self, env):
- self.env = env
- self.memory = deque(maxlen=2000)
- self.gamma = 0.85
- self.epsilon = 1.0
- self.epsilon_min = 0.01
- self.epsilon_decay = 0.995
- self.learning_rate = 0.005
- self.tau = .125
- selfself.model = self.create_model()
- selfself.target_model = self.create_model()
- def create_model(self):
- model = Sequential()
- state_shape = self.env.observation_space.shape
- model.add(Dense(24, input_dim=state_shape[0], activation="relu"))
- model.add(Dense(48, activation="relu"))
- model.add(Dense(24, activation="relu"))
- model.add(Dense(self.env.action_space.n))
- model.compile(loss="mean_squared_error",
- optimizer=Adam(lr=self.learning_rate))
- return model
- def act(self, state):
- self.epsilon *= self.epsilon_decay
- self.epsilon = max(self.epsilon_min, self.epsilon)
- if np.random.random() < self.epsilon:
- return self.env.action_space.sample()
- return np.argmax(self.model.predict(state)[0])
- def remember(self, state, action, reward, new_state, done):
- self.memory.append([state, action, reward, new_state, done])
- def replay(self):
- batch_size = 32
- if len(self.memory) < batch_size:
- return
- samples = random.sample(self.memory, batch_size)
- for sample in samples:
- state, action, reward, new_state, done = sample
- target = self.target_model.predict(state)
- if done:
- target[0][action] = reward
- else:
- Q_future = max(self.target_model.predict(new_state)[0])
- target[0][action] = reward + Q_future * self.gamma
- self.model.fit(state, target, epochs=1, verbose=0)
- def target_train(self):
- weights = self.model.get_weights()
- target_weights = self.target_model.get_weights()
- for i in range(len(target_weights)):
- target_weights[i] = weights[i] * self.tau + target_weights[i] * (1 - self.tau)
- self.target_model.set_weights(target_weights)
- def save_model(self, fn):
- self.model.save(fn)
- def main():
- env = gym.make("MountainCar-v0")
- gamma = 0.9
- epsilon = .95
- trials = 1000
- trial_len = 500
- # updateTargetNetwork = 1000
- dqn_agent = DQN(envenv=env)
- steps = []
- for trial in range(trials):
- cur_state = env.reset().reshape(1,2)
- for step in range(trial_len):
- action = dqn_agent.act(cur_state)
- new_state, reward, done, _ = env.step(action)
- # rewardreward = reward if not done else -20
- new_statenew_state = new_state.reshape(1,2)
- dqn_agent.remember(cur_state, action, reward, new_state, done)
- dqn_agent.replay() # internally iterates default (prediction) model
- dqn_agent.target_train() # iterates target model
- cur_state = new_state
- if done:
- break
- if step >= 199:
- print("Failed to complete in trial {}".format(trial))
- if step % 10 == 0:
- dqn_agent.save_model("trial-{}.model".format(trial))
- else:
- print("Completed in {} trials".format(trial))
- dqn_agent.save_model("success.model")
- break
- if __name__ == "__main__":
- main()
原文：
https://medium.com/towards-data-science/reinforcement-learning-w-keras-openai-dqns-1eed3a5338c
【本文是51CTO專欄機構“機器之心”的原創譯文，微信公眾號“機器之心( id: almosthuman2014)”】
