強化學習 8 —— DQN 算法 Tensorflow 2.0 實現

在上一篇文章強化學習——DQN介紹 中我們詳細介紹了DQN 的來源，以及對於強化學習難以收斂的問題DQN算法提出的兩個處理方法：經驗回放和固定目標值。這篇文章我們就用代碼來實現 DQN 算法

一、環境介紹

1、Gym 介紹

本算法以及以后文章要介紹的算法都會使用 由 \(OpenAI\) 推出的\(Gym\)仿真環境， \(Gym\) 是一個研究和開發強化學習相關算法的仿真平台，了許多問題和環境（或游戲）的接口，而用戶無需過多了解游戲的內部實現，通過簡單地調用就可以用來測試和仿真，並兼容常見的數值運算庫如 \(TensorFlow\) 。

import gym
env = gym.make('CartPole-v1')
env.reset()
for _ in range(1000):
    env.render()
    env.step(env.action_space.sample()) # take a random action
env.close()

運行結果如下：

以上代碼中可以看出，gym的核心接口是Env。作為統一的環境接口，Env包含下面幾個核心方法：

• reset(self)：重置環境的狀態，返回觀察。如果回合結束，就要調用此函數，重置環境信息
• step(self, action)：執行動作 action 推進一個時間步長，返回observation, reward, done, info。
  • observation表示環境觀測，也就是state
  • reward 表示獲得的獎勵
  • done表示當前回個是否結束
  • info 返回一些診斷信息，一般不經常用
• render(self, mode=‘human’, close=False)：重新繪制環境的一幀。
• close(self)：關閉環境，並清除內存。

以上代碼首先導入gym庫，第2行創建CartPole-v01環境，並在第3行重置環境狀態。在 for 循環中進行1000個時間步長(*timestep)的控制，第5行刷新每個時間步長環境畫面，第6行對當前環境狀態采取一個隨機動作（0或1），最后第7行循環結束后關閉仿真環境。

2、CartPole-v1 環境介紹

CartPole 是gym提供的一個基礎的環境，即車桿游戲，游戲里面有一個小車，上有豎着一根桿子，每次重置后的初始狀態會有所不同。小車需要左右移動來保持桿子豎直，為了保證游戲繼續進行需要滿足以下兩個條件：

• 桿子傾斜的角度 \(\theta\) 不能大於15°
• 小車移動的位置 x 需保持在一定范圍（中間到兩邊各2.4個單位長度）

對於 CartPole-v1 環境，其動作是兩個離散的動作左移（0）和右移（1），環境包括小車位置、小車速度、桿子夾角及角變化率四個變量。如下代碼所示：

import gym
env = gym.make('CartPole-v0')
print(env.action_space)  # Discrete(2)
observation = env.reset()
print(observation)  # [-0.0390601  -0.04725411  0.0466889   0.02129675]

下面以CartPole-v1 環境為例，來介紹 DQN 的實現

二、代碼實現

1、經驗回放池的實現

class ReplayBuffer:
    def __init__(self, capacity=10000):
        self.capacity = capacity
        self.buffer = []
        self.position = 0

    def push(self, state, action, reward, next_state, done):
        if len(self.buffer) < self.capacity:
            self.buffer.append(None)
        self.buffer[self.position] = (state, action, reward, next_state, done)
        self.position = int((self.position + 1) % self.capacity)

    def sample(self, batch_size = args.batch_size):
        batch = random.sample(self.buffer, batch_size)
        state, action, reward, next_state, done = map(np.stack, zip(*batch))
        return state, action, reward, next_state, done

首先定義一個經驗回放池，其容量為 10000，函數 push 就是把智能體與環境交互的到的信息添加到經驗池中，這里使用的循環隊列的實現方式，注意 position 指針的運算。當需要用數據來更新算法 時，使用 sample 從經驗隊列中隨機挑選 一個 batch_size 的數據，使用 zip 函數把每一條數據打包到一起：

zip: a=[1,2], b=[2,3], zip(a,b) => [(1, 2), (2, 3)]

然后對每一列數據使用 stack 函數轉化為列表后返回

2、網絡構造

本系列強化學習的代碼，都是使用的 tensorlayer ，就是對 tensorflow 做了一些封裝，使其更加易用，重點是還專門為強化學習內置了一些接口，下面是官網介紹：

TensorLayer 是為研究人員和工程師設計的一款基於Google TensorFlow開發的深度學習與強化學習庫。 它提供高級別的（Higher-Level）深度學習API，這樣不僅可以加快研究人員的實驗速度，也能夠減少工程師在實際開發當中的重復工作。 TensorLayer非常易於修改和擴展，這使它可以同時用於機器學習的研究與應用。

定義網絡模型：

def create_model(input_state_shape):
    input_layer = tl.layers.Input(input_state_shape)
    layer_1 = tl.layers.Dense(n_units=32, act=tf.nn.relu)(input_layer)
    layer_2 = tl.layers.Dense(n_units=16, act=tf.nn.relu)(layer_1)
    output_layer = tl.layers.Dense(n_units=self.action_dim)(layer_2)
    return tl.models.Model(inputs=input_layer, outputs=output_layer)

self.model = create_model([None, self.state_dim])
self.target_model = create_model([None, self.state_dim])
self.model.train()
self.target_model.eval()

可以看到tensorlayer 使用起來與tensorflow 大同小異，只要有tensorflow基礎一眼就能明白，在上面代碼中我們定義一個函數用來生成網絡模型。然后創建一個當前網絡model和一個目標網絡target_model ，我們知道DQN中的目標網絡是起到一個“靶子”的作用，用來評估當前的 target 值，所以我們把它設置為評估模式，調用 eval() 函數即可。而 model 網絡是我們要訓練的網絡，調用函數 train() 設置為訓練模式。

3、算法控制流程

for episode in range(train_episodes):
    total_reward, done = 0, False
    while not done:
        action = self.choose_action(state)
        next_state, reward, done, _ = self.env.step(action)
        self.buffer.push(state, action, reward, next_state, done)
        total_reward += reward
        state = next_state
        # self.render()
    if len(self.buffer.buffer) > args.batch_size:
        self.replay()
        self.target_update()

關於與環境交互過程在上面已經介紹過了，這里重點看 第 10 行的 if 語句，當經驗池的長度大於一個batch_size 時，就開始調用replay() 函數來更新網絡 model 的網絡參數，然后調用target_update() 函數把 model 網絡參數復制給 target_model 網絡。

4、網絡參數更新

def replay(self):
    for _ in range(10):
        states, actions, rewards, next_states, done = self.buffer.sample()
        # compute the target value for the sample tuple
        # target [batch_size, action_dim]
        # target represents the current fitting level
        target = self.target_model(states).numpy()
        next_q_value = tf.reduce_max(self.target_model(next_states), axis=1)
        target_q = rewards + (1 - done) * args.gamma * next_q_value
        target[range(args.batch_size), actions] = target_q

        with tf.GradientTape() as tape:
            q_pred = self.model(states)
            loss = tf.losses.mean_squared_error(target, q_pred)
        grads = tape.gradient(loss, self.model.trainable_weights)
        self.model_optim.apply_gradients(zip(grads, self.model.trainable_weights))

這部分應該就是 DQN 的核心代碼了，在replay() 函數中，我們循環更新更新當前網絡十次，目的就是改變兩個網絡的更新頻率，有利於網絡收斂。

具體的更新部分：我們知道，DQN就是把Q-Learning中的Q表格換成了神經網絡，兩者之間有很多 相似之處，我們可以類比Q-Learning 的更新方式。對於Q表格形式，我們獲取某一個狀態的動作價值Q是直接通過下標得到的，那么在神經網絡中就需要把狀態輸入神經網絡，經過前向計算得到。

\[\Delta w = \alpha (r + \gamma\;max_{a'}\; \hat{Q}(s', a', w) - \hat{Q}{(s, a, w)})\cdot \nabla_w\hat{Q}{(s, a, w)} \]

第三行首先獲取一個batch_size的數據，這個過程稱為 sample 。第7行我們首先獲取當前的動作價值，target 表示的是根據當前的網絡參數計算得到的動作價值。然后第8行先獲取當前網絡參數下 的下一個狀態的所有動作，然后使用reduce_max() 函數找出最大的動作價值。然后第9行和第10行利用下一個狀態最大的動作價值來計算出 target_q ，也就是 \(r + \gamma\;max_{a'}\; \hat{Q}(s', a', w)\) 部分，然后更新target 。注意上面我們計算target時一直在使用 target_model 網絡，target網絡只有在評估網絡狀態時才使用。

接着我們使用 q_pred = self.model(states) 網絡獲取當前 網絡的狀態，也就是 公式中的 \(\hat{Q}{(s, a, w)}\) ，利用MSE函數計算其損失函數，最后更新 model 網絡。

完整代碼請參考強化學習——DQN代碼地址 還請給個 star ，謝謝各位了

三、DQN 小結

雖然 DQN 提出的這兩個解決方案不錯，但是仍然還有問題沒有解決，比如：

• 目標 Q 值（Q Target ）計算是否准確？全部通過 \(max\;Q\) 來計算有沒有問題？
• Q 值代表動作價值，那么單純的動作價值評估會不會不准確？

對應第一個問題的改進就是 Double DQN ，第二個問題的改進是 Dueling DQN。他們都屬與DQN的改進版，我們下篇文章介紹。