如何在TensorFlow 2.0中構建強化學習智慧體
TensorFlow2.0 即將在今年上半年推出正式版 ，雖然目前有預覽版 可供使用，但是對於大多數人來說，新版本能夠帶來什麼還沒有具體的概念。本文將簡要介紹在TensorFlow2.0 上使用強化學習演算法的體驗。其中作者概述了 Keras 子類 API、Eager Execution、會話替換以及會讓開發更加方便的技巧。
對此，Keras 提出者、谷歌科學家 François Chollet 表示，這是一份非常詳盡的介紹。
在這一教程中，我們將會使用TensorFlow2.0 新特性，並藉助深度強化學習中的 A2C 智慧體解決經典 CartPole-v0 環境任務。雖然我們的目標是展示TensorFlow2.0，但與此同時我們也會盡量詳細解釋深度強化學習（DRL）的概念，其中包括這一領域的簡要概述。
TensorFlow2.0 版的宗旨是讓開發者們能夠更輕鬆，在 深度強化學習 上這一理念顯然也得到了發揚：在這個例子中，我們的智慧體原始碼不到 150 行！如果你想看看程式碼，Python 檔案格式的在這裡：https://github.com/inoryy/tensorflow2-deep-reinforcement-learning
Colab 格式在這裡：https://colab.research.google.com/drive/12QvW7VZSzoaF-Org-u-N6aiTdBN5ohNA
安裝
目前TensorFlow2.0 仍然是公開測試版，所以我們最好將其安裝在單獨的（虛擬）環境中。在這裡推薦 Anaconda，假如這樣的話：
> conda create -n tf2 python=3.6 > source activate tf2 > pip install tf-nightly-2.0-preview # tf-nightly-gpu-2.0-preview for GPU version
快速驗證一下安裝是否成功：
>>> import tensorflow as tf >>> print(tf.__version__)1.13.0-dev20190117 >>> print(tf.executing_eagerly()) True
顯示的是 1.13.x 版本？不用擔心，這是因為它還是早期預覽版。這裡需要注意的重點是預設使用 Eager 模式。
>>> print(tf.reduce_sum([1, 2, 3, 4, 5])) tf.Tensor(15, shape=(), dtype=int32)
如果你還不太熟悉 Eager 模式，實際上它意味著計算是實時執行的——而不再是通過預編譯的計算圖來執行。你可以在TensorFlow的文件中找到很好的概述：https://www.tensorflow.org/tutorials/eager/eager_basics
強化學習
強化學習指的是面向目標的演算法，這種演算法學習如何在一些具體的步驟中達到一個目標或者最大化；例如，最大化一個遊戲中通過一些行動而獲得的得分。它們可以從一個空白狀態開始，然後在合適的條件下達到超越人類水平的效能。就像被糖果和體罰刺激的小孩子一樣，當它們做出錯誤的預測時，這些演算法會受到懲罰，當它們做出正確的預測時，它們會得到獎勵—這便是強化的意義所在。
大多數人第一次聽聞強化學習概念是來自 AlphaGo，結合深度學習的強化演算法可以在圍棋和 Atari 遊戲中打敗人類冠軍。儘管這聽起來還不具有足夠的說服力，但是這已經遠遠優於它們之前的成就了，而且目前最先進的進步是很迅速的。
兩個強化學習的演算法 Deep-Q learning 和 A3C 已經在 Deeplearning4j 庫上實現了，現在，它已經可以玩《毀滅戰士（Doom）》了。
有關強化學習概念，可參閱：
通過TensorFlow2.0 實現 Actor-Critic 的優勢
這一部分主要介紹實現許多現代 DRL 演算法的基礎：Actor-Critic 智慧體。為了簡單起見，這裡並不會實現並行的工作站。不過大多數程式碼已經可以支援並行化，所以這也可以作為讀者自行練習的機會。
作為測試平臺，我們會使用 CartPole-v0 環境。雖然這個環境很簡單，但它仍然是一個很好的入門級選擇，我們經常依賴它作為實現 RL 演算法異常檢查的環境。
通過 Keras 模型 API 實現策略和價值函式
首先，我們可以在單個 Model 類下定義策略和價值估計網路：
import numpy as npimport tensorflow as tfimport tensorflow.keras.layers as kl class ProbabilityDistribution(tf.keras.Model):def call(self, logits):# sample a random categorical action from given logitsreturn tf.squeeze(tf.random.categorical(logits, 1), axis=-1) class Model(tf.keras.Model):def __init__(self, num_actions): super().__init__('mlp_policy') # no tf.get_variable(), just simple Keras API self.hidden1 = kl.Dense(128, activation='relu') self.hidden2 = kl.Dense(128, activation='relu') self.value = kl.Dense(1, name='value') # logits are unnormalized log probabilities self.logits = kl.Dense(num_actions, name='policy_logits') self.dist = ProbabilityDistribution() def call(self, inputs):# inputs is a numpy array, convert to Tensor x = tf.convert_to_tensor(inputs, dtype=tf.float32) # separate hidden layers from the same input tensor hidden_logs = self.hidden1(x) hidden_vals = self.hidden2(x) return self.logits(hidden_logs), self.value(hidden_vals) def action_value(self, obs):# executes call() under the hood logits, value = self.predict(obs) action = self.dist.predict(logits) # a simpler option, will become clear later why we don't use it# action = tf.random.categorical(logits, 1)return np.squeeze(action, axis=-1), np.squeeze(value, axis=-1)
下面就可以驗證模型是否能正常執行：
import gym env = gym.make('CartPole-v0') model = Model(num_actions=env.action_space.n) obs = env.reset()# no feed_dict or tf.Session() needed at all action, value = model.action_value(obs[None, :]) print(action, value) # [1] [-0.00145713]
這裡需要注意的是：
-
模型的層級和執行路徑是獨立定義的
-
模型並沒有「input」層，它將接收原始的 NumPy 陣列
-
兩個計算路徑可以通過函式式 API 在一個模型中定義
-
模型可以包含動作取樣等輔助性方法
-
在實時執行模式中，所有模組都從 NumPy 陣列開始執行
隨機智慧體
現在我們可以開始編寫更有意思的模組：A2CAgent 類。首先，我們可以新增 test 方法，它會在整個 episode 中執行並返回獎勵的和。
class A2CAgent:def __init__(self, model): self.model = model def test(self, env, render=True): obs, done, ep_reward = env.reset(), False, 0while not done: action, _ = self.model.action_value(obs[None, :]) obs, reward, done, _ = env.step(action) ep_reward += reward if render: env.render() return ep_reward
現在可以看看在隨機初始化權重的情況下模型計算出來的得分是多少：
agent = A2CAgent(model)
rewards_sum = agent.test(env)
print("%d out of 200" % rewards_sum) # 18 out of 200
當然隨機初始化的權重肯定是不能獲得最佳獎勵的，現在就需要定義訓練部分了。
損失或目標函式
一般而言，智慧體會通過對某些損失函式或目標函式執行梯度下降而提升策略效果。在 Actor-Critic 中，我們需要訓練三個目標函式：利用加權梯度最大化和資訊熵最大化提升策略效果，並最小化價值估計誤差。
import tensorflow.keras.losses as klsimport tensorflow.keras.optimizers as ko class A2CAgent:def __init__(self, model):# hyperparameters for loss terms self.params = {'value': 0.5, 'entropy': 0.0001} self.model = model self.model.compile( optimizer=ko.RMSprop(lr=0.0007), # define separate losses for policy logits and value estimate loss=[self._logits_loss, self._value_loss] ) def test(self, env, render=True):# unchanged from previous section ... def _value_loss(self, returns, value):# value loss is typically MSE between value estimates and returnsreturn self.params['value']*kls.mean_squared_error(returns, value) def _logits_loss(self, acts_and_advs, logits):# a trick to input actions and advantages through same API actions, advantages = tf.split(acts_and_advs, 2, axis=-1) # polymorphic CE loss function that supports sparse and weighted options# from_logits argument ensures transformation into normalized probabilities cross_entropy = kls.CategoricalCrossentropy(from_logits=True) # policy loss is defined by policy gradients, weighted by advantages# note: we only calculate the loss on the actions we've actually taken# thus under the hood a sparse version of CE loss will be executed actions = tf.cast(actions, tf.int32) policy_loss = cross_entropy(actions, logits, sample_weight=advantages) # entropy loss can be calculated via CE over itself entropy_loss = cross_entropy(logits, logits) # here signs are flipped because optimizer minimizesreturn policy_loss - self.params['entropy']*entropy_loss
上面已經完成對目標函式的定義，這樣的程式碼非常緊湊，甚至註釋行都要比程式碼本身多。
智慧體訓練迴圈
最後，我們需要定義一個訓練迴圈，它會相對長一點，但同樣也非常直觀：採集樣本、計算反饋獎勵和梯度、最後訓練並更新模型。
import tensorflow.keras.losses as klsimport tensorflow.keras.optimizers as ko class A2CAgent:def __init__(self, model):# hyperparameters for loss terms self.params = {'value': 0.5, 'entropy': 0.0001} self.model = model self.model.compile( optimizer=ko.RMSprop(lr=0.0007), # define separate losses for policy logits and value estimate loss=[self._logits_loss, self._value_loss] ) def test(self, env, render=True):# unchanged from previous section ... def _value_loss(self, returns, value):# value loss is typically MSE between value estimates and returnsreturn self.params['value']*kls.mean_squared_error(returns, value) def _logits_loss(self, acts_and_advs, logits):# a trick to input actions and advantages through same API actions, advantages = tf.split(acts_and_advs, 2, axis=-1) # polymorphic CE loss function that supports sparse and weighted options# from_logits argument ensures transformation into normalized probabilities cross_entropy = kls.CategoricalCrossentropy(from_logits=True) # policy loss is defined by policy gradients, weighted by advantages# note: we only calculate the loss on the actions we've actually taken# thus under the hood a sparse version of CE loss will be executed actions = tf.cast(actions, tf.int32) policy_loss = cross_entropy(actions, logits, sample_weight=advantages) # entropy loss can be calculated via CE over itself entropy_loss = cross_entropy(logits, logits) # here signs are flipped because optimizer minimizesreturn policy_loss - self.params['entropy']*entropy_loss
訓練和結果
現在已經預備好在 CartPole-v0 上訓練單工作站的 A2C 智慧體了，訓練過程也就需要幾分鐘。在訓練完成後，我們應該能看到智慧體成功實現了 200/200 的目標分值。
rewards_history = agent.train(env)
print("Finished training, testing...")
print("%d out of 200" % agent.test(env)) # 200 out of 200
在原始碼中，我們還實現了一些額外的輔助模組，包括列印執行 episode 的獎勵值和損失值等，並繪製出所有獎勵值的歷史變化。
靜態計算圖
有了這些令人興奮的實時執行模式，你可能會考慮以前的靜態計算圖還能行嗎？當然靜態計算圖也是可以的，我們只需要額外的程式碼行就能啟動它。
with tf.Graph().as_default(): print(tf.executing_eagerly()) # False model = Model(num_actions=env.action_space.n) agent = A2CAgent(model) rewards_history = agent.train(env) print("Finished training, testing...") print("%d out of 200" % agent.test(env)) # 200 out of 200
需要注意的是，在靜態計算圖執行的期間，我們不能只使用 Tensor。這也就是為什麼在模型定義的過程中需要使用 CategoricalDistribution 技巧。
One More Thing…
還記得我說過TensorFlow預設使用 eager 模式，甚至還用程式碼展示了一下。然而，並不是這樣的，不完全是。
如果你是用 Keras API 來構建和管理你的模型，那麼它將會將模型編譯成靜態圖。因此你最終將獲得靜態計算圖的效能和 eager execution 的靈活性。
你可以通過 model.run_eagerly 標記來檢查模型狀態，你也可以通過將這個 flag 設定為 True 來強制使用 eager 模式。如果 Keras 檢測到無法實現 eager 模式，就會使用預設的模式。
為了說明它確實以靜態圖執行，以下是一個簡單的基準：
# create a 100000 samples batch
env = gym.make('CartPole-v0')
obs = np.repeat(env.reset()[None, :], 100000, axis=0)
Eager基準
%%time
model = Model(env.action_space.n)
model.run_eagerly = True
print("Eager Execution: ", tf.executing_eagerly())
print("Eager Keras Model:", model.run_eagerly)
_ = model(obs)
######## Results #######
Eager Execution: True
Eager Keras Model: True
CPU times: user 639 ms, sys: 736 ms, total: 1.38 s
靜態基準
%%time
with tf.Graph().as_default():
model = Model(env.action_space.n)
print("Eager Execution: ", tf.executing_eagerly())
print("Eager Keras Model:", model.run_eagerly)
_ = model.predict(obs)
######## Results #######
Eager Execution: False
Eager Keras Model: False
CPU times: user 793 ms, sys: 79.7 ms, total: 873 ms
預設基準
%%time
model = Model(env.action_space.n)
print("Eager Execution: ", tf.executing_eagerly())
print("Eager Keras Model:", model.run_eagerly)
_ = model.predict(obs)
######## Results #######
Eager Execution: True
Eager Keras Model: False
CPU times: user 994 ms, sys: 23.1 ms, total: 1.02 s
正如你所看到的，Eager 模式是在靜態模式之後的，在預設情況下，模型確實是靜態執行的，所以也匹配顯示靜態圖執行的形式。
結論
希望本文可以讓你瞭解深度強化學習及其在TensorFlow2.0 中的實現方式。請注意，在文中使用的仍然是「每晚預覽版本」，它甚至還不是正式版的候選版本。一切都可能會發生改變，不過這也意味著如果你對新版本的TensorFlow有什麼不喜歡的地方，可以盡情地去提意見。
還有一個經常出現的問題：TensorFlow和 PyTorch 比誰好？現在我們還無法確定，這兩個深度學習框架都非常流行，我們很難說哪個更好。不過如果你很熟悉 PyTorch，你應該可以看得出 TenrorFlow 2.0 不僅補齊了缺點，而且還避免了 PyTorch API 的一些短板。
不論如何，深度學習框架的競爭對於使用者來說都是好事，我們可以期待未來它們會變成什麼樣子。
原文連結：http://inoryy.com/post/tensorflow2-deep-reinforcement-learning/