摘要：剛剛舉行的深度學習開發者峰會上，發布了版本，這一版新增了等一系列并行算法。專注于游戲智能少兒趣味編程兩大領域。有了貝爾曼最優方程，我們就可以通過純粹貪心的策略來確定，即僅僅把最優動作的概率設置為，其他所有非最優動作的概率都設置為。

剛剛舉行的 WAVE SUMMIT 2019 深度學習開發者峰會上，PaddlePaddle 發布了 PARL 1.1 版本，這一版新增了 IMPALA、A3C、A2C 等一系列并行算法。作者重新測試了一遍內置 example，發現卷積速度也明顯加快，從 1.0 版本的訓練一幀需大約 1 秒優化到了 0.15 秒（配置：win8，i5-6200U，GeForce-940M，batch-size=32）。

嘿嘿，超級本實現游戲智能的時代終于來臨！廢話不多說，我們趕緊試試 PARL 的官方 DQN 算法，玩一玩 Flappy-Bird。

關于作者：曹天明（kosora），2011 年畢業于天津科技大學，7 年的 PHP+Java 經驗。于2018年9月報名加入光環國際人工智能周末轉型班進行學習提升。個人研究方向——融合 CLRS 與 DRL 兩大技術體系，并行刷題和模型訓練。專注于游戲智能、少兒趣味編程兩大領域。
模擬環境

相信大家對于這個游戲并不陌生，我們需要控制一只小鳥向前飛行，只有飛翔、下落兩種操作，小鳥每穿過一根柱子，總分就會增加。由于柱子是高低不平的，所以需要想盡辦法躲避它們。一旦碰到了柱子，或者碰到了上、下邊緣，都會導致 game-over。下圖展示了未經訓練的小笨鳥，可以看到，他處于人工智障的狀態，經常撞柱子或者撞草地：

▲ 未經訓練的小笨鳥

先簡要分析一下環境 Environment 的主要代碼。

BirdEnv.py 繼承自 gym.Env，實現了 init、reset、reward、render 等標準接口。init 函數，用于加載圖片、聲音等外部文件，并初始化得分、小鳥位置、上下邊緣、水管位置等環境信息：

def __init__(self):

if not hasattr(self,"IMAGES"):
    print("InitGame!")
    self.beforeInit()
self.score = self.playerIndex = self.loopIter = 0
self.playerx = int(SCREENWIDTH * 0.3)
self.playery = int((SCREENHEIGHT - self.PLAYER_HEIGHT) / 2.25)
self.baseShift = self.IMAGES["base"].get_width() - self.BACKGROUND_WIDTH
newPipe1 = getRandomPipe(self.PIPE_HEIGHT)
newPipe2 = getRandomPipe(self.PIPE_HEIGHT)
#...other code

step 函數，執行兩個動作，0 表示不采取行動（小鳥會自動下落），1 表示飛翔；step 函數有四個返回值，image_data 表示當前狀態，也就是游戲畫面，reward 表示本次 step 的即時獎勵，terminal 表示是否是吸收狀態，{} 表示其他信息：

def step(self, input_action=0):

pygame.event.pump()
reward = 0.1
terminal = False
if input_action == 1:
    if self.playery > -2 * self.PLAYER_HEIGHT:
        self.playerVelY = self.playerFlapAcc
        self.playerFlapped = True

#...other code

image_data=self.render()
return image_data, reward, terminal,{}

獎勵 reward；初始獎勵是 +0.1，表示小鳥向前飛行一小段距離；穿過柱子，獎勵 +1；撞到柱子，獎勵為 -1，并且到達 terminal 狀態：

reward = 0.1

playerMidPos = self.playerx + self.PLAYER_WIDTH / 2
for pipe in self.upperPipes:

pipeMidPos = pipe["x"] + self.PIPE_WIDTH / 2
#穿過一個柱子獎勵加1
if pipeMidPos <= playerMidPos < pipeMidPos + 4:             
    self.score += 1
    reward = self.reward(1)

if isCrash:

#撞到邊緣或者撞到柱子,結束,并且獎勵為-1
terminal = True
reward = self.reward(-1)

reward 函數，返回即時獎勵 r：

def reward(self,r):

return r

reset 函數，調用 init，并執行一次飛翔操作，返回 observation,reward,isOver：

def reset(self,mode="train"):

self.__init__()
self.mode=mode
action0 = 1
observation, reward, isOver,_ = self.step(action0)
return observation,reward,isOver

render 函數，渲染游戲界面，并返回當前畫面：

def render(self):

image_data = pygame.surfarray.array3d(pygame.display.get_surface())
pygame.display.update()
self.FPSCLOCK.tick(FPS)
return image_data

至此，強化學習所需的狀態、動作、獎勵等功能均定義完畢。接下來簡單推導一下 DQN (Deep-Q-Network) 算法的原理。

DQN的發展過程

DQN 的進化歷史可謂源遠流長，從最開始 Bellman 在 1956 年提出的動態規劃，到后來 Watkins 在 1989 年提出的的 Q-learning，再到 DeepMind 的 Nature-2015 穩定版，最后到 Dueling DQN、Priority Replay Memory、Parameter Noise 等優化算法，橫跨整整一個甲子，凝聚了無數專家、教授們的心血。如今的我們站在先賢們的肩膀上，從以下角度逐步分析：

貝爾曼（最優）方程與 VQ 樹
Q-learning
參數化逼近
DQN 算法框架

貝爾曼 (最優) 方程與VQ樹

我們從經典的表格型強化學習（Tabular Reinforcement Learning）開始，回憶一下馬爾可夫決策（MDP）過程，MDP 可由五元組 (S,A,P,R,γ) 表示，其中：

S 狀態集合，維度為 1×|S|
A 動作集合，維度為 1×|A|
P 狀態轉移概率矩陣，經常寫成，其維度為 |S|×|A|×|S|
R 回報函數，如果依賴于狀態值函數 V，維度為 1×|S|，如果依賴于狀態-動作值函數 Q，則維度為 |S|×|A|
γ 折扣因子，用來計算帶折扣的累計回報 G(t)，維度為 1

S、A、R、γ 均不難理解，可能部分同學對有疑問——既然 S 和 A 確定了，下一個狀態 S" 不是也確定了嗎？為什么會有概率轉移矩陣呢？

其實我初學的時候也曾經被這個問題困擾過，不妨通過如下兩個例子以示區別：

恒等于 1.0 的情況。如圖 1 所示，也就是上一次我們在策略梯度算法中所使用的迷宮，假設機器人處于左上角，這時候你命令機器人向右走，那么他轉移到紅框所示位置的概率就是 1.0，不會有任何異議：

▲ 圖1. 迷宮尋寶

不等于 1.0 的情況。假設現在我們下一個飛行棋，如圖 2 所示。有兩種骰子，第一種是普通的正方體骰子，可以投出 1~6，第二種是正四面體的骰子，可以投出 1~4。現在飛機處于紅框所示的位置，現在我們選擇投擲第二種骰子這個動作，由于骰子本身具有均勻隨機性，所以飛機轉移到終點的概率僅僅是 0.25。這就說明，在某些環境中，給定 S、A 的情況下，轉移到具體哪一個 S" 其實是不確定的：

▲ 圖2. 飛行棋

除了經典的五元組外，為了研究長期回報，還經常加入三個重要的元素，分別是：

策略 π(a∣s)，維度為 |S|×|A|
狀態值函數，維度為 1×|S|，表示當智能體采用策略 π 時，累積回報在狀態 s 處的期望值：

▲ 圖3. 狀態值函數

狀態-行為值函數，也叫狀態-動作值函數，維度為 |S|×|A|，表示當智能體采取策略 π 時，累計回報在狀態 s 處并執行動作 a 時的期望值：

▲ 圖4. 狀態-行為值函數

知道了 π、v、q 的具體含義后，我們來看一個重要的概念，也就是 V、Q 的遞歸展開式。

學過動態規劃的同學都知道，動態規劃本質上是一個 bootstrap（自舉）問題，它包含最優子結構與重疊子問題兩個性質，也就是說，通常有兩種方法解決動態規劃：

將總問題劃分為 k 個子問題，遞歸求解這些子問題，然后將子問題進行合并，得到總問題的最優解；對于重復的子問題，我們可以將他們進行緩存（記憶搜索 MemorySearch，請回憶 f(n)=f(n-1)+f(n-2) 這個遞歸程序）；

計算最小的子問題，合并這些子問題產生一個更大的子問題，不斷的自底向上計算，隨著子問題的規模越來越大，我們會得到最終的總問題的最優解（打表 DP，請回憶楊輝三角中的 dp[i-1,j-1]+dp[i-1,j]=dp[i,j]）。

這兩種切題技巧，對于有過 ACM 或者 LeetCode 刷題經驗的同學，可以說是老朋友了，那么能否把以上思想遷移到強化學習呢？答案是肯定的！

分別考慮 v、q 的展開式：

處在狀態 s 時，由于有策略 π 的存在，故可以把狀態值函數 v 展開成以下形式：

▲ 圖5. v展開成q

這個公式表示：在狀態 s 處的值函數，等于采取策略 π 時，所有狀態-行為值函數的總和。

處在狀態 s、并執行動作 a，可以把狀態-行為值函數 q 展開成以下形式：

▲ 圖6. q展開成v

這個公式表示：在狀態 s 采用動作 a 的狀態行為值函數，等于回報加上后序可能產生的的狀態值函數的總和。

我們可以看到：v 可以展開成 q，同時 q 也可以展開成 v。

所以可以用以下 v、q 節點相隔的樹來表示以上兩個公式，這顆樹比純粹的公式更容易理解，我習慣上把它叫做 V-Q 樹，它顯然是一個遞歸的結構：

▲ 圖7. V-Q樹

注意畫紅圈中的兩個節點，體現了重疊子問題特性。如何理解這個性質呢？不妨回憶一下上文提到的飛行棋，假設飛機處在起點位置 1，那么無論投擲 1 號骰子還是 2 號骰子，都是有機會可以到達位置 3 的，這就是重疊子問題的一個例子。

有了這棵遞歸樹之后，就不難推導出 v 和 v"，以及 q 和 q" 自身的遞歸展開式：

▲ 圖8. 狀態值函數v自身的遞歸展開式

▲ 圖9. 狀態-行為值函數q自身的遞歸展開式

其實無論是 v 還是 q，都擁有最優子結構特性。不妨利用反證法加以證明：

假設要求總問題 V(s) 的最優解，那么它包含的每個子問題 V(s") 也必須是最優解；否則，如果某個子問題 V(s") 不是最優，那么必然有一個更優的子問題 V"(s") 存在，使得總問題 V"(s) 比原來的總問題 V(s) 更優，與我們的假設相矛盾，故最優子結構性質得證，q(s) 的最優子結構性質同理。

計算值函數的目的是為了構建學習算法得到最優策略，每個策略對應著一個狀態值函數，最優策略自然也對應著最優狀態值函數，故而定義如下兩個函數：

最優狀態值函數，表示在所有策略中最大的值函數，即：

▲ 圖10. 最優狀態值函數

最優狀態-行為值函數，表示在所有策略中最大的狀態-行為值函數：

▲ 圖11. 最優狀態-行為值函數

結合上文的遞歸展開式和最優子結構性質，可以得到 v 與 q 的貝爾曼最優方程：

▲ 圖12. v的貝爾曼最優方程

▲ 圖13. q的貝爾曼最優方程

重點理解第二個公式，也就是關于 q 的貝爾曼最優方程，它是今天的主角 Q-learning 以及 DQN 的理論基礎。

有了貝爾曼最優方程，我們就可以通過純粹貪心的策略來確定 π，即：僅僅把最優動作的概率設置為 1，其他所有非最優動作的概率都設置為 0。這樣做的好處是：當算法收斂的時候，策略 π(a|s) 必然是一個 one-hot 型的矩陣。用數學公式表達如下：

▲ 圖14. 算法收斂時候的策略π

強化學習中的動態規劃方法實質上是一種 model-based（模型已知）方法，因為 MDP 五元組是已知的，特別是狀態轉移概率矩陣是已知的。

也就是說，所有的環境信息對于我們來說是 100% 完備的，故而可以對整個解空間樹進行全局搜索，下圖展示了動態規劃方法的示意圖，在確定根節點狀態 S(t) 的最優值的時候，必須遍歷他所有的 S(t+1) 子節點并選出最優解：

▲ 圖15. 動態規劃方法的解空間搜索過程

不過，和傳統的刷題動態規劃略有不同，強化學習往往是利用值迭代（Value Iteration）、策略迭代（Policy Iteration）、策略改善（Policy Improve）等方式使 v、q、π 等元素達到收斂狀態，當然也有直接利用矩陣求逆計算解析解的方法，有興趣的同學可以參考相關文獻，這里不再贅述。

Q-learning

上文提到的動態規劃方法是一種 model-based 方法，僅僅適用于已知的情況。若狀態轉移概率矩陣未知，model-free（無模型）方法就派上用場了，上一期的 MCPG 算法就是一種典型的 model-free 方法。它搜索解空間的方式更像是 DFS（深度優先搜索），而且一條道走到黑，沒有指針回溯的操作，下圖展示了蒙特卡洛算法的求解示意圖：

▲ 圖16. MC系列方法的解空間搜索過程

雖然每次只能走一條分支，但隨機數發生器會幫助算法遍歷整個解空間，再通過大量的迭代，所有節點也會收斂到最優解。

不過，MC 類方法有兩個小缺點：

使用作為訓練標簽，其本身就是值函數準確的無偏估計。但是，這也正是它的缺點，因為 MC 方法會經歷很多隨機的狀態和動作，使得每次得到的 G(t) 隨機性很大，具有很高的方差。

由于采用的是一條道走到黑的方式從根節點遍歷到葉子節點，所以必須要等到 episode 結束才能進行訓練，而且每輪 episode 產生的數據只訓練一次，每輪 episode 產生數據的 batch-size 還不一定相同，所以在訓練過程中，MC 方法的 loss 函數（或者 TD-Error）的波動幅度較大，而數據利用效率不高。

那么，能否邊產生數據邊訓練呢？可以！時序差分（Temporal-Difference-Learning，簡稱 TD）算法應運而生了。

時序差分學習是模擬（或者經歷）一段序列，每行動一步（或者幾步）就根據新狀態的價值估計當前執行的狀態價值。大致可以分為兩個小類：

TD(0) 算法，只向后估計一個 step。其值函數更新公式為：

▲ 圖17. TD(0)算法的更新公式

其中，α 為學習率，稱為 TD 目標，MC 方法中的 G(t) 也可以叫做 TD 目標，稱為 TD-Error，當模型收斂時，TD-Error 會無限接近于 0。

Sarsa(λ) 算法，向后估計 n 步，n 為有限值，還有一個衰減因子 λ。其值函數的更新公式為：

▲ 圖18. Sarsa(λ)算法的更新公式

▲ 圖19. 的計算方法

與 MC 方法相比，TD 方法只用到了一步或者有限步隨機狀態和動作，因此它是一個有偏估計。不過，由于 TD 目標的隨機性比 MC 方法的 G(t) 要小，所以方差也比 MC 方法小的多，值函數的波動幅度較小，訓練比較穩定。

看一下 TD 方法的解空間搜索示意圖，紅框表示 TD(0)，藍框表示 Sarsa(λ)。雖然每次估計都有一定的偏差，但隨著算法的不斷迭代，所有的節點也會收斂到最優解：

▲ 圖20. TD方法的解空間搜索過程

有了 TD 的框架，既然我們要求狀態值函數 v、狀態-行為值函數 q 的最優解，那么是否能直接選擇最優的 TD 目標作為 Target 呢？答案是肯定的，這也是 Q-Learning 算法的基本思想，其公式如下所示：

▲ 圖21. Q-learning算法的學習公式

其中，動作 a 由 ε-greedy 策略選出，從而在狀態 s 處執行 a 之后產生了另一個狀態 s"，接下來選出狀態 s" 處最大的狀態-行為值函數 q(s",a")，這樣，TD 目標就可以確定為 R+γmax[a′]Q(s′,a′)。這種思想很像貪心算法中的總是選擇在當前看來最優的決策，它一開始可能會得到一個局部最優解，不過沒關系，隨著算法的不斷迭代，整個解空間樹也會收斂到全局最優解。

以下是 Q-learning 算法的偽代碼，和 on-policy 的 MC 方法對應，它是一種 off-policy（異策略）方法：

initialize Q_table[|S|,|A|] //初始化Q矩陣
for i in range(0,maxEpisode):

s=env.reset()  //初始化狀態s
for j in range(0,maxStep):
    //用ε-greedy策略在s行選一個動作a
    choose action a using ε-greedy from Q_table[s] 
    s",R,terminal,_=env.step(a) //執行動作a,得到下一個狀態s",獎勵R，是否結束terminal
    max_s_prime_action=np.max(Q_table[s",:]) //選s"對應的最大行為值函數
    td=R+γ*max_s_prime_action //計算TD目標
    Q_table[s,a]= Q_table[s,a]+α*(td-Q_table[s,a]) //學習Q(s,a)的值
    s=s" //更新s，注意，和sarsa算法不同，這里的a不用更新
    if terminal:
        break

Q-learning 是一種優秀的算法，不僅簡單直觀，而且平均速度比 MC 快。在 DRL 未出現之前，它在強化學習中的地位，差不多可以媲美 SVM 在機器學習中的地位。

參數化逼近

有了 Q-learning 算法，是否就能一招吃遍天下鮮了呢？答案是否定的，我們看一下它存在的問題。

上文所提到的，無論是 DP、MC 還是 TD，都是基于表格（tabular）的方法，當狀態空間比較小的時候，計算機內存完全可以裝下,表格式型強化學習是完全適用的。但遇到高階魔方（三階魔方的總變化數是）、圍棋（）這類問題時，S、V、Q、P 等表格均會出現維度災難，早就超出了計算機內存甚至硬盤容量。這時候，參數化逼近方法就派上用場了。

所謂參數化逼近，是指值函數可以由一組參數 θ 來近似，如 Q-learning 中的 Q(s,a) 可以寫成 Q(s,a|θ) 的形式。這樣，不但降低了存儲維度，還便于做一些額外的特征工程，而且 θ 更新的同時，Q(s,a|θ) 會進行整體更新，不僅避免了過擬合情況，還使得模型的泛化能力更強。

既然有了可訓練參數，我們就要研究損失函數了，Q-Learning 的損失函數是什么呢？

先看一下 Q-Learning 的優化目標——使得 TD-Error 最小：

▲ 圖22. Q-Learning的優化目標

加入參數 θ 之后，若將 TD 目標作為標簽 target，將 Q(s,a) 作為模型的輸出 y，則問題轉化為：

▲ 圖23. 帶參數的優化目標

這是我們所熟悉的監督學習中的回歸問題，顯然 loss 函數就是 mse，故而可以用梯度下降算法最小化 loss，從而更新參數 θ：

▲ 圖24. loss函數的梯度下降公式

注意到，TD 目標是標簽，所以 Q(s",a"|θ) 中的 θ 是不能更新的，這種方法并非完全的梯度法，只有部分梯度，稱為半梯度法，這是 NIPS-2013 的雛形。

后來，DeepMind 在 Nature-2015 版本中將 TD 網絡多帶帶分開，其參數為 θ"，它本身并不參與訓練，而是每隔固定步數將值函數逼近的網絡參數 θ 拷貝給 θ"，這樣保證了 DQN 的訓練更加穩定：

▲ 圖25. 含有目標網絡參數θ"的梯度下降公式

至此，DQN 的 Loss 函數、梯度下降公式推導完畢。

DQN算法框架

接下來，還要解決兩個問題——數據從哪里來？如何采集？

針對以上兩個問題，DeepMind 團隊提出了深度強化學習的全新訓練方法：經驗回放（experience replay）。

在強化學習過程中，智能體將數據存儲到一個 ReplayBuffer 中（任何一種集合，可以是哈希表、數組、隊列，也可以是數據庫），然后利用均勻隨機采樣的方法從 ReplayBuffer 中抽取數據，這些數據就可以進行 Mini-Batch-SGD，這樣就打破了數據之間的相關性，使得數據之間盡量符合獨立同分布原則。

DQN 的基本網絡結構如下：

▲ 圖26. DQN的基本網絡結構

要特別注意：

與參數 θ 做線性運算 (wx+b) 的僅僅是輸入狀態 s，這一步沒有動作 a 的參與；

output_1 的維度為 |A|，表示神經網絡 Q(s,θ) 的輸出；

輸入動作 a 是 one-hot，與 output_1 作哈達馬積后產生的 output_2 是一個數字，作為損失函數中的 Q(s,a|θ)，也就是 y。

以下是 DQN 算法的偽代碼：

Initialize replay_memory D as a deque,mas_size=50000

Initialize action-value function Q(s,a|θ) as Neural Network with random-weights-initializer

Initialize target action-value function Q(s,a|θ) with weights θ"=θ

for episode in range(0,max_episode=65535):

#重置環境env，得到初始狀態s
s=env.reset()

#循環事件的每一步，最多迭代max_step_limit個step
for step in range(0,max_step_limit=1024):

    #通過ε-greedy的方式選出一個動作action
    With probability ε select a random action a or select a=argmax(Q(s,θ))

    #在env中執行動作a，得到下一個狀態s"，獎勵R,是否終止terminal
    s",R,terminal,_=env.step(a)

    #將五元組(s,a,s",R,terminal)壓進隊尾
    D.addLast(s,a,s",R,terminal)

    #如果隊列滿，彈出隊頭元素
    if D.isFull():
        D.removeFirst()

    #更新狀態s
    s=s"

    #從隊列中進行隨機采樣
    batch_experience[s,a,s",R,terminal]=random_select(D,batch_size=32)

    #計算TD目標
    target = R + γ*(1- terminal) * np.max(Q(s",θ"))

    #對loss函數執行Gradient-decent，訓練參數θ
    θ=θ+α*(target-Q(s,a|θ))▽Q(s,a|θ)

    #每隔C步，同步θ與θ"的權值
    Every C steps set θ"=θ

    #是否結束
    if terminal：
       break 

我們玩的游戲 Flappy-Bird，它的輸入是一幀一幀的圖片，所以，經典的 Atari-CNN 模型就可以派上用場了：

▲ 圖27. Atari游戲的CNN網絡結構

網絡的輸入是被處理成灰度圖的最近 4 幀 84*84 圖像（4 是經驗值），經過若干 CNN 和 FullyConnect 后，輸出各個動作所對應的狀態-行為值函數 Q。以下是每一層的具體參數，由于 atari 游戲最多有 18 個動作，所以最后一層的維度是 18：

▲ 圖28. 神經網絡的具體參數

至此，理論部分推導完畢。下面，我們分析一下 PARL 中的 DQN 部分的源碼，并實現 Flappy-Bird 的游戲智能。

代碼實現

依次分析 env、model、algorithm、agent、replay_memory、train 等模塊。

BirdEnv.py，環境；上文已經分析過了。

BirdModel.py，神經網絡模型；使用三層 CNN+兩層 FC，CNN 的 padding 方式都是 valid，最后輸出狀態-行為值函數 Q，維度為 |A|。注意輸入圖片歸一化，并按照官方模板填入代碼：

class BirdModel(Model):

def __init__(self, act_dim):
    self.act_dim = act_dim
    #padding方式為valid
    p_valid=0
    self.conv1 = layers.conv2d(
        num_filters=32, filter_size=8, stride=4, padding=p_valid, act="relu")
    self.conv2 = layers.conv2d(
        num_filters=64, filter_size=4, stride=2, padding=p_valid, act="relu")
    self.conv3 = layers.conv2d(
        num_filters=64, filter_size=3, stride=1, padding=p_valid, act="relu")
    self.fc0=layers.fc(size=512)
    self.fc1 = layers.fc(size=act_dim)

def value(self, obs):
    #輸入歸一化
    obs = obs / 255.0
    out = self.conv1(obs)
    out = self.conv2(out)
    out = self.conv3(out)
    out = layers.flatten(out, axis=1)
    out = self.fc0(out)
    out = self.fc1(out)
    return out

dqn.py，算法層；官方倉庫已經提供好了，我們無需自己再寫，直接復用算法庫（parl.algorithms）里邊的 DQN 算法即可。

簡單分析一下 DQN 的源碼實現。

define_learn 函數，用于神經網絡的學習。接收 [狀態 obs, 動作 action, 即時獎勵 reward, 下一個狀態 next_obs, 是否終止 terminal] 這樣一個五元組，代碼實現如下：

pred_value = self.model.value(obs)

next_pred_value = self.target_model.value(next_obs)

best_v = layers.reduce_max(next_pred_value, dim=1)
best_v.stop_gradient = True

target = reward + (1.0 - layers.cast(terminal, dtype="float32")) self.gamma best_v

action_onehot = layers.one_hot(action, self.action_dim)
action_onehot = layers.cast(action_onehot, dtype="float32")
pred_action_value = layers.reduce_sum(layers.elementwise_mul(action_onehot, pred_value), dim=1)

cost = layers.square_error_cost(pred_action_value, target)
cost = layers.reduce_mean(cost)
optimizer = fluid.optimizer.Adam(self.lr, epsilon=1e-3)
optimizer.minimize(cost)

sync_target 函數用于同步網絡參數：

def sync_target(self, gpu_id):

""" sync parameters of self.target_model with self.model
"""
self.model.sync_params_to(self.target_model, gpu_id=gpu_id)

BirdAgent.py，智能體。其中，build_program 函數封裝了 algorithm 中的 define_predict 和 define_learn，sample 函數以 ε-greedy 策略選擇動作，predict 函數以 100% 貪心的策略選擇 argmax 動作，learn 函數接收五元組 (obs, act, reward, next_obs, terminal) 完成學習功能，這些函數和 Policy-Gradient 的寫法類似。

除了這些常用功能之外，由于游戲的訓練時間比較長，所以附加了兩個函數，save_params 用于保存模型，load_params 用于加載模型：

def save_params(self, learnDir,predictDir):

fluid.io.save_params(
    executor=self.fluid_executor,
    dirname=learnDir,
    main_program=self.learn_programs[0])   
fluid.io.save_params(
    executor=self.fluid_executor,
    dirname=predictDir,
    main_program=self.predict_programs[0])     

def load_params(self, learnDir,predictDir):

fluid.io.load_params(
    executor=self.fluid_executor,
    dirname=learnDir,
    main_program=self.learn_programs[0])  
fluid.io.load_params(
    executor=self.fluid_executor,
    dirname=predictDir,
    main_program=self.predict_programs[0]) 

另外，還有四個超參數，可以進行微調：

self.update_target_steps = 5000

self.exploration = 0.8

self.exploration_dacay=1e-6

self.min_exploration=0.05

replay_memory.py，經驗回放單元。雙端隊列 _context 是一個滑動窗口，用來記錄最近 3 幀（再加上新產生的 1 幀就是 4 幀）；state、action、reward 等用 numpy 數組存儲，因為 numpy 的功能比雙端隊列更豐富，max_size 表示 replay_memory 的最大容量：

self.state = np.zeros((self.max_size, ) + state_shape, dtype="int32")
self.action = np.zeros((self.max_size, ), dtype="int32")
self.reward = np.zeros((self.max_size, ), dtype="float32")
self.isOver = np.zeros((self.max_size, ), dtype="bool")

self._context = deque(maxlen=context_len - 1)

其他的 append、recent_state、sample_batch 等函數并不難理解，都是基于 numpy 數組的進一步封裝，略過一遍即可看懂。

Train_Test_Working_Flow.py，訓練與測試，讓環境 evn 和智能體 agent 進行交互。最重要的就是 run_train_episode 函數，體現了 DQN 的主要邏輯，重點分析注釋部分與 DQN 偽代碼的對應關系，其他都是編程細節：

def run_train_episode(env, agent, rpm):

global trainEpisode
global meanReward
total_reward = 0
all_cost = []
#重置環境
state,_, __ = env.reset()
step = 0
#循環每一步
while True:
    context = rpm.recent_state()
    context.append(resizeBirdrToAtari(state))
    context = np.stack(context, axis=0)
    #用ε-greedy的方式選一個動作
    action = agent.sample(context)
    #執行動作
    next_state, reward, isOver,_ = env.step(action)
    step += 1
    #存入replay_buffer
    rpm.append(Experience(resizeBirdrToAtari(state), action, reward, isOver))
    if rpm.size() > MEMORY_WARMUP_SIZE:
        if step % UPDATE_FREQ == 0:
            #從replay_buffer中隨機采樣
            batch_all_state, batch_action, batch_reward, batch_isOver = rpm.sample_batch(batchSize)
            batch_state = batch_all_state[:, :CONTEXT_LEN, :, :]
            batch_next_state = batch_all_state[:, 1:, :, :]
            #執行SGD,訓練參數θ
            cost=agent.learn(batch_state,batch_action, batch_reward,batch_next_state, batch_isOver)
            all_cost.append(float(cost))
    total_reward += reward
    state = next_state
    if isOver or step>=MAX_Step_Limit:
        break
if all_cost:
    trainEpisode+=1
    #以滑動平均的方式打印平均獎勵
    meanReward=meanReward+(total_reward-meanReward)/trainEpisode
    print("
 trainEpisode:{},total_reward:{:.2f}, meanReward:{:.2f} mean_cost:{:.3f}"
          .format(trainEpisode,total_reward, meanReward,np.mean(all_cost)))
return total_reward, step

除了主要邏輯外，還有一些常見的優化手段，防止訓練過程中出現 trick：

MEMORY_WARMUP_SIZE = MEMORY_SIZE//20

MAX_Step_Limit=int(1<<12)

avgQueue=deque(maxlen=16)

另外，還有其他一些超參數，比如學習率 LEARNING_RATE、衰減因子 GAMMA、記錄日志的頻率 log_freq 等等，都可以進行微調：

GAMMA = 0.99

LEARNING_RATE = 1e-3 * 0.5

log_freq=10

main 函數在這里，輸入 train 訓練網絡，輸入 test 進行測試：

if name == "__main__":

print("train or test ?")
mode=input()
print(mode)
if mode=="train":
    train()
elif mode=="test":
    test()
else:
    print("Invalid input!")

這是模型在我本機訓練的輸出日志，大概 3300 個 episode、50 萬步之后，模型就收斂了：

▲ 圖29. 模型訓練的輸出日志

平均獎勵：

▲ 圖30. 最近16次平均獎勵變化曲線

各位同學可以試著調節超參數，或者修改網絡模型，看看能不能遇到一些坑？哪些因素會影響訓練效率？如何提升收斂速度？

接下來就是見證奇跡的時刻，當初懵懂的小笨鳥，如今已修煉成精了！

▲ 訓練完的FlappyBird

觀看 4 分鐘完整版：

https://www.bilibili.com/vide...

Github源碼：

https://github.com/kosoraYint...

參考文獻

[1] Bellman, R.E. & Dreyfus, S.E. (1962). Applied dynamic programming. RAND Corporation.
[2] Sutton, R.S. (1988). Learning to predict by the methods of temporal difference.Machine Learning, 3, pp. 9–44.
[3] V. Mnih, K. Kavukcuoglu, D. Silver, A. A. Rusu, et al., "Human-level control through deep reinforcement learning," Nature, vol. 518(7540), pp. 529-533, 2015.
[4] https://leetcode.com/problems...
[5] https://leetcode.com/problems...
[6] https://github.com/yenchenlin...
[7] https://github.com/MorvanZhou...