From 966c5ba0daf8163a07ee12f128a6bc5ba334e1e4 Mon Sep 17 00:00:00 2001 From: Robert Nishihara Date: Fri, 29 Jul 2016 14:51:35 -0700 Subject: [PATCH] update rl pong app (#302) --- examples/lbfgs/README.md | 2 +- examples/rl_pong/README.md | 114 +++++++++++++++++++++++ examples/rl_pong/driver.py | 164 ++++++++++++++++++++++++++------- examples/rl_pong/functions.py | 85 ----------------- examples/rl_pong/visualizer.py | 17 ---- examples/rl_pong/worker.py | 16 ---- 6 files changed, 244 insertions(+), 154 deletions(-) create mode 100644 examples/rl_pong/README.md delete mode 100644 examples/rl_pong/functions.py delete mode 100644 examples/rl_pong/visualizer.py delete mode 100644 examples/rl_pong/worker.py diff --git a/examples/lbfgs/README.md b/examples/lbfgs/README.md index 2d4400e85..b854b97ab 100644 --- a/examples/lbfgs/README.md +++ b/examples/lbfgs/README.md @@ -4,7 +4,7 @@ This document provides a walkthrough of the L-BFGS example. To run the application, first install these dependencies. - SciPy -- [TensorFlow](https://www.tensorflow.org/). +- [TensorFlow](https://www.tensorflow.org/) Then from the directory `ray/examples/lbfgs/` run the following. diff --git a/examples/rl_pong/README.md b/examples/rl_pong/README.md new file mode 100644 index 000000000..cc02899e4 --- /dev/null +++ b/examples/rl_pong/README.md @@ -0,0 +1,114 @@ +# Learning to Play Pong + +In this example, we'll be training a neural network to play Pong using the +OpenAI Gym. This application is adapted, with minimal modifications, from Andrej +Karpathy's +[code](https://gist.github.com/karpathy/a4166c7fe253700972fcbc77e4ea32c5) (see +the accompanying [blog post](http://karpathy.github.io/2016/05/31/rl/)). To run +the application, first install this dependency. + +- [Gym](https://gym.openai.com/) + +Then from the directory `ray/examples/rl_pong/` run the following. + +``` +source ../../setup-env.sh +python driver.py +``` + +## The distributed version + +At the core of [Andrej's +code](https://gist.github.com/karpathy/a4166c7fe253700972fcbc77e4ea32c5), a +neural network is used to define a "policy" for playing Pong (that is, a +function that chooses an action given a state). In the loop, the network +repeatedly plays games of Pong and records a gradient from each game. Every ten +games, the gradients are combined together and used to update the network. + +This example is easy to parallelize because the network can play ten games in +parallel and no information needs to be shared between the games. We define a +remote function `compute_gradient`, which plays a game of pong and returns an +estimate of the gradient. Below is a simplified pseudocode version of this +function. + +```python +@ray.remote([dict], [dict, float]) +def compute_gradient(model): + # Retrieve the game environment. + env = ray.reusables.env + # Reset the game. + observation = env.reset() + while not done: + # Choose an action using policy_forward. + # Take the action and observe the new state of the world. + # Compute a gradient using policy_backward. Return the gradient and reward. + return gradient, reward_sum +``` + +Calling this remote function inside of a for loop, we launch multiple tasks to +perform rollouts and compute gradients. If we have at least ten worker +processes, then these tasks will all be executed in parallel. + +```python +model_ref = ray.put(model) +grads, reward_sums = [], [] +# Launch tasks to compute gradients from multiple rollouts in parallel. +for i in range(10): + grad_ref, reward_sum_ref = compute_gradient(model_ref) + grads.append(grad_ref) + reward_sums.append(reward_sum_ref) +``` + +### Reusing the Gym environment + +Workers are long-running Python processes, and though we'd like to think of +workers as being stateless, sometimes it's important to have a variable that +gets shared between different tasks on the same worker (perhaps because it is +expensive to initialize the variable). + +In this example, we'd like each worker to have access to a Pong environment. The +Pong environment has state that gets mutated by the task, and this state is +shared between tasks that run on the same worker, so there is some danger that +the output of the overall program will depend on which tasks are scheduled on +which workers. This can be avoided if the state of the Pong environment is reset +between tasks. + +To accomplish this, the user must mark the Pong environment as a reusable +variable. This is done by providing a method for initializing the gym, and +storing it in `ray.reusables`. + +```python +# Function for initializing the gym environment. +def env_initializer(): + return gym.make("Pong-v0") + +# Create a reusable variable for the gym environment. +ray.reusables.env = ray.Reusable(env_initializer) +``` + +A remote task can then call `ray.reusables.env` to retrieve the variable. + +By default, whenever a task uses the `ray.reusables.env` variable, the worker +that the task was scheduled on will rerun the initialization code +`env_initializer` after the task has finished so that state will not leak +between the tasks. + +However, sometimes the initialization code is expensive, and there may be a +faster way to reinitialize the variable (or maybe no reinitialization is needed +at all). In these cases, the user can provide a custom **reinitializer**, which +gets run after any task that uses the variable. + +```python +# Function for initializing the gym environment. +def env_initializer(): + return gym.make("Pong-v0") + +# Function for reinitializing the gym environment in order to guarantee that +# the state of the game is reset after each remote task. +def env_reinitializer(env): + env.reset() + return env + +# Create a reusable variable for the gym environment. +ray.reusables.env = ray.Reusable(env_initializer, env_reinitializer) +``` diff --git a/examples/rl_pong/driver.py b/examples/rl_pong/driver.py index b0bd31ae7..e03977fd7 100644 --- a/examples/rl_pong/driver.py +++ b/examples/rl_pong/driver.py @@ -3,49 +3,143 @@ import numpy as np import cPickle as pickle -import gym import ray -import os -import functions - -worker_dir = os.path.dirname(os.path.abspath(__file__)) -worker_path = os.path.join(worker_dir, "worker.py") -ray.services.start_ray_local(num_workers=10, worker_path=worker_path) +import gym # hyperparameters H = 200 # number of hidden layer neurons batch_size = 10 # every how many episodes to do a param update? learning_rate = 1e-4 +gamma = 0.99 # discount factor for reward decay_rate = 0.99 # decay factor for RMSProp leaky sum of grad^2 resume = False # resume from previous checkpoint? -running_reward = None -batch_num = 1 -D = functions.D # input dimensionality: 80x80 grid -if resume: - model = pickle.load(open("save.p", "rb")) -else: - model = {} - model["W1"] = np.random.randn(H, D) / np.sqrt(D) # "Xavier" initialization - model["W2"] = np.random.randn(H) / np.sqrt(H) -grad_buffer = {k: np.zeros_like(v) for k, v in model.iteritems()} # update buffers that add up gradients over a batch -rmsprop_cache = {k: np.zeros_like(v) for k, v in model.iteritems()} # rmsprop memory +D = 80 * 80 # input dimensionality: 80x80 grid -while True: - modelref = ray.put(model) - grads = [] - for i in range(batch_size): - grads.append(functions.compgrad(modelref)) - for i in range(batch_size): - grad = ray.get(grads[i]) - for k in model: grad_buffer[k] += grad[0][k] # accumulate grad over batch - running_reward = grad[1] if running_reward is None else running_reward * 0.99 + grad[1] * 0.01 - print "Batch {}. episode reward total was {}. running mean: {}".format(batch_num, grad[1], running_reward) - for k, v in model.iteritems(): - g = grad_buffer[k] # gradient - rmsprop_cache[k] = decay_rate * rmsprop_cache[k] + (1 - decay_rate) * g ** 2 - model[k] += learning_rate * g / (np.sqrt(rmsprop_cache[k]) + 1e-5) - grad_buffer[k] = np.zeros_like(v) # reset batch gradient buffer - batch_num += 1 - if batch_num % 10 == 0: pickle.dump(model, open("save.p", "wb")) +# Function for initializing the gym environment. +def env_initializer(): + return gym.make("Pong-v0") + +# Function for reinitializing the gym environment in order to guarantee that +# the state of the game is reset after each remote task. +def env_reinitializer(env): + env.reset() + return env + +# Create a reusable variable for the gym environment. +ray.reusables.env = ray.Reusable(env_initializer, env_reinitializer) + +def sigmoid(x): + return 1.0 / (1.0 + np.exp(-x)) # sigmoid "squashing" function to interval [0,1] + +def preprocess(I): + """preprocess 210x160x3 uint8 frame into 6400 (80x80) 1D float vector""" + I = I[35:195] # crop + I = I[::2,::2,0] # downsample by factor of 2 + I[I == 144] = 0 # erase background (background type 1) + I[I == 109] = 0 # erase background (background type 2) + I[I != 0] = 1 # everything else (paddles, ball) just set to 1 + return I.astype(np.float).ravel() + +def discount_rewards(r): + """take 1D float array of rewards and compute discounted reward""" + discounted_r = np.zeros_like(r) + running_add = 0 + for t in reversed(xrange(0, r.size)): + if r[t] != 0: running_add = 0 # reset the sum, since this was a game boundary (pong specific!) + running_add = running_add * gamma + r[t] + discounted_r[t] = running_add + return discounted_r + +def policy_forward(x, model): + h = np.dot(model["W1"], x) + h[h < 0] = 0 # ReLU nonlinearity + logp = np.dot(model["W2"], h) + p = sigmoid(logp) + return p, h # return probability of taking action 2, and hidden state + +def policy_backward(eph, epx, epdlogp, model): + """backward pass. (eph is array of intermediate hidden states)""" + dW2 = np.dot(eph.T, epdlogp).ravel() + dh = np.outer(epdlogp, model["W2"]) + dh[eph <= 0] = 0 # backpro prelu + dW1 = np.dot(dh.T, epx) + return {"W1": dW1, "W2": dW2} + +@ray.remote([dict], [dict, float]) +def compute_gradient(model): + env = ray.reusables.env + observation = env.reset() + prev_x = None # used in computing the difference frame + xs, hs, dlogps, drs = [], [], [], [] + reward_sum = 0 + done = False + while not done: + cur_x = preprocess(observation) + x = cur_x - prev_x if prev_x is not None else np.zeros(D) + prev_x = cur_x + + aprob, h = policy_forward(x, model) + action = 2 if np.random.uniform() < aprob else 3 # roll the dice! + + xs.append(x) # observation + hs.append(h) # hidden state + y = 1 if action == 2 else 0 # a "fake label" + dlogps.append(y - aprob) # grad that encourages the action that was taken to be taken (see http://cs231n.github.io/neural-networks-2/#losses if confused) + + observation, reward, done, info = env.step(action) + reward_sum += reward + + drs.append(reward) # record reward (has to be done after we call step() to get reward for previous action) + + epx = np.vstack(xs) + eph = np.vstack(hs) + epdlogp = np.vstack(dlogps) + epr = np.vstack(drs) + xs, hs, dlogps, drs = [], [], [], [] # reset array memory + + # compute the discounted reward backwards through time + discounted_epr = discount_rewards(epr) + # standardize the rewards to be unit normal (helps control the gradient estimator variance) + discounted_epr -= np.mean(discounted_epr) + discounted_epr /= np.std(discounted_epr) + epdlogp *= discounted_epr # modulate the gradient with advantage (PG magic happens right here.) + return policy_backward(eph, epx, epdlogp, model), reward_sum + +if __name__ == "__main__": + ray.services.start_ray_local(num_workers=10) + + # Run the reinforcement learning + running_reward = None + batch_num = 1 + if resume: + model = pickle.load(open("save.p", "rb")) + else: + model = {} + model["W1"] = np.random.randn(H, D) / np.sqrt(D) # "Xavier" initialization + model["W2"] = np.random.randn(H) / np.sqrt(H) + grad_buffer = {k: np.zeros_like(v) for k, v in model.iteritems()} # update buffers that add up gradients over a batch + rmsprop_cache = {k: np.zeros_like(v) for k, v in model.iteritems()} # rmsprop memory + + while True: + model_ref = ray.put(model) + grads, reward_sums = [], [] + # Launch tasks to compute gradients from multiple rollouts in parallel. + for i in range(batch_size): + grad_ref, reward_sum_ref = compute_gradient(model_ref) + grads.append(grad_ref) + reward_sums.append(reward_sum_ref) + for i in range(batch_size): + grad = ray.get(grads[i]) + reward_sum = ray.get(reward_sums[i]) + for k in model: grad_buffer[k] += grad[k] # accumulate grad over batch + running_reward = reward_sum if running_reward is None else running_reward * 0.99 + reward_sum * 0.01 + print "Batch {}. episode reward total was {}. running mean: {}".format(batch_num, reward_sum, running_reward) + for k, v in model.iteritems(): + g = grad_buffer[k] # gradient + rmsprop_cache[k] = decay_rate * rmsprop_cache[k] + (1 - decay_rate) * g ** 2 + model[k] += learning_rate * g / (np.sqrt(rmsprop_cache[k]) + 1e-5) + grad_buffer[k] = np.zeros_like(v) # reset batch gradient buffer + batch_num += 1 + if batch_num % 10 == 0: pickle.dump(model, open("save.p", "wb")) diff --git a/examples/rl_pong/functions.py b/examples/rl_pong/functions.py deleted file mode 100644 index 85cf4a858..000000000 --- a/examples/rl_pong/functions.py +++ /dev/null @@ -1,85 +0,0 @@ -# This code is copied and adapted from Andrej Karpathy's code for learning to -# play Pong https://gist.github.com/karpathy/a4166c7fe253700972fcbc77e4ea32c5. - -import ray -import numpy as np -import gym - -env = gym.make("Pong-v0") -D = 80 * 80 -gamma = 0.99 # discount factor for reward -def sigmoid(x): - return 1.0 / (1.0 + np.exp(-x)) # sigmoid "squashing" function to interval [0,1] - -def preprocess(I): - """preprocess 210x160x3 uint8 frame into 6400 (80x80) 1D float vector""" - I = I[35:195] # crop - I = I[::2,::2,0] # downsample by factor of 2 - I[I == 144] = 0 # erase background (background type 1) - I[I == 109] = 0 # erase background (background type 2) - I[I != 0] = 1 # everything else (paddles, ball) just set to 1 - return I.astype(np.float).ravel() - -def discount_rewards(r): - """take 1D float array of rewards and compute discounted reward""" - discounted_r = np.zeros_like(r) - running_add = 0 - for t in reversed(xrange(0, r.size)): - if r[t] != 0: running_add = 0 # reset the sum, since this was a game boundary (pong specific!) - running_add = running_add * gamma + r[t] - discounted_r[t] = running_add - return discounted_r - -def policy_forward(x, model): - h = np.dot(model["W1"], x) - h[h < 0] = 0 # ReLU nonlinearity - logp = np.dot(model["W2"], h) - p = sigmoid(logp) - return p, h # return probability of taking action 2, and hidden state - -def policy_backward(eph, epx, epdlogp, model): - """backward pass. (eph is array of intermediate hidden states)""" - dW2 = np.dot(eph.T, epdlogp).ravel() - dh = np.outer(epdlogp, model["W2"]) - dh[eph <= 0] = 0 # backpro prelu - dW1 = np.dot(dh.T, epx) - return {"W1": dW1, "W2": dW2} - -@ray.remote([dict], [tuple]) -def compgrad(model): - observation = env.reset() - prev_x = None # used in computing the difference frame - xs, hs, dlogps, drs = [], [], [], [] - reward_sum = 0 - done = False - while not done: - cur_x = preprocess(observation) - x = cur_x - prev_x if prev_x is not None else np.zeros(D) - prev_x = cur_x - - aprob, h = policy_forward(x,model) - action = 2 if np.random.uniform() < aprob else 3 # roll the dice! - - xs.append(x) # observation - hs.append(h) # hidden state - y = 1 if action == 2 else 0 # a "fake label" - dlogps.append(y - aprob) # grad that encourages the action that was taken to be taken (see http://cs231n.github.io/neural-networks-2/#losses if confused) - - observation, reward, done, info = env.step(action) - reward_sum += reward - - drs.append(reward) # record reward (has to be done after we call step() to get reward for previous action) - - epx = np.vstack(xs) - eph = np.vstack(hs) - epdlogp = np.vstack(dlogps) - epr = np.vstack(drs) - xs, hs, dlogps, drs = [], [], [], [] # reset array memory - - # compute the discounted reward backwards through time - discounted_epr = discount_rewards(epr) - # standardize the rewards to be unit normal (helps control the gradient estimator variance) - discounted_epr -= np.mean(discounted_epr) - discounted_epr /= np.std(discounted_epr) - epdlogp *= discounted_epr # modulate the gradient with advantage (PG magic happens right here.) - return (policy_backward(eph, epx, epdlogp, model), reward_sum) diff --git a/examples/rl_pong/visualizer.py b/examples/rl_pong/visualizer.py deleted file mode 100644 index 83eeef1b9..000000000 --- a/examples/rl_pong/visualizer.py +++ /dev/null @@ -1,17 +0,0 @@ -import matplotlib -from matplotlib import pyplot as plt -import pickle -import numpy as np - -matplotlib.use("Agg") - -logs = pickle.load(open("logs_rl_original.p", "rb")) -times_og = range(1, (len(logs) + 1)) -reward_og = map(lambda x:x[2], logs) -plt.plot(times_og, reward_og) -plt.savefig("original_batchnum_graph") -logs = pickle.load(open("logs_rl_ray.p", "rb")) -times_ray = range(1, (len(logs) + 1)) -reward_ray = map(lambda x: x[2], logs) -plt.plot(times_ray, reward_ray) -plt.savefig("rl_pong_graph") diff --git a/examples/rl_pong/worker.py b/examples/rl_pong/worker.py deleted file mode 100644 index d742a1060..000000000 --- a/examples/rl_pong/worker.py +++ /dev/null @@ -1,16 +0,0 @@ -import argparse -import ray -import gym - -import functions - -parser = argparse.ArgumentParser(description="Parse addresses for the worker to connect to.") -parser.add_argument("--scheduler-address", default="127.0.0.1:10001", type=str, help="the scheduler's address") -parser.add_argument("--objstore-address", default="127.0.0.1:20001", type=str, help="the objstore's address") -parser.add_argument("--worker-address", default="127.0.0.1:40001", type=str, help="the worker's address") - -if __name__ == "__main__": - args = parser.parse_args() - ray.connect(args.scheduler_address, args.objstore_address, args.worker_address) - ray.register_module(functions) - ray.worker.main_loop()