本教程講解如何使用深度強化學習訓練一個可以在 CartPole 游戲中獲勝的模型。研究人員使用 tf.keras、OpenAI 訓練了一個使用「異步優勢動作評價」（Asynchronous Advantage Actor Critic，A3C）算法的智能體，通過 A3C 的實現解決了 CartPole 游戲問題，過程中使用了貪婪執行、模型子類和自定義訓練循環。
該過程圍繞以下概念運行：
- 貪婪執行——貪婪執行是一個必要的、由運行定義的接口，此處的運算一旦從 Python 調用，就要立刻執行。這使得以 TensorFLow 開始變得更加容易，還可以使研究和開發變得更加直觀。
- 模型子類——模型子類允許通過編寫 tf.keras.Model 子類以及定義自己的正向傳導通路自定義模型。由於可以強制寫入前向傳導，模型子類在貪婪執行啟用時尤其有用。
- 自定義訓練循環。
本教程遵循的基本工作流程如下：
- 建立主要的智能體監管
- 建立工作智能體
- 實現 A3C 算法
- 訓練智能體
- 將模型表現可視化
本教程面向所有對強化學習感興趣的人，不會涉及太深的機器學習基礎，但主題中涵蓋了高級策略網絡和價值網絡的相關知識。此外，我建議閱讀 Voldymyr Mnih 的《Asynchronous Methods for Deep Reinforcement Learning》（https://arxiv.org/abs/1602.01783），這篇文章很值得一讀，而且文中涉及到本教程采用的算法的很多細節。
什么是 Cartpole？
Cartpole 是一個游戲。在該游戲中，一根桿通過非驅動關節連接到小車上，小車沿無摩擦的軌道滑動。初始狀態（推車位置、推車速度、桿的角度和桿子頂端的速度）隨機初始化為 +/-0.05。通過對車施加 +1 或 -1（車向左或向右移動）的力對該系統進行控制。桿開始的時候是直立的，游戲目標是防止桿倒下。桿保持直立過程中的每個時間步都會得到 +1 的獎勵。當桿傾斜 15 度以上或小車與中間位置相隔 2.4 個單位時游戲結束。
代碼
- 完整代碼：https://github.com/tensorflow/models/blob/master/research/a3c_blogpost/a3c_cartpole.py
- 安裝指南：https://github.com/tensorflow/models/tree/master/research/a3c_blogpost
建立基線
為了正確判斷模型的實際性能以及評估模型的度量標准，建立一個基線通常非常有用。舉個例子，如果返回的分數很高，你就會覺得模型表現不錯，但事實上，我們很難確定高分是由好的算法還是隨機行為帶來的。在分類問題的樣例中，可以通過簡單分析類別分布以及預測最常見的類別來建立基線。但我們該如何針對強化學習建立基線呢？可以創建隨機的智能體，該智能體可以在我們的環境中做出一些隨機行為。
class RandomAgent: """Random Agent that will play the specified game Arguments: env_name: Name of the environment to be played max_eps: Maximum number of episodes to run agent for. """ def __init__(self, env_name, max_eps): self.env = gym.make(env_name) self.max_episodes = max_eps self.global_moving_average_reward = 0 self.res_queue = Queue() def run(self): reward_avg = 0 for episode in range(self.max_episodes): done = False self.env.reset() reward_sum = 0.0 steps = 0 while not done: # Sample randomly from the action space and step _, reward, done, _ = self.env.step(self.env.action_space.sample()) steps += 1 reward_sum += reward # Record statistics self.global_moving_average_reward = record(episode, reward_sum, 0, self.global_moving_average_reward, self.res_queue, 0, steps) reward_avg += reward_sum final_avg = reward_avg / float(self.max_episodes) print("Average score across {} episodes: {}".format(self.max_episodes, final_avg)) return final_avg
就 CartPole 這個游戲而言，我們在 4000 個循環中得到了 ～20 的平均值。為了運行隨機的智能體，要先運行 python 文件： python a3c_cartpole.py—algorithm=random—max-eps=4000。
什么是異步優勢動作評價算法
異步優勢動作評價算法是一個非常拗口的名字。我們將這個名字拆開，算法的機制就自然而然地顯露出來了：
- 異步：該算法是一種異步算法，其中並行訓練多個工作智能體，每一個智能體都有自己的模型和環境副本。由於有更多的工作智能體並行訓練，我們的算法不僅訓練得更快，而且可以獲得更多樣的訓練經驗，因為每一個工作體的經驗都是獨立的。
- 優勢：優勢是一個評價行為好壞和行為輸出結果如何的指標，允許算法關注網絡預測值缺乏什么。直觀地講，這使得我們可以衡量在給定時間步時遵循策略 π 采取行為 a 的優勢。
- 動作-評價：算法的動作-評價用了在策略函數和價值函數間共享層的架構。
它是如何起作用的？
在更高級別上，A3C 算法可以采用異步更新策略，該策略可以在固定的經驗時間步上進行操作。它將使用這些片段計算獎勵和優勢函數的估計值。每一個工作智能體都會遵循下述工作流程：
- 獲取全局網絡參數
- 通過遵循最小化（t_max：到終極狀態的步長）步長數的局部策略與環境進行交互
- 計算價值損失和策略損失
- 從損失中得到梯度
- 用梯度更新全局網絡
- 重復
在這樣的訓練配置下，我們期望看到智能體的數量以線性速度增長。但你的機器可以支持的智能體數量受可用 CPU 核的限制。此外，A3C 可以擴展到多個機器上，有一些較新的研究（像是 IMPALA（https://deepmind.com/blog/impala-scalable-distributed-deeprl-dmlab-30/））甚至支持它更進一步擴展。但添加太多機器可能會對速度和性能產生一些不利影響。參閱這篇文章（https://arxiv.org/abs/1602.01783）以獲取更深入的信息。
重新審視策略函數和價值函數
如果你已經對策略梯度有所了解，那么就可以跳過這一節。如果你不知道什么是策略或價值，或是想要快速復習一些策略或價值，請繼續閱讀。
策略的思想是在給定輸入狀態的情況下參數化行為概率分布。我們通過創建一個網絡來了解游戲的狀態並決定我們應該做什么，以此來實現這個想法。因此，當智能體進行游戲時，每當它看到某些狀態（或是相似的狀態），它就可以在給定輸入狀態下計算出每一個可能的行為的概率，然后再根據概率分布對行為進行采樣。從更深入的數學角度進行分析，策略梯度是更為通用的分數函數梯度估計的特例。一般情況下將期望表示為 p(x | ) [f(x)]；但在我們的例子中，獎勵（優勢）函數的期望，f，在某些策略網絡中，p。然后再用對數導數方法，算出如何更新我們的網絡參數，使得行為樣本能獲得更高的獎勵並以 ∇ Ex[f(x)] =Ex[f(x) ∇ log p(x)] 結束。這個等式解釋了如何根據獎勵函數 f 在梯度方向上轉換 θ 使得分最大化。
價值函數基本上就可以判斷某種狀態的好壞程度。從形式上講，價值函數定義了當以狀態 s 開始，遵循策略 p 時得到獎勵的期望總和。這是模型中「評價」部分相關之處。智能體使用價值估計（評價）來更新策略（動作）。
實現
我們首先來定義一下要使用的模型。主智能體有全局網絡，每個局部的工作體在它們自己的程序中擁有該網絡的的副本。我們用模型子類實例化該模型。模型子類為我們提供了更高的靈活度，而代價是冗余度也更高。
public class MyActivity extends AppCompatActivity { @Override //override the function protected void onCreate(@Nullable Bundle savedInstanceState) { super.onCreate(savedInstanceState); try { OkhttpManager.getInstance().setTrustrCertificates(getAssets().open("mycer.cer"); OkHttpClient mOkhttpClient= OkhttpManager.getInstance().build(); } catch (IOException e) { e.printStackTrace(); } }
從前向傳遞中可以看出，模型得到輸入后會返回策略概率 logits 和 values。
主智能體——主線程
我們來了解一下該操作的主體部分。主智能體有可以更新全局網絡的共享優化器。該智能體實例化了每個工作智能體將要更新的全局網絡以及用來更新它的優化器。這樣每個工作智能體和我們將使用的優化器就可以對其進行更新。A3C 對學習率的傳遞是很有彈性的，但針對 Cart Pole 我們還是要用學習率為 5e-4 的 AdamOptimizer（https://www.tensorflow.org/api_docs/python/tf/train/AdamOptimizer）。
class MasterAgent(): def __init__(self): self.game_name = 'CartPole-v0' save_dir = args.save_dir self.save_dir = save_dir if not os.path.exists(save_dir): os.makedirs(save_dir) env = gym.make(self.game_name) self.state_size = env.observation_space.shape[0] self.action_size = env.action_space.n self.opt = tf.train.AdamOptimizer(args.lr, use_locking=True) print(self.state_size, self.action_size) self.global_model = ActorCriticModel(self.state_size, self.action_size) # global network self.global_model(tf.convert_to_tensor(np.random.random((1, self.state_size)), dtype=tf.float32))
主智能體將運行訓練函數以實例化並啟動每一個智能體。主智能體負責協調和監管每一個智能體。每一個智能體都將異步運行。（因為這是在 Python 中運行的，從技術上講這不能稱為真正的異步，由於 GIL（全局解釋器鎖）的原因，一個單獨的 Python 過程不能並行多個線程（利用多核）。但可以同時運行它們（在 I/O 密集型操作過程中轉換上下文）。我們用線程簡單而清晰地實現了樣例。
def train(self): if args.algorithm == 'random': random_agent = RandomAgent(self.game_name, args.max_eps) random_agent.run() return res_queue = Queue() workers = [Worker(self.state_size, self.action_size, self.global_model, self.opt, res_queue, i, game_name=self.game_name, save_dir=self.save_dir) for i in range(multiprocessing.cpu_count())] for i, worker in enumerate(workers): print("Starting worker {}".format(i)) worker.start() moving_average_rewards = [] # record episode reward to plot while True: reward = res_queue.get() if reward is not None: moving_average_rewards.append(reward) else: break [w.join() for w in workers] plt.plot(moving_average_rewards) plt.ylabel('Moving average ep reward') plt.xlabel('Step') plt.savefig(os.path.join(self.save_dir, '{} Moving Average.png'.format(self.game_name))) plt.show()
Memory 類——存儲我們的經驗
此外，為了更簡單地追蹤模型，我們用了 Memory 類。該類的功能是追蹤每一步的行為、獎勵和狀態。
class Memory: def __init__(self): self.states = [] self.actions = [] self.rewards = [] def store(self, state, action, reward): self.states.append(state) self.actions.append(action) self.rewards.append(reward) def clear(self): self.states = [] self.actions = [] self.rewards = []
現在我們已經知道了算法的關鍵：工作智能體。工作智能體繼承自 threading 類，我們重寫了來自 Thread 的 run 方法。這使我們得以實現 A3C 中的第一個 A——異步。我們先通過實例化局部模型和設置特定的訓練參數開始。
運行算法
下一步是要實現 run 函數。這是要真正運行我們的算法了。我們將針對給定的全局最大運行次數運行所有線程。這是 A3C 中的「動作」所起的作用。我們的智能體會在「評價」判斷行為時根據策略函數采取「行動」，這是我們的價值函數。盡管這一節的代碼看起來很多，但實際上沒有進行太多操作。在每一個 episode 中，代碼只簡單地做了這些：
1. 基於現有框架得到策略（行為概率分布）。
2. 根據策略選擇行動。
3. 如果智能體已經做了一些操作（args.update_freq）或者說智能體已經達到了終端狀態（結束），那么：
a. 用從局部模型計算得到的梯度更新全局模型。
4. 重復
def run(self): total_step = 1 mem = Memory() while Worker.global_episode < args.max_eps: current_state = self.env.reset() mem.clear() ep_reward = 0. ep_steps = 0 self.ep_loss = 0 time_count = 0 done = False while not done: logits, _ = self.local_model( tf.convert_to_tensor(current_state[None, :], dtype=tf.float32)) probs = tf.nn.softmax(logits) action = np.random.choice(self.action_size, p=probs.numpy()[0]) new_state, reward, done, _ = self.env.step(action) if done: reward = -1 ep_reward += reward mem.store(current_state, action, reward) if time_count == args.update_freq or done: # Calculate gradient wrt to local model. We do so by tracking the # variables involved in computing the loss by using tf.GradientTape with tf.GradientTape() as tape: total_loss = self.compute_loss(done, new_state, mem, args.gamma) self.ep_loss += total_loss # Calculate local gradients grads = tape.gradient(total_loss, self.local_model.trainable_weights) # Push local gradients to global model self.opt.apply_gradients(zip(grads, self.global_model.trainable_weights)) # Update local model with new weights self.local_model.set_weights(self.global_model.get_weights()) mem.clear() time_count = 0 if done: # done and print information Worker.global_moving_average_reward = \ record(Worker.global_episode, ep_reward, self.worker_idx, Worker.global_moving_average_reward, self.result_queue, self.ep_loss, ep_steps) # We must use a lock to save our model and to print to prevent data races. if ep_reward > Worker.best_score: with Worker.save_lock: print("Saving best model to {}, " "episode score: {}".format(self.save_dir, ep_reward)) self.global_model.save_weights( os.path.join(self.save_dir, 'model_{}.h5'.format(self.game_name)) ) Worker.best_score = ep_reward Worker.global_episode += 1 ep_steps += 1 time_count += 1 current_state = new_state total_step += 1 self.result_queue.put(None)
如何計算損失？
工作智能體通過計算損失得到所有相關網絡參數的梯度。這是 A3C 中最后一個 A——advantage（優勢）所起的作用。將這些應用於全局網絡。損失計算如下：
- 價值損失：L=∑(R—V(s))²
- 策略損失：L=-log(?(s)) * A(s)
式中 R 是折扣獎勵，V 是價值函數（輸入狀態），? 是策略函數（輸入狀態），A 是優勢函數。我們用折扣獎勵估計 Q 值，因為我們不能直接用 A3C 決定 Q 值。
def compute_loss(self, done, new_state, memory, gamma=0.99): if done: reward_sum = 0. # terminal else: reward_sum = self.local_model( tf.convert_to_tensor(new_state[None, :], dtype=tf.float32))[-1].numpy()[0] # Get discounted rewards discounted_rewards = [] for reward in memory.rewards[::-1]: # reverse buffer r reward_sum = reward + gamma * reward_sum discounted_rewards.append(reward_sum) discounted_rewards.reverse() logits, values = self.local_model( tf.convert_to_tensor(np.vstack(memory.states), dtype=tf.float32)) # Get our advantages advantage = tf.convert_to_tensor(np.array(discounted_rewards)[:, None], dtype=tf.float32) - values # Value loss value_loss = advantage ** 2 # Calculate our policy loss actions_one_hot = tf.one_hot(memory.actions, self.action_size, dtype=tf.float32) policy = tf.nn.softmax(logits) entropy = tf.reduce_sum(policy * tf.log(policy + 1e-20), axis=1) policy_loss = tf.nn.softmax_cross_entropy_with_logits_v2(labels=actions_one_hot, logits=logits) policy_loss *= tf.stop_gradient(advantage) policy_loss -= 0.01 * entropy total_loss = tf.reduce_mean((0.5 * value_loss + policy_loss)) return total_loss
工作智能體將重復在全局網絡中重置網絡參數和與環境進行交互、計算損失再將梯度應用於全局網絡的過程。通過運行下列命令訓練算法：python a3c_cartpole.py—train。
測試算法
通過啟用新環境和簡單遵循訓練出來的模型得到的策略輸出測試算法。這將呈現出我們的環境和模型產生的策略分布中的樣本。
def play(self): env = gym.make(self.game_name).unwrapped state = env.reset() model = self.global_model model_path = os.path.join(self.save_dir, 'model_{}.h5'.format(self.game_name)) print('Loading model from: {}'.format(model_path)) model.load_weights(model_path) done = False step_counter = 0 reward_sum = 0 try: while not done: env.render(mode='rgb_array') policy, value = model(tf.convert_to_tensor(state[None, :], dtype=tf.float32)) policy = tf.nn.softmax(policy) action = np.argmax(policy) state, reward, done, _ = env.step(action) reward_sum += reward print("{}. Reward: {}, action: {}".format(step_counter, reward_sum, action)) step_counter += 1 except KeyboardInterrupt: print("Received Keyboard Interrupt. Shutting down.") finally: env.close()
你可以在模型訓練好后運行下列命令：python a3c_cartpole.py。
檢查模型所得分數的滑動平均：
我們應該看到得分 >200 后收斂了。該游戲連續試驗 100 次平均獲得了 195.0 的獎勵，至此稱得上「解決」了該游戲。
在新環境中的表現：
關鍵點
該教程涵蓋的內容：
- 通過 A3C 的實現解決了 CartPole。
- 使用了貪婪執行、模型子類和自定義訓練循環。
- Eager 使開發訓練循環變得簡單，因為可以直接打印和調試張量，這使編碼變得更容易也更清晰。
- 通過策略網絡和價值網絡對強化學習的基礎進行了學習，並將其結合在一起以實現 A3C
- 通過應用 tf.gradient 得到的優化器更新規則迭代更新了全局網絡。