# 十八、強化學習 > 譯者：[@SeanCheney](https://www.jianshu.com/u/130f76596b02) 強化學習（RL）如今是機器學習的一大令人激動的領域，也是最老的領域之一。自從 1950 年被發明出來后，它被用于一些有趣的應用，尤其是在游戲（例如 TD-Gammon，一個西洋雙陸棋程序）和機器控制領域，但是從未弄出什么大新聞。直到 2013 年一個革命性的發展：來自英國的研究者發起了 Deepmind 項目，這個項目可以學習去玩任何從頭開始的 Atari 游戲，在多數游戲中，比人類玩的還好，它僅使用像素作為輸入而沒有使用游戲規則的任何先驗知識。這是一系列令人驚嘆的壯舉中的第一個，并在 2016 年 3 月以他們的系統 AlphaGo 戰勝了世界圍棋冠軍李世石而告終。從未有程序能勉強打敗這個游戲的大師，更不用說世界冠軍了。今天，RL 的整個領域正在沸騰著新的想法，其都具有廣泛的應用范圍。DeepMind 在 2014 被谷歌以超過 5 億美元收購。 DeepMind 是怎么做到的呢？事后看來，原理似乎相當簡單：他們將深度學習運用到強化學習領域，結果卻超越了他們最瘋狂的設想。在本章中，我們將首先解釋強化學習是什么，以及它擅長于什么，然后我們將介紹兩個在深度強化學習領域最重要的技術：策略梯度和深度 Q 網絡（DQN），包括討論馬爾可夫決策過程（MDP）。我們將使用這些技術來訓練一個模型來平衡移動車上的桿子；然后，我會介紹 TF-Agents 庫，這個庫利用先進的算法，可以大大簡化創建 RL 系統，然后我們會用這個系統來玩 Breakout，一個著名的 Atari 游戲。本章最后，會介紹強化學習領域的最新進展。 ## 學習優化獎勵 在強化學習中，智能體在環境（environment）中觀察（observation）并且做出決策（action），隨后它會得到獎勵（reward）。它的目標是去學習如何行動能最大化**期望獎勵**。如果你不在意擬人化的話，可以認為正獎勵是愉快，負獎勵是痛苦（這樣的話獎勵一詞就有點誤導了）。總之，智能體在環境中行動，并且在實驗和錯誤中去學習最大化它的愉快，最小化它的痛苦。 這是一個相當廣泛的設置，可以適用于各種各樣的任務。以下是幾個例子（詳見圖 16-1）： 1. 智能體可以是控制一個機器人的程序。在此例中，環境就是真實的世界，智能體通過許多的傳感器例如攝像機或者觸覺傳感器來觀察，它可以通過給電機發送信號來行動。它可以被編程設置為如果到達了目的地就得到正獎勵，如果浪費時間，或者走錯方向，或摔倒了就得到負獎勵。 2. 智能體可以是控制 Ms.Pac-Man 的程序。在此例中，環境是 Atari 游戲的模擬器，行為是 9 個操縱桿位（上下左右中間等等），觀察是屏幕，回報就是游戲點數。 3. 相似地，智能體也可以是棋盤游戲的程序，例如圍棋。 4. 智能體也可以不用去控制一個實體（或虛擬的）去移動。例如它可以是一個智能恒溫器，當它調整到目標溫度以節能時會得到正獎勵，當人們需要自己去調節溫度時它會得到負獎勵，所以智能體必須學會預見人們的需要。 5. 智能體也可以去觀測股票市場價格以實時決定買賣。獎勵的依據為掙錢或者賠錢。 圖 18-1 強化學習案例：（a）行走機器人，（b）Ms.Pac-Man 游戲，（c）圍棋玩家，（d）恒溫器，（e）自動交易員 其實沒有正獎勵也是可以的，例如智能體在迷宮內移動，它每分每秒都得到一個負獎勵，所以它要盡可能快的找到出口！還有很多適合強化學習的領域，例如自動駕駛汽車，推薦系統，在網頁上放廣告，或者控制一個圖像分類系統讓它明白它應該關注于什么。 ## 策略搜索 智能體用于改變行為的算法稱為策略。例如，策略可以是一個把觀測當輸入，行為當做輸出的神經網絡（見圖 16-2）。 圖 18-2 用神經網絡策略做加強學習 這個策略可以是你能想到的任何算法，它甚至可以是非確定性的。事實上，在某些任務中，策略根本不必觀察環境！舉個例子，例如，考慮一個真空吸塵器，它的獎勵是在 30 分鐘內撿起的灰塵數量。它的策略可以是每秒以概率`p`向前移動，或者以概率`1-p`隨機地向左或向右旋轉。旋轉角度將是`-r`和`+r`之間的隨機角度，因為該策略涉及一些隨機性，所以稱為隨機策略。機器人將有一個不確定的軌跡，它保證它最終會到達任何可以到達的地方，并撿起所有的灰塵。問題是：30 分鐘后它會撿起多少灰塵？ 怎么訓練這樣的機器人？你能調整的策略參數只有兩個：概率`p`和角度范圍`r`。一個想法是這些參數嘗試許多不同的值，并選擇執行最佳的組合（見圖 18-3）。這是一個策略搜索的例子，在這種情況下使用暴力方法。然而，當策略空間太大（通常情況下），以這樣的方式找到一組好的參數就像是大海撈針。 圖 18-3 策略空間中的四個點以及機器人的對應行為 另一種搜尋策略空間的方法是遺傳算法。例如你可以隨機創造一個包含 100 個策略的第一代基因，隨后殺死 80 個糟糕的策略，隨后讓 20 個幸存策略繁衍 4 代。一個后代只是它父輩基因的復制品加上一些隨機變異。幸存的策略加上他們的后代共同構成了第二代。你可以繼續以這種方式迭代代，直到找到一個好的策略。 另一種方法是使用優化技術，通過評估獎勵關于策略參數的梯度，然后通過跟隨梯度向更高的獎勵（梯度上升）調整這些參數。這種方法被稱為策略梯度（policy gradient, PG），我們將在本章后面詳細討論。例如，回到真空吸塵器機器人，你可以稍微增加概率 P 并評估這是否增加了機器人在 30 分鐘內拾起的灰塵的量；如果確實增加了，就相對應增加`p`，否則減少`p`。我們將使用 Tensorflow 來實現 PG 算法，但是在這之前我們需要為智能體創造一個生存的環境，所以現在是介紹 OpenAI Gym 的時候了。 ## OpenAI Gym 介紹 強化學習的一個挑戰是，為了訓練對象，首先需要有一個工作環境。如果你想設計一個可以學習 Atari 游戲的程序，你需要一個 Atari 游戲模擬器。如果你想設計一個步行機器人，那么環境就是真實的世界，你可以直接在這個環境中訓練你的機器人，但是這有其局限性：如果機器人從懸崖上掉下來，你不能僅僅點擊“撤消”。你也不能加快時間；增加更多的計算能力不會讓機器人移動得更快。一般來說，同時訓練 1000 個機器人是非常昂貴的。簡而言之，訓練在現實世界中是困難和緩慢的，所以你通常需要一個模擬環境，至少需要引導訓練。例如，你可以使用 PyBullet 或 MuJoCo 來做 3D 物理模擬。 OpenAI Gym 是一個工具包，它提供各種各樣的模擬環境（Atari 游戲，棋盤游戲，2D 和 3D 物理模擬等等），所以你可以訓練，比較，或開發新的 RL 算法。 安裝之前，如果你是用虛擬環境創建的獨立的環境，需要先激活： ```py $ cd $ML_PATH # 工作目錄 (e.g., $HOME/ml) $ source my_env/bin/activate # Linux or MacOS $ .\my_env\Scripts\activate # Windows ``` 接下來安裝 OpenAI gym。可通過`pip`安裝： ```py $ python3 -m pip install --upgrade gym ``` 取決于系統，你可能還要安裝 Mesa OpenGL Utility（GLU）庫（比如，在 Ubuntu 18.04 上，你需要運行`apt install libglu1-mesa`）。這個庫用來渲染第一個環境。接著，打開一個 Python 終端或 Jupyter notebook，用`make()`創建一個環境： ```py >>> import gym >>> env = gym.make("CartPole-v1") >>> obs = env.reset() >>> obs array([-0.01258566, -0.00156614, 0.04207708, -0.00180545]) ``` 這里創建了一個 CartPole 環境。這是一個 2D 模擬，其中推車可以被左右加速，以平衡放置在它上面的平衡桿（見圖 18-4）。你可以用`gym.envs.registry.all()`獲得所有可用的環境。在創建環境之后，需要使用`reset()`初始化。這會返回第一個觀察結果。觀察取決于環境的類型。對于 CartPole 環境，每個觀測是包含四個浮點數的 1D Numpy 向量：這些浮點數代表推車的水平位置（0.0 為中心）、速度（正是右）、桿的角度（0.0 為垂直）及角速度（正為順時針）。 用`render()`方法展示環境（見圖 18-4）。在 Windows 上，這需要安裝 X Server，比如 VcXsrv 或 Xming： ```py >>> env.render() True ``` 圖 18-4 CartPole 環境 > 提示：如果你在使用無頭服務器（即，沒有顯示器），比如云上的虛擬機，渲染就會失敗。解決的唯一方法是使用假 X server，比如 Xvfb 或 Xdummy。例如，裝好 Xvfb 之后（Ubuntu 或 Debian 上運行`apt install xvfb`），用這條命令啟動 Python：`xvfb-run -s "-screen 0 1400x900x24" python3`。或者，安裝 Xvfb 和[`pyvirtualdisplay` 庫](https://links.jianshu.com/go?to=https%3A%2F%2Fhoml.info%2Fpyvd)（這個庫包裝了 Xvfb），在程序啟動處運行`pyvirtualdisplay.Display(visible=0, size=(1400, 900)).start()`。 如果你想讓`render()`讓圖像以一個 Numpy 數組格式返回，可以將`mode`參數設置為`rgb_array`（注意，這個環境會渲染環境到屏幕上）： ```py >>> img = env.render(mode="rgb_array") >>> img.shape # height, width, channels (3=RGB) (800, 1200, 3) ``` 詢問環境，可以采取的可能行動： ```py >>> env.action_space Discrete(2) ``` `Discrete(2)`的意思是可能的行動是整數 0 和 1，表示向左（0）或向右（1）加速。其它的環境可能有其它離散的行動，或其它種類的行動（例如，連續性行動）。因為棍子是向右偏的（`obs[2] > 0`），讓車子向右加速： ```py >>> action = 1 # accelerate right >>> obs, reward, done, info = env.step(action) >>> obs array([-0.01261699, 0.19292789, 0.04204097, -0.28092127]) >>> reward 1.0 >>> done False >>> info {} ``` `step()`方法執行給定的動作并返回四個值： `obs`: 這是新的觀測，小車現在正在向右走（`obs[1]>0`，注：當前速度為正，向右為正）。平衡桿仍然向右傾斜（`obs[2]>0`），但是他的角速度現在為負（`obs[3]<0`），所以它在下一步后可能會向左傾斜。 `reward`： 在這個環境中，無論你做什么，每一步都會得到 1.0 獎勵，所以游戲的目標就是盡可能長的運行。 `done`： 當游戲結束時這個值會為`True`。當平衡桿傾斜太多、或越過屏幕、或超過 200 步時會發生這種情況。之后，必須重新設置環境才能重新使用。 `info`： 該字典可以在其他環境中提供額外信息用于調試或訓練。例如，在一些游戲中，可以指示 agent 還剩多少條命。 > 提示：使用完環境后，應當調用它的`close()`方法釋放資源。 讓我們硬編碼一個簡單的策略，當桿向左傾斜時向左邊加速，當桿向右傾斜時加速向右邊加速。我們使用這個策略來獲得超過 500 步的平均回報： ```py def basic_policy(obs): angle = obs[2] return 0 if angle < 0 else 1 totals = [] for episode in range(500): episode_rewards = 0 obs = env.reset() for step in range(200): action = basic_policy(obs) obs, reward, done, info = env.step(action) episode_rewards += reward if done: break totals.append(episode_rewards) ``` 這段代碼不難。讓我們看看結果： ```py >>> import numpy as np >>> np.mean(totals), np.std(totals), np.min(totals), np.max(totals) (41.718, 8.858356280936096, 24.0, 68.0) ``` 即使有 500 次嘗試，這一策略從未使平衡桿在超過 68 個連續的步驟里保持直立。結果太好。如果你看一下 Juyter Notebook 中的模擬，你會發現，推車越來越強烈地左右擺動，直到平衡桿傾斜過度。讓我們看看神經網絡是否能提出更好的策略。 ## 神經網絡策略 讓我們創建一個神經網絡策略。就像之前我們編碼的策略一樣，這個神經網絡將把觀察作為輸入，輸出要執行的動作。更確切地說，它將估計每個動作的概率，然后我們將根據估計的概率隨機地選擇一個動作（見圖 18-5）。在 CartPole 環境中，只有兩種可能的動作（左或右），所以我們只需要一個輸出神經元。它將輸出動作 0（左）的概率`p`，動作 1（右）的概率顯然將是`1 - p`。例如，如果它輸出 0.7，那么我們將以 70% 的概率選擇動作 0，以 30% 的概率選擇動作 1。 圖 18-5 神經網絡策略 你可能奇怪為什么我們根據神經網絡給出的概率來選擇隨機的動作，而不是選擇最高分數的動作。這種方法使智能體在**探索新的行為**和**利用那些已知可行的行動**之間找到正確的平衡。舉個類比：假設你第一次去餐館，所有的菜看起來同樣吸引人，所以你隨機挑選一個。如果菜好吃，你可以增加下一次點它的概率，但是你不應該把這個概率提高到 100%，否則你將永遠不會嘗試其他菜肴，其中一些甚至比你嘗試的更好。 還要注意，在這個特定的環境中，過去的動作和觀察可以被放心地忽略，因為每個觀察都包含環境的完整狀態。如果有一些隱藏狀態，那么你也需要考慮過去的行為和觀察。例如，如果環境僅僅揭示了推車的位置，而不是它的速度，那么你不僅要考慮當前的觀測，還要考慮先前的觀測，以便估計當前的速度。另一個例子是當觀測是有噪聲的的，在這種情況下，通常你想用過去的觀察來估計最可能的當前狀態。因此，CartPole 問題是簡單的；觀測是無噪聲的，而且它們包含環境的全狀態。 下面是用 tf.keras 創建這個神經網絡策略的代碼： ```py import tensorflow as tf from tensorflow import keras n_inputs = 4 # == env.observation_space.shape[0] model = keras.models.Sequential([ keras.layers.Dense(5, activation="elu", input_shape=[n_inputs]), keras.layers.Dense(1, activation="sigmoid"), ]) ``` 在導入之后，我們使用`Sequential`模型定義策略網絡。輸入的數量是觀測空間的大小（在 CartPole 的情況下是 4 個），我們只有 5 個隱藏單元，并且我們只有 1 個輸出概率（向左的概率），所以輸出層只需一個使用 sigmoid 的神經元就成。如果超過兩個動作，每個動作就要有一個神經元，然后使用 softmax 激活函數。 好了，現在我們有一個可以觀察和輸出動作的神經網絡了，那我們怎么訓練它呢？ ## 評價行為：信用分配問題 如果我們知道每一步的最佳動作，我們可以像通常一樣訓練神經網絡，通過最小化估計概率和目標概率之間的交叉熵。這只是通常的監督學習。然而，在強化學習中，智能體獲得的指導的唯一途徑是通過獎勵，獎勵通常是稀疏的和延遲的。例如，如果智能體在 100 個步驟內設法平衡桿，它怎么知道它采取的 100 個行動中的哪一個是好的，哪些是壞的？它所知道的是，在最后一次行動之后，桿子墜落了，但最后一次行動肯定不是負全責的。這被稱為信用分配問題：當智能體得到獎勵時，很難知道哪些行為應該被信任（或責備）。如果一只狗在表現優秀幾小時后才得到獎勵，它會明白它做對了什么嗎？ 為了解決這個問題，一個通常的策略是基于這個動作后得分的總和來評估這個個動作，通常在每個步驟中應用衰減因子`r`。例如（見圖 18-6），如果一個智能體決定連續三次向右，在第一步之后得到 +10 獎勵，第二步后得到 0，最后在第三步之后得到 -50，然后假設我們使用衰減率`r=0.8`，那么第一個動作將得到`10 +r×0 + r2×(-50)=-22`的分數。如果衰減率接近 0，那么與即時獎勵相比，未來的獎勵不會有多大意義。相反，如果衰減率接近 1，那么對未來的獎勵幾乎等于即時回報。典型的衰減率通常從 0.9 到 0.99 之間。如果衰減率為 0.95，那么未來 13 步的獎勵大約是即時獎勵的一半（`0.9513×0.5`），而當衰減率為 0.99，未來 69 步的獎勵是即時獎勵的一半。在 CartPole 環境下，行為具有相當短期的影響，因此選擇 0.95 的折扣率是合理的。 圖 18-6 計算行動的回報：未來衰減求和 當然，一個好的動作可能會緊跟著一串壞動作，這些動作會導致平衡桿迅速下降，從而導致一個好的動作得到一個低分數（類似的，一個好行動者有時會在一部爛片中扮演主角）。然而，如果我們花足夠多的時間來訓練游戲，平均下來好的行為會得到比壞的更好的分數。因此，為了獲得相當可靠的動作分數，我們必須運行很多次并將所有動作分數歸一化（通過減去平均值并除以標準偏差）。之后，我們可以合理地假設消極得分的行為是壞的，而積極得分的行為是好的。現在我們有一個方法來評估每一個動作，我們已經準備好使用策略梯度來訓練我們的第一個智能體。讓我們看看如何做。 ## 策略梯度 正如前面所討論的，PG 算法通過遵循更高回報的梯度來優化策略參數。一種流行的 PG 算法，稱為增強算法，在 1929 由 Ronald Williams 提出。這是一個常見的變體： 1. 首先，讓神經網絡策略玩幾次游戲，并在每一步計算梯度，這使得智能體更可能選擇行為，但不應用這些梯度。 2. 運行幾次后，計算每個動作的得分（使用前面段落中描述的方法）。 3. 如果一個動作的分數是正的，這意味著動作是好的，可應用較早計算的梯度，以便將來有更大的的概率選擇這個動作。但是，如果分數是負的，這意味著動作是壞的，要應用相反梯度來使得這個動作在將來采取的可能性更低。我們的方法就是簡單地將每個梯度向量乘以相應的動作得分。 4. 最后，計算所有得到的梯度向量的平均值，并使用它來執行梯度下降步驟。 讓我們使用 tf.keras 實現這個算法。我們將訓練我們早先建立的神經網絡策略，讓它學會平衡車上的平衡桿。首先，需要一個能執行一步的函數。假定做出的動作都是對的，激素親戚損失和梯度（梯度會保存一會，根據動作的結果再對其修改）： ```py def play_one_step(env, obs, model, loss_fn): with tf.GradientTape() as tape: left_proba = model(obs[np.newaxis]) action = (tf.random.uniform([1, 1]) > left_proba) y_target = tf.constant([[1.]]) - tf.cast(action, tf.float32) loss = tf.reduce_mean(loss_fn(y_target, left_proba)) grads = tape.gradient(loss, model.trainable_variables) obs, reward, done, info = env.step(int(action[0, 0].numpy())) return obs, reward, done, grads ``` 逐行看代碼： * 在`GradientTape`代碼塊內，先調用模型，傳入一個觀察（將觀察變形為包含單個實例的批次）。輸出是向左的概率。 * 然后，選取一個 0 到 1 之間的浮點數，檢查是否大于`left_proba`。概率為`left_proba`時，`action`是`False`；概率為`1-left_proba`時，`action`是`True`。當將這個布爾值轉變為數字時，動作是 0（左）或 1（右）及對應的概率。 * 接著，定義向左的目標概率：1 減去動作（浮點值）。如果動作是 0（左），則向左的目標概率等于 1。如果動作是 1（右），則目標概率等于 0。 * 然后使用損失函數計算損失，使用記錄器計算模型可訓練變量的損失梯度。這些梯度會在后面應用前，根據動作的結果做微調。 * 最后，執行選擇的動作，無論是否結束，返回新的觀察、獎勵，和剛剛計算的梯度。 現在，創建另一個函數基于`play_one_step()`的多次執行函數，返回所有獎勵和每個周期和步驟的梯度： ```py def play_multiple_episodes(env, n_episodes, n_max_steps, model, loss_fn): all_rewards = [] all_grads = [] for episode in range(n_episodes): current_rewards = [] current_grads = [] obs = env.reset() for step in range(n_max_steps): obs, reward, done, grads = play_one_step(env, obs, model, loss_fn) current_rewards.append(reward) current_grads.append(grads) if done: break all_rewards.append(current_rewards) all_grads.append(current_grads) return all_rewards, all_grads ``` 這段代碼返回了獎勵列表（每個周期一個獎勵列表，每個步驟一個獎勵），還有一個梯度列表（每個周期一個梯度列表，每個步驟一個梯度元組，每個元組每個變臉有一個梯度張量）。 算法會使用`play_multiple_episodes()`函數，多次執行游戲（比如，10 次），然后會檢查所有獎勵，做衰減，然后歸一化。要這么做，需要多個函數：第一個計算每個步驟的未來衰減獎勵的和，第二個歸一化所有這些衰減獎勵（減去平均值，除以標準差）： ```py def discount_rewards(rewards, discount_factor): discounted = np.array(rewards) for step in range(len(rewards) - 2, -1, -1): discounted[step] += discounted[step + 1] * discount_factor return discounted def discount_and_normalize_rewards(all_rewards, discount_factor): all_discounted_rewards = [discount_rewards(rewards, discount_factor) for rewards in all_rewards] flat_rewards = np.concatenate(all_discounted_rewards) reward_mean = flat_rewards.mean() reward_std = flat_rewards.std() return [(discounted_rewards - reward_mean) / reward_std for discounted_rewards in all_discounted_rewards] ``` 檢測其是否有效： ```py >>> discount_rewards([10, 0, -50], discount_factor=0.8) array([-22, -40, -50]) >>> discount_and_normalize_rewards([[10, 0, -50], [10, 20]], ... discount_factor=0.8) ... [array([-0.28435071, -0.86597718, -1.18910299]), array([1.26665318, 1.0727777 ])] ``` 調用`discount_rewards()`，返回了我們想要的結果（見圖 18-6）。可以確認函數`discount_and_normalize_rewards()`返回了每個周期每個步驟的歸一化的行動的結果。可以看到，第一個周期的表現比第二個周期的表現糟糕，所以歸一化的結果都是負的；第一個周期中的動作都是不好的，而第二個周期中的動作被認為是好的。 可以準備運行算法了！現在定義超參數。運行 150 個訓練迭代，每次迭代完成 10 次周期，每個周期最多 200 個步驟。衰減因子是 0.95： ```py n_iterations = 150 n_episodes_per_update = 10 n_max_steps = 200 discount_factor = 0.95 ``` 還需要一個優化器和損失函數。優化器用普通的 Adam 就成，學習率用 0.01，因為是二元分類器，使用二元交叉熵損失函數： ```py optimizer = keras.optimizers.Adam(lr=0.01) loss_fn = keras.losses.binary_crossentropy ``` 接下來創建和運行訓練循環。 ```py for iteration in range(n_iterations): all_rewards, all_grads = play_multiple_episodes( env, n_episodes_per_update, n_max_steps, model, loss_fn) all_final_rewards = discount_and_normalize_rewards(all_rewards, discount_factor) all_mean_grads = [] for var_index in range(len(model.trainable_variables)): mean_grads = tf.reduce_mean( [final_reward * all_grads[episode_index][step][var_index] for episode_index, final_rewards in enumerate(all_final_rewards) for step, final_reward in enumerate(final_rewards)], axis=0) all_mean_grads.append(mean_grads) optimizer.apply_gradients(zip(all_mean_grads, model.trainable_variables)) ``` 逐行看下代碼： * 在每次訓練迭代，循環調用`play_multiple_episodes()`，這個函數玩 10 次游戲，返回每個周期和步驟的獎勵和梯度。 * 然后調用`discount_and_normalize_rewards()`計算每個動作的歸一化結果（代碼中是`final_reward`）。這樣可以測量每個動作的好壞結果。 * 接著，循環每個可訓練變量，計算每個變量的梯度加權平均，權重是`final_reward`。 * 最后，將這些平均梯度應用于優化器：微調模型的變量。 就是這樣。這段代碼可以訓練神經網絡策略，模型可以學習保持棍子的平衡（可以嘗試 notebook 中的“策略梯度”部分）。每個周期的平均獎勵會非常接近 200（200 是環境默認的最大值）。成功！ > 提示：研究人員試圖找到一種即使當智能體最初對環境一無所知時也能很好工作的算法。然而，除非你正在寫論文，否則你應該盡可能多地將先前的知識注入到智能體中，因為它會極大地加速訓練。例如，因為知道棍子要盡量垂直，你可以添加與棍子角度成正比的負獎勵。這可以讓獎勵不那么分散，是訓練加速。此外，如果你已經有一個相當好的策略，你可以訓練神經網絡模仿它，然后使用策略梯度來改進它。 盡管它相對簡單，但是該算法是非常強大的。你可以用它來解決更難的問題，而不僅僅是平衡一輛手推車上的平衡桿。事實上，因為樣本不足，必須多次玩游戲，才能取得更大進展。但這個算法是更強大算法的基礎，比如 Actor-Critic 算法（后面會介紹）。 現在我們來看看另一個流行的算法族。與 PG 算法直接嘗試優化策略以增加獎勵相反，我們現在看的算法不那么直接：智能體學習去估計每個狀態的未來衰減獎勵的期望總和，或者在每個狀態中的每個行為未來衰減獎勵的期望和。然后，使用這些知識來決定如何行動。為了理解這些算法，我們必須首先介紹馬爾可夫決策過程（MDP）。 ## 馬爾可夫決策過程 在二十世紀初，數學家 Andrey Markov 研究了沒有記憶的隨機過程，稱為馬爾可夫鏈。這樣的過程具有固定數量的狀態，并且在每個步驟中隨機地從一個狀態演化到另一個狀態。它從狀態`S`演變為狀態`S'`的概率是固定的，它只依賴于`(S, S')`對，而不是依賴于過去的狀態（系統沒有記憶）。 圖 18-7 展示了一個具有四個狀態的馬爾可夫鏈的例子。假設該過程從狀態`S0`開始，并且在下一步驟中有 70% 的概率保持在該狀態不變中。最終，它必然離開那個狀態，并且永遠不會回來，因為沒有其他狀態回到`S0`。如果它進入狀態`S1`，那么它很可能會進入狀態`S2`（90% 的概率），然后立即回到狀態`S1`（以 100% 的概率）。它可以在這兩個狀態之間交替多次，但最終它會落入狀態`S3`并永遠留在那里（這是一個終端狀態）。馬爾可夫鏈可以有非常不同的動力學，它們在熱力學、化學、統計學等方面有著廣泛的應用。 圖 18-7 馬爾科夫鏈案例 馬爾可夫決策過程最初是在 20 世紀 50 年代由 Richard Bellman [描述](https://links.jianshu.com/go?to=https%3A%2F%2Fhoml.info%2F133)的。它們類似于馬爾可夫鏈，但有一個不同：在狀態轉移的每一步中，一個智能體可以選擇幾種可能的動作中的一個，并且過渡概率取決于所選擇的動作。此外，一些狀態過渡返回一些獎勵（正或負），智能體的目標是找到一個策略，隨著時間的推移將最大限度地提高獎勵。 例如，圖 18-8 中所示的 MDP 在每個步驟中具有三個狀態（用圓圈表示）和三個可能的離散動作（用菱形表示）。 圖 18-8 馬爾科夫決策過程案例 如果從狀態`S0`開始，可以在動作`A0`、`A1`或`A2`之間進行選擇。如果它選擇動作`A1`，它就保持在狀態`S0`中，并且沒有任何獎勵。因此，如果愿意的話，它可以決定永遠呆在那里。但是，如果它選擇動作`A0`，它有 70% 的概率獲得 +10 獎勵，并保持在狀態`S0`。然后，它可以一次又一次地嘗試獲得盡可能多的獎勵。但它將在狀態`S1`中結束這樣的行為。在狀態`S1`中，它只有兩種可能的動作：`A0`或`A2`。它可以通過反復選擇動作`A0`來選擇停留，或者它可以選擇動作`A2`移動到狀態`S2`并得到 -50 獎勵。在狀態`S2`中，除了采取行動`A1`之外，別無選擇，這將最有可能引導它回到狀態`S0`，在途中獲得 +40 的獎勵。通過觀察這個 MDP，你能猜出哪一個策略會隨著時間的推移而獲得最大的回報嗎？在狀態`S0`中，`A0`是最好的選擇，在狀態`S2`中，智能體別無選擇，只能采取行動`A1`，但是在狀態`S1`中，智能體否應該保持不動（`A0`）或通過火（`A2`），這是不明確的。 Bellman 找到了一種估計任何狀態`S`的最佳狀態值的方法，記作`V(s)`，它是智能體在其采取最佳行為達到狀態`s`后所有衰減未來獎勵的總和的平均期望。他證明，如果智能體的行為最佳，那么就適用于貝爾曼最優性公式（見公式 18-1）。這個遞歸公式表示，如果智能體最優地運行，那么當前狀態的最優值等于在采取一個最優動作之后平均得到的獎勵，加上該動作可能導致的所有可能的下一個狀態的期望最優值。 公式 18-1 貝爾曼最優性公式 其中： * `T(s, a, s′)`為智能體選擇動作`a`時從狀態`s`到狀態`s'`的概率。例如，圖 18-8 中，T(s2, a1, s0) = 0.8。 * `R(s, a, s′)`為智能體選擇以動作`a`從狀態`s`到狀態`s'`的過程中得到的獎勵。例如圖 18-8 中，R(s2, a1, s0) = +40。 * `γ`為衰減率。 這個等式直接引出了一種算法，該算法可以精確估計每個可能狀態的最優狀態值：首先將所有狀態值估計初始化為零，然后用數值迭代算法迭代更新它們（見公式 18-2）。一個顯著的結果是，給定足夠的時間，這些估計保證收斂到最優狀態值，對應于最優策略。 公式 18-2 數值迭代算法 在這個公式中，V_k(s)是在`k`次算法迭代對狀態`s`的估計。 > 筆記：該算法是動態規劃的一個例子，它將了一個復雜的問題（在這種情況下，估計潛在的未來衰減獎勵的總和）變為可處理的子問題，可以迭代地處理（在這種情況下，找到最大化平均報酬與下一個衰減狀態值的和的動作） 了解最佳狀態值可能是有用的，特別是評估策略，但它沒有明確地告訴智能體要做什么。幸運的是，Bellman 發現了一種非常類似的算法來估計最優狀態-動作值（*state-action values*），通常稱為 Q 值。狀態行動`(S, A)`對的最優 Q 值，記為`Q*(s, a)`，是智能體在到達狀態`S`，然后選擇動作`A`之后平均衰減未來獎勵的期望的總和。但是在它看到這個動作的結果之前，假設它在該動作之后的動作是最優的。 下面是它的工作原理：再次，通過初始化所有的 Q 值估計為零，然后使用 Q 值迭代算法更新它們（參見公式 18-3）。 公式 18-3 Q 值迭代算法 一旦你有了最佳的 Q 值，定義最優的策略`π*(s)`，就沒什么作用了：當智能體處于狀態`S`時，它應該選擇具有最高 Q 值的動作，用于該狀態：  讓我們把這個算法應用到圖 18-8 所示的 MDP 中。首先，我們需要定義 MDP： ```py transition_probabilities = [ # shape=[s, a, s'] [[0.7, 0.3, 0.0], [1.0, 0.0, 0.0], [0.8, 0.2, 0.0]], [[0.0, 1.0, 0.0], None, [0.0, 0.0, 1.0]], [None, [0.8, 0.1, 0.1], None]] rewards = [ # shape=[s, a, s'] [[+10, 0, 0], [0, 0, 0], [0, 0, 0]], [[0, 0, 0], [0, 0, 0], [0, 0, -50]], [[0, 0, 0], [+40, 0, 0], [0, 0, 0]]] possible_actions = [[0, 1, 2], [0, 2], [1]] ``` 例如，要想知道經過動作 a1，從 s2 到 s0 的過渡概率，我們需要查詢`transition_probabilities[2][1][0]`（等于 0.8）。相似的，要得到獎勵，需要查詢`rewards[2][1][0]`（等于 +40）。要得到 s2 的可能的動作，需要查詢`possible_actions[2]`（結果是 a1）。然后，必須將 Q-值初始化為 0（對于不可能的動作，Q-值設為 –∞）： ```py Q_values = np.full((3, 3), -np.inf) # -np.inf for impossible actions for state, actions in enumerate(possible_actions): Q_values[state, actions] = 0.0 # for all possible actions ``` 現在運行 Q-值迭代算法。它反復對 Q-值的每個狀態和可能的動作應用公式 18-3： ```py gamma = 0.90 # the discount factor for iteration in range(50): Q_prev = Q_values.copy() for s in range(3): for a in possible_actions[s]: Q_values[s, a] = np.sum([ transition_probabilities[s][a][sp] * (rewards[s][a][sp] + gamma * np.max(Q_prev[sp])) for sp in range(3)]) ``` Q-值的結果如下： ```py >>> Q_values array([[18.91891892, 17.02702702, 13.62162162], [ 0\. , -inf, -4.87971488], [ -inf, 50.13365013, -inf]]) ``` 例如，當智能體處于狀態 s0，選擇動 a1，衰減未來獎勵的期望和大約是 17.0。 對于每個狀態，查詢擁有最高 Q-值的動作： ```py >>> np.argmax(Q_values, axis=1) # optimal action for each state array([0, 0, 1]) ``` 這樣就得到了衰減因子等于 0.9 時，這個 MDP 的最佳策略是什么：狀態 s0 時選擇動作 a0；在狀態 s1 時選擇動作 a0；在狀態 s2 時選擇動作 a1。有趣的是，如果將衰減因子提高到 0.95，最佳策略發生了改變：在狀態 s1 時，最佳動作變為 a2（通過火！）。道理很明顯，如果未來期望越高，忍受當前的痛苦是值得的。 ## 時間差分學習 具有離散動作的強化學習問題通常可以被建模為馬爾可夫決策過程，但是智能體最初不知道轉移概率是什么（它不知道`T(s, a, s′)`），并且它不知道獎勵會是什么（它不知道`R(s, a, s′)`）。它必須經歷每一個狀態和每一次轉變并且至少知道一次獎勵，并且如果要對轉移概率進行合理的估計，就必須經歷多次。 時間差分學習（TD 學習）算法與數值迭代算法非常類似，但考慮到智能體僅具有 MDP 的部分知識。一般來說，我們假設智能體最初只知道可能的狀態和動作，沒有更多了。智能體使用探索策略，例如，純粹的隨機策略來探索 MDP，并且隨著它的發展，TD 學習算法基于實際觀察到的轉換和獎勵來更新狀態值的估計（見公式 18-4）。 公式 18-4 TD 學習算法 在這個公式中： * `α`是學習率（例如 0.01）。 * `r + γ · Vk(s′)`被稱為 TD 目標。 * `δk(s, r, s′)`被稱為 TD 誤差。 公式的第一種形式的更為準確的表達，是使用  ，它的意思是 ak+1 ← (1 – α) · ak + α ·bk，公式 18-4 的第一行可以重寫為：  > 提示：TD 學習和隨機梯度下降有許多相似點，特別是 TD 學習每次只處理一個樣本。另外，和隨機梯度下降一樣，如果逐漸降低學習率，是能做到收斂的（否則，會在最佳 Q-值附近反復跳躍）。 對于每個狀態`S`，該算法只跟蹤智能體離開該狀態時立即獲得的獎勵的平均值，再加上它期望稍后得到的獎勵（假設它的行為最佳）。 ## Q-學習 類似地，Q-學習算法是 Q 值迭代算法的改編版本，其適應轉移概率和回報在初始未知的情況（見公式 18-5）。Q-學習通過觀察智能體玩游戲，逐漸提高 Q-值的估計。一旦有了準確（或接近）的 Q-值估計，則選擇具有最高 Q-值的動作（即，貪婪策略）。 公式 18-5 Q 學習算法 對于每一個狀態動作對`(s,a)`，該算法跟蹤智能體在以動作`A`離開狀態`S`時獲得的即時獎勵平均值`R`，加上它期望稍后得到的獎勵。由于目標策略將最優地運行，所以我們取下一狀態的 Q 值估計的最大值。 以下是如何實現 Q-學習算法。首先，需要讓一個智能體探索環境。要這么做的話，我們需要一個步驟函數，好讓智能體執行一個動作，并返回結果狀態和獎勵： ```py def step(state, action): probas = transition_probabilities[state][action] next_state = np.random.choice([0, 1, 2], p=probas) reward = rewards[state][action][next_state] return next_state, reward ``` 現在，實現智能體的探索策略。因為狀態空間很小，使用簡單隨機策略就可以。如果長時間運行算法，智能體會多次訪問每個狀態，也會多次嘗試每個可能的動作： ```py def exploration_policy(state): return np.random.choice(possible_actions[state]) ``` 然后，和之前一樣初始化 Q-值，使用學習率遞降的方式運行 Q-學習算法（使用第 11 章介紹過的指數調度算法）： ```py alpha0 = 0.05 # initial learning rate decay = 0.005 # learning rate decay gamma = 0.90 # discount factor state = 0 # initial state for iteration in range(10000): action = exploration_policy(state) next_state, reward = step(state, action) next_value = np.max(Q_values[next_state]) alpha = alpha0 / (1 + iteration * decay) Q_values[state, action] *= 1 - alpha Q_values[state, action] += alpha * (reward + gamma * next_value) state = next_state ``` 算法會覆蓋最優 Q-值，但會經歷多次迭代，可能有許多超參數調節。見圖 18-9，Q-值迭代算法（左）覆蓋速度很快，只用了不到 20 次迭代，而 Q-學習算法（右）用了 8000 次迭代才覆蓋完。很明顯，不知道過渡概率或獎勵，使得找到最佳策略顯著變難！ 圖 18-9 Q-值迭代算法（左）對比 Q-學習算法（右） Q-學習被稱為離線策略算法，因為正在訓練的策略不是正在執行的策略：在前面的例子中，被執行的策略（探索策略）是完全隨機的，而訓練的算法總會選擇具有最高 Q-值的動作。相反的，策略梯度下降算法是在線算法：使用訓練的策略探索世界。令人驚訝的是，該算法能夠通過觀察智能體的隨機行為（例如當你的老師是一個醉猴子時，學習打高爾夫球）學習最佳策略。我們能做得更好嗎？ ### 探索策略 當然，只有在探索策略充分探索 MDP 的情況下，Q 學習才能起作用。盡管一個純粹的隨機策略保證最終訪問每一個狀態和每個轉換多次，但可能需要很長的時間這樣做。因此，一個更好的選擇是使用 ε 貪婪策略：在每個步驟中，它以概率`ε`隨機地或以概率為`1-ε`貪婪地選擇具有最高 Q 值的動作。ε 貪婪策略的優點（與完全隨機策略相比）是，它將花費越來越多的時間來探索環境中有趣的部分，因為 Q 值估計越來越好，同時仍花費一些時間訪問 MDP 的未知區域。以`ε`為很高的值（例如，1）開始，然后逐漸減小它（例如，下降到 0.05）是很常見的。 或者，不依賴于探索的可能性，另一種方法是鼓勵探索策略來嘗試它以前沒有嘗試過的行動。這可以被實現為加到 Q-值估計的獎勵，如公式 18-6 所示。 公式 18-6 使用探索函數的 Q-學習 在這個公式中： * `N(s′, a′)`計算了在狀態`s`時選擇動作`a`的次數 * `f(Q, N)`是一個探索函數，例如`f(Q, N) = Q + κ/(1 + N)`，其中`κ`是一個好奇超參數，它測量智能體被吸引到未知狀態的程度。 ### 近似 Q 學習和深度 Q-學習 Q 學習的主要問題是，它不能很好地擴展到具有許多狀態和動作的大（甚至中等）的 MDP。例如，假如你想用 Q 學習來訓練一個智能體去玩 Ms. Pac-Man（圖 18-1）。Ms. Pac-Man 可以吃超過 150 粒粒子，每一粒都可以存在或不存在（即已經吃過）。因此，可能狀態的數目大于 21⁵⁰ ≈ 10⁴⁵。空間大小比地球的的總原子數要多得多，所以你絕對無法追蹤每一個 Q 值的估計值。 解決方案是找到一個函數 Q_θ(s,a)，使用可管理數量的參數（根據矢量θ）來近似 Q 值。這被稱為近似 Q 學習。多年來，人們都是手工在狀態中提取并線性組合特征（例如，最近的鬼的距離，它們的方向等）來估計 Q 值，但在 2013 年， DeepMind 表明使用深度神經網絡可以工作得更好，特別是對于復雜的問題。它不需要任何特征工程。用于估計 Q 值的 DNN 被稱為深度 Q 網絡（DQN），并且使用近似 Q 學習的 DQN 被稱為深度 Q 學習。 如何訓練 DQN 呢？這里用 DQN 在給定的狀態動作對(s,a)，來估計 Q-值。感謝 Bellman，我們知道這個近似 Q-值要接近在狀態 s 執行動作 a 的獎勵 r，加上之前的衰減獎勵。要估計未來衰減獎勵的和，我們只需在下一個狀態 s'，對于所有可能的動作 a‘執行 DQN。針對每個可能的動作，獲得了近似的 Q-值。然后挑選最高的，并做衰減，就得到了未來衰減獎勵的和。通過將獎勵 r 和未來衰減獎勵估計相加，得到了狀態動作對(s, a)的目標 Q-值 y(s, a)，見公式 18-7。 公式 18-7 目標 Q-值 有了這個目標 Q-值，可以使用梯度下降運行一步訓練算法。具體地，要最小化 Q-值 Q(s, a)和目標 Q-值的平方根方差（或使用 Huber 損失降低算法對大誤差的敏感度）。這就是基礎的深度 Q-學習算法。下面用其處理平衡車問題。 ## 實現深度 Q-學習 首先需要的是一個深度 Q-網絡。理論上，需要一個輸入是狀態-動作對、輸出是近似 Q-值的神經網絡，但在實際中，使用輸入是狀態、輸出是每個可能動作的近似 Q 值的神經網絡，會更加高效。要處理 CartPole 環境，我們不需要非常復雜的神經網絡；只要幾個隱藏層就夠了： ```py env = gym.make("CartPole-v0") input_shape = [4] # == env.observation_space.shape n_outputs = 2 # == env.action_space.n model = keras.models.Sequential([ keras.layers.Dense(32, activation="elu", input_shape=input_shape), keras.layers.Dense(32, activation="elu"), keras.layers.Dense(n_outputs) ]) ``` 使用這個 DQN 選擇一個動作，選擇 Q-值最大的動作。要保證智能體探索環境，使用的是ε-貪婪策略（即，選擇概率為ε的隨機動作）： ```py def epsilon_greedy_policy(state, epsilon=0): if np.random.rand() < epsilon: return np.random.randint(2) else: Q_values = model.predict(state[np.newaxis]) return np.argmax(Q_values[0]) ``` 不僅只根據最新的經驗訓練 DQN，將所有經驗存儲在接力緩存（或接力記憶）中，每次訓練迭代，從中隨機采樣一個批次。這樣可以降低訓練批次中的經驗相關性，可以極大的提高訓練效果。如下，使用雙端列表實現： ```py from collections import deque replay_buffer = deque(maxlen=2000) ``` > 提示：雙端列表是一個鏈表，每個元素指向后一個和前一個元素。插入和刪除元素都非常快，但雙端列表越長，隨機訪問越慢。如果需要一個非常大的接力緩存，可以使用環狀緩存；見 notebook 中的“Deque vs Rotating List”章節。 每個經驗包含五個元素：狀態，智能體選擇的動作，獎勵，下一個狀態，一個知識是否結束的布爾值（`done`）。需要一個小函數從接力緩存隨機采樣。返回的是五個 NumPy 數組，對應五個經驗： ```py def sample_experiences(batch_size): indices = np.random.randint(len(replay_buffer), size=batch_size) batch = [replay_buffer[index] for index in indices] states, actions, rewards, next_states, dones = [ np.array([experience[field_index] for experience in batch]) for field_index in range(5)] return states, actions, rewards, next_states, dones ``` 再創建一個使用ε-貪婪策略的單次玩游戲函數，然后將結果經驗存儲在接力緩存中： ```py def play_one_step(env, state, epsilon): action = epsilon_greedy_policy(state, epsilon) next_state, reward, done, info = env.step(action) replay_buffer.append((state, action, reward, next_state, done)) return next_state, reward, done, info ``` 最后，再創建最后一個批次采樣函數，用單次梯度下降訓練這個 DQN： ```py batch_size = 32 discount_factor = 0.95 optimizer = keras.optimizers.Adam(lr=1e-3) loss_fn = keras.losses.mean_squared_error def training_step(batch_size): experiences = sample_experiences(batch_size) states, actions, rewards, next_states, dones = experiences next_Q_values = model.predict(next_states) max_next_Q_values = np.max(next_Q_values, axis=1) target_Q_values = (rewards + (1 - dones) * discount_factor * max_next_Q_values) mask = tf.one_hot(actions, n_outputs) with tf.GradientTape() as tape: all_Q_values = model(states) Q_values = tf.reduce_sum(all_Q_values * mask, axis=1, keepdims=True) loss = tf.reduce_mean(loss_fn(target_Q_values, Q_values)) grads = tape.gradient(loss, model.trainable_variables) optimizer.apply_gradients(zip(grads, model.trainable_variables)) ``` 逐行看下代碼： * 首先定義一些超參數，并創建優化器和損失函數。 * 然后創建`training_step()`函數。先采樣經驗批次，然后使用 DQN 預測每個可能動作的每個經驗的下一狀態的 Q-值。因為假定智能體采取最佳行動，所以只保留下一狀態的最大 Q-值。接著，我們使用公式 18-7 計算每個經驗的狀態-動作對的目標 Q-值。 * 接著，使用 DQN 計算每個有經驗的狀態-動作對的 Q-值。但是，DQN 還會輸出其它可能動作的 Q-值，不僅是智能體選擇的動作。所以，必須遮掩不需要的 Q-值。`tf.one_hot()`函數可以方便地將動作下標的數組轉別為 mask。例如，如果前三個經驗分別包含動作 1，1，0，則 mask 會以`[[0, 1], [0, 1], [1, 0],...]`開頭。然后將 DQN 的輸出乘以這個 mask，就可以排除所有不需要的 Q-值。然后，按列求和，去除所有的零，只保留有經驗的狀態-動作對的 Q-值。得到張量`Q_values`，包含批次中每個經驗的預測的 Q-值。 * 然后，計算損失：即有經驗的狀態-動作對的目標 Q-值和預測 Q-值的均方誤差。 * 最后，對可訓練變量，用梯度下降步驟減小損失。 這是最難的部分。現在，訓練模型就簡單了： ```py for episode in range(600): obs = env.reset() for step in range(200): epsilon = max(1 - episode / 500, 0.01) obs, reward, done, info = play_one_step(env, obs, epsilon) if done: break if episode > 50: training_step(batch_size) ``` 跑 600 次游戲，每次最多 200 步。在每一步，先計算ε-貪婪策略的`epsilon`值：這個值在 500 個周期內，從 1 線性降到 0.01。然后調用`play_one_step()`函數，用ε-貪婪策略挑選動作，然后執行并在接力緩存中記錄經驗。如果周期結束，就退出循環。最后，如果超過了 50 個周期，就調用`training_step()`函數，用從接力緩存取出的批次樣本訓練模型。玩 50 個周期，而不訓練的原因是給接力緩存一些時間來填充（如果等待的不夠久，則接力緩存中的樣本散度太小）。像上面這樣，我們就實現了深度 Q-學習算法。 圖 18-10 展示了智能體在每個周期獲得的總獎勵。 圖 18-10 深度 Q 學習算法的學習曲線 可以看到，在前 300 個周期，算法的進步不大（部分是因為ε在一開始時非常高），然后表現突然提升到了 200（環境最高值）。這說明算法效果不錯，并且比策略梯度算法快得多！但僅僅幾個周期之后，性能就驟降到了 25。這被稱為“災難性遺忘”，這是所有 RL 算法都面臨的大問題：隨著智能體探索環境，不斷更新策略，但是在環境的一部分學到的內容可能和之前學到的內容相悖。經驗是關聯的，學習環境不斷改變 —— 這不利于梯度下降！如果增加接力緩存的大小，可以減輕這個問題。但真實的情況是，強化學習很難：訓練通常不穩定，需要嘗試許多超參數值和隨機種子。例如，如果改變每層神經元的數量，從 32 到 30 或 34，模型表現不會超過 100（DQN 只有一個隱藏層時，可能更穩定）。 > 筆記：強化學習非常困難，很大程度是因為訓練的不穩定性，以及巨大的超參數和隨機種子的不穩定性。就像 Andrej Karpathy 說的：“監督學習自己就能工作，強化學習被迫工作”。你需要時間、耐心、毅力，還有一點運氣。這是為什么強化學習不是常用的深度學習算法的原因。除了 AlphaGo 和 Atari 游戲，還有一些其它應用：例如，Google 使用 RL 優化數據中心的費用，也用于一些機器人應用的超參數調節，和推薦系統。 你可能想為什么我們不畫出損失。事實證明損失不是模型表現的好指標。就算損失下降，智能體的表現也可能更糟（例如，智能體困在了環境中，則 DQN 開始對區域過擬合）。相反的，損失可能變大，但智能體表現不錯（例如，如果 DQN 知道 Q-值，就能提高預測的質量，智能體就能表現得更好，得到更多獎勵，但因為 DQN 還設置了更大的目標，所以誤差增加了）。 我們現在學的基本的深度 Q-學習算法，在玩 Atari 時太不穩定。DeepMind 是怎么做的呢？他們調節了算法。 ## 深度 Q-學習的變體 下面看幾個深度 Q-學習算法的變體，它們不僅訓練穩定而且很快。 ### 固定 Q-值目標 在基本的深度 Q-學習算法中，模型不僅做預測還自己設置目標。有點像一只狗追自己的尾巴。反饋循環使得網絡不穩定：會發生分叉、搖擺、凍結，等等。要解決問題，DeepMind 在 2013 年的論文中使用了兩個 DQN，而不是一個：第一個是在線模型，它在每一步進行學習，并移動智能體；另一個是目標模型只定義目標。目標模型只是在線模型的克隆： ```py target = keras.models.clone_model(model) target.set_weights(model.get_weights()) ``` 然后，在`training_step()`函數中，只需要變動一行，使用目標模型計算接下來狀態的 Q-值： ```py next_Q_values = target.predict(next_states) ``` 最后，在訓練循環中，必須每隔一段周期（比如，每 50 個周期），將在線模型的權重復制到目標模型中： ```py if episode % 50 == 0: target.set_weights(model.get_weights()) ``` 因為目標模型更新的沒有在線模型頻繁，Q-值目標更加穩定，前面討論反饋循環減弱了。這個方法是 DeepMind 在 2013 年的論文中提出的方法之一，可以讓智能體從零學習 Atari 游戲。要穩定訓練，他們使用的學習率是 0.00025，很小，每隔 10000 步才更新目標模型，接力緩存的大小是 1 百萬。并且`epsilon`降低的很慢，用 1 百萬步從 1 降到 0.1，他們讓算法運行了 5000 萬步。 本章后面會用這些超參數，使用 TF-Agents 庫訓練 DQN 智能體來玩 Breakout。在此之前，再看另一個性能更好的 DQN 變體。 ### 雙 DQN 在[2015 年的論文](https://links.jianshu.com/go?to=https%3A%2F%2Fhoml.info%2Fdoubledqn)中，DeepMind 調節了他們的 DQN 算法，提高了性能，也穩定化了訓練。他們稱這個變體為雙 DQN。算法更新的原因，是觀察到目標網絡傾向于高估 Q-值。事實上，假設所有動作都一樣好：目標模型預測的 Q-值應該一樣，但因為是估計值，其中一些可能存在更大的幾率。目標模型會選擇最大的 Q-值，最大的 Q-值要比平均 Q-值稍大，就像高估真正的 Q 值（就像在測量池塘深度時，測量隨機水波的最高峰）。要修正這個問題，他們提出使用在線模型，而不是目標模型，來選擇下一狀態的最佳動作，只用目標模型估計這些最佳動作的 Q-值。下面是改善后的`training_step()`函數： ```py def training_step(batch_size): experiences = sample_experiences(batch_size) states, actions, rewards, next_states, dones = experiences next_Q_values = model.predict(next_states) best_next_actions = np.argmax(next_Q_values, axis=1) next_mask = tf.one_hot(best_next_actions, n_outputs).numpy() next_best_Q_values = (target.predict(next_states) * next_mask).sum(axis=1) target_Q_values = (rewards + (1 - dones) * discount_factor * next_best_Q_values) mask = tf.one_hot(actions, n_outputs) [...] # the rest is the same as earlier ``` 幾個月之后，人們又提出了另一個改進的 DQN 算法。 ### 優先經驗接力 除了均勻地從接力緩存采樣經驗，如果更頻繁地采樣重要經驗如何呢？這個主意被稱為重要性采樣（importance sampling，IS）或優先經驗接力（prioritized experience replay，PER），是在[2015 年的論文中](https://links.jianshu.com/go?to=https%3A%2F%2Fhoml.info%2Fprioreplay)由 DeepMind 發表的。 更具體的，可以導致快速學習成果的經驗被稱為重要經驗。但如何估計呢？一個可行的方法是測量 TD 誤差的大小 δ = r + γ·V(s′) – V(s) 。大 TD 誤差說明過渡 (s, r, s′) 很值得學習。當經驗記錄在接力緩存中，它的重要性被設為非常大的值，保證可以快速采樣。但是，一旦被采樣（以及每次采樣時），就計算 RD 誤差δ，這個經驗的優先度設為 p = |δ| （加上一個小常數，保證每個經驗的采樣概率不是零）。采樣優先度為 p 的概率 P 正比于 p^ζ，ζ是調整采樣貪婪度的超參數：當ζ=0 時，就是均勻采樣，ζ=1 時，就是完全的重要性采樣。在論文中，作者使用的是ζ=0.6，最優值取決于任務。 但有一點要注意，因為樣本偏向重要經驗，必須要在訓練時，根據重要性降低經驗的重要性，否則模型會對重要經驗過擬合。更加清楚的講，重要經驗采樣更頻繁，但訓練時的權重要小。要這么做，將每個經驗的訓練權重定義為 w = (n P)^–β，n 是接力緩存的經驗數，β是平衡重要性偏向的超參數（0 是不偏向，1 是完全偏向）。在論文中，作者一開始使用的是β=0.4，在訓練結束，提高到了β=1。最佳值取決于任務，如果你提高了一個，也要提高其它的值。 接下來是最后一個重要的 DQN 算法的變體。 ### 對決 DQN 對決 DQN 算法（DDQN，不要與雙 DQN 混淆）是 DeepMind 在另一篇[2015 年的論文](https://links.jianshu.com/go?to=https%3A%2F%2Fhoml.info%2Fddqn)中提出的。要明白原理，首先狀態-動作對（s,a）的 Q-值，可以表示為 Q(s, a) = V(s) + A(s, a)，其中 V(s)是狀態 s 的值，A(s, a)是狀態 s 采取行動 a 的結果。另外，狀態的值等于狀態最佳動作 a*的 Q-值（因為最優策略會選最佳動作），因此 V(s) = Q(s, a*)，即 A(s, a*) = 0。在對決 DQN 中，模型估計狀態值和每個動作的結果。因為最佳動作的結果是 0，模型減去最大預測結果。下面是一個簡單的對決 DQN，用 Functional API 實現： ```py K = keras.backend input_states = keras.layers.Input(shape=[4]) hidden1 = keras.layers.Dense(32, activation="elu")(input_states) hidden2 = keras.layers.Dense(32, activation="elu")(hidden1) state_values = keras.layers.Dense(1)(hidden2) raw_advantages = keras.layers.Dense(n_outputs)(hidden2) advantages = raw_advantages - K.max(raw_advantages, axis=1, keepdims=True) Q_values = state_values + advantages model = keras.Model(inputs=[input_states], outputs=[Q_values]) ``` 算法的其余部分和之前一樣。事實上，你可以創建一個雙對決 DQN，并結合優先經驗隊列！更為一般地，許多 RL 方法都可以結合起來，就像 DeepMind 在[2017 年的論文](https://links.jianshu.com/go?to=https%3A%2F%2Fhoml.info%2Frainbow)展示的。論文的作者將六個不同的方法結合起來，訓練了一個智能體，稱為“彩虹”，表現很好。 不過，要實現所有這些方法，進行調試、微調，并且訓練模型需要很多工作。因此，不要重新草輪子，最好的方法是復用可擴展的、使用效果好的庫，比如 TF-Agents。 ## TF-Agents 庫 [TF-Agents 庫](https://links.jianshu.com/go?to=https%3A%2F%2Fgithub.com%2Ftensorflow%2Fagents)是基于 TensorFlow 實現的強化學習庫，Google 開發并在 2018 年開源。和 OpenAI Gym 一樣，它提供了許多現成的環境（包括了 OpenAI Gym 環境的包裝），還支持 PyBullet 庫（用于 3D 物理模擬），DeepMind 的 DM 控制庫（基于 MuJoCo 的物理引擎），Unity 的 ML-Agents 庫（模擬了許多 3D 環境）。它還使用了許多 RL 算法，包括 REINFORCE、DQN、DDQN，和各種 RL 組件，比如高效接力緩存和指標。TF-Agents 速度快、可擴展、便于使用、可自定義：你可以創建自己的環境和神經網絡，可以對任意組件自定義。在這一節，我們使用 TF-Agents 訓練一個智能體玩 Breakout，一個有名的 Atari 游戲（見圖 18-11），使用的是 DQN 算法（可以換成任何你想用的算法）。 圖 18-11 Breakout 游戲 ### 安裝 TF-Agents 先安裝 TF-Agents 。可以使用 pip 安裝（如果使用的是虛擬環境，一定要先激活；如果不激活，要使用選項`--user`，或用管理員權限）： ```py $ python3 -m pip install --upgrade tf-agents ``` > 警告：寫作本書時，TF-Agents 還很新，每天都有新進展，因此 API 可能會和現在有所不同 —— 但大體相同。如果代碼不能運行，我會更新 Jupyter notebook。 然后，創建一個 TF-Agents 包裝了 OpenAI GGym 的 Breakout 的環境。要這么做，需要先安裝 OpenAI Gym 的 Atari 依賴： ```py $ python3 -m pip install --upgrade 'gym[atari]' ``` 這條命令安裝了`atari-py`，這是 Arcade 學習環境的 Python 接口，這個學習環境是基于 Atari 2600 模擬器 Stella。 ### TF-Agents 環境 如果一切正常，就能引入 TF-Agents，創建 Breakout 環境了： ```py >>> from tf_agents.environments import suite_gym >>> env = suite_gym.load("Breakout-v4") >>> env <tf_agents.environments.wrappers.TimeLimit at 0x10c523c18> ``` 這是 OpenAI Gym 環境的包裝，可以通過屬性`gym`訪問： ```py >>> env.gym <gym.envs.atari.atari_env.AtariEnv at 0x24dcab940> ``` TF-Agents 環境和 OpenAI Gym 環境非常相似，但有些差別。首先，`reset()`方法不返回觀察；返回的是`TimeStep`對象，它包裝了觀察，和一些其它信息： ```py >>> env.reset() TimeStep(step_type=array(0, dtype=int32), reward=array(0., dtype=float32), discount=array(1., dtype=float32), observation=array([[[0., 0., 0.], [0., 0., 0.],...]]], dtype=float32)) ``` `step()`方法返回的也是`TimeStep`對象： ```py >>> env.step(1) # Fire TimeStep(step_type=array(1, dtype=int32), reward=array(0., dtype=float32), discount=array(1., dtype=float32), observation=array([[[0., 0., 0.], [0., 0., 0.],...]]], dtype=float32)) ``` 屬性`reward`和`observation`是獎勵和觀察，與 OpenAI Gym 相同（除了`reward`表示為 NumPy 數組）。對于周期的第一個時間步，屬性`step_type`等于 0，1 是中間步，2 后最后一步。可以調用時間步的`is_last()`方法，檢測是否是最后一步。最后，`discount`屬性指明了在這個時間步的衰減率。在這個例子中的值等于 1，所以沒有任何衰減。可以通過在加載環境時設置`discount`參數，定義衰減因子。 > 筆記：在任何時候，你可以通過調用方法`current_time_step() method.`訪問環境的當前時間步。 ### 環境配置 TF-Agents 環境提供了配置，包括觀察、動作、時間步，以及它們的形狀、數據類型、名字，還有最小值和最大值： ```py >>> env.observation_spec() BoundedArraySpec(shape=(210, 160, 3), dtype=dtype('float32'), name=None, minimum=[[[0\. 0\. 0.], [0\. 0\. 0.],...]], maximum=[[[255., 255., 255.], [255., 255., 255.], ...]]) >>> env.action_spec() BoundedArraySpec(shape=(), dtype=dtype('int64'), name=None, minimum=0, maximum=3) >>> env.time_step_spec() TimeStep(step_type=ArraySpec(shape=(), dtype=dtype('int32'), name='step_type'), reward=ArraySpec(shape=(), dtype=dtype('float32'), name='reward'), discount=BoundedArraySpec(shape=(), ..., minimum=0.0, maximum=1.0), observation=BoundedArraySpec(shape=(210, 160, 3), ...)) ``` 可以看到，觀察就是 Atari 屏幕的截圖，用形狀是 [210, 160, 3] 的 NumPy 數組表示。要渲染環境，可以調用`env.render(mode="human")`，如果想用 NumPy 數組的形式返回圖片，可以調用`env.render(mode="rgb_array")`（與 OpenAI Gym 不同，這是默認模式）。 有四個可能的動作。Gym 的 Atari 環境有另一個方法，可以知道每個動作對應什么： ```py >>> env.gym.get_action_meanings() ['NOOP', 'FIRE', 'RIGHT', 'LEFT'] ``` > 提示：配置是配置類的一個實例，可以是嵌套列表、字典。如果配置是嵌套的，則配置對象必須匹配配置的嵌套結構。例如，如果觀察配置是 `{"sensors": ArraySpec(shape=[2]), "camera": ArraySpec(shape=[100, 100])}` ，有效觀察應該是 `{"sensors": np.array([1.5, 3.5]), "camera": np.array(...)}`。`tf.nest`包提供了工具處理嵌套結構（即，`nests`）。 觀察結果很大，所以需要做降采樣，并轉換成灰度。這樣可以加速訓練，減少內存使用。要這么做，要使用環境包裝器。 ### 環境包裝器和 Atari 預處理 TF-Agents 在`tf_agents.environments.wrappers`中，提供了一些環境包裝器。正如名字，它們可以包裝環境，轉發每個調用，還可以添加其它功能。以下是一些常見的包裝器： `ActionClipWrapper` * 根據動作配置裁剪動作。 `ActionDiscretizeWrapper` * 將連續動作空間量化到離散的動作空間。例如，如果原始環境的動作空間是 -1.0 到 +1.0 的連續范圍，但是如果想用算法支持離散的動作空間，比如 DQN，就可以用`discrete_env = ActionDiscretizeWrapper(env, num_actions=5)`包裝環境，新的`discrete_env`有離散的可能動作空間：0、1、2、3、4。這些動作對應原始環境的動作-1.0、-0.5、0.0、0.5、1.0。 `ActionRepeat` * 將每個動作重復 n 次，并積累獎勵。在許多環境中，這么做可以顯著加速訓練。 `RunStats` * 記錄環境數據，比如步驟數和周期數。 `TimeLimit` * 超過最大的時間步數，則中斷環境。 `VideoWrapper` * 記錄環境的視頻。 要創建包裝環境，需要先創建一個包裝器，將包裝過的環境傳遞給構造器。例如，下面的代碼將一個環境包裝在`ActionRepeat`中，讓每個動作重復四次： ```py from tf_agents.environments.wrappers import ActionRepeat repeating_env = ActionRepeat(env, times=4) ``` OpenAI Gym 在`gym.wrappers`中有一些環境包裝器。但它們是用來包裝 Gym 環境，不是 TF-Agents 環境，所以要使用的話，必須用 Gym 包裝器包裝 Gym 環境，再用 TF-Agents 包裝器再包裝起來。`suite_gym.wrap_env()`函數可以實現，只要傳入 Gym 環境和 Gym 包裝器列表，和/或 TF-Agents 包裝器的列表。另外，`suite_gym.load()`函數既能創建 Gym 環境，如果傳入包裝器，也能做包裝。每個包裝器在包裝時沒有參數，所以如果想設置參數，必須傳入`lambda`。例如，下面的代碼創建了一個 Breakout 環境，每個周期最多運行 10000 步，每個動作重復四次： ```py from gym.wrappers import TimeLimit limited_repeating_env = suite_gym.load( "Breakout-v4", gym_env_wrappers=[lambda env: TimeLimit(env, max_episode_steps=10000)], env_wrappers=[lambda env: ActionRepeat(env, times=4)]) ``` 對于 Atari 環境，大多數論文使用了標準預處理步驟，TF-Agents 提供了便捷的`AtariPreprocessing`包裝器做預處理。以下是支持的預處理： 灰度和降采樣 * 將觀察轉換為灰度，并降采樣（默認是 84 × 84 像素） 最大池化 * 游戲的最后兩幀使用 1 × 1 過濾器做最大池化。是為了去除閃爍點。 跳幀 * 智能體每隔 n 個幀做一次觀察（默認是 4），對于每一幀，動作都要重復幾次，并收集所有的獎勵。這么做可以有效加速游戲，因為獎勵延遲降低，訓練也加速了。 丟命損失 在某些游戲中，獎勵是基于得分的，所以智能體死掉的話，不會立即受到懲罰。一種方法是當死掉時，立即結束游戲。這種做法有些爭議，所以默認是關掉的。 因為默認 Atari 環境已經應用了隨機跳幀和最大池化，我們需要加載原生不跳幀的變體，`BreakoutNoFrameskip-v4`。另外，從 Breakout 游戲中的一幀并不能知道球的方向和速度，這會使得智能體很難玩好游戲（除非這是一個 RNN 智能體，它可以在步驟之間保存狀態）。應對方法之一是使用一個環境包裝器，沿著每個頻道維度，將多個幀疊起來做輸出。`FrameStack4`包裝器實現了這個策略，返回四個幀的棧式結果。下面就創建一個包裝過的 Atari 環境。 ```py from tf_agents.environments import suite_atari from tf_agents.environments.atari_preprocessing import AtariPreprocessing from tf_agents.environments.atari_wrappers import FrameStack4 max_episode_steps = 27000 # <=> 108k ALE frames since 1 step = 4 frames environment_name = "BreakoutNoFrameskip-v4" env = suite_atari.load( environment_name, max_episode_steps=max_episode_steps, gym_env_wrappers=[AtariPreprocessing, FrameStack4]) ``` 預處理的結果展示在圖 18-12 中。可以看到解析度更低了，但足夠玩游戲了。另外，幀沿著頻道維度疊起來，所以紅色表示的是三步之前到現在的幀，綠色是從兩步之前，藍色是前一幀，粉色是當前幀。根據這一幀的觀察，智能體可以看到球是像左下角移動的，所以應該繼續將板子向左移動（和前面一步相同）。 圖 18-12 預處理 Breakout 觀察 最后，可以將環境包裝進`TFPyEnvironment`： ```py from tf_agents.environments.tf_py_environment import TFPyEnvironment tf_env = TFPyEnvironment(env) ``` 這樣就能在 TensorFlow 圖中使用這個環境（在底層，這個類使用的是`tf.py_function()`，這可以讓圖調用任何 Python 代碼）。有了`TFPyEnvironment`類，TF-Agents 支持純 Python 環境和基于 TensorFlow 環境。更為一般的，TF-Agents 支持并提供了純 Python 和基于 TensorFlow 的組件（智能體，接力緩存，指標，等等）。 有了 Breakout 環境，預處理和 TensorFlow 支持，我們必須創建 DQN 智能體，和其它要訓練的組件。下面看看系統架構。 ### 訓練架構 TF-Agents 訓練程序通常分為兩個并行運行的部分，見圖 18-13：左邊，driver 使用收集策略探索環境選擇動作，并收集軌跡（即，經驗），將軌跡發送給 observer，observer 將軌跡存儲到接力緩存中；右邊，智能體從接力緩存中取出軌跡批次，然后訓練網絡。總而言之，左邊的部分探索環境、收集軌跡，右邊的部分學習更新收集策略。 圖 18-13 一個典型的 TF-Agents 訓練架構 這張圖有些疑惑點，回答如下： * 為什么使用多個環境呢？這是為了讓 driver 并行探索多個環境的復制，發揮 CPU、GPU 的能力，給訓練算法提供低關聯的軌跡。 * 什么是軌跡？這是從一個時間步向下一個時間步過渡的簡潔表征，或是一連串連續的從時間步 n 到時間步 n+t 的過渡。driver 收集的軌跡傳給 observer，再將其存入接力緩存，接著再被采樣用來訓練。 * 為什么需要 observer？driver 不能直接保存軌跡嗎？事實上，driver 可以直接保存軌跡，但這么做的話，會使得架構不夠靈活。例如，如果不想使用接力緩存，該怎么做呢？如果想用軌跡做一些其它事情，比如計算指標，該怎么做呢？事實上，observer 是使用軌跡作為參數的任意函數。可以用 observer 將軌跡存入接力緩存，或保存為 TFRecord 文件，或計算指標，或其它事情。另外，可以將多個 observer 傳給 driver，廣播軌跡。 > 提示：盡管這個架構是最常見的，但是可以盡情自定義，可以更換成自己的組件。事實上，除非是研究新的 RL 算法，更適合使用自定義的環境來做自己的任務。要這么做，需要創建一個自定義類，繼承自`tf_agents.environments.py_environment`包的`PyEnvironment`類，并重寫一些方法，比如`action_spec()`、`observation_spec()`、`_reset()`、`_step()`（見 notebook 的章節 Creating a Custom TF_Agents Environment）。 現在創建好了所有組件：先是深度 Q-網絡，然后是 DQN 智能體（負責創建收集策略），然后是接力緩存和 observer，一些訓練指標，driver，最后是數據集。有了所有組件之后，先用一些軌跡填充接力緩存，然后運行主訓練循環。因此，從創建深度 Q-網絡開始。 ### 創建深度 Q-網絡 TF-Agents 庫在`tf_agents.networks`包和子包中提供了許多網絡。我們使用`tf_agents.networks.q_network.QNetwork`類： ```py from tf_agents.networks.q_network import QNetwork preprocessing_layer = keras.layers.Lambda( lambda obs: tf.cast(obs, np.float32) / 255.) conv_layer_params=[(32, (8, 8), 4), (64, (4, 4), 2), (64, (3, 3), 1)] fc_layer_params=[512] q_net = QNetwork( tf_env.observation_spec(), tf_env.action_spec(), preprocessing_layers=preprocessing_layer, conv_layer_params=conv_layer_params, fc_layer_params=fc_layer_params) ``` 這個`QNetwork`的輸入是觀察，每個動作輸出一個 Q-值，所以必須給出觀察和動作的配置。先是預處理層：一個`lambda`層將觀察轉換為 32 位浮點數，并做歸一化（范圍落到 0.0 和 1.0 之間）。觀察包含無符號字節，占用空間是 32 位浮點數的四分之一，這就是為什么不在前面將觀察轉換為 32 位浮點數；我們要節省接力緩存的內存空間。接著，網絡使用三個卷積層：第一個有 32 個 8 × 8 過濾器，步長是 4，第二個有 64 個 4 × 4 過濾器，步長是 2，第三個層有 64 個 3 × 3 的過濾器，步長為 1。最后，使用一個有 512 個神經元的緊密層，然后是一個有 4 個神經元的緊密輸出層，輸出是 Q-值（每個動作一個 Q-值）。所有卷積層和除了輸出層的緊密層使用 ReLU 激活函數（可以通過設置參數`activation_fn`改變）。輸出層不使用激活函數。 `QNetwork`的底層包含兩個部分：一個處理觀察的編碼網絡，和一個輸出 Q-值的輸出層。TF-Agent 的`EncodingNetwork`類實現了多種智能體都使用了的神經網絡架構（見圖 18-14）。 圖 18-14 編碼網絡架構 可能有一個或多個輸入。例如，如果每個觀察包括傳感器數據加上攝像頭圖片，就有兩個輸入。每個輸入可能需要一些預處理步驟，你可以通過`preprocessing_layers`參數指定 Keras 層列表，每個輸入有一個預處理層，網絡會將層應用到每個對應的輸入上（如果輸入需要多個預處理層，可以傳入一個完整模型，因為 Keras 模型也可以用作層）。如果有兩個或更多輸入，必須通過參數`preprocessing_combiner`傳入其它層，將預處理層的輸出合并成一個輸出。 然后，編碼層會順序應用一列卷積層，只要指定參數`conv_layer_params`。這是一個包含 3 個元組的列表（每個卷積層一個元組），指明過濾器的數量、核大小，步長。卷積層之后，如果設置參數`fc_layer_params`，編碼網絡可以應用一串緊密層：參數`fc_layer_params`是一個列表，包含每個緊密層的神經元數。另外，通過參數`dropout_layer_params`，還可以傳入 dropout 率列表（每個緊密層一個），給每個緊密成設置 dropout。`QNetwork`將編碼網絡的輸出傳入給緊密輸出層（每個動作一個神經元）。 > 筆記：`QNetwork`類非常靈活，可以創建許多不同的架構，如果需要更多的靈活性，還以通過創建自己的類：擴展類`tf_agents.networks.Network`，像常規自定義 Keras 層一樣實現。`tf_agents.networks.Network`類是`keras.layers.Layer`類的子類，前者添加了一些智能體需要的功能，比如創建網絡的淺復制（即，只復制架構，不復制權重）。例如，`DQNAgent`使用這個功能創建在線模型。 有了 DQN，接下來創建 DQN 智能體。 ### 創建 DQN 智能體 利用`tf_agents?.agents`包和它的子包，TF-Agents 庫實現了多種類型的智能體。我們使用類`tf_agents.agents?.dqn.dqn_agent.DqnAgent`： ```py from tf_agents.agents.dqn.dqn_agent import DqnAgent train_step = tf.Variable(0) update_period = 4 # train the model every 4 steps optimizer = keras.optimizers.RMSprop(lr=2.5e-4, rho=0.95, momentum=0.0, epsilon=0.00001, centered=True) epsilon_fn = keras.optimizers.schedules.PolynomialDecay( initial_learning_rate=1.0, # initial ε decay_steps=250000 // update_period, # <=> 1,000,000 ALE frames end_learning_rate=0.01) # final ε agent = DqnAgent(tf_env.time_step_spec(), tf_env.action_spec(), q_network=q_net, optimizer=optimizer, target_update_period=2000, # <=> 32,000 ALE frames td_errors_loss_fn=keras.losses.Huber(reduction="none"), gamma=0.99, # discount factor train_step_counter=train_step, epsilon_greedy=lambda: epsilon_fn(train_step)) agent.initialize() ``` 逐行看下代碼： * 首先創建計算訓練步驟數的變量。 * 然后創建優化器，使用 2015 DQN 論文相同的超參數。 * 接著，創建對象`PolynomialDecay`，根據當前的訓練步驟（用于降低學習率，也可以是其它值），用于計算ε-貪婪收集策略的ε值。在 100 萬 ALE 幀內（等于 250000 步驟，因為跳幀周期等于 4），將ε值從 1 降到 0.01（也是 2015 DQN 論文的用值）。另外，每隔 4 步（即，16 個 ALE 幀），所以ε值是在 62500 個訓練步內下降的。 * 然后創建`DQNAgent`，傳入時間步和動作配置、`QNetwork`、優化器、目標模型更新間的訓練步驟數、損失函數、衰減率、變量`train_step`、返回ε值的函數（不接受參數，這就是為什么使用匿名函數傳入`train_step`的原因）。注意，損失函數對每個實例返回一個誤差，不是平均誤差，所以要設置`reduction="none"`。 * 最后，啟動智能體。 然后，創建接力緩存和 observer。 ### 創建接力緩存和 observer TF-Agents 庫在`tf_agents.replay_buffers`包實現了多種接力緩存。一些是用純 Python 寫的（模塊名開頭是`py_`），其它是基于 TensorFlow 的（開頭是`tf_`）。我們使用`tf_agents.replay_buffers.tf_uniform_replay_buffer`包追蹤的`TFUniformReplayBuffer`類。它實現了高效均勻采樣的接力緩存： ```py from tf_agents.replay_buffers import tf_uniform_replay_buffer replay_buffer = tf_uniform_replay_buffer.TFUniformReplayBuffer( data_spec=agent.collect_data_spec, batch_size=tf_env.batch_size, max_length=1000000) ``` 看一下這些參數： `data_spec` * 數據的配置會存儲在接力緩存中。DQN 智能體知道收集數據什么樣，通過屬性`collect_data_spec`做數據配置。 `batch_size` * 軌跡數量添加到每個步驟。在這個例子中，軌跡數是 1，因為 driver 每個步驟執行一個動作收集一個軌跡。如果環境是一個批次化的環境（環境在每個時間步接收批次動作，返回批次觀察），則 driver 必須在每個時間步保存批次的軌跡。因為使用的是 TensorFlow 接力緩存，需要知道批次大小（創建計算圖）。批次化環境的一個例子是`ParallelPyEnvironment`（出自包`tf_agents.environments.parallel_py_environment`）：用獨立進程并行運行多個環境（對于相同的動作和觀察配置，進程可以不同），每個步驟接收批次化的動作，并在環境中執行（每個環境一個動作），然后返回所有觀察結果。 `max_length` * 接力緩存的最大大小。我們創建一個可以存儲 100 萬個軌跡的接力緩存（和 2015 DQN 論文一樣）。這需要不少內存。 > 提示：當存儲兩個連續的軌跡，它們包含兩個連續的觀察，每個觀察有四個幀（因為包裝器是`FrameStack4`），但是第二個觀察中的三個幀是多余的（第一個觀察中已經存在了）。換句話說，使用的內存大小是必須的四倍。要避免這個問題，可以使用包`tf_agents.replay_buffers.py_hashed_replay_buffer`的`PyHashedReplayBuffer`：它能沿著觀察的最后一個軸對存儲的軌跡去重。 現在創建向接力緩存寫入軌跡的 observer。observer 就是一個接收軌跡參數的函數（或是調用對象），所以可以直接使用方法`add_method()`（綁定`replay_buffer`對象）作為 observer： ```py replay_buffer_observer = replay_buffer.add_batch ``` 如果想創建自己的 observer，可以一個包含參數`trajectory`的函數。如果必須有狀態，可以寫一個包含方法`__call__(self, trajectory)`的類。例如，下面是一個每次調用，計數器都會加 1 的 observer（除了軌跡表示周期間的邊界，不算成一步），每隔 100 次累加，顯示總數（`\r`和`end=""`保證展示的計數器處于一條線）。 ```py class ShowProgress: def __init__(self, total): self.counter = 0 self.total = total def __call__(self, trajectory): if not trajectory.is_boundary(): self.counter += 1 if self.counter % 100 == 0: print("\r{}/{}".format(self.counter, self.total), end="") ``` 接下來創建一些訓練指標。 ### 創建訓練指標 TF-Agents 庫再`tf_agents.metrics`包中實現了幾個 RL 指標，一些是基于純 Python 的，一些是基于 TensorFlow 的。創建一些指標統計周期數、步驟數、周期的平均數、平均周期長度： ```py from tf_agents.metrics import tf_metrics train_metrics = [ tf_metrics.NumberOfEpisodes(), tf_metrics.EnvironmentSteps(), tf_metrics.AverageReturnMetric(), tf_metrics.AverageEpisodeLengthMetric(), ] ``` > 筆記：訓練或實現策略時，對獎勵做衰減是合理的，這是為了平衡當前獎勵與未來獎勵的平衡。但是，當周期結束時，可以通過對所有未衰減的獎勵求和來做評估。出于這個原因，`AverageReturnMetric`計算了每個周期未衰減獎勵的和，并追蹤平均值。 任何時候，可以調用`result()`方法獲取指標（例如，`train_metrics[0].result()`）。或者，可以調用`log_metrics(train_metrics)`記錄所有指標（這個函數位于`tf_agents.eval.metric_utils`包）： ```py >>> from tf_agents.eval.metric_utils import log_metrics >>> import logging >>> logging.get_logger().set_level(logging.INFO) >>> log_metrics(train_metrics) [...] NumberOfEpisodes = 0 EnvironmentSteps = 0 AverageReturn = 0.0 AverageEpisodeLength = 0.0 ``` 接下來創建收集 driver。 ### 創建收集 driver 正如圖 18-13，driver 是使用給定策略探索環境的對象，收集經驗，并廣播給 observer。在每一步，發生的事情如下： * driver 將當前時間步傳給收集策略，收集策略使用時間步選擇動作，并返回包含動作的動作步對象。 * driver 然后將動作傳給環境，環境返回下一個時間步。 * 最后，driver 創建一個軌跡對象表示過渡，并廣播給所有觀察。 一些策略，比如 RNN 策略，是有狀態的：策略根據給定的時間步和內部狀態選擇動作。有狀態策略在動作步返回自己的狀態，driver 會在下一個時間步將這個狀態返回給策略。另外，driver 將策略狀態保存到軌跡中（在字段`policy_info`中）：當智能體采樣一條軌跡，它必須設置策略的狀態設為采樣時間步時的狀態。 另外，就像前面討論的，環境可能是批次化的環境，這種情況下，driver 將批次化的時間步傳給策略（即，時間步對象包含批次觀察、批次步驟類型、批次獎勵、批次衰減，這四個批次的大小相同）。driver 還傳遞前一批次的策略狀態。然后，策略返回去批次動作步，包含著批次動作和批次策略狀態。最后，driver 創建批次化軌跡（即，軌跡包含批次步驟類型、批次觀察、批次動作、批次獎勵，更一般地，每個軌跡屬性一個批次，所有批次大小相同）。 有兩個主要的 driver 類：`DynamicStepDriver`和`DynamicEpisodeDriver`。第一個收集給定數量步驟的經驗，第二個收集給定數量周期數的經驗。我們想收集每個訓練迭代的四個步驟的經驗（正如 2015 DQN 論文），所以創建一個`DynamicStepDriver`： ```py from tf_agents.drivers.dynamic_step_driver import DynamicStepDriver collect_driver = DynamicStepDriver( tf_env, agent.collect_policy, observers=[replay_buffer_observer] + training_metrics, num_steps=update_period) # collect 4 steps for each training iteration ``` 傳入環境、智能體的收集策略、observer 列表（包括接力緩存 observer 和訓練指標），最后是要運行的步驟數（這個例子中是 4）。現在可以調用方法`run()`來運行，但最好先用純隨機策略收集的經驗先填充接力緩存。要這么做，可以使用類`RandomTFPolicy`創建第二個 driver，運行 20000 步這個策略（等于 80000 個模擬幀，正如 2015 DQN 論文）。可以用`ShowProgress` observer 展示進展： ```py from tf_agents.policies.random_tf_policy import RandomTFPolicy initial_collect_policy = RandomTFPolicy(tf_env.time_step_spec(), tf_env.action_spec()) init_driver = DynamicStepDriver( tf_env, initial_collect_policy, observers=[replay_buffer.add_batch, ShowProgress(20000)], num_steps=20000) # <=> 80,000 ALE frames final_time_step, final_policy_state = init_driver.run() ``` 快要能運行訓練循環了。只需要最后一個組件：數據集。 ### 創建數據集 要從接力緩存采樣批次的軌跡，可以調用`get_next()`方法。這返回了軌跡的批次，還返回了含有樣本 id 和采樣概率的`BufferInfo`對象（可能對有些算法有用，比如 PER）。例如，下面的代碼采樣了一個包含兩條軌跡的批次（子周期），每個包含三個連續步。這些子周期見圖 18-15（每行包含一個周期的三個連續步）： ```py >>> trajectories, buffer_info = replay_buffer.get_next( ... sample_batch_size=2, num_steps=3) ... >>> trajectories._fields ('step_type', 'observation', 'action', 'policy_info', 'next_step_type', 'reward', 'discount') >>> trajectories.observation.shape TensorShape([2, 3, 84, 84, 4]) >>> trajectories.step_type.numpy() array([[1, 1, 1], [1, 1, 1]], dtype=int32) ``` `trajectories`對象是一個命名元組，有 7 個字段。每個字段包含一個張量，前兩個維度是 2 和 3（因為有兩條軌跡，每個三個時間步）。這解釋了為什么`observation`字段的形狀是 [2, 3, 84, 84, 4] ：這是兩條軌跡，每條軌跡三個時間步，每步的觀察是 84 × 84 × 4。相似的，`step_type`張量的形狀是 [2, 3] ：在這個例子中，兩條軌跡包含三個連續步驟，步驟是在周期的中部，（類型是 1, 1, 1）。在第二條軌跡中，看不到第一個觀察中左下方的球，在接下來的兩個觀察中，球消失了，所以智能體會死，但周期不會馬上結束，因為還剩幾條命。 圖 18-15 包含三個連續步驟的兩條軌跡 每條軌跡是連續時間步和動作步的簡潔表征，初衷是為了避免繁瑣，怎么做呢？見圖 18-16，過渡 n 由時間步 n、動作步 n、時間步 n+1 組成，而過渡 n+1 由時間步 n+1、動作步 n+1、時間步 n+2。如果將這兩個過渡直接存入接力緩存，時間步 n+1 是重復的。為了避免重復，n^th軌跡步只包括時間步 n 的類型和觀察（不是獎勵和衰減），不包括時間步 n+1 的觀察（但是，不包括下一個時間步類型的復制）。 圖 18-16 軌跡，過渡，時間步和動作步 因此，如果有批次軌跡，每個軌跡有 t+1 步驟（從時間步 n 到時間步 n+t），包含從時間步 n 到時間步 n+t 的所有數據，但沒有獎勵和時間步 n 的衰減（但包括時間步 n+t+1 的獎勵和衰減）。這表示 t 過渡（n 到 n + 1, n + 1 到 n + 2, …, n + t – 1 到 n + t） 模塊`tf_agents.trajectories.trajectory`中的函數`to_transition()`將批次化的軌跡轉變為包含批次`time_step`、`action_step`、`next_time_step`的列表。注意，第二個維度是 2，而不是 3，這是因為 t + 1 個時間步之間有 t 個過渡： ```py >>> from tf_agents.trajectories.trajectory import to_transition >>> time_steps, action_steps, next_time_steps = to_transition(trajectories) >>> time_steps.observation.shape TensorShape([2, 2, 84, 84, 4]) # 3 time steps = 2 transitions ``` > 筆記：采樣的軌跡可能會將兩個（或多個）周期重疊！這種情況下，會包含邊界過渡，意味著過渡的`step_type`等于 2（結束），`next_step_type`等于 0（開始）。當然，TF-Agents 可以妥善處理這些軌跡（例如，通過在碰到邊界時重新設置策略狀態）。軌跡的方法`is_boundary()`返回只是每一步是否是邊界的張量。 對于主訓練循環，不使用`get_next()`，而是用`tf.data.Dataset`。這樣，就能借助 Data API 的高效（并行計算和預提取）。要這么做，可以調用接力緩存的`as_dataset()`方法： ```py dataset = replay_buffer.as_dataset( sample_batch_size=64, num_steps=2, num_parallel_calls=3).prefetch(3) ``` 在每個訓練步驟，提取包含 64 條軌跡的批次（和 2015 DQN 論文一樣），每條軌跡有兩步（即，2 步 = 1 個完整過渡，包括下一步的觀察）。這個數據集能并行處理三條軌跡，預提取三條軌跡。 > 筆記：對于策略算法，比如策略梯度，每個經驗只需采樣一次，訓練完就可以丟掉。在這個例子中，你還可以使用一個接力緩存，但使用接力緩存的`gather_all()`方法，在每個訓練迭代獲取軌跡張量，訓練完，再動過`clear()`方法清空接力緩存。 有了所有組件之后，就可以訓練模型了。 ### 創建訓練循環 要加速訓練，將主函數轉換為 TensorFlow 函數。可以使用函數`tf_agents.utils.common.function()`，它包裝了`tf.function()`，還有一些其它選項： ```py from tf_agents.utils.common import function collect_driver.run = function(collect_driver.run) agent.train = function(agent.train) ``` 寫一個小函數，可以`n_iterations`次運行主訓練循環： ```py def train_agent(n_iterations): time_step = None policy_state = agent.collect_policy.get_initial_state(tf_env.batch_size) iterator = iter(dataset) for iteration in range(n_iterations): time_step, policy_state = collect_driver.run(time_step, policy_state) trajectories, buffer_info = next(iterator) train_loss = agent.train(trajectories) print("\r{} loss:{:.5f}".format( iteration, train_loss.loss.numpy()), end="") if iteration % 1000 == 0: log_metrics(train_metrics) ``` 這個函數先向收集策略詢問初始狀態（給定環境批次大小，這個例子中是 1）。因為策略是無狀態的，返回的是空元組（所以可以寫成`policy_state = ()`）。然后，創建一個數據集的迭代器，并運行訓練循環。在每個迭代，調用 driver 的`run()`方法，傳入當前的時間步（最初是`None`）和當前的策略狀態。運行收集策略，收集四步的經驗，將收集到的軌跡廣播給接力緩存和指標。然后，從數據集采樣一個批次軌跡，傳給智能體的`train()`方法。返回對象`train_loss`，可能根據智能體的類型有變動。接著，展示迭代數和訓練損失，每隔 1000 次迭代，輸出所有指標的日志。現在可以調用`train_agent()`做一些迭代，智能體就能逐漸學會玩 Breakout 了。 ```py train_agent(10000000) ``` 訓練需要大量算力和極大的耐心（根據硬件，可能需要幾個小時甚至幾天），可能還需要用不同的隨機種子多次運行，以得到更好的結果，但是訓練完成后，智能體在玩 Breakout 就比人厲害了。你還可以在其它 Atari 游戲上訓練這個 DQN 智能體：智能體對于大多數動作游戲都可以超越人的表現，但是智能體對長故事線游戲不擅長。 ## 流行 RL 算法概覽 本章結束前，快速瀏覽一些流行的 RL 算法： Actor-Critic 算法 * 將策略梯度和深度 Q-網絡結合而成 RL 算法族。Actor-Critic 智能體包含兩個神經網絡：一個策略網絡和一個 DQN。用智能體的經驗正常訓練 DQN。與常規 PG 相比，策略網絡的學習有所不同：智能體（actor）依賴 DQN（critic）估計的動作值。就像運動員（智能體）在教練（DQN）的幫助下學習。 異步優勢 Actor-Critic 算法（A3C） * 這是 DeepMind 在 2016 年推出的重要的 Actor-Critic 算法的變體，其中多個智能體并行學習，探索環境的不同復制。每隔一段間隔，每個智能體異步更新主網權重，然后從網絡拉取最新權重。每個智能體都對網絡產生共現，也從其它智能體學習。另外，DQN 不估計 Q-值，而是估計每個動作的優勢，這樣可以穩定訓練。 優勢 Actor-Critic 算法（A2C） * A3C 算法的變體，去除了異步。所有模型更新是同步的，所以梯度更新傾向于大批次，可以讓模型更好地利用 GPU。 軟 Actor-Critic 算法（SAC） * Tuomas Haarnoja 和其它 UC Berkeley 研究員在 2018 年提出的 Actor-Critic 變體。這個算法不僅學習獎勵，還最大化其動作的熵。換句話說，在盡可能獲取更多獎勵的同時，盡量不可預測。這樣可以鼓勵智能體探索環境，可以加速訓練。在 DQN 的估計不好時，可以避免重復執行相同的動作。這個算法采樣非常高效（與前面的算法相反，前者采樣慢）。TF-Agents 中有 SAC。 近似策略優化（PPO） * 基于 A2C 的算法，它能裁剪函數的損失，避免過量權重更新（會導致訓練不穩定）。PPO 是信任區域策略優化（TRPO）的簡化版本，作者是 John Schulman 和其它 OpenAI 研究員。OpenAI 在 2019 年四月弄了個大新聞，他們用基于 PPO 的 OpenAI Five 打敗了多人游戲 Dota2 的世界冠軍。TF-Agents 中有 PPO。 基于好奇探索 * RL 算法中反復出現的問題是獎勵過于稀疏，這使得學習太慢且低效。Deepak Pathak 和其它 UC Berkeley 的研究員提出了解決方法：忽略獎勵，讓智能體極度好奇地探索環境？獎勵變為了智能體的一部分，而不是來自環境。相似的，讓孩子變得更好奇，比純粹的獎勵孩子，能取得更好的結果。怎么實現呢？智能體不斷地預測動作的結果，并探索結果不匹配預測的環境。換句話說，智能體想得到驚喜。如果結果是可預測的（枯燥），智能體就去其它地方。但是，如果結果不可預測，智能體發現無法控制結果，也會變得無聊。只用好奇心，作者成功地訓練智能體玩電子游戲：即使智能體失敗不會受懲罰，游戲也會結束，智能體是玩膩了。 這一章學習了許多主題：策略梯度、馬爾科夫鏈、馬爾科夫決策過程、Q-學習、近似 Q-學習、深度 Q-學習及其變體（固定 Q-值目標、雙 DQN、對決 DQN、優先經驗接力）。還討論了如何使用 TF-Agents 訓練智能體，最后瀏覽了一些流行的算法。強化學習是一個龐大且令人興奮的領域，每天都有新主意和新算法冒出來，希望這章能激發你的好奇心！ ## 練習 1. 如何定義強化學習？它與傳統的監督和非監督學習有什么不同？ 2. 你能想到什么本章沒有提到過的強化學習的應用？環境是什么？智能體是什么？什么是可能的動作，什么是獎勵？ 3. 什么是衰減率？如果你修改了衰減率那最優策略會變化嗎？ 4. 如何測量強化學習智能體的表現？ 5. 什么是信用分配問題？它怎么出現的？怎么解決？ 6. 使用接力緩存的目的是什么？ 7. 什么是 off 策略 RL 算法？ 8. 使用策略梯度處理 OpenAI gym 的“LunarLander-v2” 環境。需要安裝`Box2D`依賴（`python3 -m pip install gym[box2d]`）。 9. 用任何可行的算法，使用 TF-Agents 訓練可以達到人類水平的可以玩 SpaceInvaders-v4 的智能體。 10. 如果你有大約 100 美元備用，你可以購買 Raspberry Pi 3 再加上一些便宜的機器人組件，在 Pi 上安裝 TensorFlow，然后讓我們嗨起來~！舉個例子，看看 Lukas Biewald 的這個[有趣的帖子](https://links.jianshu.com/go?to=https%3A%2F%2Fhoml.info%2F2)，或者看看 GoPiGo 或 BrickPi。從簡單目標開始，比如讓機器人轉向最亮的角度（如果有光傳感器）或最近的物體（如果有聲吶傳感器），并移動。然后可以使用深度學習：比如，如果機器人有攝像頭，可以實現目標檢測算法，檢測人并向人移動。還可以利用 RL 算法讓智能體自己學習使用馬達達到目的。 參考答案見附錄 A。