Merge pull request #69 from ruihuili/master

MorvanZhou · web-flow · commit 601990e175de · 2018-07-31T11:06:08.000+08:00
DPPO with discrete action space
diff --git a/contents/12_Proximal_Policy_Optimization/discrete_DPP0.py b/contents/12_Proximal_Policy_Optimization/discrete_DPP0.py
@@ -0,0 +1,231 @@
+"""
+A simple version of OpenAI's Proximal Policy Optimization (PPO). [https://arxiv.org/abs/1707.06347]
+
+Distributing workers in parallel to collect data, then stop worker's roll-out and train PPO on collected data.
+Restart workers once PPO is updated.
+
+The global PPO updating rule is adopted from DeepMind's paper (DPPO):
+Emergence of Locomotion Behaviours in Rich Environments (Google Deepmind): [https://arxiv.org/abs/1707.02286]
+
+View more on my tutorial website: https://morvanzhou.github.io/tutorials
+
+Dependencies:
+tensorflow r1.3
+gym 0.9.2
+"""
+
+import tensorflow as tf
+import numpy as np
+import matplotlib.pyplot as plt
+import gym, threading, queue
+import time
+
+EP_MAX = 1000
+EP_LEN = 500
+N_WORKER = 4                # parallel workers
+GAMMA = 0.9                 # reward discount factor
+A_LR = 0.0001               # learning rate for actor
+C_LR = 0.0001               # learning rate for critic
+MIN_BATCH_SIZE = 64         # minimum batch size for updating PPO
+UPDATE_STEP = 10            # loop update operation n-steps
+EPSILON = 0.2               # for clipping surrogate objective
+GAME = 'CartPole-v0'
+
+env = gym.make(GAME)
+S_DIM = env.observation_space.shape[0]
+A_DIM = env.action_space.n
+
+
+class PPONet(object):
+    def __init__(self):
+        self.sess = tf.Session()
+        self.tfs = tf.placeholder(tf.float32, [None, S_DIM], 'state')
+
+        # critic
+        w_init = tf.random_normal_initializer(0., .1)
+        lc = tf.layers.dense(self.tfs, 200, tf.nn.relu, kernel_initializer=w_init, name='lc')
+        self.v = tf.layers.dense(lc, 1)
+        self.tfdc_r = tf.placeholder(tf.float32, [None, 1], 'discounted_r')
+        self.advantage = self.tfdc_r - self.v
+        self.closs = tf.reduce_mean(tf.square(self.advantage))
+        self.ctrain_op = tf.train.AdamOptimizer(C_LR).minimize(self.closs)
+
+        # actor
+        self.pi, self.pi_params = self._build_anet('pi', trainable=True)
+        oldpi, oldpi_params = self._build_anet('oldpi', trainable=False)
+        
+        self.update_oldpi_op = [oldp.assign(p) for p, oldp in zip(self.pi_params, oldpi_params)]
+
+        self.tfa = tf.placeholder(tf.int32, [None,], 'action')
+        
+        self.tfadv = tf.placeholder(tf.float32, [None, 1], 'advantage')
+
+        #debug
+        self.val1 = tf.reduce_sum(self.pi * tf.one_hot(self.tfa, A_DIM, dtype=tf.float32), axis=1, keep_dims=True)
+        self.val2 = tf.reduce_sum(oldpi * tf.one_hot(self.tfa, A_DIM, dtype=tf.float32), axis=1, keep_dims=True)
+        #debug
+        
+        ratio = self.val1/self.val2
+        
+        surr = ratio * self.tfadv                       # surrogate loss
+
+        self.aloss = -tf.reduce_mean(tf.minimum(        # clipped surrogate objective
+            surr,
+            tf.clip_by_value(ratio, 1. - EPSILON, 1. + EPSILON) * self.tfadv))
+
+        self.atrain_op = tf.train.AdamOptimizer(A_LR).minimize(self.aloss)
+        self.sess.run(tf.global_variables_initializer())
+
+    def update(self):
+        global GLOBAL_UPDATE_COUNTER
+        while not COORD.should_stop():
+            if GLOBAL_EP < EP_MAX:
+                UPDATE_EVENT.wait()                     # wait until get batch of data
+                self.sess.run(self.update_oldpi_op)     # copy pi to old pi
+                s, a, r = [],[],[]
+                for iter in range(QUEUE.qsize()):
+                    data = QUEUE.get()
+                    if iter == 0:
+                        s = data['bs']
+                        a = data['ba']
+                        r = data['br']
+                    else:
+                        s = np.append(s, data['bs'], axis=0)
+                        a = np.append(a, data['ba'], axis=0)
+                        r = np.append(r, data['br'], axis=0)
+
+                adv = self.sess.run(self.advantage, {self.tfs: s, self.tfdc_r: r})
+
+                # update actor and critic in a update loop
+                [self.sess.run(self.atrain_op, {self.tfs: s, self.tfa: a, self.tfadv: adv}) for _ in range(UPDATE_STEP)]
+                [self.sess.run(self.ctrain_op, {self.tfs: s, self.tfdc_r: r}) for _ in range(UPDATE_STEP)]
+                UPDATE_EVENT.clear()        # updating finished
+                GLOBAL_UPDATE_COUNTER = 0   # reset counter
+                ROLLING_EVENT.set()         # set roll-out available
+
+    def _build_anet(self, name, trainable):
+        w_init = tf.random_normal_initializer(0., .1)
+
+        with tf.variable_scope(name):
+            l_a = tf.layers.dense(self.tfs, 200, tf.nn.relu, trainable=trainable)
+            a_prob = tf.layers.dense(l_a, A_DIM, tf.nn.softmax, trainable=trainable, name='ap')
+        params = tf.get_collection(tf.GraphKeys.GLOBAL_VARIABLES, scope=name)
+        return a_prob, params
+
+    def choose_action(self, s):  # run by a local
+        prob_weights = self.sess.run(self.pi, feed_dict={self.tfs: s[np.newaxis, :]})
+        action = np.random.choice(range(prob_weights.shape[1]),
+                                      p=prob_weights.ravel())  # select action w.r.t the actions prob
+        return action
+    
+    def get_v(self, s):
+        if s.ndim < 2: s = s[np.newaxis, :]
+        return self.sess.run(self.v, {self.tfs: s})[0, 0]
+
+
+class Worker(object):
+    def __init__(self, wid):
+        self.wid = wid
+        self.env = gym.make(GAME).unwrapped
+        self.ppo = GLOBAL_PPO
+
+    def work(self):
+        global GLOBAL_EP, GLOBAL_RUNNING_R, GLOBAL_UPDATE_COUNTER
+        while not COORD.should_stop():
+            s = self.env.reset()#new episode
+            ep_r = 0
+            buffer_s, buffer_a, buffer_r = [], [], []
+            for t in range(EP_LEN):
+                if not ROLLING_EVENT.is_set():                  # while global PPO is updating
+                    ROLLING_EVENT.wait()                        # wait until PPO is updated
+                    buffer_s, buffer_a, buffer_r = [], [], []   # clear history buffer, use new policy to collect data
+                
+                a = self.ppo.choose_action(s)
+                s_, r, done, _ = self.env.step(a)
+                if done: r = -5
+                buffer_s.append(s)
+                buffer_a.append(a)
+                buffer_r.append((r + 8) / 8)                    # normalize reward, find to be useful
+                s = s_
+                ep_r += r
+
+                GLOBAL_UPDATE_COUNTER += 1               # count to minimum batch size, no need to wait other workers
+                if t == EP_LEN - 1 or GLOBAL_UPDATE_COUNTER >= MIN_BATCH_SIZE or done:
+                    
+                    if done:
+                        v_s_ = 0 #episode ends
+                    else:
+                        v_s_ = self.ppo.get_v(s_)
+                    
+                    discounted_r = []                           # compute discounted reward
+                    for r in buffer_r[::-1]:
+                        v_s_ = r + GAMMA * v_s_
+                        discounted_r.append(v_s_)
+                    discounted_r.reverse()
+        
+                    bs, ba, br = np.vstack(buffer_s), np.array(buffer_a), np.array(discounted_r)[:, np.newaxis]
+                    
+                    buffer_s, buffer_a, buffer_r = [], [], []
+                    
+                    q_in = dict([('bs', bs), ('ba', ba), ('br', br)])
+#                    q_in = dict([('bs', list(bs)), ('ba', list(ba)), ('br', list(br))])
+
+                    QUEUE.put(q_in)
+
+                    if GLOBAL_UPDATE_COUNTER >= MIN_BATCH_SIZE:
+                        ROLLING_EVENT.clear()       # stop collecting data
+                        UPDATE_EVENT.set()          # globalPPO update
+                    
+                    if GLOBAL_EP >= EP_MAX:         # stop training
+                        COORD.request_stop()
+                        break
+        
+                    if done:break
+
+            # record reward changes, plot later
+            if len(GLOBAL_RUNNING_R) == 0: GLOBAL_RUNNING_R.append(ep_r)
+            else: GLOBAL_RUNNING_R.append(GLOBAL_RUNNING_R[-1]*0.9+ep_r*0.1)
+            GLOBAL_EP += 1
+            print("EP", GLOBAL_EP,'|W%i' % self.wid, '|step %i' %t, '|Ep_r: %.2f' % ep_r,)
+            np.save("Global_return",GLOBAL_RUNNING_R)
+            np.savez("PI_PARA",self.ppo.sess.run(GLOBAL_PPO.pi_params))
+
+
+if __name__ == '__main__':
+    GLOBAL_PPO = PPONet()
+    UPDATE_EVENT, ROLLING_EVENT = threading.Event(), threading.Event()
+    UPDATE_EVENT.clear()            # not update now
+    ROLLING_EVENT.set()             # start to roll out
+    workers = [Worker(wid=i) for i in range(N_WORKER)]
+    
+    start = time.time()
+    
+    GLOBAL_UPDATE_COUNTER, GLOBAL_EP = 0, 0
+    GLOBAL_RUNNING_R = []
+    COORD = tf.train.Coordinator()
+    QUEUE = queue.Queue()           # workers putting data in this queue
+    threads = []
+    for worker in workers:          # worker threads
+        t = threading.Thread(target=worker.work, args=())
+        t.start()                   # training
+        threads.append(t)
+    # add a PPO updating thread
+    threads.append(threading.Thread(target=GLOBAL_PPO.update,))
+    threads[-1].start()
+    COORD.join(threads)
+
+    end = time.time()
+    print "Total time ", (end - start)
+
+    # plot reward change and test
+    plt.plot(np.arange(len(GLOBAL_RUNNING_R)), GLOBAL_RUNNING_R)
+    plt.xlabel('Episode'); plt.ylabel('Moving reward'); plt.ion(); plt.show()
+    env = gym.make('CartPole-v0')
+    while True:
+        s = env.reset()
+        for t in range(1000):
+            env.render()
+            s, r, done, info = env.step(GLOBAL_PPO.choose_action(s))
+            if done:
+                break
+
