mirror of
https://github.com/wassname/ray.git
synced 2026-08-05 13:21:03 +08:00
update rl pong app (#302)
This commit is contained in:
Robert Nishihara
committed by
Philipp Moritz
Philipp Moritz
parent
bcd0e3781f
commit
966c5ba0da
@@ -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.
|
||||
|
||||
|
||||
@@ -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)
|
||||
```
|
||||
+129
-35
@@ -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"))
|
||||
|
||||
@@ -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)
|
||||
@@ -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")
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user