寫在前面：我是根據莫煩的視訊學習的Reinforce learning，具體程式碼實現包括Q-learning，SARSA，DQN，Policy-Gradient，Actor-Critic以及A3C。（莫凡老師的網站：https://morvanzhou.github.io/tutorials/machine-learning/reinforcement-learning/）
今天先發表最後一個也是最複雜的一個，將來會應用到自己課設中的A3C演算法，以後會把前面幾個程式碼和註解一起釋出。

演算法理解：首先說一下A3C演算法的個人理解，他是基於Actor-Critic的一種改進演算法，主要是利用多執行緒維持多個agent學習並分享各自的經驗，達到更快的學習速率和更好的訓練效用。類似於三個工人和一個管理者共同解決一個難題，三個工人在解決過程中向管理者彙報自己的解題經驗，而管理者彙總三個人的經驗再將總結後的經驗發放給三個工人，讓他們獲得集體的經驗並繼續解決問題。
具體到A3C演算法上，這個演算法維持著一箇中央大腦global，以及n個獨立個體worker（n一般取cpu數量，有多少個cpu就有多少個worker）。global和worker各自維持著結構完全相同的網路，Actor網路和Critic網路。worker通過actor—critic網路獲取近幾個回合所獲得的經驗並上傳到global網路中，global彙集幾個worker的經驗後下發給每一個worker，這裡所說的經驗，其實就是actor和critic網路的引數。
需要著重注意的兩點，首先是這個上傳經驗，上傳的是損失loss對神經網路引數求導所得到的梯度，global從worker中獲取到這個梯度並更新自己的網路引數。下載經驗，是worker直接將global中的神經網路引數全盤接受，並覆蓋當前自己的神經網路actor和critic。
另外，相較於普通的Actor-critic網路，每一個worker沒有自我學習的過程，在Actor-critic中，worker通過optimizer最小化損失loss並更新自己的網路引數。而在A3C演算法中，這一過程被global-net取代了，更新自己的網路引數只需從global中獲取最新的引數即可（集體的智慧）。

具體程式碼：
環境：開發環境python3.7。 遊戲環境：gym中的 ‘CartPole-v0’ 一級倒立擺

import multiprocessing  #多執行緒模組
import threading  #執行緒模組
import tensorflow as tf
import numpy as np
import gym
import os
import shutil  #拷貝檔案用
import matplotlib.pyplot as plt

Game='CartPole-v0'
N_workers=multiprocessing.cpu_count()    #獨立玩家個體數為cpu數
MAX_GLOBAL_EP=2000  #中央大腦最大回合數
GLOBALE_NET_SCOPE='Globale_Net' #中央大腦的名字
UPDATE_GLOBALE_ITER=10   #中央大腦每N次提升一次
GAMMA=0.9    #衰減度
LR_A=0.0001   #Actor網路學習率
LR_C=0.001    #Critic 網路學習率

GLOBALE_RUNNING_R=[]   #儲存總的reward
GLOBALE_EP=0   #中央大腦步數

env=gym.make(Game)   #定義遊戲環境


N_S=env.observation_space.shape[0]  #觀測值個數
N_A=env.action_space.n              #行為值個數


class ACnet(object):     #這個class即可用於生產global net，也可生成 worker net，因為結構相同
    def __init__(self,scope,globalAC=None):   #scope 用於確定生成什麼網路
        if scope==GLOBALE_NET_SCOPE:   #建立中央大腦
            with tf.variable_scope(scope):
                self.s=tf.placeholder(tf.float32,[None,N_S],'S')   #初始化state，None代表batch，N—S是每個state的觀測值個數
                self.build_net(scope)                               #建立中央大腦神經網路
                self.a_params=tf.get_collection(tf.GraphKeys.TRAINABLE_VARIABLES,scope=scope+'/actor')
                self.c_params = tf.get_collection(tf.GraphKeys.TRAINABLE_VARIABLES, scope=scope + '/critic')
                #定義中央大腦actor和critic的引數


        else:                        #建立worker兩個網路的具體步驟
            with tf.variable_scope(scope):    #這裡的scope傳入的是worker的名字
                self.s=tf.placeholder(tf.float32,[None,N_S],'S')  #初始化state
                self.a_his = tf.placeholder(tf.int32, [None,1], 'A_his')         #初始化action,是一個[batch，1]的矩陣，第二個維度為1，
                                                                                  #格式類似於[[1],[2],[3]]
                self.v_target=tf.placeholder(tf.float32,[None,1],'Vtarget')     #初始化v現實(V_target)，資料格式和上面相同


                self.acts_prob,self.v=self.build_net(scope)   #建立神經網路，acts_prob為返回的概率值,v為返回的評價值
                self.a_params = tf.get_collection(tf.GraphKeys.TRAINABLE_VARIABLES, scope=scope + '/actor')
                self.c_params = tf.get_collection(tf.GraphKeys.TRAINABLE_VARIABLES, scope=scope + '/critic')


                td=tf.subtract(self.v_target,self.v,name='TD_error')  #計算td—error即v現實和v估計之差
                                                                      #v—target和v都是一串值，v-target（現實）已經計算好並傳入了，v估計由傳入的
                                                                      #一系列state送入critic網路確定

                with tf.name_scope('c_loss'):    #計算Critic網路的loss
                    self.c_loss=tf.reduce_mean(tf.square(td))  #Critic的loss就是td—error加平方避免負數


                with tf.name_scope('a_loss'):    #計算actor網路的損失
                    log_prob = tf.reduce_sum(tf.log(self.acts_prob +1e-5)*tf.one_hot(self.a_his,N_A,dtype=tf.float32),axis=1,keep_dims=True)
                    #這裡是矩陣乘法，目的是篩選出本batch曾進行的一系列選擇的概率值，acts—prob類似於一個向量[0.3,0.8,0.5]，
                    #one—hot是在本次進行的的操作置位1，其他位置置為0，比如走了三次a—his為[1,0,3],N—A是4，則one—hot就是[[0,1,0,0],[1,0,0,0],[0,0,0,1]]
                    #相乘以後就是[[0,0.3,0,0],[0.8,0,0,0],[0,0,0,0.5]],log_prob就是計算這一系列選擇的log值。

                    self.exp_v = log_prob * td     #td決定梯度下降的方向
                    self.a_loss=tf.reduce_mean(-self.exp_v)    #計算actor網路的損失a-loss


                with tf.name_scope('local_grad'):
                    self.a_grads=tf.gradients(self.a_loss,self.a_params)   #實現a_loss對a_params每一個引數的求導，返回一個list
                    self.c_grads=tf.gradients(self.c_loss,self.c_params)   #實現c_loss對c_params每一個引數的求導，返回一個list


            with tf.name_scope('sync'):   #worker和global的同步過程
                with tf.name_scope('pull'):   #獲取global引數,複製到local—net
                    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'):   #將引數傳送到gloabl中去
                    self.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))
                    #其中傳送的是local—net的actor和critic的引數梯度grads,具體計算在上面定義
                    #apply_gradients是tf.train.Optimizer中自帶的功能函式，將求得的梯度引數更新到global中


    def build_net(self,scope): #建立神經網路過程
        w_init=tf.random_normal_initializer(0.,.1)  #初始化神經網路weights
        with tf.variable_scope('actor'):            #actor神經網路結構
            l_a=tf.layers.dense(inputs=self.s,units=200,activation=tf.nn.relu6,
                                kernel_initializer=w_init,bias_initializer=tf.constant_initializer(0.1),name='la')  #建立第一層神經網路
            acts_prob=tf.layers.dense(inputs=l_a,units=N_A,activation=tf.nn.softmax,
                               kernel_initializer=w_init,bias_initializer=tf.constant_initializer(0.1),name='act_prob')  #第二層神經網路其中之一輸出為動作的均值

        with tf.variable_scope('critic'):     #critic神經網路結構,輸入為位置的觀測值，輸出為評價值v
            l_c=tf.layers.dense(self.s,20,tf.nn.relu6,kernel_initializer=w_init,bias_initializer=tf.constant_initializer(0.1),name='lc')  #建立第一層神經網路
            v=tf.layers.dense(l_c,1,kernel_initializer=w_init,bias_initializer=tf.constant_initializer(0.1),name='v')   #第二層神經網路

        return acts_prob,v    #建立神經網路後返回的是輸入當前state得到的actor網路的動作概率和critic網路的v估計


    def update_global(self,feed_dict):    #定義更新global引數函式
        SESS.run([self.update_a_op,self.update_c_op],feed_dict)    #分別更新actor和critic網路

    def pull_global(self):   #定義更新local引數函式
        SESS.run([self.pull_a_params_op,self.pull_c_params_op])

    def choose_action(self,s):   #定義選擇動作函式
        s=s[np.newaxis, :]
        probs=SESS.run(self.acts_prob, feed_dict={self.s: s})
        return np.random.choice(range(probs.shape[1]), p=probs.ravel())   #從probs中按概率選取出某一個動作



class Worker(object):
    def __init__(self,name,globalAC):    #傳入的name是worker的名字，globalAC是已經建立好的中央大腦GLOBALE—AC
        self.env=gym.make(Game).unwrapped
        self.name=name                   #worker的名字
        self.AC=ACnet(name,globalAC)     #第二個引數當傳入的是已經建立好的GLOBALE—AC時建立的是local net
                                         #建立worker的AC網路

    def work(self):   #定義worker執行的的具體過程
        global  GLOBALE_RUNNING_R,GLOBALE_EP   #兩個全域性變數，R是所有worker的總reward，ep是所有worker的總episode
        total_step=1                            #本worker的總步數
        buffer_s,buffer_a,buffer_r=[],[],[]    #state,action,reward的快取

        while not COORD.should_stop() and GLOBALE_EP<MAX_GLOBAL_EP:   #停止本worker執行的條件
                                                                     #本迴圈一次是一個回合

            s=self.env.reset()       #初始化環境
            ep_r=0                   #本回合總的reward

            while True:      #本迴圈一次是一步
                if self.name=='W_0':    #只有worker0才將動畫影象顯示
                    self.env.render()

                a=self.AC.choose_action(s)    #將當前狀態state傳入AC網路選擇動作action

                s_,r,done,info=self.env.step(a)   #行動並獲得新的狀態和回報等資訊

                if done:r=-5    #如果結束了，reward給一個懲罰數

                ep_r+=r              #記錄本回合總體reward
                buffer_s.append(s)   #將當前狀態，行動和回報加入快取
                buffer_a.append(a)
                buffer_r.append(r)


                if total_step % UPDATE_GLOBALE_ITER==0 or done:  #每iter步完了或者或者到達終點了，進行同步sync操作
                    if done:
                        v_s_=0   #如果結束了，設定對未來的評價值為0
                    else:
                        v_s_=SESS.run(self.AC.v,feed_dict={self.AC.s:s_[np.newaxis,:]})[0,0]   #如果是中間步驟，則用AC網路分析下一個state的v評價

                    buffer_v_target=[]
                    for r in buffer_r[::-1]:    #將下一個state的v評價進行一個反向衰減傳遞得到每一步的v現實
                        v_s_=r + GAMMA* v_s_
                        buffer_v_target.append(v_s_)  #將每一步的v現實都加入快取中
                    buffer_v_target.reverse()    #反向後，得到本系列操作每一步的v現實(v-target)

                    buffer_s,buffer_a,buffer_v_target=np.vstack(buffer_s),np.vstack(buffer_a),np.vstack(buffer_v_target)

                    feed_dict={
                        self.AC.s:buffer_s,                 #本次走過的所有狀態，用於計算v估計
                        self.AC.a_his:buffer_a,             #本次進行過的所有操作，用於計算a—loss
                        self.AC.v_target:buffer_v_target    #走過的每一個state的v現實值，用於計算td
                    }

                    self.AC.update_global(feed_dict)  #update—global的具體過程在AC類中定義，feed-dict如上

                    buffer_s,buffer_a,buffer_r=[],[],[]   #清空快取

                    self.AC.pull_global()    #從global—net提取出引數賦值給local—net

                s=s_   #跳轉到下一個狀態
                total_step+=1  #本回合總步數加1


                if done:   #如果本回合結束了
                    if len(GLOBALE_RUNNING_R)==0:  #如果尚未記錄總體running
                        GLOBALE_RUNNING_R.append(ep_r)
                    else:
                        GLOBALE_RUNNING_R.append(0.9*GLOBALE_RUNNING_R[-1]+0.1*ep_r)

                    print(self.name,'EP:',GLOBALE_EP)
                    GLOBALE_EP+=1       #加一回合
                    break   #結束本回合



if __name__=='__main__':
    SESS=tf.Session()

    with tf.device('/cpu:0'):
        OPT_A=tf.train.RMSPropOptimizer(LR_A,name='RMSPropA')    #定義actor訓練過程,後續主要是使用該optimizer中的apply—gradients操作
        OPT_C = tf.train.RMSPropOptimizer(LR_C, name='RMSPropC')  #定義critic訓練過程
        GLOBALE_AC=ACnet(GLOBALE_NET_SCOPE)  #建立中央大腦GLOBALE_AC，只建立結構(A和C的引數)
        workers=[]
        for i in range(N_workers):    #N—workers等於cpu數量
            i_name='W_%i'%i   #worker name
            workers.append(Worker(name=i_name,globalAC=GLOBALE_AC))   #建立獨立的worker

        COORD=tf.train.Coordinator()    #多執行緒
        SESS.run(tf.global_variables_initializer())   #初始化所有引數

        worker_threads=[]
        for worker in workers:    #並行過程
            job= lambda:worker.work()   #worker的工作目標,此處呼叫Worker類中的work
            t=threading.Thread(target=job)  #每一個執行緒完成一個worker的工作目標
            t.start()                 # 啟動每一個worker
            worker_threads.append(t)   #每一個worker的工作都加入thread中
        COORD.join(worker_threads)     #合併幾個worker,當每一個worker都執行完再繼續後面步驟

        plt.plot(np.arange(len(GLOBALE_RUNNING_R)),GLOBALE_RUNNING_R)   #繪製reward影象
        plt.xlabel('step')
        plt.ylabel('Total moving reward')
        plt.show()

 

寫在後面：這是A3C演算法的實現，也算是做過裡面最複雜的一個了，需要Actor-critic的基礎（後面還會貼上其他幾個RL演算法）。當中也遇到了很多問題，首先是tensorflow運用不熟練，由於是從RL才開始真正使用tensorflow框架，所以對其中很多東西理解不深，比如張量的資料型別，由於不像numpy等可以直接輸出，張量的資料型別全靠理解，佔位符的各種運算也基於對張量的型別理解，因此比較吃力。其次是TF中眾多的函式還需要加深理解，像本程式碼中的tf.train.Optimizer中地apply-gradient這些函式過去都沒有遇到過。另外是面向物件的思想，由於java是我們的選修課，儘管放假時再次學習了java基礎，但對於面向物件的程式設計使用的不多，而這樣的程式設計又是現在軟體開發的主流思想，因此也需要多多學習。
在RL的學習過程中，萬分感謝莫煩老師的教程，相比於其他尚不成熟的RL學習書籍，莫煩老師的教程更加清晰明瞭，配合程式碼能夠理解的較為深入。其次是學長聰哥的答疑解惑，很多問題都是在聰哥的點撥下才理解清楚，在這裡表示感謝！另外有任何指正或者建議意見都可以留在下面一起討論！謝謝！