########################################## # A3C做出的改進(jìn)： # 解決AC難以收斂的問(wèn)題 # 不一樣的地方： #import threading # import tensorflow as tf import tensorflow.compat.v1 as tftf.compat.v1.disable_eager_execution() import numpy as np import gym import os import shutil import matplotlib.pyplot as pltGAME = 'CartPole-v0' # 游戲名稱 OUTPUT_GRAPH = True LOG_DIR = '../log' # 一個(gè)文件名 N_WORKERS = 3 # 有3個(gè)并行的worker MAX_GLOBAL_EP = 3000 # 最大執(zhí)行回合數(shù) GLOBAL_NET_SCOPE = 'Global_Net' # object名稱 UPDATE_GLOBAL_ITER = 100 # 每UPDATE_GLOBAL_ITER 步 或者回合完了，進(jìn)行sync操作 GAMMA = 0.9 # 折扣因子 ENTROPY_BETA = 0.001 LR_A = 0.001 # learning rate for actor Actor的學(xué)習(xí)率 LR_C = 0.001 # learning rate for critic Actor的學(xué)習(xí)率 GLOBAL_RUNNING_R = [] # 游戲整體的這個(gè)游戲里面一共進(jìn)行了多少步，這個(gè)數(shù)組長(zhǎng)度就是多少 GLOBAL_EP = 0 # 整體進(jìn)行的回合數(shù) STEP = 3000 # Step limitation in an episode TODO 這里應(yīng)該改成200,因?yàn)?00步游戲就達(dá)到終止條件了 TEST = 10 # 實(shí)驗(yàn)次數(shù)每100回合測(cè)試一次 TODO 這里應(yīng)該是100吧？env = gym.make(GAME) N_S = env.observation_space.shape[0] # 狀態(tài)維度（？，4） N_A = env.action_space.n # 動(dòng)作維度（2）# 這個(gè) class 可以被調(diào)用生成一個(gè) global net. # 也能被調(diào)用生成一個(gè) worker 的 net, 因?yàn)樗麄兊慕Y(jié)構(gòu)是一樣的, # 所以這個(gè) class 可以被重復(fù)利用. class ACNet(object):def __init__(self, scope, globalAC=None): # 這里的scope是網(wǎng)絡(luò)名稱# 當(dāng)創(chuàng)建worker網(wǎng)絡(luò)的時(shí)候，我們傳入之前創(chuàng)建的globalAC給這個(gè)worker# 判斷當(dāng)下建立的網(wǎng)絡(luò)是 local 還是 globalif scope == GLOBAL_NET_SCOPE: # 如果建立的網(wǎng)絡(luò)是Global網(wǎng)絡(luò)# 關(guān)于tf.name_scope和tf.variable_scopewith tf.variable_scope(scope):self.s = tf.placeholder(tf.float32, [None, N_S], 'S') # 定義一組狀態(tài)# 這里的a_params具體代表求導(dǎo)時(shí)的那個(gè)分母，也就是對(duì)什么求導(dǎo)self.a_params, self.c_params = self._build_net(scope)[-2:] # 取這個(gè)數(shù)組從第一個(gè)到倒數(shù)第三個(gè)，最后兩個(gè)數(shù)被丟棄了else: # local net, calculate losseswith tf.variable_scope(scope):# 接著計(jì)算 critic loss 和 actor loss# 用這兩個(gè) loss 計(jì)算要推送的 gradientsself.s = tf.placeholder(tf.float32, [None, N_S], 'S') # 定義一組狀態(tài)self.a_his = tf.placeholder(tf.int32, [None, ], 'A') # 定義了一組動(dòng)作self.v_target = tf.placeholder(tf.float32, [None, 1], 'Vtarget') # 目標(biāo)價(jià)值self.a_prob, self.v, self.a_params, self.c_params = self._build_net(scope)# 接著計(jì)算 critic loss 和 actor loss# 用這兩個(gè) loss 計(jì)算要推送的 gradients# 張量減法運(yùn)算：tf.subtract函數(shù)是一個(gè)算術(shù)運(yùn)算符, 用于表示減法, 返回x - y的元素td = tf.subtract(self.v_target, self.v, name='TD_error') # td_error = v_target - vwith tf.name_scope('c_loss'): # c網(wǎng)絡(luò)loss函數(shù)self.c_loss = tf.reduce_mean(tf.square(td)) # td_error先求平方然后取平均with tf.name_scope('a_loss'): # 下面這些操作都封裝到一個(gè)叫a_loss的作用域中，用來(lái)計(jì)算a_loss,方便可視化# 下面這些應(yīng)該是在實(shí)現(xiàn)a_loss的那個(gè)公式log_prob = tf.reduce_sum(tf.log(self.a_prob + 1e-5) * tf.one_hot(self.a_his, N_A, dtype=tf.float32),axis=1, keep_dims=True)# tf.stop_gradient: 停止梯度運(yùn)算，當(dāng)在一個(gè)圖中執(zhí)行時(shí)， 這個(gè)op按原樣輸出它的輸入張量。# 當(dāng)構(gòu)建ops來(lái)計(jì)算梯度時(shí)，該op會(huì)阻止將其輸入貢獻(xiàn)考慮在內(nèi)。exp_v = log_prob * tf.stop_gradient(td)entropy = -tf.reduce_sum(self.a_prob * tf.log(self.a_prob + 1e-5),axis=1, keep_dims=True) # encourage explorationself.exp_v = ENTROPY_BETA * entropy + exp_v # expect valueself.a_loss = tf.reduce_mean(-self.exp_v) # actor_losswith tf.name_scope('local_grad'):# tf.gradients 實(shí)現(xiàn)a_loss對(duì)a_params求導(dǎo) 梯度self.a_grads = tf.gradients(self.a_loss, self.a_params)self.c_grads = tf.gradients(self.c_loss, self.c_params)with tf.name_scope('sync'): # 同步with tf.name_scope('pull'): # 調(diào)用pull，這個(gè)worker就會(huì)從global_net中獲取到最新的參數(shù)# assign相當(dāng)于連線，一般是將一個(gè)變量的值不間斷地賦值給另一個(gè)變量，就像把這兩個(gè)變量連在一起，所以習(xí)慣性的當(dāng)做連線用，比如把一個(gè)模塊的輸出給另一個(gè)模塊當(dāng)輸入。self.pull_a_params_op = [l_p.assign(g_p) for l_p, g_p in zip(self.a_params, globalAC.a_params)]self.pull_c_params_op = [l_p.assign(g_p) for l_p, g_p in zip(self.c_params, globalAC.c_params)]with tf.name_scope('push'): # 調(diào)用push，這個(gè)worker就會(huì)將自己的個(gè)人更新推送去global_netself.update_a_op = OPT_A.apply_gradients(zip(self.a_grads, globalAC.a_params))self.update_c_op = OPT_C.apply_gradients(zip(self.c_grads, globalAC.c_params))def _build_net(self, scope): # 這里搭建Actor和Critic網(wǎng)絡(luò)w_init = tf.random_normal_initializer(0., .1) # 生成一組符合標(biāo)準(zhǔn)正態(tài)分布的 tensor 對(duì)象,初始化張量with tf.variable_scope('actor'): # 搭建一個(gè)actor網(wǎng)絡(luò)"""這個(gè)actor網(wǎng)絡(luò)有兩層，第一層輸入是狀態(tài)，輸出一個(gè)維度為200的東西，第一層的輸出l_a是第二層的輸入，第一層的激活函數(shù)和第二層不一樣"""# https://blog.csdn.net/yangfengling1023/article/details/81774580/# 全連接層，曾加了一個(gè)層，全連接層執(zhí)行操作 outputs = activation(inputs.kernel+bias) 如果執(zhí)行結(jié)果不想進(jìn)行激活操作，則設(shè)置activation=None# self.s:輸入該網(wǎng)絡(luò)層的數(shù)據(jù) 200：輸出的維度大小，改變inputs的最后一維 tf.nn.relu6：激活函數(shù)，即神經(jīng)網(wǎng)絡(luò)的非線性變化# kernel_initializer=w_init：卷積核的初始化器 name:層的名字l_a = tf.layers.dense(self.s, 200, tf.nn.relu6, kernel_initializer=w_init, name='la')a_prob = tf.layers.dense(l_a, N_A, tf.nn.softmax, kernel_initializer=w_init, name='ap')with tf.variable_scope('critic'): # 搭建critic網(wǎng)絡(luò)"""這個(gè)網(wǎng)絡(luò)也是有兩層，第一層有100個(gè)輸出，第二層只有一個(gè)輸出，"""l_c = tf.layers.dense(self.s, 100, tf.nn.relu6, kernel_initializer=w_init, name='lc')v = tf.layers.dense(l_c, 1, kernel_initializer=w_init, name='v') # state value# tf.get_collection 用來(lái)獲取一個(gè)名稱是‘key’的集合中的所有元素，返回的是一個(gè)列表，列表的順序是按照變量放入集合中的先后;# scope參數(shù)可選，表示的是名稱空間（名稱域），如果指定，就返回名稱域中所有放入‘key’的變量的列表，不指定則返回所有變量。# TODO 這個(gè)含義看不懂。。。TRAINABLE_VARIABLES：將由優(yōu)化程序訓(xùn)練的Variable對(duì)象的子集a_params = tf.get_collection(tf.GraphKeys.TRAINABLE_VARIABLES, scope=scope + '/actor')c_params = tf.get_collection(tf.GraphKeys.TRAINABLE_VARIABLES, scope=scope + '/critic')return a_prob, v, a_params, c_params # 返回均值、方差，v:state_value TODO a_params, c_params 這兩個(gè)參數(shù)什么意思啊def update_global(self, feed_dict): # run by a local 進(jìn)行push操作SESS.run([self.update_a_op, self.update_c_op], feed_dict) # 將本地def pull_global(self): # run by a local #進(jìn)行pull操作SESS.run([self.pull_a_params_op, self.pull_c_params_op])def choose_action(self, s): # run by a local 根據(jù)s選擇動(dòng)作prob_weights = SESS.run(self.a_prob, feed_dict={self.s: s[np.newaxis, :]})action = np.random.choice(range(prob_weights.shape[1]),p=prob_weights.ravel()) # select action w.r.t the actions prob 根據(jù)概率選擇動(dòng)作return actionclass Worker(object):def __init__(self, name, globalAC):self.env = gym.make(GAME).unwrapped # 創(chuàng)建自己的環(huán)境self.name = name # 自己的名字self.AC = ACNet(name, globalAC) # 自己的local net，并綁定上globalACdef work(self): # worker網(wǎng)絡(luò)的主體部分# s,a,r 的緩存，用于n_steps更新global GLOBAL_RUNNING_R, GLOBAL_EPtotal_step = 1 # 初始化步數(shù)buffer_s, buffer_a, buffer_r = [], [], [] # state,action,和reward的緩存# 當(dāng) COORD不需要停止的時(shí)候，或者當(dāng)GLOBAL_EP比MAX_GLOBAL_EP小的時(shí)候while not COORD.should_stop() and GLOBAL_EP < MAX_GLOBAL_EP: # MAX_GLOBAL_EP最大訓(xùn)練EPs = self.env.reset() # 重置環(huán)境ep_r = 0 # 統(tǒng)計(jì)ep的總rewardwhile True:# if self.name == 'W_0':# self.env.render()a = self.AC.choose_action(s) # 選擇動(dòng)作s_, r, done, info = self.env.step(a) # 與環(huán)境互動(dòng)if done: r = -5 # 把reward-100的時(shí)候，改為-5ep_r += r # 保存數(shù)據(jù) ep_r 總的rewardbuffer_s.append(s) # 添加各種緩存buffer_a.append(a)buffer_r.append(r)# 每UPDATE_GLOBAL_ITER 步 或者回合完了，進(jìn)行sync操作，也就是學(xué)習(xí)的步驟if total_step % UPDATE_GLOBAL_ITER == 0 or done: # update global and assign to local net# 獲得用于計(jì)算TD_error 的下一state的value# 下面對(duì)應(yīng)價(jià)值函數(shù)的那個(gè)公式，如果是terminal state，則對(duì)未來(lái)的期望等于0if done:v_s_ = 0 # terminalelse:v_s_ = SESS.run(self.AC.v, {self.AC.s: s_[np.newaxis, :]})[0, 0]buffer_v_target = [] # 下 state value的緩存，用于計(jì)算TD# 計(jì)算每個(gè)state的V(s')# print(a[::-1]) ### 取從后向前（相反）的元素[1 2 3 4 5]-->[ 5 4 3 2 1 ]# （莫煩說(shuō)）對(duì)MDP的一個(gè)反向的計(jì)算，對(duì)未來(lái)的reward的一個(gè)遞解的步驟for r in buffer_r[::-1]:v_s_ = r + GAMMA * v_s_buffer_v_target.append(v_s_)buffer_v_target.reverse() # 先反轉(zhuǎn)填充再轉(zhuǎn)回來(lái)# np.vstack:按垂直方向（行順序）堆疊數(shù)組構(gòu)成一個(gè)新的數(shù)組buffer_s, buffer_a, buffer_v_target = np.vstack(buffer_s), np.array(buffer_a), np.vstack(buffer_v_target)# feed_dict的作用是給使用placeholder創(chuàng)建出來(lái)的tensor賦值feed_dict = {self.AC.s: buffer_s,self.AC.a_his: buffer_a,self.AC.v_target: buffer_v_target,}self.AC.update_global(feed_dict) # 推送更新去globalACbuffer_s, buffer_a, buffer_r = [], [], [] # 清空緩存self.AC.pull_global() # 獲取globalAC的最新參數(shù)s = s_ # 更新?tīng)顟B(tài)total_step += 1 # 整體部署+1if done:# 達(dá)到游戲終止條件，更新rewardif len(GLOBAL_RUNNING_R) == 0: # 如果是第一回合GLOBAL_RUNNING_R.append(ep_r)else: # 不是第一回合GLOBAL_RUNNING_R.append(0.99 * GLOBAL_RUNNING_R[-1] + 0.01 * ep_r)print(self.name,"Ep:", GLOBAL_EP,"| Ep_r: %i" % GLOBAL_RUNNING_R[-1],)GLOBAL_EP += 1 # 加一回合break # 結(jié)束這回合if __name__ == '__main__':SESS = tf.Session() # 創(chuàng)建一個(gè)會(huì)話# 下面是真正的重點(diǎn)！！worker并行計(jì)算# 使用 tf.device() 指定模型運(yùn)行的具體設(shè)備，可以指定運(yùn)行在GPU還是CPU上，以及哪塊GPU上。with tf.device("/cpu:0"): # 以下部分，都在CPU0完成# tf.train.RMSPropOptimizerSHI是一種優(yōu)化算法，有很多種優(yōu)化算法，具體見(jiàn)下面這個(gè)文檔，有空好好學(xué)習(xí)下# https://www.cnblogs.com/bigcome/p/10084220.htmlOPT_A = tf.train.RMSPropOptimizer(LR_A, name='RMPropA') # 定義了一個(gè)actor優(yōu)化器OPT_C = tf.train.RMSPropOptimizer(LR_C, name='RMPropC') # 定義了一個(gè)Critic優(yōu)化器GLOBAL_AC = ACNet(GLOBAL_NET_SCOPE) # we only need its params 定義了一個(gè)總的AC網(wǎng)絡(luò)workers = [] # 定義workersfor i in range(N_WORKERS): # 創(chuàng)建worker，之后再并行i_name = 'W_%i' % i # worker的名字workers.append(Worker(i_name, GLOBAL_AC)) # 每個(gè)worker都有共享這個(gè)global AC# 調(diào)用 tf.train.Coordinator() 來(lái)創(chuàng)建一個(gè)線程協(xié)調(diào)器，用來(lái)管理之后在Session中啟動(dòng)的所有線程;COORD = tf.train.Coordinator() # Tensorflow用于并行的工具SESS.run(tf.global_variables_initializer())# 執(zhí)行到這兒的意思是，OUTPUT_GRAPH這個(gè)參量只出現(xiàn)了一次，而且是true，也就是說(shuō)，刪除path這個(gè)路徑里面的LOG_DIR文件，然后保存一張新的進(jìn)去，也就是咱們收斂情況的圖if OUTPUT_GRAPH:# os.path.exists()就是判斷括號(hào)里的文件是否存在的意思，path代表路徑，括號(hào)內(nèi)的可以是文件名。if os.path.exists(LOG_DIR): # 這句是說(shuō)，如果LOG_DIR這個(gè)文件存在，就將它刪除# 在python文件中，使用代碼刪除文件夾以及里面的文件，可以使用shutil.rmtree，遞歸地刪除文件夾以及里面的文件。shutil.rmtree(LOG_DIR)# tf.summary.FileWriter 指定一個(gè)文件用來(lái)保存圖 ，指定LOG_DIR這個(gè)文件來(lái)保存SESS.graph這個(gè)圖tf.summary.FileWriter(LOG_DIR, SESS.graph)# 開(kāi)啟tf線程worker_threads = []for worker in workers: # 執(zhí)行每一個(gè)workerjob = lambda: worker.work() # 有一個(gè)工人（worker）,有一個(gè)方法（work）,這句是說(shuō)讓這個(gè)工人worker去執(zhí)行work這個(gè)方法t = threading.Thread(target=job) # 添加一個(gè)工作線程t.start() # 開(kāi)始這個(gè)線程worker_threads.append(t) # 把這個(gè)線程添加到worker_threads中COORD.join(worker_threads) # 當(dāng)所有的worker都運(yùn)行完了才會(huì)進(jìn)行下面的步驟，如果沒(méi)有這一句，那么每一個(gè)worker運(yùn)行完就會(huì)進(jìn)行下面的步驟testWorker = Worker("test", GLOBAL_AC) # 創(chuàng)建一個(gè)測(cè)試worker TODO 不知道這個(gè)worker有什么作用testWorker.AC.pull_global() # 對(duì)testworker執(zhí)行pull操作# 下面應(yīng)該是測(cè)試部分total_reward = 0for i in range(TEST):state = env.reset() # 初始化狀態(tài)for j in range(STEP):env.render() # env.render()函數(shù)用于渲染出當(dāng)前的智能體以及環(huán)境的狀態(tài)action = testWorker.AC.choose_action(state) # 選擇一個(gè)測(cè)試動(dòng)作state, reward, done, _ = env.step(action)total_reward += reward # reward疊加if done: # 達(dá)到終止條件就跳出循環(huán)breakave_reward = total_reward / TEST# 打印內(nèi)容：整個(gè)過(guò)程一共進(jìn)行了多少回合,平均獎(jiǎng)勵(lì)是多少print('episode: ', GLOBAL_EP, 'Evaluation Average Reward:', ave_reward)plt.plot(np.arange(len(GLOBAL_RUNNING_R)), GLOBAL_RUNNING_R) # TODO GLOBAL_RUNNING_R這個(gè)參量再研究一下plt.xlabel('step')plt.ylabel('Total moving reward')plt.show()

總結(jié)

以上是生活随笔為你收集整理的【强化学习】A3C代码注释版本的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。