Keras+OpenAI強(qiáng)化學(xué)習(xí)實(shí)踐：行為-評判模型
像之前的教程一樣，我們首先快速了解一下已取得的驚人成果：在一個(gè)連續(xù)的輸出空間場景下，從完全不明白「勝利」的含義開始，現(xiàn)在我們可以探索環(huán)境并「完成」試驗(yàn)。
將自身置身于模擬環(huán)境中。這就相當(dāng)于要求你在沒有游戲說明書和特定目標(biāo)的場景下玩一場游戲，且不可中斷，直到你贏得整個(gè)游戲(這有些近乎殘忍)。不僅如此：一系列動作可能產(chǎn)生的結(jié)果狀態(tài)是無窮無盡的(即連續(xù)觀測空間)!然而，DQN 通過調(diào)控并緩慢更新各動作內(nèi)部參數(shù)的值可以快速解決這個(gè)看似不可能的任務(wù)。
更復(fù)雜的環(huán)境
從以前的 MountainCar 環(huán)境向 Pendulum 環(huán)境的升級與 CartPole 到 MountainCar 的升級極其相似：我們正在從一個(gè)離散的環(huán)境擴(kuò)展到連續(xù)的環(huán)境。Pendulum 環(huán)境具有無限的輸入空間，這意味著你在任何給定時(shí)間可以進(jìn)行不限次數(shù)的動作。為何 DQN 不再適用此類環(huán)境了?DQN 的實(shí)現(xiàn)不是完全獨(dú)立于環(huán)境動作的結(jié)構(gòu)嗎?
不同于 MountainCar-v0，Pendulum-v0 通過給我們提供了無窮大的輸入空間而構(gòu)成了更大的挑戰(zhàn)。
雖然它與動作無關(guān)，但 DQN 的基本原理是擁有有限的輸出空間。畢竟，我們要考慮該如何構(gòu)建代碼：預(yù)測會在每個(gè)時(shí)間步長(給定當(dāng)前環(huán)境狀態(tài))下為每個(gè)可能的動作分配一個(gè)分?jǐn)?shù)，并簡單地采用得分最高的動作。我們之前已經(jīng)簡化了強(qiáng)化學(xué)習(xí)的問題，以便高效地為每個(gè)動作分配分?jǐn)?shù)。但是，如果我們輸入空間無限大，這還有可能嗎?我們需要一個(gè)無限大的表格來記錄所有的 Q 值!
無限大的數(shù)據(jù)表聽起來很不靠譜!
我們該如何著手解決這個(gè)看似不可能的任務(wù)呢?畢竟，我們現(xiàn)在要做比以前更加瘋狂的事：不僅僅只是贏下一場沒有攻略的游戲，現(xiàn)在我們還要應(yīng)對一個(gè)被無數(shù)條指令控制的游戲!讓我們來看看為什么 DQN 只接收有限數(shù)量的動作輸入。
我們從模型的搭建方式來分析。我們必須能夠在每個(gè)時(shí)間步迭代更新特定的動作位置的改變方式。這正是為什么我們讓模型預(yù)測 Q 值，而不是直接預(yù)測下一步的動作。如果選擇了后者，我們不知道如何更新模型以更好地預(yù)測，以及從對未來的預(yù)測中獲利。
因此，本質(zhì)問題源于一個(gè)事實(shí)——類似于模型已經(jīng)輸出與所有可能發(fā)生的行動相關(guān)的獎(jiǎng)勵(lì)的列表運(yùn)算結(jié)果。如果我們把這個(gè)模型拆解開會怎樣?如果我們有兩個(gè)獨(dú)立的模型：一個(gè)輸出期望的動作(在連續(xù)空間中)，另一個(gè)以它的輸出作為輸入，以產(chǎn)生 DQN 的 Q 值?這貌似解決了我們的問題，而且這正是行為-評判模型(actor-critic model)的基本原理!
行為-評判模型原理
不同于 DQN 算法，行為-評判模型(如名字所示)有兩個(gè)獨(dú)立的網(wǎng)絡(luò)：一個(gè)基于當(dāng)前的環(huán)境狀態(tài)預(yù)測出即將被采用的動作，另一個(gè)用于計(jì)算狀態(tài)和動作下的價(jià)值。
正如上節(jié)所述，整個(gè)行為—評判(AC)方法可行的前提是有兩個(gè)交互模型。多個(gè)神經(jīng)網(wǎng)絡(luò)之間相互關(guān)聯(lián)的主題在強(qiáng)化學(xué)習(xí)和監(jiān)督學(xué)習(xí)(即 GAN、AC、A3C、DDQN(升級版 DQN)等)中越發(fā)凸顯。初次了解這些架構(gòu)可能有些困難，但這絕對值得去做：你將能夠理解和編程實(shí)現(xiàn)一些現(xiàn)代領(lǐng)域研究前沿的算法!
回到主題，AC 模型有兩個(gè)恰如其分的名字：行為和評判。前者接受當(dāng)前的環(huán)境狀態(tài)，并決定從哪個(gè)狀態(tài)獲得最佳動作。它以非常類似于人類自身的行為方式來實(shí)現(xiàn) DQN 算法。評判模塊通過從 DQN 中接受環(huán)境狀態(tài)和動作并返回一個(gè)表征動作狀態(tài)的分?jǐn)?shù)來完成評判功能。
把這想象成是一個(gè)孩子(「行為模塊」)與其父母(「評判模塊」)的游樂場。孩子正在環(huán)顧四周，探索周邊環(huán)境中的所有可能選擇，例如滑動幻燈片，蕩秋千，在草地上玩耍。父母會照看孩子并基于其所為，對孩子給出批評或補(bǔ)充。父母的決定依賴于環(huán)境的事實(shí)無可否認(rèn)：畢竟，如果孩子試圖在真實(shí)的秋千上玩耍，相比于試圖在幻燈片上這樣做，孩子更值得表揚(yáng)!
簡介：鏈?zhǔn)椒▌t(可選)
你需要理解的主要理論在很大程度上支撐著現(xiàn)代機(jī)器學(xué)習(xí)：鏈?zhǔn)椒▌t。毫不夸張的說鏈?zhǔn)椒▌t可能是掌握理解實(shí)用機(jī)器學(xué)習(xí)的最關(guān)鍵的(即使有些簡單)的想法之一。事實(shí)上，如果只是直觀地了解鏈?zhǔn)椒▌t的概念，你并不需要很深厚的數(shù)學(xué)背景。我會非?？焖俚刂v解鏈?zhǔn)椒▌t，但如果你已了解，請隨意跳到下一部分，下一部分我們將看到開發(fā) AC 模型的實(shí)際概述，以及鏈條法則如何適用于該規(guī)劃。
一個(gè)看似可能來自你第一節(jié)微積分課堂上的簡單概念，構(gòu)成了實(shí)用機(jī)器學(xué)習(xí)的現(xiàn)代基礎(chǔ)，因?yàn)樗诜聪蛲扑愫皖愃扑惴ㄖ杏兄钊穗y以置信的加速運(yùn)算效果。
這個(gè)等式看起來非常直觀：畢竟只是「重寫了分子/分母」。這個(gè)「直觀的解釋」有一個(gè)主要問題：等式中的推導(dǎo)完全是倒退的!關(guān)鍵是要記住，數(shù)學(xué)中引入直觀的符號是為了幫助我們理解概念。因此，由于鏈?zhǔn)椒▌t的計(jì)算方式非常類似于簡化分?jǐn)?shù)的運(yùn)算過程，人們才引入這種「分?jǐn)?shù)」符號。那么試圖通過符號來解釋概念的人正在跳過關(guān)鍵的一步：為什么這些符號可以通用?如同這里，為什么要像這樣進(jìn)行求導(dǎo)?
可以借助經(jīng)典的彈簧實(shí)例可視化運(yùn)動中的鏈條規(guī)則。
基本概念實(shí)際上并不比這個(gè)符號更難理解。想象一下，我們把一捆繩子一根根地系在一起，類似于把一堆彈簧串聯(lián)起來。假設(shè)你固定了這個(gè)彈簧系統(tǒng)的一端，你的目標(biāo)是以 10 英尺/秒的速度搖晃另一端，那么你可以用這個(gè)速度搖動你的末端，并把它傳播到另一端?；蛘吣憧梢赃B接一些中間系統(tǒng)，以較低的速率搖動中間連接，例如，5 英尺/秒。也就是說，在 5 英尺/秒的情況下，你只需要以 2 英尺/秒的速度搖動末端，因?yàn)槟銖拈_始到終點(diǎn)做的任何運(yùn)動都會被傳遞到終點(diǎn)位置。這是因?yàn)槲锢磉B接迫使一端的運(yùn)動被傳遞到末端。注意：和其它類比一樣，這里有一些不當(dāng)之處，但這主要是為了可視化。
類似地，如果我們有兩個(gè)系統(tǒng)，其中一個(gè)系統(tǒng)的輸出是另一個(gè)系統(tǒng)的輸入，微調(diào)「反饋網(wǎng)絡(luò)」的參數(shù)將會影響其輸出，該輸出會被傳播下去并乘以任何進(jìn)一步的變化值并貫穿整個(gè)網(wǎng)絡(luò)。
AC 模型概述
因此，我們必須制定一個(gè) ActorCritic 類，它包含一些之前實(shí)現(xiàn)過的 DQN，但是其訓(xùn)練過程更復(fù)雜。因?yàn)槲覀冃枰恍└呒壍墓δ?，我們必須使用包含了基礎(chǔ)庫 Keras 的開源框架：Tensorflow。注意：你也可以在 Theano 中實(shí)現(xiàn)這一點(diǎn)，但是我以前沒有使用過它，所以沒有包含其代碼。如果你選擇這么做，請隨時(shí)向 Theano 提交此代碼的擴(kuò)展。
模型實(shí)現(xiàn)包含四個(gè)主要部分，其直接并行如何實(shí)現(xiàn) DQN 代理：
- 模型參數(shù)/配置
- 訓(xùn)練代碼
- 預(yù)測代碼
1. AC 參數(shù)
第一步，導(dǎo)入需要的庫
- import gym
- import numpy as np
- from keras.models import Sequential, Model
- from keras.layers import Dense, Dropout, Input
- from keras.layers.merge import Add, Multiply
- from keras.optimizers import Adam
- import keras.backend as K
- import tensorflow as tf
- import random
- from collections import deque
參數(shù)與 DQN 中的參數(shù)非常類似。畢竟，這個(gè)行為-評判模型除了兩個(gè)獨(dú)立的模塊之外，還要做與 DQN 相同的任務(wù)。我們還繼續(xù)使用我們在 DQN 報(bào)告中討論的「目標(biāo)網(wǎng)絡(luò)攻擊」，以確保網(wǎng)絡(luò)成功收斂。唯一的新參數(shù)是「tau」，并且涉及在這種情況下如何進(jìn)行目標(biāo)網(wǎng)絡(luò)學(xué)習(xí)的細(xì)微變化：
- class ActorCritic:
- def __init__(self, env, sess):
- self.env = env
- self.sess = sess
- self.learning_rate = 0.001
- self.epsilon = 1.0
- self.epsilon_decay = .995
- self.gamma = .95
- self.tau = .125
- self.memory = deque(maxlen=2000)
在以下的訓(xùn)練部分中，詳細(xì)解釋了 tau 參數(shù)的準(zhǔn)確用法，它的作用其實(shí)就是推動預(yù)測模型向目標(biāo)模型逐步轉(zhuǎn)換。現(xiàn)在，我們找到了主要的關(guān)注點(diǎn)：定義模型。正如我們所描述的，我們有兩個(gè)獨(dú)立的模型，每個(gè)模型都與它自己的目標(biāo)網(wǎng)絡(luò)相關(guān)聯(lián)。
我們從定義行為模型開始。行為模型的目的是根據(jù)當(dāng)前環(huán)境狀態(tài)，得出應(yīng)當(dāng)采取的最佳動作。再次，這個(gè)模型需要處理我們提供的數(shù)字?jǐn)?shù)據(jù)，這意味著沒有空間也沒有必要在網(wǎng)絡(luò)中添加任何比我們迄今為止使用的密集/完全連接層更復(fù)雜的層。因此，行為模型只是一系列全連接層，將環(huán)境觀察的狀態(tài)映射到環(huán)境空間上的一個(gè)點(diǎn)：
- def create_actor_model(self):
- state_input = Input(shape=self.env.observation_space.shape)
- h1 = Dense(24, activation='relu')(state_input)
- h2 = Dense(48, activation='relu')(h1)
- h3 = Dense(24, activation='relu')(h2)
- output = Dense(self.env.action_space.shape[0],
- activation='relu')(h3)
- model = Model(input=state_input, outputoutput=output)
- adam = Adam(lr=0.001)
- model.compile(loss="mse", optimizer=adam)
- return state_input, model
主要的區(qū)別是我們返回了一個(gè)對輸入層的引用。本節(jié)結(jié)尾對此原因的解釋十分清楚，但簡而言之，這解釋了我們?yōu)槭裁磳π袨槟Ｐ偷挠?xùn)練過程采取不同的處理。
行為模型中的棘手部分是決定如何訓(xùn)練它，這就是鏈?zhǔn)椒▌t發(fā)揮作用的地方。但在討論之前，讓我們考慮一下為什么它與標(biāo)準(zhǔn)評論/ DQN 網(wǎng)絡(luò)的訓(xùn)練不同。畢竟，我們不是簡單地去適應(yīng)在 DQN(根據(jù)當(dāng)前狀態(tài)擬合模型)的情況下，基于當(dāng)前的和打折的未來獎(jiǎng)勵(lì)得出接下來采用的最佳動作是哪一個(gè)?問題在于：如果我們能夠按照要求去做，那么這個(gè)問題將會解決。問題在于我們?nèi)绾未_定「最佳動作」是什么，因?yàn)? Q 值現(xiàn)在是在評判網(wǎng)絡(luò)中單獨(dú)計(jì)算出來的。
所以，為了解決這個(gè)問題，我們選擇了一種替代方法。不同于找到「最佳選擇」和擬合，我們實(shí)際上選擇了爬山算法(梯度上升)。對于不熟悉這個(gè)算法的人來說，登山是一個(gè)形象的比喻：從你當(dāng)?shù)氐? POV，找到斜率最大的傾斜方向，并沿著該方向逐步移動。換句話說，爬山正試圖通過原始的沖動并沿著局部斜率最大的方向來達(dá)到全局最大值?？梢韵胂笤谀承┣闆r下，該方法大錯(cuò)特錯(cuò)，但通常情況下，它具備很好的實(shí)用性。
因此，我們想使用該算法來更新我們的行為模型：我們想確定(行為模型中的)參數(shù)的什么變化會導(dǎo)致 Q 值最大幅度的增加(由評判模型預(yù)測得出)。由于行為模型的輸出是動作，評判模型通過環(huán)境狀態(tài)+動作對來評估，我們在此可以看到鏈?zhǔn)椒▌t如何發(fā)揮作用。我們想看看如何改變行為模型的參數(shù)才會改變最終的 Q 值，使用行為網(wǎng)絡(luò)的輸出作為我們的「中間鏈接」(下面的代碼全部在「__init __(self)」方法中)：
- self.actor_state_input, self.actor_model = \
- self.create_actor_model()
- _, selfself.target_actor_model = self.create_actor_model()
- self.actor_critic_grad = tf.placeholder(tf.float32,
- [None, self.env.action_space.shape[0]])
- actor_model_weights = self.actor_model.trainable_weights
- self.actor_grads = tf.gradients(self.actor_model.output,
- actor_model_weights, -self.actor_critic_grad)
- grads = zip(self.actor_grads, actor_model_weights)
- self.optimize = tf.train.AdamOptimizer(
- self.learning_rate).apply_gradients(grads)
我們看到在這里我們持有模型權(quán)重和輸出(動作)之間的梯度。我們還通過負(fù)的 self.actor_critic_grad(因?yàn)槲覀兿朐谶@種情況下使用梯度上升)來放縮它，梯度由占位符持有。對于那些不熟悉 Tensorflow 或首次接觸的讀者，你只需要知道運(yùn)行 Tensorflow 會話時(shí)，占位符將扮演「輸入數(shù)據(jù)」的角色。我不會詳細(xì)介紹它的工作原理，因?yàn)?tensorflow.org 教程的材料相當(dāng)全面。
再來看看評判網(wǎng)絡(luò)，基本上我們面臨著相反的問題。即，網(wǎng)絡(luò)定義稍微復(fù)雜一些，但是訓(xùn)練比較簡單。評判網(wǎng)絡(luò)旨在將環(huán)境狀態(tài)和動作作為輸入，并計(jì)算出相應(yīng)的估值。我們通過合并一系列全連接層以及得出最終的 Q 值預(yù)測之前的中間層來實(shí)現(xiàn)這一點(diǎn)：
- def create_critic_model(self):
- state_input = Input(shape=self.env.observation_space.shape)
- state_h1 = Dense(24, activation='relu')(state_input)
- state_h2 = Dense(48)(state_h1)
- action_input = Input(shape=self.env.action_space.shape)
- action_h1 = Dense(48)(action_input)
- merged = Add()([state_h2, action_h1])
- merged_h1 = Dense(24, activation='relu')(merged)
- output = Dense(1, activation='relu')(merged_h1)
- model = Model(input=[state_input,action_input],
- outputoutput=output)
- adam = Adam(lr=0.001)
- model.compile(loss="mse", optimizer=adam)
- return state_input, action_input, model
需要注意的是如何處理輸入和返回的不對稱性。對于第一點(diǎn)，我們在環(huán)境狀態(tài)輸入中有一個(gè)額外的 FC(全連接)層。我這樣做是因?yàn)檫@是推薦 AC 網(wǎng)絡(luò)使用的結(jié)構(gòu)，但它可能與處理兩個(gè)輸入的 FC 層效果差不多(或稍差)。至于后面一點(diǎn)(我們正在返回的值)，我們需要保留輸入狀態(tài)和動作的引用，因?yàn)槲覀冃枰褂盟鼈兏滦袨榫W(wǎng)絡(luò)：
- self.critic_state_input, self.critic_action_input, \
- selfself.critic_model = self.create_critic_model()
- _, _, selfself.target_critic_model = self.create_critic_model()
- self.critic_grads = tf.gradients(self.critic_model.output,
- self.critic_action_input)
- # Initialize for later gradient calculations
- self.sess.run(tf.initialize_all_variables())
我們在這里設(shè)置要計(jì)算的缺失梯度：關(guān)于動作權(quán)重的輸出 Q。這是訓(xùn)練代碼中直接調(diào)用的，我們現(xiàn)在來深入探討。
2. AC 模型訓(xùn)練
該代碼的最后一個(gè)主要與 DQN 不同的部分是實(shí)際的訓(xùn)練代碼。然而，我們使用了從記憶(LSTM 結(jié)構(gòu))中吸取教訓(xùn)和學(xué)習(xí)的基本結(jié)構(gòu)。由于我們有兩種訓(xùn)練方法，我們將代碼分成不同的訓(xùn)練函數(shù)，并將它們稱為：
- def train(self):
- batch_size = 32
- if len(self.memory) < batch_size:
- return
- rewards = []
- samples = random.sample(self.memory, batch_size)
- self._train_critic(samples)
- self._train_actor(samples)
現(xiàn)在我們定義兩種訓(xùn)練方法。不過，與 DQN 非常相似：我們只是簡單地找到未來打折的獎(jiǎng)勵(lì)和訓(xùn)練方法。唯一的區(qū)別是，我們正在對狀態(tài)/動作對進(jìn)行訓(xùn)練，并使用 target_critic_model 來預(yù)測未來的獎(jiǎng)勵(lì)，而不是僅使用行為來預(yù)測：
- def _train_critic(self, samples):
- for sample in samples:
- cur_state, action, reward, new_state, done = sample
- if not done:
- target_action =
- self.target_actor_model.predict(new_state)
- future_reward = self.target_critic_model.predict(
- [new_state, target_action])[0][0]
- reward += self.gamma * future_reward
- self.critic_model.fit([cur_state, action],
- reward, verbose=0)
對于行為模型，我們幸運(yùn)地解決了所有難題!我們已經(jīng)設(shè)置了梯度如何在網(wǎng)絡(luò)中運(yùn)作，現(xiàn)在只需傳入當(dāng)前的動作和狀態(tài)并調(diào)用該函數(shù)：
- def _train_actor(self, samples):
- for sample in samples:
- cur_state, action, reward, new_state, _ = sample
- predicted_action = self.actor_model.predict(cur_state)
- grads = self.sess.run(self.critic_grads, feed_dict={
- self.critic_state_input: cur_state,
- self.critic_action_input: predicted_action
- })[0]
- self.sess.run(self.optimize, feed_dict={
- self.actor_state_input: cur_state,
- self.actor_critic_grad: grads
- })
如上所述，我們利用了目標(biāo)模型。所以我們必須在每個(gè)時(shí)間步更新其權(quán)重。但是，更新過程太慢了。具體說，我們將目標(biāo)模型的估值保持在一個(gè)分?jǐn)?shù) self.tau 上，并將其更新為余數(shù)(1-self.tau)分?jǐn)?shù)的相應(yīng)模型權(quán)重。行為/評判模型均如此處理，但下面只給出行為模型的代碼(你可以在文章底部的完整代碼中看到評判模型代碼)：
- def _update_actor_target(self):
- actor_model_weights = self.actor_model.get_weights()
- actor_target_weights =self.target_critic_model.get_weights()
- for i in range(len(actor_target_weights)):
- actor_target_weights[i] = actor_model_weights[i]
- self.target_critic_model.set_weights(actor_target_weights
3. AC 模型預(yù)測
這與我們在 DQN 中的做法一樣，所以沒有什么好說的：
- def act(self, cur_state):
- self.epsilon *= self.epsilon_decay
- if np.random.random() < self.epsilon:
- return self.env.action_space.sample()
- return self.actor_model.predict(cur_state)
4. 預(yù)測代碼
預(yù)測代碼也與之前的強(qiáng)化學(xué)習(xí)算法相同。也就是說，我們只需反復(fù)試驗(yàn)，并對代理進(jìn)行預(yù)測、記憶和訓(xùn)練：
- def main():
- sess = tf.Session()
- K.set_session(sess)
- env = gym.make("Pendulum-v0")
- actor_critic = ActorCritic(env, sess)
- num_trials = 10000
- trial_len = 500
- cur_state = env.reset()
- action = env.action_space.sample()
- while True:
- env.render()
- cur_statecur_state = cur_state.reshape((1,
- env.observation_space.shape[0]))
- action = actor_critic.act(cur_state)
- actionaction = action.reshape((1, env.action_space.shape[0]))
- new_state, reward, done, _ = env.step(action)
- new_statenew_state = new_state.reshape((1,
- env.observation_space.shape[0]))
- actor_critic.remember(cur_state, action, reward,
- new_state, done)
- actor_critic.train()
- cur_state = new_state
完整代碼
這是使用 AC(Actor-Critic)對「Pendulum-v0」環(huán)境進(jìn)行訓(xùn)練的完整代碼!
- """
- solving pendulum using actor-critic model
- """
- import gym
- import numpy as np
- from keras.models import Sequential, Model
- from keras.layers import Dense, Dropout, Input
- from keras.layers.merge import Add, Multiply
- from keras.optimizers import Adam
- import keras.backend as K
- import tensorflow as tf
- import random
- from collections import deque
- # determines how to assign values to each state, i.e. takes the state
- # and action (two-input model) and determines the corresponding value
- class ActorCritic:
- def __init__(self, env, sess):
- self.env = env
- self.sess = sess
- self.learning_rate = 0.001
- self.epsilon = 1.0
- self.epsilon_decay = .995
- self.gamma = .95
- self.tau = .125
- # ===================================================================== #
- # Actor Model #
- # Chain rule: find the gradient of chaging the actor network params in #
- # getting closest to the final value network predictions, i.e. de/dA #
- # Calculate de/dA as = de/dC * dC/dA, where e is error, C critic, A act #
- # ===================================================================== #
- self.memory = deque(maxlen=2000)
- self.actor_state_input, selfself.actor_model = self.create_actor_model()
- _, selfself.target_actor_model = self.create_actor_model()
- self.actor_critic_grad = tf.placeholder(tf.float32,
- [None, self.env.action_space.shape[0]]) # where we will feed de/dC (from critic)
- actor_model_weights = self.actor_model.trainable_weights
- self.actor_grads = tf.gradients(self.actor_model.output,
- actor_model_weights, -self.actor_critic_grad) # dC/dA (from actor)
- grads = zip(self.actor_grads, actor_model_weights)
- self.optimize = tf.train.AdamOptimizer(self.learning_rate).apply_gradients(grads)
- # ===================================================================== #
- # Critic Model #
- # ===================================================================== #
- self.critic_state_input, self.critic_action_input, \
- selfself.critic_model = self.create_critic_model()
- _, _, selfself.target_critic_model = self.create_critic_model()
- self.critic_grads = tf.gradients(self.critic_model.output,
- self.critic_action_input) # where we calcaulte de/dC for feeding above
- # Initialize for later gradient calculations
- self.sess.run(tf.initialize_all_variables())
- # ========================================================================= #
- # Model Definitions #
- # ========================================================================= #
- def create_actor_model(self):
- state_input = Input(shape=self.env.observation_space.shape)
- h1 = Dense(24, activation='relu')(state_input)
- h2 = Dense(48, activation='relu')(h1)
- h3 = Dense(24, activation='relu')(h2)
- output = Dense(self.env.action_space.shape[0], activation='relu')(h3)
- model = Model(input=state_input, outputoutput=output)
- adam = Adam(lr=0.001)
- model.compile(loss="mse", optimizer=adam)
- return state_input, model
- def create_critic_model(self):
- state_input = Input(shape=self.env.observation_space.shape)
- state_h1 = Dense(24, activation='relu')(state_input)
- state_h2 = Dense(48)(state_h1)
- action_input = Input(shape=self.env.action_space.shape)
- action_h1 = Dense(48)(action_input)
- merged = Add()([state_h2, action_h1])
- merged_h1 = Dense(24, activation='relu')(merged)
- output = Dense(1, activation='relu')(merged_h1)
- model = Model(input=[state_input,action_input], outputoutput=output)
- adam = Adam(lr=0.001)
- model.compile(loss="mse", optimizer=adam)
- return state_input, action_input, model
- # ========================================================================= #
- # Model Training #
- # ========================================================================= #
- def remember(self, cur_state, action, reward, new_state, done):
- self.memory.append([cur_state, action, reward, new_state, done])
- def _train_actor(self, samples):
- for sample in samples:
- cur_state, action, reward, new_state, _ = sample
- predicted_action = self.actor_model.predict(cur_state)
- grads = self.sess.run(self.critic_grads, feed_dict={
- self.critic_state_input: cur_state,
- self.critic_action_input: predicted_action
- })[0]
- self.sess.run(self.optimize, feed_dict={
- self.actor_state_input: cur_state,
- self.actor_critic_grad: grads
- })
- def _train_critic(self, samples):
- for sample in samples:
- cur_state, action, reward, new_state, done = sample
- if not done:
- target_action = self.target_actor_model.predict(new_state)
- future_reward = self.target_critic_model.predict(
- [new_state, target_action])[0][0]
- reward += self.gamma * future_reward
- self.critic_model.fit([cur_state, action], reward, verbose=0)
- def train(self):
- batch_size = 32
- if len(self.memory) < batch_size:
- return
- rewards = []
- samples = random.sample(self.memory, batch_size)
- self._train_critic(samples)
- self._train_actor(samples)
- # ========================================================================= #
- # Target Model Updating #
- # ========================================================================= #
- def _update_actor_target(self):
- actor_model_weights = self.actor_model.get_weights()
- actor_target_weights = self.target_critic_model.get_weights()
- for i in range(len(actor_target_weights)):
- actor_target_weights[i] = actor_model_weights[i]
- self.target_critic_model.set_weights(actor_target_weights)
- def _update_critic_target(self):
- critic_model_weights = self.critic_model.get_weights()
- critic_target_weights = self.critic_target_model.get_weights()
- for i in range(len(critic_target_weights)):
- critic_target_weights[i] = critic_model_weights[i]
- self.critic_target_model.set_weights(critic_target_weights)
- def update_target(self):
- self._update_actor_target()
- self._update_critic_target()
- # ========================================================================= #
- # Model Predictions #
- # ========================================================================= #
- def act(self, cur_state):
- self.epsilon *= self.epsilon_decay
- if np.random.random() < self.epsilon:
- return self.env.action_space.sample()
- return self.actor_model.predict(cur_state)
- def main():
- sess = tf.Session()
- K.set_session(sess)
- env = gym.make("Pendulum-v0")
- actor_critic = ActorCritic(env, sess)
- num_trials = 10000
- trial_len = 500
- cur_state = env.reset()
- action = env.action_space.sample()
- while True:
- env.render()
- cur_statecur_state = cur_state.reshape((1, env.observation_space.shape[0]))
- action = actor_critic.act(cur_state)
- actionaction = action.reshape((1, env.action_space.shape[0]))
- new_state, reward, done, _ = env.step(action)
- new_statenew_state = new_state.reshape((1, env.observation_space.shape[0]))
- actor_critic.remember(cur_state, action, reward, new_state, done)
- actor_critic.train()
- cur_state = new_state
- if __name__ == "__main__":
- main()
原文地址：
https://medium.com/towards-data-science/reinforcement-learning-w-keras-openai-actor-critic-models-f084612cfd69
【本文是51CTO專欄機(jī)構(gòu)“機(jī)器之心”的原創(chuàng)譯文，微信公眾號“機(jī)器之心( id: almosthuman2014)”】
