|
| 1 | +import multiprocessing |
| 2 | +import threading |
| 3 | +import tensorflow as tf |
| 4 | +import numpy as np |
| 5 | +import gym |
| 6 | +import os |
| 7 | +import shutil |
| 8 | +import matplotlib.pyplot as plt |
| 9 | + |
| 10 | + |
| 11 | +GAME = 'CartPole-v0' |
| 12 | +OUTPUT_GRAPH = True |
| 13 | +LOG_DIR = './log' |
| 14 | +N_WORKERS = multiprocessing.cpu_count() |
| 15 | +MAX_GLOBAL_EP = 1000 |
| 16 | +GLOBAL_NET_SCOPE = 'Global_Net' |
| 17 | +UPDATE_GLOBAL_ITER = 10 |
| 18 | +GAMMA = 0.9 |
| 19 | +ENTROPY_BETA = 0.001 |
| 20 | +LR_A = 0.001 # learning rate for actor |
| 21 | +LR_C = 0.001 # learning rate for critic |
| 22 | +GLOBAL_RUNNING_R = [] |
| 23 | +GLOBAL_EP = 0 |
| 24 | + |
| 25 | +env = gym.make(GAME) |
| 26 | +N_S = env.observation_space.shape[0] |
| 27 | +N_A = env.action_space.n |
| 28 | + |
| 29 | + |
| 30 | +class ACNet(object): |
| 31 | + def __init__(self, scope, globalAC=None): |
| 32 | + |
| 33 | + if scope == GLOBAL_NET_SCOPE: # get global network |
| 34 | + with tf.variable_scope(scope): |
| 35 | + self.s = tf.placeholder(tf.float32, [None, N_S], 'S') |
| 36 | + self.a_params, self.c_params = self._build_net(scope)[-2:] |
| 37 | + else: # local net, calculate losses |
| 38 | + with tf.variable_scope(scope): |
| 39 | + self.s = tf.placeholder(tf.float32, [None, N_S], 'S') |
| 40 | + self.a_his = tf.placeholder(tf.int32, [None, ], 'A') |
| 41 | + self.v_target = tf.placeholder(tf.float32, [None, 1], 'Vtarget') |
| 42 | + |
| 43 | + self.a_prob, self.v, self.a_params, self.c_params = self._build_net(scope) |
| 44 | + |
| 45 | + td = tf.subtract(self.v_target, self.v, name='TD_error') |
| 46 | + with tf.name_scope('c_loss'): |
| 47 | + self.c_loss = tf.reduce_mean(tf.square(td)) # critic的loss是平方loss |
| 48 | + |
| 49 | + with tf.name_scope('a_loss'): |
| 50 | + # Q * log( |
| 51 | + log_prob = tf.reduce_sum(tf.log(self.a_prob + 1e-5) * |
| 52 | + tf.one_hot(self.a_his, N_A, dtype=tf.float32), |
| 53 | + axis=1, keep_dims=True) |
| 54 | + exp_v = log_prob * tf.stop_gradient(td) # 这里的td不再求导,当作是常数 |
| 55 | + entropy = -tf.reduce_sum(self.a_prob * tf.log(self.a_prob + 1e-5), |
| 56 | + axis=1, keep_dims=True) # encourage exploration |
| 57 | + self.exp_v = ENTROPY_BETA * entropy + exp_v |
| 58 | + self.a_loss = tf.reduce_mean(-self.exp_v) |
| 59 | + |
| 60 | + with tf.name_scope('local_grad'): |
| 61 | + self.a_grads = tf.gradients(self.a_loss, self.a_params) |
| 62 | + self.c_grads = tf.gradients(self.c_loss, self.c_params) |
| 63 | + |
| 64 | + with tf.name_scope('sync'): |
| 65 | + with tf.name_scope('pull'): # 把主网络的参数赋予各子网络 |
| 66 | + self.pull_a_params_op = [l_p.assign(g_p) for l_p, g_p in zip(self.a_params, globalAC.a_params)] |
| 67 | + self.pull_c_params_op = [l_p.assign(g_p) for l_p, g_p in zip(self.c_params, globalAC.c_params)] |
| 68 | + with tf.name_scope('push'): # 使用子网络的梯度对主网络参数进行更新 |
| 69 | + self.update_a_op = OPT_A.apply_gradients(zip(self.a_grads, globalAC.a_params)) |
| 70 | + self.update_c_op = OPT_C.apply_gradients(zip(self.c_grads, globalAC.c_params)) |
| 71 | + |
| 72 | + def _build_net(self, scope): |
| 73 | + w_init = tf.random_normal_initializer(0., .1) |
| 74 | + with tf.variable_scope('actor'): |
| 75 | + l_a = tf.layers.dense(self.s, 200, tf.nn.relu6, kernel_initializer=w_init, name='la') |
| 76 | + a_prob = tf.layers.dense(l_a, N_A, tf.nn.softmax, kernel_initializer=w_init, name='ap') # 得到每个动作的选择概率 |
| 77 | + with tf.variable_scope('critic'): |
| 78 | + l_c = tf.layers.dense(self.s, 100, tf.nn.relu6, kernel_initializer=w_init, name='lc') |
| 79 | + v = tf.layers.dense(l_c, 1, kernel_initializer=w_init, name='v') # 得到每个状态的价值函数 |
| 80 | + a_params = tf.get_collection(tf.GraphKeys.TRAINABLE_VARIABLES, scope=scope + '/actor') |
| 81 | + c_params = tf.get_collection(tf.GraphKeys.TRAINABLE_VARIABLES, scope=scope + '/critic') |
| 82 | + return a_prob, v, a_params, c_params |
| 83 | + |
| 84 | + def update_global(self, feed_dict): # run by a local |
| 85 | + SESS.run([self.update_a_op, self.update_c_op], feed_dict) # local grads applies to global net |
| 86 | + |
| 87 | + def pull_global(self): # run by a local |
| 88 | + SESS.run([self.pull_a_params_op, self.pull_c_params_op]) |
| 89 | + |
| 90 | + def choose_action(self, s): # run by a local |
| 91 | + prob_weights = SESS.run(self.a_prob, feed_dict={self.s: s[np.newaxis, :]}) |
| 92 | + action = np.random.choice(range(prob_weights.shape[1]), |
| 93 | + p=prob_weights.ravel()) # select action w.r.t the actions prob |
| 94 | + return action |
| 95 | + |
| 96 | + |
| 97 | +class Worker(object): |
| 98 | + def __init__(self, name, globalAC): |
| 99 | + self.env = gym.make(GAME).unwrapped |
| 100 | + self.name = name |
| 101 | + self.AC = ACNet(name, globalAC) |
| 102 | + |
| 103 | + def work(self): |
| 104 | + global GLOBAL_RUNNING_R, GLOBAL_EP |
| 105 | + total_step = 1 |
| 106 | + buffer_s, buffer_a, buffer_r = [], [], [] |
| 107 | + while not COORD.should_stop() and GLOBAL_EP < MAX_GLOBAL_EP: |
| 108 | + s = self.env.reset() |
| 109 | + ep_r = 0 |
| 110 | + while True: |
| 111 | + a = self.AC.choose_action(s) |
| 112 | + s_, r, done, info = self.env.step(a) |
| 113 | + if done: r = -5 |
| 114 | + ep_r += r |
| 115 | + buffer_s.append(s) |
| 116 | + buffer_a.append(a) |
| 117 | + buffer_r.append(r) |
| 118 | + |
| 119 | + if total_step % UPDATE_GLOBAL_ITER == 0 or done: # update global and assign to local net |
| 120 | + if done: |
| 121 | + v_s_ = 0 # terminal |
| 122 | + else: |
| 123 | + v_s_ = SESS.run(self.AC.v, {self.AC.s: s_[np.newaxis, :]})[0, 0] |
| 124 | + buffer_v_target = [] |
| 125 | + for r in buffer_r[::-1]: # reverse buffer r |
| 126 | + v_s_ = r + GAMMA * v_s_ # 使用v(s) = r + v(s+1)计算target_v |
| 127 | + buffer_v_target.append(v_s_) |
| 128 | + buffer_v_target.reverse() |
| 129 | + |
| 130 | + buffer_s, buffer_a, buffer_v_target = np.vstack(buffer_s), np.array(buffer_a), np.vstack(buffer_v_target) |
| 131 | + feed_dict = { |
| 132 | + self.AC.s: buffer_s, |
| 133 | + self.AC.a_his: buffer_a, |
| 134 | + self.AC.v_target: buffer_v_target, |
| 135 | + } |
| 136 | + self.AC.update_global(feed_dict) |
| 137 | + |
| 138 | + buffer_s, buffer_a, buffer_r = [], [], [] |
| 139 | + self.AC.pull_global() |
| 140 | + |
| 141 | + s = s_ |
| 142 | + total_step += 1 |
| 143 | + if done: |
| 144 | + if len(GLOBAL_RUNNING_R) == 0: # record running episode reward |
| 145 | + GLOBAL_RUNNING_R.append(ep_r) |
| 146 | + else: |
| 147 | + GLOBAL_RUNNING_R.append(0.99 * GLOBAL_RUNNING_R[-1] + 0.01 * ep_r) |
| 148 | + print( |
| 149 | + self.name, |
| 150 | + "Ep:", GLOBAL_EP, |
| 151 | + "| Ep_r: %i" % GLOBAL_RUNNING_R[-1], |
| 152 | + ) |
| 153 | + GLOBAL_EP += 1 |
| 154 | + break |
| 155 | + |
| 156 | +if __name__ == "__main__": |
| 157 | + SESS = tf.Session() |
| 158 | + |
| 159 | + with tf.device("/cpu:0"): |
| 160 | + OPT_A = tf.train.RMSPropOptimizer(LR_A, name='RMSPropA') |
| 161 | + OPT_C = tf.train.RMSPropOptimizer(LR_C, name='RMSPropC') |
| 162 | + GLOBAL_AC = ACNet(GLOBAL_NET_SCOPE) # we only need its params |
| 163 | + workers = [] |
| 164 | + # Create worker |
| 165 | + for i in range(N_WORKERS): |
| 166 | + i_name = 'W_%i' % i # worker name |
| 167 | + workers.append(Worker(i_name, GLOBAL_AC)) |
| 168 | + |
| 169 | + # Coordinator类用来管理在Session中的多个线程, |
| 170 | + # 使用 tf.train.Coordinator()来创建一个线程管理器(协调器)对象。 |
| 171 | + COORD = tf.train.Coordinator() |
| 172 | + SESS.run(tf.global_variables_initializer()) |
| 173 | + |
| 174 | + if OUTPUT_GRAPH: |
| 175 | + if os.path.exists(LOG_DIR): |
| 176 | + shutil.rmtree(LOG_DIR) |
| 177 | + tf.summary.FileWriter(LOG_DIR, SESS.graph) |
| 178 | + |
| 179 | + worker_threads = [] |
| 180 | + for worker in workers: |
| 181 | + job = lambda: worker.work() |
| 182 | + t = threading.Thread(target=job) # 创建一个线程,并分配其工作 |
| 183 | + t.start() # 开启线程 |
| 184 | + worker_threads.append(t) |
| 185 | + COORD.join(worker_threads) #把开启的线程加入主线程,等待threads结束 |
| 186 | + |
| 187 | + plt.plot(np.arange(len(GLOBAL_RUNNING_R)), GLOBAL_RUNNING_R) |
| 188 | + plt.xlabel('step') |
| 189 | + plt.ylabel('Total moving reward') |
| 190 | + plt.show() |
0 commit comments