学习笔记：强化学习之A3C代码详解
写在前面：我是根据莫烦的视频学习的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学习书籍，莫烦老师的教程更加清晰明了，配合代码能够理解的较为深入。其次是学长聪哥的答疑解惑，很多问题都是在聪哥的点拨下才理解清楚，在这里表示感谢！另外有任何指正或者建议意见都可以留在下面一起讨论！谢谢！