From 3ebfd850e116c88fd5e045ab1f4a068ae1324b01 Mon Sep 17 00:00:00 2001 From: Robert Nishihara Date: Tue, 16 May 2017 14:12:18 -0700 Subject: [PATCH] Make example applications pep8 compliant. (#553) * Test examples for pep8 compliance. * Make rl_pong example pep8 compliant. * Make policy gradient example pep8 compliant. * Make lbfgs example pep8 compliant. * Make hyperopt example pep8 compliant. * Make a3c example pep8 compliant. * Make evolution strategies example pep8 compliant. * Make resnet example pep8 compliant. * Fix. --- .travis.yml | 2 +- examples/a3c/LSTM.py | 154 ++++--- examples/a3c/driver.py | 116 +++--- examples/a3c/envs.py | 148 +++---- examples/a3c/misc.py | 36 +- examples/a3c/policy.py | 202 ++++----- examples/a3c/runner.py | 247 +++++------ .../evolution_strategies.py | 36 +- examples/evolution_strategies/policies.py | 386 +++++++++--------- .../evolution_strategies/tabular_logger.py | 295 +++++++------ examples/evolution_strategies/tf_util.py | 355 ++++++++-------- examples/hyperopt/hyperopt_adaptive.py | 23 +- examples/hyperopt/hyperopt_simple.py | 14 +- examples/hyperopt/objective.py | 31 +- examples/lbfgs/driver.py | 28 +- examples/policy_gradient/examples/example.py | 38 +- examples/policy_gradient/reinforce/agent.py | 13 +- .../reinforce/distributions.py | 59 +-- examples/policy_gradient/reinforce/env.py | 20 +- examples/policy_gradient/reinforce/filter.py | 23 +- .../policy_gradient/reinforce/models/fcnet.py | 31 +- .../reinforce/models/visionnet.py | 4 +- examples/policy_gradient/reinforce/policy.py | 37 +- examples/policy_gradient/reinforce/rollout.py | 38 +- examples/policy_gradient/reinforce/tfutils.py | 30 -- examples/policy_gradient/reinforce/utils.py | 6 +- examples/policy_gradient/tests/test.py | 3 + examples/resnet/cifar_input.py | 35 +- examples/resnet/resnet_main.py | 136 +++--- examples/resnet/resnet_model.py | 16 +- examples/rl_pong/driver.py | 22 +- 31 files changed, 1392 insertions(+), 1192 deletions(-) delete mode 100644 examples/policy_gradient/reinforce/tfutils.py diff --git a/.travis.yml b/.travis.yml index 89e5ac493..2b63ab1e4 100644 --- a/.travis.yml +++ b/.travis.yml @@ -38,7 +38,7 @@ matrix: - cd .. # Run Python linting. - flake8 --ignore=E111,E114 - --exclude=python/ray/core/src/common/flatbuffers_ep-prefix/,python/ray/core/generated/,src/numbuf/thirdparty/,src/common/format/,examples/,doc/source/conf.py + --exclude=python/ray/core/src/common/flatbuffers_ep-prefix/,python/ray/core/generated/,src/numbuf/thirdparty/,src/common/format/,doc/source/conf.py - os: linux dist: trusty env: VALGRIND=1 PYTHON=2.7 diff --git a/examples/a3c/LSTM.py b/examples/a3c/LSTM.py index bf317aa56..2b8d79f3d 100644 --- a/examples/a3c/LSTM.py +++ b/examples/a3c/LSTM.py @@ -6,87 +6,105 @@ import numpy as np import tensorflow as tf import tensorflow.contrib.rnn as rnn import distutils.version -import ray -from policy import * -use_tf100_api = distutils.version.LooseVersion(tf.VERSION) >= distutils.version.LooseVersion('1.0.0') +from policy import (categorical_sample, conv2d, linear, flatten, + normalized_columns_initializer, Policy) + +use_tf100_api = (distutils.version.LooseVersion(tf.VERSION) >= + distutils.version.LooseVersion("1.0.0")) class LSTMPolicy(Policy): - def setup_graph(self, ob_space, ac_space): - """Setup model used for Policy (in this A3C, both the Critic and the Actor share the model)""" - self.x = x = tf.placeholder(tf.float32, [None] + list(ob_space)) + def setup_graph(self, ob_space, ac_space): + """Setup model used for Policy. - for i in range(4): - x = tf.nn.elu(conv2d(x, 32, "l{}".format(i + 1), [3, 3], [2, 2])) - # introduce a "fake" batch dimension of 1 after flatten so that we can do LSTM over time dim - x = tf.expand_dims(flatten(x), [0]) + In this A3C implementation, both the Critic and the Actor share the model. + """ + self.x = x = tf.placeholder(tf.float32, [None] + list(ob_space)) - size = 256 - if use_tf100_api: - lstm = rnn.BasicLSTMCell(size, state_is_tuple=True) - else: - lstm = rnn.rnn_cell.BasicLSTMCell(size, state_is_tuple=True) - self.state_size = lstm.state_size - step_size = tf.shape(self.x)[:1] + for i in range(4): + x = tf.nn.elu(conv2d(x, 32, "l{}".format(i + 1), [3, 3], [2, 2])) + # Introduce a "fake" batch dimension of 1 after flatten so that we can do + # LSTM over the time dim. + x = tf.expand_dims(flatten(x), [0]) - c_init = np.zeros((1, lstm.state_size.c), np.float32) - h_init = np.zeros((1, lstm.state_size.h), np.float32) - self.state_init = [c_init, h_init] - c_in = tf.placeholder(tf.float32, [1, lstm.state_size.c]) - h_in = tf.placeholder(tf.float32, [1, lstm.state_size.h]) - self.state_in = [c_in, h_in] + size = 256 + if use_tf100_api: + lstm = rnn.BasicLSTMCell(size, state_is_tuple=True) + else: + lstm = rnn.rnn_cell.BasicLSTMCell(size, state_is_tuple=True) + self.state_size = lstm.state_size + step_size = tf.shape(self.x)[:1] - if use_tf100_api: - state_in = rnn.LSTMStateTuple(c_in, h_in) - else: - state_in = rnn.rnn_cell.LSTMStateTuple(c_in, h_in) - lstm_outputs, lstm_state = tf.nn.dynamic_rnn( - lstm, x, initial_state=state_in, sequence_length=step_size, - time_major=False) - lstm_c, lstm_h = lstm_state - x = tf.reshape(lstm_outputs, [-1, size]) - self.logits = linear(x, ac_space, "action", normalized_columns_initializer(0.01)) - self.vf = tf.reshape(linear(x, 1, "value", normalized_columns_initializer(1.0)), [-1]) - self.state_out = [lstm_c[:1, :], lstm_h[:1, :]] - self.sample = categorical_sample(self.logits, ac_space)[0, :] - self.var_list = tf.get_collection(tf.GraphKeys.TRAINABLE_VARIABLES, tf.get_variable_scope().name) - self.global_step = tf.get_variable("global_step", [], tf.int32, initializer=tf.constant_initializer(0, dtype=tf.int32), - trainable=False) + c_init = np.zeros((1, lstm.state_size.c), np.float32) + h_init = np.zeros((1, lstm.state_size.h), np.float32) + self.state_init = [c_init, h_init] + c_in = tf.placeholder(tf.float32, [1, lstm.state_size.c]) + h_in = tf.placeholder(tf.float32, [1, lstm.state_size.h]) + self.state_in = [c_in, h_in] - def get_gradients(self, batch): - """Computing the gradient is actually model-dependent. - The LSTM needs its hidden states in order to compute the gradient accurately.""" - feed_dict = { - self.x: batch.si, - self.ac: batch.a, - self.adv: batch.adv, - self.r: batch.r, - self.state_in[0]: batch.features[0], - self.state_in[1]: batch.features[1], - } - self.local_steps += 1 - return self.sess.run(self.grads, feed_dict=feed_dict) + if use_tf100_api: + state_in = rnn.LSTMStateTuple(c_in, h_in) + else: + state_in = rnn.rnn_cell.LSTMStateTuple(c_in, h_in) + lstm_outputs, lstm_state = tf.nn.dynamic_rnn( + lstm, x, initial_state=state_in, sequence_length=step_size, + time_major=False) + lstm_c, lstm_h = lstm_state + x = tf.reshape(lstm_outputs, [-1, size]) + self.logits = linear(x, ac_space, "action", + normalized_columns_initializer(0.01)) + self.vf = tf.reshape(linear(x, 1, "value", + normalized_columns_initializer(1.0)), [-1]) + self.state_out = [lstm_c[:1, :], lstm_h[:1, :]] + self.sample = categorical_sample(self.logits, ac_space)[0, :] + self.var_list = tf.get_collection(tf.GraphKeys.TRAINABLE_VARIABLES, + tf.get_variable_scope().name) + self.global_step = tf.get_variable( + "global_step", [], tf.int32, + initializer=tf.constant_initializer(0, dtype=tf.int32), + trainable=False) - def act(self, ob, c, h): - return self.sess.run([self.sample, self.vf] + self.state_out, - {self.x: [ob], self.state_in[0]: c, self.state_in[1]: h}) + def get_gradients(self, batch): + """Computing the gradient is actually model-dependent. - def value(self, ob, c, h): - return self.sess.run(self.vf, {self.x: [ob], self.state_in[0]: c, self.state_in[1]: h})[0] + The LSTM needs its hidden states in order to compute the gradient + accurately. + """ + feed_dict = { + self.x: batch.si, + self.ac: batch.a, + self.adv: batch.adv, + self.r: batch.r, + self.state_in[0]: batch.features[0], + self.state_in[1]: batch.features[1] + } + self.local_steps += 1 + return self.sess.run(self.grads, feed_dict=feed_dict) - def get_initial_features(self): - return self.state_init + def act(self, ob, c, h): + return self.sess.run([self.sample, self.vf] + self.state_out, + {self.x: [ob], + self.state_in[0]: c, + self.state_in[1]: h}) + + def value(self, ob, c, h): + return self.sess.run(self.vf, {self.x: [ob], + self.state_in[0]: c, + self.state_in[1]: h})[0] + + def get_initial_features(self): + return self.state_init class RawLSTMPolicy(LSTMPolicy): - def get_weights(self): - if not hasattr(self, "_weights"): - self._weights = self.variables.get_weights() - return self._weights + def get_weights(self): + if not hasattr(self, "_weights"): + self._weights = self.variables.get_weights() + return self._weights - def set_weights(self, weights): - self._weights = weights + def set_weights(self, weights): + self._weights = weights - def model_update(self, grads): - for var, grad in zip(self.var_list, grads): - self._weights[var.name[:-2]] -= 1e-4 * grad + def model_update(self, grads): + for var, grad in zip(self.var_list, grads): + self._weights[var.name[:-2]] -= 1e-4 * grad diff --git a/examples/a3c/driver.py b/examples/a3c/driver.py index e4fc0624a..74af71ea8 100644 --- a/examples/a3c/driver.py +++ b/examples/a3c/driver.py @@ -3,77 +3,81 @@ from __future__ import division from __future__ import print_function import ray -import numpy as np from runner import RunnerThread, process_rollout from LSTM import LSTMPolicy import tensorflow as tf import six.moves.queue as queue -import gym import sys import os -from datetime import datetime, timedelta -from misc import timestamp, time_string from envs import create_env + @ray.remote class Runner(object): - """Actor object to start running simulation on workers. - Gradient computation is also executed from this object.""" - def __init__(self, env_name, actor_id, logdir="results/", start=True): - env = create_env(env_name) - self.id = actor_id - num_actions = env.action_space.n - self.policy = LSTMPolicy(env.observation_space.shape, num_actions, actor_id) - self.runner = RunnerThread(env, self.policy, 20) - self.env = env - self.logdir = logdir - if start: - self.start() + """Actor object to start running simulation on workers. - def pull_batch_from_queue(self): - """ self explanatory: take a rollout from the queue of the thread runner. """ - rollout = self.runner.queue.get(timeout=600.0) - while not rollout.terminal: - try: - rollout.extend(self.runner.queue.get_nowait()) - except queue.Empty: - break - return rollout + The gradient computation is also executed from this object. + """ + def __init__(self, env_name, actor_id, logdir="results/", start=True): + env = create_env(env_name) + self.id = actor_id + num_actions = env.action_space.n + self.policy = LSTMPolicy(env.observation_space.shape, num_actions, + actor_id) + self.runner = RunnerThread(env, self.policy, 20) + self.env = env + self.logdir = logdir + if start: + self.start() - def start(self): - summary_writer = tf.summary.FileWriter(os.path.join(self.logdir, "agent_%d" % self.id)) - self.summary_writer = summary_writer - self.runner.start_runner(self.policy.sess, summary_writer) + def pull_batch_from_queue(self): + """Take a rollout from the queue of the thread runner.""" + rollout = self.runner.queue.get(timeout=600.0) + while not rollout.terminal: + try: + rollout.extend(self.runner.queue.get_nowait()) + except queue.Empty: + break + return rollout - def compute_gradient(self, params): - self.policy.set_weights(params) - rollout = self.pull_batch_from_queue() - batch = process_rollout(rollout, gamma=0.99, lambda_=1.0) - gradient = self.policy.get_gradients(batch) - info = {"id": self.id, - "size": len(batch.a)} - return gradient, info + def start(self): + summary_writer = tf.summary.FileWriter( + os.path.join(self.logdir, "agent_%d" % self.id)) + self.summary_writer = summary_writer + self.runner.start_runner(self.policy.sess, summary_writer) + + def compute_gradient(self, params): + self.policy.set_weights(params) + rollout = self.pull_batch_from_queue() + batch = process_rollout(rollout, gamma=0.99, lambda_=1.0) + gradient = self.policy.get_gradients(batch) + info = {"id": self.id, + "size": len(batch.a)} + return gradient, info def train(num_workers, env_name="PongDeterministic-v3"): - env = create_env(env_name) - policy = LSTMPolicy(env.observation_space.shape, env.action_space.n, 0) - agents = [Runner.remote(env_name, i) for i in range(num_workers)] + env = create_env(env_name) + policy = LSTMPolicy(env.observation_space.shape, env.action_space.n, 0) + agents = [Runner.remote(env_name, i) for i in range(num_workers)] + parameters = policy.get_weights() + gradient_list = [agent.compute_gradient.remote(parameters) + for agent in agents] + steps = 0 + obs = 0 + while True: + done_id, gradient_list = ray.wait(gradient_list) + gradient, info = ray.get(done_id)[0] + policy.model_update(gradient) parameters = policy.get_weights() - gradient_list = [agent.compute_gradient.remote(parameters) for agent in agents] - steps = 0 - obs = 0 - while True: - done_id, gradient_list = ray.wait(gradient_list) - gradient, info = ray.get(done_id)[0] - policy.model_update(gradient) - parameters = policy.get_weights() - steps += 1 - obs += info["size"] - gradient_list.extend([agents[info["id"]].compute_gradient.remote(parameters)]) - return policy + steps += 1 + obs += info["size"] + gradient_list.extend( + [agents[info["id"]].compute_gradient.remote(parameters)]) + return policy -if __name__ == '__main__': - num_workers = int(sys.argv[1]) - ray.init(num_cpus=num_workers) - train(num_workers) + +if __name__ == "__main__": + num_workers = int(sys.argv[1]) + ray.init(num_cpus=num_workers) + train(num_workers) diff --git a/examples/a3c/envs.py b/examples/a3c/envs.py index d5d60a9f5..ba29c408e 100644 --- a/examples/a3c/envs.py +++ b/examples/a3c/envs.py @@ -5,7 +5,6 @@ from __future__ import print_function import cv2 import gym from gym.spaces.box import Box -from gym import spaces import logging import numpy as np import time @@ -13,95 +12,96 @@ import time logger = logging.getLogger(__name__) logger.setLevel(logging.INFO) + def create_env(env_id): - env = gym.make(env_id) - env = AtariProcessing(env) - env = Diagnostic(env) - return env + env = gym.make(env_id) + env = AtariProcessing(env) + env = Diagnostic(env) + return env + def _process_frame42(frame): - frame = frame[34:(34+160), :160] - # Resize by half, then down to 42x42 (essentially mipmapping). If - # we resize directly we lose pixels that, when mapped to 42x42, - # aren't close enough to the pixel boundary. - frame = cv2.resize(frame, (80, 80)) - frame = cv2.resize(frame, (42, 42)) - frame = frame.mean(2) - frame = frame.astype(np.float32) - frame *= (1.0 / 255.0) - frame = np.reshape(frame, [42, 42, 1]) - return frame + frame = frame[34:(34 + 160), :160] + # Resize by half, then down to 42x42 (essentially mipmapping). If we resize + # directly we lose pixels that, when mapped to 42x42, aren't close enough to + # the pixel boundary. + frame = cv2.resize(frame, (80, 80)) + frame = cv2.resize(frame, (42, 42)) + frame = frame.mean(2) + frame = frame.astype(np.float32) + frame *= (1.0 / 255.0) + frame = np.reshape(frame, [42, 42, 1]) + return frame + class AtariProcessing(gym.ObservationWrapper): - def __init__(self, env=None): - super(AtariProcessing, self).__init__(env) - self.observation_space = Box(0.0, 1.0, [42, 42, 1]) + def __init__(self, env=None): + super(AtariProcessing, self).__init__(env) + self.observation_space = Box(0.0, 1.0, [42, 42, 1]) + + def _observation(self, observation): + return _process_frame42(observation) - def _observation(self, observation): - return _process_frame42(observation) class Diagnostic(gym.Wrapper): - def __init__(self, env=None): - super(Diagnostic, self).__init__(env) - self.diagnostics = DiagnosticsLogger() + def __init__(self, env=None): + super(Diagnostic, self).__init__(env) + self.diagnostics = DiagnosticsLogger() - def _reset(self): - observation = self.env.reset() - return self.diagnostics._after_reset(observation) + def _reset(self): + observation = self.env.reset() + return self.diagnostics._after_reset(observation) - def _step(self, action): - results = self.env.step(action) - return self.diagnostics._after_step(*results) + def _step(self, action): + results = self.env.step(action) + return self.diagnostics._after_step(*results) -class DiagnosticsLogger(): - def __init__(self, log_interval=503): +class DiagnosticsLogger(object): + def __init__(self, log_interval=503): + self._episode_time = time.time() + self._last_time = time.time() + self._local_t = 0 + self._log_interval = log_interval + self._episode_reward = 0 + self._episode_length = 0 + self._all_rewards = [] + self._last_episode_id = -1 - self._episode_time = time.time() - self._last_time = time.time() - self._local_t = 0 - self._log_interval = log_interval - self._episode_reward = 0 - self._episode_length = 0 - self._all_rewards = [] - self._last_episode_id = -1 + def _after_reset(self, observation): + logger.info("Resetting environment") + self._episode_reward = 0 + self._episode_length = 0 + self._all_rewards = [] + return observation - def _after_reset(self, observation): - logger.info('Resetting environment') - self._episode_reward = 0 - self._episode_length = 0 - self._all_rewards = [] - return observation + def _after_step(self, observation, reward, done, info): + to_log = {} + if self._episode_length == 0: + self._episode_time = time.time() - def _after_step(self, observation, reward, done, info): - to_log = {} - if self._episode_length == 0: - self._episode_time = time.time() + self._local_t += 1 - self._local_t += 1 + if self._local_t % self._log_interval == 0: + cur_time = time.time() + self._last_time = cur_time - if self._local_t % self._log_interval == 0: - cur_time = time.time() - elapsed = cur_time - self._last_time - fps = self._log_interval / elapsed - self._last_time = cur_time + if reward is not None: + self._episode_reward += reward + if observation is not None: + self._episode_length += 1 + self._all_rewards.append(reward) - if reward is not None: - self._episode_reward += reward - if observation is not None: - self._episode_length += 1 - self._all_rewards.append(reward) - - if done: - logger.info('Episode terminating: episode_reward=%s episode_length=%s', self._episode_reward, self._episode_length) - total_time = time.time() - self._episode_time - to_log["global/episode_reward"] = self._episode_reward - to_log["global/episode_length"] = self._episode_length - to_log["global/episode_time"] = total_time - to_log["global/reward_per_time"] = self._episode_reward / total_time - self._episode_reward = 0 - self._episode_length = 0 - self._all_rewards = [] - - return observation, reward, done, to_log + if done: + logger.info("Episode terminating: episode_reward=%s episode_length=%s", + self._episode_reward, self._episode_length) + total_time = time.time() - self._episode_time + to_log["global/episode_reward"] = self._episode_reward + to_log["global/episode_length"] = self._episode_length + to_log["global/episode_time"] = total_time + to_log["global/reward_per_time"] = self._episode_reward / total_time + self._episode_reward = 0 + self._episode_length = 0 + self._all_rewards = [] + return observation, reward, done, to_log diff --git a/examples/a3c/misc.py b/examples/a3c/misc.py index 421964fa5..ab915ce0f 100644 --- a/examples/a3c/misc.py +++ b/examples/a3c/misc.py @@ -2,28 +2,32 @@ from __future__ import absolute_import from __future__ import division from __future__ import print_function -import numpy as np from datetime import datetime -import cProfile, pstats, io +import cProfile +import io +import pstats + def timestamp(): - return datetime.now().timestamp() + return datetime.now().timestamp() + def time_string(): - return datetime.now().strftime("%Y%m%d_%H_%M_%f") + return datetime.now().strftime("%Y%m%d_%H_%M_%f") + class Profiler(object): - def __init__(self): - self.pr = cProfile.Profile() - pass + def __init__(self): + self.pr = cProfile.Profile() + pass - def __enter__(self): - self.pr.enable() + def __enter__(self): + self.pr.enable() - def __exit__(self ,type, value, traceback): - self.pr.disable() - s = io.StringIO() - sortby = 'cumtime' - ps = pstats.Stats(self.pr, stream=s).sort_stats(sortby) - ps.print_stats(.2) - print(s.getvalue()) + def __exit__(self, type, value, traceback): + self.pr.disable() + s = io.StringIO() + sortby = "cumtime" + ps = pstats.Stats(self.pr, stream=s).sort_stats(sortby) + ps.print_stats(.2) + print(s.getvalue()) diff --git a/examples/a3c/policy.py b/examples/a3c/policy.py index 188bbbe46..b77ff6506 100644 --- a/examples/a3c/policy.py +++ b/examples/a3c/policy.py @@ -4,130 +4,142 @@ from __future__ import print_function import numpy as np import tensorflow as tf -import tensorflow.contrib.rnn as rnn -import distutils.version import ray -use_tf100_api = distutils.version.LooseVersion(tf.VERSION) >= distutils.version.LooseVersion('1.0.0') + class Policy(object): - """Policy base class""" + """The policy base class.""" + def __init__(self, ob_space, ac_space, task, name="local"): + self.local_steps = 0 + worker_device = "/job:localhost/replica:0/task:0/cpu:0" + self.g = tf.Graph() + with self.g.as_default(), tf.device(worker_device): + with tf.variable_scope(name): + self.setup_graph(ob_space, ac_space) + assert all([hasattr(self, attr) + for attr in ["vf", "logits", "x", "var_list"]]) + print("Setting up loss") + self.setup_loss(ac_space) + self.initialize() - def __init__(self, ob_space, ac_space, task, name="local"): - self.local_steps = 0 - worker_device = "/job:localhost/replica:0/task:0/cpu:0" - self.g = tf.Graph() - with self.g.as_default(), tf.device(worker_device): - with tf.variable_scope(name): - self.setup_graph(ob_space, ac_space) - assert all([hasattr(self, attr) for attr in ["vf", "logits", "x", "var_list"]]) - print("Setting up loss") - self.setup_loss(ac_space) - self.initialize() + def setup_graph(self): + raise NotImplementedError - def setup_graph(self): - raise NotImplementedError + def setup_loss(self, num_actions, summarize=True): + self.ac = tf.placeholder(tf.float32, [None, num_actions], name="ac") + self.adv = tf.placeholder(tf.float32, [None], name="adv") + self.r = tf.placeholder(tf.float32, [None], name="r") - def setup_loss(self, num_actions, summarize=True): - self.ac = tf.placeholder(tf.float32, [None, num_actions], name="ac") - self.adv = tf.placeholder(tf.float32, [None], name="adv") - self.r = tf.placeholder(tf.float32, [None], name="r") + log_prob_tf = tf.nn.log_softmax(self.logits) + prob_tf = tf.nn.softmax(self.logits) - log_prob_tf = tf.nn.log_softmax(self.logits) - prob_tf = tf.nn.softmax(self.logits) + # The "policy gradients" loss: its derivative is precisely the policy + # gradient. Notice that self.ac is a placeholder that is provided + # externally. adv will contain the advantages, as calculated in + # process_rollout. + pi_loss = - tf.reduce_sum(tf.reduce_sum(log_prob_tf * self.ac, + [1]) * self.adv) - # the "policy gradients" loss: its derivative is precisely the policy gradient - # notice that self.ac is a placeholder that is provided externally. - # adv will contain the advantages, as calculated in process_rollout - pi_loss = - tf.reduce_sum(tf.reduce_sum(log_prob_tf * self.ac, [1]) * self.adv) + # loss of value function + vf_loss = 0.5 * tf.reduce_sum(tf.square(self.vf - self.r)) + vf_loss = tf.Print(vf_loss, [vf_loss], "Value Fn Loss") + entropy = - tf.reduce_sum(prob_tf * log_prob_tf) - # loss of value function - vf_loss = 0.5 * tf.reduce_sum(tf.square(self.vf - self.r)) - vf_loss = tf.Print(vf_loss, [vf_loss], "Value Fn Loss") - entropy = - tf.reduce_sum(prob_tf * log_prob_tf) + bs = tf.to_float(tf.shape(self.x)[0]) + self.loss = pi_loss + 0.5 * vf_loss - entropy * 0.01 - bs = tf.to_float(tf.shape(self.x)[0]) - self.loss = pi_loss + 0.5 * vf_loss - entropy * 0.01 + grads = tf.gradients(self.loss, self.var_list) + self.grads, _ = tf.clip_by_global_norm(grads, 40.0) - grads = tf.gradients(self.loss, self.var_list) - self.grads, _ = tf.clip_by_global_norm(grads, 40.0) + grads_and_vars = list(zip(self.grads, self.var_list)) + opt = tf.train.AdamOptimizer(1e-4) + self._apply_gradients = opt.apply_gradients(grads_and_vars) - grads_and_vars = list(zip(self.grads, self.var_list)) - opt = tf.train.AdamOptimizer(1e-4) - self._apply_gradients = opt.apply_gradients(grads_and_vars) + if summarize: + tf.summary.scalar("model/policy_loss", pi_loss / bs) + tf.summary.scalar("model/value_loss", vf_loss / bs) + tf.summary.scalar("model/entropy", entropy / bs) + tf.summary.image("model/state", self.x) + self.summary_op = tf.summary.merge_all() - if summarize: - tf.summary.scalar("model/policy_loss", pi_loss / bs) - tf.summary.scalar("model/value_loss", vf_loss / bs) - tf.summary.scalar("model/entropy", entropy / bs) - tf.summary.image("model/state", self.x) - self.summary_op = tf.summary.merge_all() + def initialize(self): + self.sess = tf.Session(graph=self.g, config=tf.ConfigProto( + intra_op_parallelism_threads=1, inter_op_parallelism_threads=2)) + self.variables = ray.experimental.TensorFlowVariables(self.loss, self.sess) + self.sess.run(tf.global_variables_initializer()) - def initialize(self): - self.sess = tf.Session(graph=self.g, config=tf.ConfigProto(intra_op_parallelism_threads=1, inter_op_parallelism_threads=2)) - self.variables = ray.experimental.TensorFlowVariables(self.loss, self.sess) - self.sess.run(tf.global_variables_initializer()) + def model_update(self, grads): + feed_dict = {self.grads[i]: grads[i] + for i in range(len(grads))} + self.sess.run(self._apply_gradients, feed_dict=feed_dict) - def model_update(self, grads): - feed_dict = {self.grads[i]: grads[i] - for i in range(len(grads))} - self.sess.run(self._apply_gradients, feed_dict=feed_dict) + def get_weights(self): + weights = self.variables.get_weights() + return weights - def get_weights(self): - weights = self.variables.get_weights() - return weights + def set_weights(self, weights): + self.variables.set_weights(weights) - def set_weights(self, weights): - self.variables.set_weights(weights) + def get_gradients(self, batch): + raise NotImplementedError - def get_gradients(self, batch): - raise NotImplementedError + def get_vf_loss(self): + raise NotImplementedError - def get_vf_loss(self): - raise NotImplementedError + def act(self, ob): + raise NotImplementedError + def value(self, ob): + raise NotImplementedError - def act(self, ob): - raise NotImplementedError - - def value(self, ob): - raise NotImplementedError def normalized_columns_initializer(std=1.0): - def _initializer(shape, dtype=None, partition_info=None): - out = np.random.randn(*shape).astype(np.float32) - out *= std / np.sqrt(np.square(out).sum(axis=0, keepdims=True)) - return tf.constant(out) - return _initializer + def _initializer(shape, dtype=None, partition_info=None): + out = np.random.randn(*shape).astype(np.float32) + out *= std / np.sqrt(np.square(out).sum(axis=0, keepdims=True)) + return tf.constant(out) + return _initializer + def flatten(x): - return tf.reshape(x, [-1, np.prod(x.get_shape().as_list()[1:])]) + return tf.reshape(x, [-1, np.prod(x.get_shape().as_list()[1:])]) -def conv2d(x, num_filters, name, filter_size=(3, 3), stride=(1, 1), pad="SAME", dtype=tf.float32, collections=None): - with tf.variable_scope(name): - stride_shape = [1, stride[0], stride[1], 1] - filter_shape = [filter_size[0], filter_size[1], int(x.get_shape()[3]), num_filters] - # there are "num input feature maps * filter height * filter width" - # inputs to each hidden unit - fan_in = np.prod(filter_shape[:3]) - # each unit in the lower layer receives a gradient from: - # "num output feature maps * filter height * filter width" / - # pooling size - fan_out = np.prod(filter_shape[:2]) * num_filters - # initialize weights with random weights - w_bound = np.sqrt(6. / (fan_in + fan_out)) +def conv2d(x, num_filters, name, filter_size=(3, 3), stride=(1, 1), pad="SAME", + dtype=tf.float32, collections=None): + with tf.variable_scope(name): + stride_shape = [1, stride[0], stride[1], 1] + filter_shape = [filter_size[0], filter_size[1], int(x.get_shape()[3]), + num_filters] + + # There are "num input feature maps * filter height * filter width" + # inputs to each hidden unit. + fan_in = np.prod(filter_shape[:3]) + # Each unit in the lower layer receives a gradient from: + # "num output feature maps * filter height * filter width" / pooling size. + fan_out = np.prod(filter_shape[:2]) * num_filters + # Initialize weights with random weights. + w_bound = np.sqrt(6 / (fan_in + fan_out)) + + w = tf.get_variable("W", filter_shape, dtype, + tf.random_uniform_initializer(-w_bound, w_bound), + collections=collections) + b = tf.get_variable("b", [1, 1, 1, num_filters], + initializer=tf.constant_initializer(0.0), + collections=collections) + return tf.nn.conv2d(x, w, stride_shape, pad) + b - w = tf.get_variable("W", filter_shape, dtype, tf.random_uniform_initializer(-w_bound, w_bound), - collections=collections) - b = tf.get_variable("b", [1, 1, 1, num_filters], initializer=tf.constant_initializer(0.0), - collections=collections) - return tf.nn.conv2d(x, w, stride_shape, pad) + b def linear(x, size, name, initializer=None, bias_init=0): - w = tf.get_variable(name + "/w", [x.get_shape()[1], size], initializer=initializer) - b = tf.get_variable(name + "/b", [size], initializer=tf.constant_initializer(bias_init)) - return tf.matmul(x, w) + b + w = tf.get_variable(name + "/w", [x.get_shape()[1], size], + initializer=initializer) + b = tf.get_variable(name + "/b", [size], + initializer=tf.constant_initializer(bias_init)) + return tf.matmul(x, w) + b + def categorical_sample(logits, d): - value = tf.squeeze(tf.multinomial(logits - tf.reduce_max(logits, [1], keep_dims=True), 1), [1]) - return tf.one_hot(value, d) + value = tf.squeeze(tf.multinomial(logits - tf.reduce_max(logits, [1], + keep_dims=True), + 1), [1]) + return tf.one_hot(value, d) diff --git a/examples/a3c/runner.py b/examples/a3c/runner.py index 70ce47ce1..f787b79ea 100644 --- a/examples/a3c/runner.py +++ b/examples/a3c/runner.py @@ -5,156 +5,159 @@ from __future__ import print_function from collections import namedtuple import numpy as np import tensorflow as tf -from LSTM import LSTMPolicy import six.moves.queue as queue import scipy.signal import threading -import distutils.version -use_tf12_api = distutils.version.LooseVersion(tf.VERSION) >= distutils.version.LooseVersion('0.12.0') -from datetime import datetime + def discount(x, gamma): - return scipy.signal.lfilter([1], [1, -gamma], x[::-1], axis=0)[::-1] + return scipy.signal.lfilter([1], [1, -gamma], x[::-1], axis=0)[::-1] + def process_rollout(rollout, gamma, lambda_=1.0): - """ -given a rollout, compute its returns and the advantage -""" - batch_si = np.asarray(rollout.states) - batch_a = np.asarray(rollout.actions) - rewards = np.asarray(rollout.rewards) - vpred_t = np.asarray(rollout.values + [rollout.r]) + """Given a rollout, compute its returns and the advantage.""" + batch_si = np.asarray(rollout.states) + batch_a = np.asarray(rollout.actions) + rewards = np.asarray(rollout.rewards) + vpred_t = np.asarray(rollout.values + [rollout.r]) - rewards_plus_v = np.asarray(rollout.rewards + [rollout.r]) - batch_r = discount(rewards_plus_v, gamma)[:-1] - delta_t = rewards + gamma * vpred_t[1:] - vpred_t[:-1] - # this formula for the advantage comes "Generalized Advantage Estimation": - # https://arxiv.org/abs/1506.02438 - batch_adv = discount(delta_t, gamma * lambda_) + rewards_plus_v = np.asarray(rollout.rewards + [rollout.r]) + batch_r = discount(rewards_plus_v, gamma)[:-1] + delta_t = rewards + gamma * vpred_t[1:] - vpred_t[:-1] + # This formula for the advantage comes "Generalized Advantage Estimation": + # https://arxiv.org/abs/1506.02438 + batch_adv = discount(delta_t, gamma * lambda_) + + features = rollout.features[0] + return Batch(batch_si, batch_a, batch_adv, batch_r, rollout.terminal, + features) - features = rollout.features[0] - return Batch(batch_si, batch_a, batch_adv, batch_r, rollout.terminal, features) Batch = namedtuple("Batch", ["si", "a", "adv", "r", "terminal", "features"]) + class PartialRollout(object): - """ -a piece of a complete rollout. We run our agent, and process its experience -once it has processed enough steps. -""" - def __init__(self): - self.states = [] - self.actions = [] - self.rewards = [] - self.values = [] - self.r = 0.0 - self.terminal = False - self.features = [] + """A piece of a complete rollout. - def add(self, state, action, reward, value, terminal, features): - self.states += [state] - self.actions += [action] - self.rewards += [reward] - self.values += [value] - self.terminal = terminal - self.features += [features] + We run our agent, and process its experience once it has processed enough + steps. + """ + def __init__(self): + self.states = [] + self.actions = [] + self.rewards = [] + self.values = [] + self.r = 0.0 + self.terminal = False + self.features = [] + + def add(self, state, action, reward, value, terminal, features): + self.states += [state] + self.actions += [action] + self.rewards += [reward] + self.values += [value] + self.terminal = terminal + self.features += [features] + + def extend(self, other): + assert not self.terminal + self.states.extend(other.states) + self.actions.extend(other.actions) + self.rewards.extend(other.rewards) + self.values.extend(other.values) + self.r = other.r + self.terminal = other.terminal + self.features.extend(other.features) - def extend(self, other): - assert not self.terminal - self.states.extend(other.states) - self.actions.extend(other.actions) - self.rewards.extend(other.rewards) - self.values.extend(other.values) - self.r = other.r - self.terminal = other.terminal - self.features.extend(other.features) class RunnerThread(threading.Thread): - """ This thread constantly interacts with the environment and tell it what to do. """ - def __init__(self, env, policy, num_local_steps, visualise=False): - threading.Thread.__init__(self) - self.queue = queue.Queue(5) - self.num_local_steps = num_local_steps - self.env = env - self.last_features = None - self.policy = policy - self.daemon = True - self.sess = None - self.summary_writer = None - self.visualise = visualise + """This thread interacts with the environment and tells it what to do.""" + def __init__(self, env, policy, num_local_steps, visualise=False): + threading.Thread.__init__(self) + self.queue = queue.Queue(5) + self.num_local_steps = num_local_steps + self.env = env + self.last_features = None + self.policy = policy + self.daemon = True + self.sess = None + self.summary_writer = None + self.visualise = visualise - def start_runner(self, sess, summary_writer): - self.sess = sess - self.summary_writer = summary_writer - self.start() + def start_runner(self, sess, summary_writer): + self.sess = sess + self.summary_writer = summary_writer + self.start() - def run(self): - with self.sess.as_default(): - self._run() - - def _run(self): - rollout_provider = env_runner(self.env, self.policy, self.num_local_steps, self.summary_writer, self.visualise) - while True: - # the timeout variable exists because apparently, if one worker dies, the other workers - # won't die with it, unless the timeout is set to some large number. This is an empirical - # observation. - self.queue.put(next(rollout_provider), timeout=600.0) - # print("Current Q count: %d" % len(self.queue.queue)) + def run(self): + with self.sess.as_default(): + self._run() + def _run(self): + rollout_provider = env_runner(self.env, self.policy, self.num_local_steps, + self.summary_writer, self.visualise) + while True: + # The timeout variable exists because apparently, if one worker dies, the + # other workers won't die with it, unless the timeout is set to some + # large number. This is an empirical observation. + self.queue.put(next(rollout_provider), timeout=600.0) def env_runner(env, policy, num_local_steps, summary_writer, render): - """ -The logic of the thread runner. In brief, it constantly keeps on running -the policy, and as long as the rollout exceeds a certain length, the thread -runner appends the policy to the queue. -""" - last_state = env.reset() - last_features = policy.get_initial_features() - length = 0 - rewards = 0 - rollout_number = 0 + """This impleents the logic of the thread runner. - while True: - terminal_end = False - rollout = PartialRollout() + It continually runs the policy, and as long as the rollout exceeds a certain + length, the thread runner appends the policy to the queue. + """ + last_state = env.reset() + last_features = policy.get_initial_features() + length = 0 + rewards = 0 + rollout_number = 0 - for _ in range(num_local_steps): - fetched = policy.act(last_state, *last_features) - action, value_, features = fetched[0], fetched[1], fetched[2:] - # argmax to convert from one-hot - state, reward, terminal, info = env.step(action.argmax()) - if render: - env.render() + while True: + terminal_end = False + rollout = PartialRollout() - # collect the experience - rollout.add(last_state, action, reward, value_, terminal, last_features) - length += 1 - rewards += reward + for _ in range(num_local_steps): + fetched = policy.act(last_state, *last_features) + action, value_, features = fetched[0], fetched[1], fetched[2:] + # Argmax to convert from one-hot. + state, reward, terminal, info = env.step(action.argmax()) + if render: + env.render() - last_state = state - last_features = features + # Collect the experience. + rollout.add(last_state, action, reward, value_, terminal, last_features) + length += 1 + rewards += reward - if info: - summary = tf.Summary() - for k, v in info.items(): - summary.value.add(tag=k, simple_value=float(v)) - summary_writer.add_summary(summary, rollout_number) - summary_writer.flush() + last_state = state + last_features = features - timestep_limit = env.spec.tags.get('wrapper_config.TimeLimit.max_episode_steps') - if terminal or length >= timestep_limit: - terminal_end = True - if length >= timestep_limit or not env.metadata.get('semantics.autoreset'): - last_state = env.reset() - last_features = policy.get_initial_features() - rollout_number += 1 - length = 0 - rewards = 0 - break + if info: + summary = tf.Summary() + for k, v in info.items(): + summary.value.add(tag=k, simple_value=float(v)) + summary_writer.add_summary(summary, rollout_number) + summary_writer.flush() - if not terminal_end: - rollout.r = policy.value(last_state, *last_features) + timestep_limit = env.spec.tags.get("wrapper_config.TimeLimit" + ".max_episode_steps") + if terminal or length >= timestep_limit: + terminal_end = True + if length >= timestep_limit or not env.metadata.get("semantics" + ".autoreset"): + last_state = env.reset() + last_features = policy.get_initial_features() + rollout_number += 1 + length = 0 + rewards = 0 + break - # once we have enough experience, yield it, and have the ThreadRunner place it on a queue - yield rollout + if not terminal_end: + rollout.r = policy.value(last_state, *last_features) + + # Once we have enough experience, yield it, and have the ThreadRunner + # place it on a queue. + yield rollout diff --git a/examples/evolution_strategies/evolution_strategies.py b/examples/evolution_strategies/evolution_strategies.py index c01ce88da..948d5b862 100644 --- a/examples/evolution_strategies/evolution_strategies.py +++ b/examples/evolution_strategies/evolution_strategies.py @@ -100,6 +100,7 @@ class Worker(object): task_ob_stat = utils.RunningStat(self.env.observation_space.shape, eps=0) # Perform some rollouts with noise. + task_tstart = time.time() while (len(noise_inds) == 0 or time.time() - task_tstart < self.min_task_runtime): noise_idx = self.noise.sample_index(self.rs, self.policy.num_params) @@ -122,15 +123,15 @@ class Worker(object): lengths.append([len_pos, len_neg]) return Result( - noise_inds_n=np.array(noise_inds), - returns_n2=np.array(returns, dtype=np.float32), - sign_returns_n2=np.array(sign_returns, dtype=np.float32), - lengths_n2=np.array(lengths, dtype=np.int32), - eval_return=None, - eval_length=None, - ob_sum=(None if task_ob_stat.count == 0 else task_ob_stat.sum), - ob_sumsq=(None if task_ob_stat.count == 0 else task_ob_stat.sumsq), - ob_count=task_ob_stat.count) + noise_inds_n=np.array(noise_inds), + returns_n2=np.array(returns, dtype=np.float32), + sign_returns_n2=np.array(sign_returns, dtype=np.float32), + lengths_n2=np.array(lengths, dtype=np.int32), + eval_return=None, + eval_length=None, + ob_sum=(None if task_ob_stat.count == 0 else task_ob_stat.sum), + ob_sumsq=(None if task_ob_stat.count == 0 else task_ob_stat.sumsq), + ob_count=task_ob_stat.count) if __name__ == "__main__": @@ -168,11 +169,11 @@ if __name__ == "__main__": episode_cutoff_mode="env_default") policy_params = { - "ac_bins": "continuous:", - "ac_noise_std": 0.01, - "nonlin_type": "tanh", - "hidden_dims": [256, 256], - "connection_type": "ff" + "ac_bins": "continuous:", + "ac_noise_std": 0.01, + "nonlin_type": "tanh", + "hidden_dims": [256, 256], + "connection_type": "ff" } # Create the shared noise table. @@ -208,10 +209,9 @@ if __name__ == "__main__": # Use the actors to do rollouts, note that we pass in the ID of the policy # weights. rollout_ids = [worker.do_rollouts.remote( - theta_id, - ob_stat.mean if policy.needs_ob_stat else None, - ob_stat.std if policy.needs_ob_stat else None) - for worker in workers] + theta_id, + ob_stat.mean if policy.needs_ob_stat else None, + ob_stat.std if policy.needs_ob_stat else None) for worker in workers] # Get the results of the rollouts. results = ray.get(rollout_ids) diff --git a/examples/evolution_strategies/policies.py b/examples/evolution_strategies/policies.py index df1b85f50..96f676e58 100644 --- a/examples/evolution_strategies/policies.py +++ b/examples/evolution_strategies/policies.py @@ -18,234 +18,224 @@ logger = logging.getLogger(__name__) class Policy: - def __init__(self, *args, **kwargs): - self.args, self.kwargs = args, kwargs - self.scope = self._initialize(*args, **kwargs) - self.all_variables = tf.get_collection(tf.GraphKeys.GLOBAL_VARIABLES, self.scope.name) + def __init__(self, *args, **kwargs): + self.args, self.kwargs = args, kwargs + self.scope = self._initialize(*args, **kwargs) + self.all_variables = tf.get_collection(tf.GraphKeys.GLOBAL_VARIABLES, + self.scope.name) - self.trainable_variables = tf.get_collection(tf.GraphKeys.TRAINABLE_VARIABLES, self.scope.name) - self.num_params = sum(int(np.prod(v.get_shape().as_list())) for v in self.trainable_variables) - self._setfromflat = U.SetFromFlat(self.trainable_variables) - self._getflat = U.GetFlat(self.trainable_variables) + self.trainable_variables = tf.get_collection( + tf.GraphKeys.TRAINABLE_VARIABLES, self.scope.name) + self.num_params = sum(int(np.prod(v.get_shape().as_list())) + for v in self.trainable_variables) + self._setfromflat = U.SetFromFlat(self.trainable_variables) + self._getflat = U.GetFlat(self.trainable_variables) - logger.info('Trainable variables ({} parameters)'.format(self.num_params)) - for v in self.trainable_variables: - shp = v.get_shape().as_list() - logger.info('- {} shape:{} size:{}'.format(v.name, shp, np.prod(shp))) - logger.info('All variables') - for v in self.all_variables: - shp = v.get_shape().as_list() - logger.info('- {} shape:{} size:{}'.format(v.name, shp, np.prod(shp))) + logger.info('Trainable variables ({} parameters)'.format(self.num_params)) + for v in self.trainable_variables: + shp = v.get_shape().as_list() + logger.info('- {} shape:{} size:{}'.format(v.name, shp, np.prod(shp))) + logger.info('All variables') + for v in self.all_variables: + shp = v.get_shape().as_list() + logger.info('- {} shape:{} size:{}'.format(v.name, shp, np.prod(shp))) - placeholders = [tf.placeholder(v.value().dtype, v.get_shape().as_list()) for v in self.all_variables] - self.set_all_vars = U.function( - inputs=placeholders, - outputs=[], - updates=[tf.group(*[v.assign(p) for v, p in zip(self.all_variables, placeholders)])] - ) + placeholders = [tf.placeholder(v.value().dtype, v.get_shape().as_list()) + for v in self.all_variables] + self.set_all_vars = U.function( + inputs=placeholders, + outputs=[], + updates=[tf.group(*[v.assign(p) for v, p + in zip(self.all_variables, placeholders)])] + ) - def _initialize(self, *args, **kwargs): - raise NotImplementedError + def _initialize(self, *args, **kwargs): + raise NotImplementedError - def save(self, filename): - assert filename.endswith('.h5') - with h5py.File(filename, 'w') as f: - for v in self.all_variables: - f[v.name] = v.eval() - # TODO: it would be nice to avoid pickle, but it's convenient to pass Python objects to _initialize - # (like Gym spaces or numpy arrays) - f.attrs['name'] = type(self).__name__ - f.attrs['args_and_kwargs'] = np.void(pickle.dumps((self.args, self.kwargs), protocol=-1)) + def save(self, filename): + assert filename.endswith('.h5') + with h5py.File(filename, 'w') as f: + for v in self.all_variables: + f[v.name] = v.eval() + # TODO: It would be nice to avoid pickle, but it's convenient to pass + # Python objects to _initialize (like Gym spaces or numpy arrays). + f.attrs['name'] = type(self).__name__ + f.attrs['args_and_kwargs'] = np.void(pickle.dumps((self.args, + self.kwargs), + protocol=-1)) - @classmethod - def Load(cls, filename, extra_kwargs=None): - with h5py.File(filename, 'r') as f: - args, kwargs = pickle.loads(f.attrs['args_and_kwargs'].tostring()) - if extra_kwargs: - kwargs.update(extra_kwargs) - policy = cls(*args, **kwargs) - policy.set_all_vars(*[f[v.name][...] for v in policy.all_variables]) - return policy + @classmethod + def Load(cls, filename, extra_kwargs=None): + with h5py.File(filename, 'r') as f: + args, kwargs = pickle.loads(f.attrs['args_and_kwargs'].tostring()) + if extra_kwargs: + kwargs.update(extra_kwargs) + policy = cls(*args, **kwargs) + policy.set_all_vars(*[f[v.name][...] for v in policy.all_variables]) + return policy - # === Rollouts/training === + # === Rollouts/training === - def rollout(self, env, *, render=False, timestep_limit=None, save_obs=False, random_stream=None): - """ - If random_stream is provided, the rollout will take noisy actions with noise drawn from that stream. - Otherwise, no action noise will be added. - """ - env_timestep_limit = env.spec.tags.get('wrapper_config.TimeLimit.max_episode_steps') - timestep_limit = env_timestep_limit if timestep_limit is None else min(timestep_limit, env_timestep_limit) - rews = [] - t = 0 - if save_obs: - obs = [] - ob = env.reset() - for _ in range(timestep_limit): - ac = self.act(ob[None], random_stream=random_stream)[0] - if save_obs: - obs.append(ob) - ob, rew, done, _ = env.step(ac) - rews.append(rew) - t += 1 - if render: - env.render() - if done: - break - rews = np.array(rews, dtype=np.float32) - if save_obs: - return rews, t, np.array(obs) - return rews, t + def rollout(self, env, *, render=False, timestep_limit=None, save_obs=False, + random_stream=None): + """Do a rollout. - def act(self, ob, random_stream=None): - raise NotImplementedError + If random_stream is provided, the rollout will take noisy actions with + noise drawn from that stream. Otherwise, no action noise will be added. + """ + env_timestep_limit = env.spec.tags.get("wrapper_config.TimeLimit" + ".max_episode_steps") + timestep_limit = (env_timestep_limit if timestep_limit is None + else min(timestep_limit, env_timestep_limit)) + rews = [] + t = 0 + if save_obs: + obs = [] + ob = env.reset() + for _ in range(timestep_limit): + ac = self.act(ob[None], random_stream=random_stream)[0] + if save_obs: + obs.append(ob) + ob, rew, done, _ = env.step(ac) + rews.append(rew) + t += 1 + if render: + env.render() + if done: + break + rews = np.array(rews, dtype=np.float32) + if save_obs: + return rews, t, np.array(obs) + return rews, t - def set_trainable_flat(self, x): - self._setfromflat(x) + def act(self, ob, random_stream=None): + raise NotImplementedError - def get_trainable_flat(self): - return self._getflat() + def set_trainable_flat(self, x): + self._setfromflat(x) - @property - def needs_ob_stat(self): - raise NotImplementedError + def get_trainable_flat(self): + return self._getflat() - def set_ob_stat(self, ob_mean, ob_std): - raise NotImplementedError + @property + def needs_ob_stat(self): + raise NotImplementedError + + def set_ob_stat(self, ob_mean, ob_std): + raise NotImplementedError def bins(x, dim, num_bins, name): - scores = U.dense(x, dim * num_bins, name, U.normc_initializer(0.01)) - scores_nab = tf.reshape(scores, [-1, dim, num_bins]) - return tf.argmax(scores_nab, 2) # 0 ... num_bins-1 + scores = U.dense(x, dim * num_bins, name, U.normc_initializer(0.01)) + scores_nab = tf.reshape(scores, [-1, dim, num_bins]) + return tf.argmax(scores_nab, 2) class MujocoPolicy(Policy): - def _initialize(self, ob_space, ac_space, ac_bins, ac_noise_std, nonlin_type, hidden_dims, connection_type): - self.ac_space = ac_space - self.ac_bins = ac_bins - self.ac_noise_std = ac_noise_std - self.hidden_dims = hidden_dims - self.connection_type = connection_type + def _initialize(self, ob_space, ac_space, ac_bins, ac_noise_std, nonlin_type, + hidden_dims, connection_type): + self.ac_space = ac_space + self.ac_bins = ac_bins + self.ac_noise_std = ac_noise_std + self.hidden_dims = hidden_dims + self.connection_type = connection_type - assert len(ob_space.shape) == len(self.ac_space.shape) == 1 - assert np.all(np.isfinite(self.ac_space.low)) and np.all(np.isfinite(self.ac_space.high)), \ - 'Action bounds required' + assert len(ob_space.shape) == len(self.ac_space.shape) == 1 + assert (np.all(np.isfinite(self.ac_space.low)) and + np.all(np.isfinite(self.ac_space.high))), "Action bounds required" - self.nonlin = {'tanh': tf.tanh, 'relu': tf.nn.relu, 'lrelu': U.lrelu, 'elu': tf.nn.elu}[nonlin_type] + self.nonlin = {'tanh': tf.tanh, + 'relu': tf.nn.relu, + 'lrelu': U.lrelu, + 'elu': tf.nn.elu}[nonlin_type] - with tf.variable_scope(type(self).__name__) as scope: - # Observation normalization - ob_mean = tf.get_variable( - 'ob_mean', ob_space.shape, tf.float32, tf.constant_initializer(np.nan), trainable=False) - ob_std = tf.get_variable( - 'ob_std', ob_space.shape, tf.float32, tf.constant_initializer(np.nan), trainable=False) - in_mean = tf.placeholder(tf.float32, ob_space.shape) - in_std = tf.placeholder(tf.float32, ob_space.shape) - self._set_ob_mean_std = U.function([in_mean, in_std], [], updates=[ - tf.assign(ob_mean, in_mean), - tf.assign(ob_std, in_std), - ]) + with tf.variable_scope(type(self).__name__) as scope: + # Observation normalization. + ob_mean = tf.get_variable( + 'ob_mean', ob_space.shape, tf.float32, + tf.constant_initializer(np.nan), trainable=False) + ob_std = tf.get_variable( + 'ob_std', ob_space.shape, tf.float32, + tf.constant_initializer(np.nan), trainable=False) + in_mean = tf.placeholder(tf.float32, ob_space.shape) + in_std = tf.placeholder(tf.float32, ob_space.shape) + self._set_ob_mean_std = U.function([in_mean, in_std], [], updates=[ + tf.assign(ob_mean, in_mean), + tf.assign(ob_std, in_std), + ]) - # Policy network - o = tf.placeholder(tf.float32, [None] + list(ob_space.shape)) - a = self._make_net(tf.clip_by_value((o - ob_mean) / ob_std, -5.0, 5.0)) - self._act = U.function([o], a) - return scope + # Policy network. + o = tf.placeholder(tf.float32, [None] + list(ob_space.shape)) + a = self._make_net(tf.clip_by_value((o - ob_mean) / ob_std, -5.0, 5.0)) + self._act = U.function([o], a) + return scope - def _make_net(self, o): - # Process observation - if self.connection_type == 'ff': - x = o - for ilayer, hd in enumerate(self.hidden_dims): - x = self.nonlin(U.dense(x, hd, 'l{}'.format(ilayer), U.normc_initializer(1.0))) - else: - raise NotImplementedError(self.connection_type) + def _make_net(self, o): + # Process observation. + if self.connection_type == 'ff': + x = o + for ilayer, hd in enumerate(self.hidden_dims): + x = self.nonlin(U.dense(x, hd, 'l{}'.format(ilayer), + U.normc_initializer(1.0))) + else: + raise NotImplementedError(self.connection_type) - # Map to action - adim, ahigh, alow = self.ac_space.shape[0], self.ac_space.high, self.ac_space.low - assert isinstance(self.ac_bins, str) - ac_bin_mode, ac_bin_arg = self.ac_bins.split(':') + # Map to action. + adim = self.ac_space.shape[0] + ahigh = self.ac_space.high + alow = self.ac_space.low + assert isinstance(self.ac_bins, str) + ac_bin_mode, ac_bin_arg = self.ac_bins.split(':') - if ac_bin_mode == 'uniform': - # Uniformly spaced bins, from ac_space.low to ac_space.high - num_ac_bins = int(ac_bin_arg) - aidx_na = bins(x, adim, num_ac_bins, 'out') # 0 ... num_ac_bins-1 - ac_range_1a = (ahigh - alow)[None, :] - a = 1. / (num_ac_bins - 1.) * tf.to_float(aidx_na) * ac_range_1a + alow[None, :] + if ac_bin_mode == 'uniform': + # Uniformly spaced bins, from ac_space.low to ac_space.high. + num_ac_bins = int(ac_bin_arg) + aidx_na = bins(x, adim, num_ac_bins, 'out') + ac_range_1a = (ahigh - alow)[None, :] + a = (1. / (num_ac_bins - 1.) * tf.to_float(aidx_na) * ac_range_1a + + alow[None, :]) - elif ac_bin_mode == 'custom': - # Custom bins specified as a list of values from -1 to 1 - # The bins are rescaled to ac_space.low to ac_space.high - acvals_k = np.array(list(map(float, ac_bin_arg.split(','))), dtype=np.float32) - logger.info('Custom action values: ' + ' '.join('{:.3f}'.format(x) for x in acvals_k)) - assert acvals_k.ndim == 1 and acvals_k[0] == -1 and acvals_k[-1] == 1 - acvals_ak = ( - (ahigh - alow)[:, None] / (acvals_k[-1] - acvals_k[0]) * (acvals_k - acvals_k[0])[None, :] - + alow[:, None] - ) + elif ac_bin_mode == 'custom': + # Custom bins specified as a list of values from -1 to 1. + # The bins are rescaled to ac_space.low to ac_space.high. + acvals_k = np.array(list(map(float, ac_bin_arg.split(','))), + dtype=np.float32) + logger.info('Custom action values: ' + ' '.join('{:.3f}'.format(x) + for x in acvals_k)) + assert acvals_k.ndim == 1 and acvals_k[0] == -1 and acvals_k[-1] == 1 + acvals_ak = ((ahigh - alow)[:, None] / (acvals_k[-1] - acvals_k[0]) * + (acvals_k - acvals_k[0])[None, :] + alow[:, None]) - aidx_na = bins(x, adim, len(acvals_k), 'out') # values in [0, k-1] - a = tf.gather_nd( - acvals_ak, - tf.concat([ - tf.tile(np.arange(adim)[None, :, None], [tf.shape(aidx_na)[0], 1, 1]), - 2, - tf.expand_dims(aidx_na, -1) - ]) # (n,a,2) - ) # (n,a) - elif ac_bin_mode == 'continuous': - a = U.dense(x, adim, 'out', U.normc_initializer(0.01)) - else: - raise NotImplementedError(ac_bin_mode) + aidx_na = bins(x, adim, len(acvals_k), 'out') # Values in [0, k-1]. + a = tf.gather_nd( + acvals_ak, + tf.concat([ + tf.tile(np.arange(adim)[None, :, None], + [tf.shape(aidx_na)[0], 1, 1]), + 2, + tf.expand_dims(aidx_na, -1) + ]) # (n, a, 2) + ) # (n, a) + elif ac_bin_mode == 'continuous': + a = U.dense(x, adim, 'out', U.normc_initializer(0.01)) + else: + raise NotImplementedError(ac_bin_mode) - return a + return a - def act(self, ob, random_stream=None): - a = self._act(ob) - if random_stream is not None and self.ac_noise_std != 0: - a += random_stream.randn(*a.shape) * self.ac_noise_std - return a + def act(self, ob, random_stream=None): + a = self._act(ob) + if random_stream is not None and self.ac_noise_std != 0: + a += random_stream.randn(*a.shape) * self.ac_noise_std + return a - @property - def needs_ob_stat(self): - return True + @property + def needs_ob_stat(self): + return True - @property - def needs_ref_batch(self): - return False + @property + def needs_ref_batch(self): + return False - def set_ob_stat(self, ob_mean, ob_std): - self._set_ob_mean_std(ob_mean, ob_std) - - def initialize_from(self, filename, ob_stat=None): - """ - Initializes weights from another policy, which must have the same architecture (variable names), - but the weight arrays can be smaller than the current policy. - """ - with h5py.File(filename, 'r') as f: - f_var_names = [] - f.visititems(lambda name, obj: f_var_names.append(name) if isinstance(obj, h5py.Dataset) else None) - assert set(v.name for v in self.all_variables) == set(f_var_names), 'Variable names do not match' - - init_vals = [] - for v in self.all_variables: - shp = v.get_shape().as_list() - f_shp = f[v.name].shape - assert len(shp) == len(f_shp) and all(a >= b for a, b in zip(shp, f_shp)), \ - 'This policy must have more weights than the policy to load' - init_val = v.eval() - # ob_mean and ob_std are initialized with nan, so set them manually - if 'ob_mean' in v.name: - init_val[:] = 0 - init_mean = init_val - elif 'ob_std' in v.name: - init_val[:] = 0.001 - init_std = init_val - # Fill in subarray from the loaded policy - init_val[tuple([np.s_[:s] for s in f_shp])] = f[v.name] - init_vals.append(init_val) - self.set_all_vars(*init_vals) - - if ob_stat is not None: - ob_stat.set_from_init(init_mean, init_std, init_count=1e5) + def set_ob_stat(self, ob_mean, ob_std): + self._set_ob_mean_std(ob_mean, ob_std) diff --git a/examples/evolution_strategies/tabular_logger.py b/examples/evolution_strategies/tabular_logger.py index c6ad3de0f..d92adf164 100644 --- a/examples/evolution_strategies/tabular_logger.py +++ b/examples/evolution_strategies/tabular_logger.py @@ -7,7 +7,6 @@ from __future__ import print_function from collections import OrderedDict import os -import shutil import sys import time @@ -23,172 +22,200 @@ ERROR = 40 DISABLED = 50 -class TbWriter(object): - """ - Based on SummaryWriter, but changed to allow for a different prefix - and to get rid of multithreading - oops, ended up using the same prefix anyway. - """ - def __init__(self, dir, prefix): - self.dir = dir - self.step = 1 # Start at 1, because EvWriter automatically generates an object with step=0 - self.evwriter = pywrap_tensorflow.EventsWriter(compat.as_bytes(os.path.join(dir, prefix))) - def write_values(self, key2val): - summary = tf.Summary(value=[tf.Summary.Value(tag=k, simple_value=float(v)) - for (k, v) in key2val.items()]) - event = event_pb2.Event(wall_time=time.time(), summary=summary) - event.step = self.step # is there any reason why you'd want to specify the step? - self.evwriter.WriteEvent(event) - self.evwriter.Flush() - self.step += 1 - def close(self): - self.evwriter.Close() -# ================================================================ +class TbWriter(object): + """Based on SummaryWriter, but changed to allow for a different prefix.""" + def __init__(self, dir, prefix): + self.dir = dir + # Start at 1, because EvWriter automatically generates an object with + # step = 0. + self.step = 1 + self.evwriter = pywrap_tensorflow.EventsWriter( + compat.as_bytes(os.path.join(dir, prefix))) + + def write_values(self, key2val): + summary = tf.Summary(value=[tf.Summary.Value(tag=k, simple_value=float(v)) + for (k, v) in key2val.items()]) + event = event_pb2.Event(wall_time=time.time(), summary=summary) + event.step = self.step + self.evwriter.WriteEvent(event) + self.evwriter.Flush() + self.step += 1 + + def close(self): + self.evwriter.Close() + # API -# ================================================================ + def start(dir): - """ - dir: directory to put all output files - force: if dir already exists, should we delete it, or throw a RuntimeError? - """ - if _Logger.CURRENT is not _Logger.DEFAULT: - sys.stderr.write("WARNING: You asked to start logging (dir=%s), but you never stopped the previous logger (dir=%s).\n"%(dir, _Logger.CURRENT.dir)) - _Logger.CURRENT = _Logger(dir=dir) + if _Logger.CURRENT is not _Logger.DEFAULT: + sys.stderr.write("WARNING: You asked to start logging (dir=%s), but you " + "never stopped the previous logger (dir=%s)." + "\n" % (dir, _Logger.CURRENT.dir)) + _Logger.CURRENT = _Logger(dir=dir) + def stop(): - if _Logger.CURRENT is _Logger.DEFAULT: - sys.stderr.write("WARNING: You asked to stop logging, but you never started any previous logger.\n"%(dir, _Logger.CURRENT.dir)) - return - _Logger.CURRENT.close() - _Logger.CURRENT = _Logger.DEFAULT + if _Logger.CURRENT is _Logger.DEFAULT: + sys.stderr.write("WARNING: You asked to stop logging, but you never " + "started any previous logger." + "\n" % (dir, _Logger.CURRENT.dir)) + return + _Logger.CURRENT.close() + _Logger.CURRENT = _Logger.DEFAULT + def record_tabular(key, val): - """ - Log a value of some diagnostic - Call this once for each diagnostic quantity, each iteration - """ - _Logger.CURRENT.record_tabular(key, val) + """Log a value of some diagnostic. + + Call this once for each diagnostic quantity, each iteration. + """ + _Logger.CURRENT.record_tabular(key, val) + def dump_tabular(): - """ - Write all of the diagnostics from the current iteration + """Write all of the diagnostics from the current iteration.""" + _Logger.CURRENT.dump_tabular() - level: int. (see logger.py docs) If the global logger level is higher than - the level argument here, don't print to stdout. - """ - _Logger.CURRENT.dump_tabular() def log(*args, level=INFO): - """ - Write the sequence of args, with no separators, to the console and output files (if you've configured an output file). - """ - _Logger.CURRENT.log(*args, level=level) + """Write the sequence of args, with no separators. + + This is written to the console and output files (if you've configured an + output file). + """ + _Logger.CURRENT.log(*args, level=level) + def debug(*args): - log(*args, level=DEBUG) + log(*args, level=DEBUG) + + def info(*args): - log(*args, level=INFO) + log(*args, level=INFO) + + def warn(*args): - log(*args, level=WARN) + log(*args, level=WARN) + + def error(*args): - log(*args, level=ERROR) + log(*args, level=ERROR) + def set_level(level): - """ - Set logging threshold on current logger. - """ - _Logger.CURRENT.set_level(level) + """ + Set logging threshold on current logger. + """ + _Logger.CURRENT.set_level(level) + def get_dir(): - """ - Get directory that log files are being written to. - will be None if there is no output directory (i.e., if you didn't call start) - """ - return _Logger.CURRENT.get_dir() + """ + Get directory that log files are being written to. + will be None if there is no output directory (i.e., if you didn't call start) + """ + return _Logger.CURRENT.get_dir() + def get_expt_dir(): - sys.stderr.write("get_expt_dir() is Deprecated. Switch to get_dir()\n") - return get_dir() + sys.stderr.write("get_expt_dir() is Deprecated. Switch to get_dir()\n") + return get_dir() -# ================================================================ # Backend -# ================================================================ + class _Logger(object): - DEFAULT = None # A logger with no output files. (See right below class definition) - # So that you can still log to the terminal without setting up any output files - CURRENT = None # Current logger being used by the free functions above + # A logger with no output files. (See right below class definition) so that + # you can still log to the terminal without setting up any output files. + DEFAULT = None + # Current logger being used by the free functions above. + CURRENT = None - def __init__(self, dir=None): - self.name2val = OrderedDict() # values this iteration - self.level = INFO - self.dir = dir - self.text_outputs = [sys.stdout] - if dir is not None: - os.makedirs(dir, exist_ok=True) - self.text_outputs.append(open(os.path.join(dir, "log.txt"), "w")) - self.tbwriter = TbWriter(dir=dir, prefix="events") - else: - self.tbwriter = None + def __init__(self, dir=None): + self.name2val = OrderedDict() # Values this iteration. + self.level = INFO + self.dir = dir + self.text_outputs = [sys.stdout] + if dir is not None: + os.makedirs(dir, exist_ok=True) + self.text_outputs.append(open(os.path.join(dir, "log.txt"), "w")) + self.tbwriter = TbWriter(dir=dir, prefix="events") + else: + self.tbwriter = None - # Logging API, forwarded - # ---------------------------------------- - def record_tabular(self, key, val): - self.name2val[key] = val - def dump_tabular(self): - # Create strings for printing - key2str = OrderedDict() - for (key,val) in self.name2val.items(): - if hasattr(val, "__float__"): valstr = "%-8.3g"%val - else: valstr = val - key2str[self._truncate(key)]=self._truncate(valstr) - keywidth = max(map(len, key2str.keys())) - valwidth = max(map(len, key2str.values())) - # Write to all text outputs - self._write_text("-"*(keywidth+valwidth+7), "\n") - for (key,val) in key2str.items(): - self._write_text("| ", key, " "*(keywidth-len(key)), " | ", val, " "*(valwidth-len(val)), " |\n") - self._write_text("-"*(keywidth+valwidth+7), "\n") - for f in self.text_outputs: - try: f.flush() - except OSError: sys.stderr.write('Warning! OSError when flushing.\n') - # Write to tensorboard - if self.tbwriter is not None: - self.tbwriter.write_values(self.name2val) - self.name2val.clear() - def log(self, *args, level=INFO): - if self.level <= level: - self._do_log(*args) + # Logging API, forwarded - # Configuration - # ---------------------------------------- - def set_level(self, level): - self.level = level - def get_dir(self): - return self.dir + def record_tabular(self, key, val): + self.name2val[key] = val - def close(self): - for f in self.text_outputs[1:]: f.close() - if self.tbwriter: self.tbwriter.close() + def dump_tabular(self): + # Create strings for printing. + key2str = OrderedDict() + for (key, val) in self.name2val.items(): + if hasattr(val, "__float__"): + valstr = "%-8.3g" % val + else: + valstr = val + key2str[self._truncate(key)] = self._truncate(valstr) + keywidth = max(map(len, key2str.keys())) + valwidth = max(map(len, key2str.values())) + # Write to all text outputs + self._write_text("-" * (keywidth + valwidth + 7), "\n") + for (key, val) in key2str.items(): + self._write_text("| ", key, " " * (keywidth - len(key)), " | ", val, + " " * (valwidth - len(val)), " |\n") + self._write_text("-" * (keywidth + valwidth + 7), "\n") + for f in self.text_outputs: + try: + f.flush() + except OSError: + sys.stderr.write('Warning! OSError when flushing.\n') + # Write to tensorboard + if self.tbwriter is not None: + self.tbwriter.write_values(self.name2val) + self.name2val.clear() + + def log(self, *args, level=INFO): + if self.level <= level: + self._do_log(*args) + + # Configuration + + def set_level(self, level): + self.level = level + + def get_dir(self): + return self.dir + + def close(self): + for f in self.text_outputs[1:]: + f.close() + if self.tbwriter: + self.tbwriter.close() + + # Misc + + def _do_log(self, *args): + self._write_text(*args, '\n') + for f in self.text_outputs: + try: + f.flush() + except OSError: + print('Warning! OSError when flushing.') + + def _write_text(self, *strings): + for f in self.text_outputs: + for string in strings: + f.write(string) + + def _truncate(self, s): + if len(s) > 33: + return s[:30] + "..." + else: + return s - # Misc - # ---------------------------------------- - def _do_log(self, *args): - self._write_text(*args, '\n') - for f in self.text_outputs: - try: f.flush() - except OSError: print('Warning! OSError when flushing.') - def _write_text(self, *strings): - for f in self.text_outputs: - for string in strings: - f.write(string) - def _truncate(self, s): - if len(s) > 33: - return s[:30] + "..." - else: - return s _Logger.DEFAULT = _Logger() _Logger.CURRENT = _Logger.DEFAULT diff --git a/examples/evolution_strategies/tf_util.py b/examples/evolution_strategies/tf_util.py index 0a1baf1a0..0ef934bb2 100644 --- a/examples/evolution_strategies/tf_util.py +++ b/examples/evolution_strategies/tf_util.py @@ -7,9 +7,7 @@ from __future__ import print_function import numpy as np import tensorflow as tf -import builtins import functools -import copy import os # ================================================================ @@ -19,242 +17,267 @@ import os clip = tf.clip_by_value # Make consistent with numpy -# ---------------------------------------- + def sum(x, axis=None, keepdims=False): - return tf.reduce_sum(x, reduction_indices=None if axis is None else [axis], keep_dims = keepdims) + return tf.reduce_sum(x, reduction_indices=None if axis is None else [axis], + keep_dims=keepdims) + + def mean(x, axis=None, keepdims=False): - return tf.reduce_mean(x, reduction_indices=None if axis is None else [axis], keep_dims = keepdims) + return tf.reduce_mean(x, reduction_indices=None if axis is None else [axis], + keep_dims=keepdims) + + def var(x, axis=None, keepdims=False): - meanx = mean(x, axis=axis, keepdims=keepdims) - return mean(tf.square(x - meanx), axis=axis, keepdims=keepdims) + meanx = mean(x, axis=axis, keepdims=keepdims) + return mean(tf.square(x - meanx), axis=axis, keepdims=keepdims) + + def std(x, axis=None, keepdims=False): - return tf.sqrt(var(x, axis=axis, keepdims=keepdims)) + return tf.sqrt(var(x, axis=axis, keepdims=keepdims)) + + def max(x, axis=None, keepdims=False): - return tf.reduce_max(x, reduction_indices=None if axis is None else [axis], keep_dims = keepdims) + return tf.reduce_max(x, reduction_indices=None if axis is None else [axis], + keep_dims=keepdims) + + def min(x, axis=None, keepdims=False): - return tf.reduce_min(x, reduction_indices=None if axis is None else [axis], keep_dims = keepdims) + return tf.reduce_min(x, reduction_indices=None if axis is None else [axis], + keep_dims=keepdims) + + def concatenate(arrs, axis=0): - return tf.concat(arrs, axis) + return tf.concat(arrs, axis) + + def argmax(x, axis=None): - return tf.argmax(x, dimension=axis) - -def switch(condition, then_expression, else_expression): - '''Switches between two operations depending on a scalar value (int or bool). - Note that both `then_expression` and `else_expression` - should be symbolic tensors of the *same shape*. - - # Arguments - condition: scalar tensor. - then_expression: TensorFlow operation. - else_expression: TensorFlow operation. - ''' - x_shape = copy.copy(then_expression.get_shape()) - x = tf.cond(tf.cast(condition, 'bool'), - lambda: then_expression, - lambda: else_expression) - x.set_shape(x_shape) - return x + return tf.argmax(x, dimension=axis) # Extras -# ---------------------------------------- -def l2loss(params): - if len(params) == 0: - return tf.constant(0.0) - else: - return tf.add_n([sum(tf.square(p)) for p in params]) -def lrelu(x, leak=0.2): - f1 = 0.5 * (1 + leak) - f2 = 0.5 * (1 - leak) - return f1 * x + f2 * abs(x) -def categorical_sample_logits(X): - # https://github.com/tensorflow/tensorflow/issues/456 - U = tf.random_uniform(tf.shape(X)) - return argmax(X - tf.log(-tf.log(U)), axis=1) -# ================================================================ + +def l2loss(params): + if len(params) == 0: + return tf.constant(0.0) + else: + return tf.add_n([sum(tf.square(p)) for p in params]) + + +def lrelu(x, leak=0.2): + f1 = 0.5 * (1 + leak) + f2 = 0.5 * (1 - leak) + return f1 * x + f2 * abs(x) + + +def categorical_sample_logits(X): + # https://github.com/tensorflow/tensorflow/issues/456 + U = tf.random_uniform(tf.shape(X)) + return argmax(X - tf.log(-tf.log(U)), axis=1) + # Global session -# ================================================================ + def get_session(): - return tf.get_default_session() + return tf.get_default_session() + def single_threaded_session(): - tf_config = tf.ConfigProto( - inter_op_parallelism_threads=1, - intra_op_parallelism_threads=1) - return tf.Session(config=tf_config) + tf_config = tf.ConfigProto(inter_op_parallelism_threads=1, + intra_op_parallelism_threads=1) + return tf.Session(config=tf_config) + ALREADY_INITIALIZED = set() + + def initialize(): - new_variables = set(tf.global_variables()) - ALREADY_INITIALIZED - get_session().run(tf.variables_initializer(new_variables)) - ALREADY_INITIALIZED.update(new_variables) + new_variables = set(tf.global_variables()) - ALREADY_INITIALIZED + get_session().run(tf.variables_initializer(new_variables)) + ALREADY_INITIALIZED.update(new_variables) def eval(expr, feed_dict=None): - if feed_dict is None: feed_dict = {} - return get_session().run(expr, feed_dict=feed_dict) + if feed_dict is None: + feed_dict = {} + return get_session().run(expr, feed_dict=feed_dict) + def set_value(v, val): - get_session().run(v.assign(val)) + get_session().run(v.assign(val)) + def load_state(fname): - saver = tf.train.Saver() - saver.restore(get_session(), fname) + saver = tf.train.Saver() + saver.restore(get_session(), fname) + def save_state(fname): - os.makedirs(os.path.dirname(fname), exist_ok=True) - saver = tf.train.Saver() - saver.save(get_session(), fname) + os.makedirs(os.path.dirname(fname), exist_ok=True) + saver = tf.train.Saver() + saver.save(get_session(), fname) -# ================================================================ # Model components -# ================================================================ def normc_initializer(std=1.0): - def _initializer(shape, dtype=None, partition_info=None): #pylint: disable=W0613 - out = np.random.randn(*shape).astype(np.float32) - out *= std / np.sqrt(np.square(out).sum(axis=0, keepdims=True)) - return tf.constant(out) - return _initializer + def _initializer(shape, dtype=None, partition_info=None): + out = np.random.randn(*shape).astype(np.float32) + out *= std / np.sqrt(np.square(out).sum(axis=0, keepdims=True)) + return tf.constant(out) + return _initializer + def dense(x, size, name, weight_init=None, bias=True): - w = tf.get_variable(name + "/w", [x.get_shape()[1], size], initializer=weight_init) - ret = tf.matmul(x, w) - if bias: - b = tf.get_variable(name + "/b", [size], initializer=tf.zeros_initializer()) - return ret + b - else: - return ret + w = tf.get_variable(name + "/w", [x.get_shape()[1], size], + initializer=weight_init) + ret = tf.matmul(x, w) + if bias: + b = tf.get_variable(name + "/b", [size], + initializer=tf.zeros_initializer()) + return ret + b + else: + return ret -# ================================================================ # Basic Stuff -# ================================================================ + def function(inputs, outputs, updates=None, givens=None): - if isinstance(outputs, list): - return _Function(inputs, outputs, updates, givens=givens) - elif isinstance(outputs, dict): - f = _Function(inputs, outputs.values(), updates, givens=givens) - return lambda *inputs : dict(zip(outputs.keys(), f(*inputs))) - else: - f = _Function(inputs, [outputs], updates, givens=givens) - return lambda *inputs : f(*inputs)[0] + if isinstance(outputs, list): + return _Function(inputs, outputs, updates, givens=givens) + elif isinstance(outputs, dict): + f = _Function(inputs, outputs.values(), updates, givens=givens) + return lambda *inputs: dict(zip(outputs.keys(), f(*inputs))) + else: + f = _Function(inputs, [outputs], updates, givens=givens) + return lambda *inputs: f(*inputs)[0] + class _Function(object): - def __init__(self, inputs, outputs, updates, givens, check_nan=False): - assert all(len(i.op.inputs)==0 for i in inputs), "inputs should all be placeholders" - self.inputs = inputs - updates = updates or [] - self.update_group = tf.group(*updates) - self.outputs_update = list(outputs) + [self.update_group] - self.givens = {} if givens is None else givens - self.check_nan = check_nan - def __call__(self, *inputvals): - assert len(inputvals) == len(self.inputs) - feed_dict = dict(zip(self.inputs, inputvals)) - feed_dict.update(self.givens) - results = get_session().run(self.outputs_update, feed_dict=feed_dict)[:-1] - if self.check_nan: - if any(np.isnan(r).any() for r in results): - raise RuntimeError("Nan detected") - return results + def __init__(self, inputs, outputs, updates, givens, check_nan=False): + assert all(len(i.op.inputs) == 0 for i in inputs), ("inputs should all be " + "placeholders") + self.inputs = inputs + updates = updates or [] + self.update_group = tf.group(*updates) + self.outputs_update = list(outputs) + [self.update_group] + self.givens = {} if givens is None else givens + self.check_nan = check_nan + + def __call__(self, *inputvals): + assert len(inputvals) == len(self.inputs) + feed_dict = dict(zip(self.inputs, inputvals)) + feed_dict.update(self.givens) + results = get_session().run(self.outputs_update, feed_dict=feed_dict)[:-1] + if self.check_nan: + if any(np.isnan(r).any() for r in results): + raise RuntimeError("Nan detected") + return results + -# ================================================================ # Graph traversal -# ================================================================ VARIABLES = {} -# ================================================================ # Flat vectors -# ================================================================ + def var_shape(x): - out = [k.value for k in x.get_shape()] - assert all(isinstance(a, int) for a in out), \ - "shape function assumes that shape is fully known" - return out + out = [k.value for k in x.get_shape()] + assert all(isinstance(a, int) for a in out), ("shape function assumes that " + "shape is fully known") + return out + def numel(x): - return intprod(var_shape(x)) + return intprod(var_shape(x)) + def intprod(x): - return int(np.prod(x)) + return int(np.prod(x)) + def flatgrad(loss, var_list): - grads = tf.gradients(loss, var_list) - return tf.concat([tf.reshape(grad, [numel(v)], 0) - for (v, grad) in zip(var_list, grads)]) + grads = tf.gradients(loss, var_list) + return tf.concat([tf.reshape(grad, [numel(v)], 0) + for (v, grad) in zip(var_list, grads)]) + class SetFromFlat(object): - def __init__(self, var_list, dtype=tf.float32): - assigns = [] - shapes = list(map(var_shape, var_list)) - total_size = np.sum([intprod(shape) for shape in shapes]) + def __init__(self, var_list, dtype=tf.float32): + assigns = [] + shapes = list(map(var_shape, var_list)) + total_size = np.sum([intprod(shape) for shape in shapes]) + + self.theta = theta = tf.placeholder(dtype, [total_size]) + start = 0 + assigns = [] + for (shape, v) in zip(shapes, var_list): + size = intprod(shape) + assigns.append(tf.assign(v, tf.reshape(theta[start:start + size], + shape))) + start += size + assert start == total_size + self.op = tf.group(*assigns) + + def __call__(self, theta): + get_session().run(self.op, feed_dict={self.theta: theta}) - self.theta = theta = tf.placeholder(dtype,[total_size]) - start=0 - assigns = [] - for (shape,v) in zip(shapes,var_list): - size = intprod(shape) - assigns.append(tf.assign(v, tf.reshape(theta[start:start+size],shape))) - start+=size - assert start == total_size - self.op = tf.group(*assigns) - def __call__(self, theta): - get_session().run(self.op, feed_dict={self.theta:theta}) class GetFlat(object): - def __init__(self, var_list): - self.op = tf.concat([tf.reshape(v, [numel(v)]) for v in var_list], 0) - def __call__(self): - return get_session().run(self.op) + def __init__(self, var_list): + self.op = tf.concat([tf.reshape(v, [numel(v)]) for v in var_list], 0) + + def __call__(self): + return get_session().run(self.op) -# ================================================================ # Misc -# ================================================================ + def scope_vars(scope, trainable_only): - """ - Get variables inside a scope - The scope can be specified as a string - """ - return tf.get_collection( - tf.GraphKeys.TRAINABLE_VARIABLES if trainable_only else tf.GraphKeys.GLOBAL_VARIABLES, - scope=scope if isinstance(scope, str) else scope.name - ) + """Get variables inside a scope. The scope can be specified as a string.""" + return tf.get_collection((tf.GraphKeys.TRAINABLE_VARIABLES if trainable_only + else tf.GraphKeys.GLOBAL_VARIABLES), + scope=(scope if isinstance(scope, str) + else scope.name)) + def in_session(f): - @functools.wraps(f) - def newfunc(*args, **kwargs): - with tf.Session(): - f(*args, **kwargs) - return newfunc + @functools.wraps(f) + def newfunc(*args, **kwargs): + with tf.Session(): + f(*args, **kwargs) + return newfunc + + +# A mapping from name -> (placeholder, dtype, shape). +_PLACEHOLDER_CACHE = {} -_PLACEHOLDER_CACHE = {} # name -> (placeholder, dtype, shape) def get_placeholder(name, dtype, shape): - print("calling get_placeholder", name) - if name in _PLACEHOLDER_CACHE: - out, dtype1, shape1 = _PLACEHOLDER_CACHE[name] - assert dtype1==dtype and shape1==shape - return out - else: - out = tf.placeholder(dtype=dtype, shape=shape, name=name) - _PLACEHOLDER_CACHE[name] = (out,dtype,shape) - return out + print("calling get_placeholder", name) + if name in _PLACEHOLDER_CACHE: + out, dtype1, shape1 = _PLACEHOLDER_CACHE[name] + assert dtype1 == dtype and shape1 == shape + return out + else: + out = tf.placeholder(dtype=dtype, shape=shape, name=name) + _PLACEHOLDER_CACHE[name] = (out, dtype, shape) + return out + + def get_placeholder_cached(name): - return _PLACEHOLDER_CACHE[name][0] + return _PLACEHOLDER_CACHE[name][0] + def flattenallbut0(x): - return tf.reshape(x, [-1, intprod(x.get_shape().as_list()[1:])]) + return tf.reshape(x, [-1, intprod(x.get_shape().as_list()[1:])]) + def reset(): - global _PLACEHOLDER_CACHE - global VARIABLES - _PLACEHOLDER_CACHE = {} - VARIABLES = {} - tf.reset_default_graph() + global _PLACEHOLDER_CACHE + global VARIABLES + _PLACEHOLDER_CACHE = {} + VARIABLES = {} + tf.reset_default_graph() diff --git a/examples/hyperopt/hyperopt_adaptive.py b/examples/hyperopt/hyperopt_adaptive.py index c48d50b00..a2620e293 100644 --- a/examples/hyperopt/hyperopt_adaptive.py +++ b/examples/hyperopt/hyperopt_adaptive.py @@ -7,16 +7,24 @@ from collections import defaultdict import numpy as np import ray -import tensorflow as tf from tensorflow.examples.tutorials.mnist import input_data import objective -parser = argparse.ArgumentParser(description="Run the hyperparameter optimization example.") -parser.add_argument("--num-starting-segments", default=5, type=int, help="The number of training segments to start in parallel.") -parser.add_argument("--num-segments", default=10, type=int, help="The number of additional training segments to perform.") -parser.add_argument("--steps-per-segment", default=20, type=int, help="The number of steps of training to do per training segment.") -parser.add_argument("--redis-address", default=None, type=str, help="The Redis address of the cluster.") +parser = argparse.ArgumentParser(description="Run the hyperparameter " + "optimization example.") +parser.add_argument("--num-starting-segments", default=5, type=int, + help="The number of training segments to start in " + "parallel.") +parser.add_argument("--num-segments", default=10, type=int, + help="The number of additional training segments to " + "perform.") +parser.add_argument("--steps-per-segment", default=20, type=int, + help="The number of steps of training to do per training " + "segment.") +parser.add_argument("--redis-address", default=None, type=str, + help="The Redis address of the cluster.") + if __name__ == "__main__": args = parser.parse_args() @@ -51,7 +59,8 @@ if __name__ == "__main__": else: # The experiment is promising if the second half of the accuracies are # better than the first half of the accuracies. - return np.mean(accuracies[:len(accuracies) // 2]) < np.mean(accuracies[len(accuracies) // 2:]) + return (np.mean(accuracies[:len(accuracies) // 2]) < + np.mean(accuracies[len(accuracies) // 2:])) # Otherwise, continue running the experiment if it is in the top half of # experiments we've seen so far at this point in time. return np.mean(accuracy > np.array(comparable_accuracies)) > 0.5 diff --git a/examples/hyperopt/hyperopt_simple.py b/examples/hyperopt/hyperopt_simple.py index 9f8a9ae2f..03c6b81b5 100644 --- a/examples/hyperopt/hyperopt_simple.py +++ b/examples/hyperopt/hyperopt_simple.py @@ -6,15 +6,19 @@ import numpy as np import ray import argparse -import tensorflow as tf from tensorflow.examples.tutorials.mnist import input_data import objective -parser = argparse.ArgumentParser(description="Run the hyperparameter optimization example.") -parser.add_argument("--trials", default=2, type=int, help="The number of random trials to do.") -parser.add_argument("--steps", default=10, type=int, help="The number of steps of training to do per network.") -parser.add_argument("--redis-address", default=None, type=str, help="The Redis address of the cluster.") +parser = argparse.ArgumentParser(description="Run the hyperparameter " + "optimization example.") +parser.add_argument("--trials", default=2, type=int, + help="The number of random trials to do.") +parser.add_argument("--steps", default=10, type=int, + help="The number of steps of training to do per network.") +parser.add_argument("--redis-address", default=None, type=str, + help="The Redis address of the cluster.") + if __name__ == "__main__": args = parser.parse_args() diff --git a/examples/hyperopt/objective.py b/examples/hyperopt/objective.py index 6381a43a1..efee7510d 100644 --- a/examples/hyperopt/objective.py +++ b/examples/hyperopt/objective.py @@ -1,15 +1,15 @@ # Most of the tensorflow code is adapted from Tensorflow's tutorial on using # CNNs to train MNIST -# https://www.tensorflow.org/versions/r0.9/tutorials/mnist/pros/index.html#build-a-multilayer-convolutional-network. +# https://www.tensorflow.org/versions/r0.9/tutorials/mnist/pros/index.html#build-a-multilayer-convolutional-network. # noqa: E501 from __future__ import absolute_import from __future__ import division from __future__ import print_function import ray -import numpy as np import tensorflow as tf + def get_batch(data, batch_index, batch_size): # This method currently drops data when num_data is not divisible by # batch_size. @@ -18,19 +18,25 @@ def get_batch(data, batch_index, batch_size): batch_index %= num_batches return data[(batch_index * batch_size):((batch_index + 1) * batch_size)] + def weight(shape, stddev): initial = tf.truncated_normal(shape, stddev=stddev) return tf.Variable(initial) + def bias(shape): initial = tf.constant(0.1, shape=shape) return tf.Variable(initial) + def conv2d(x, W): return tf.nn.conv2d(x, W, strides=[1, 1, 1, 1], padding="SAME") + def max_pool_2x2(x): - return tf.nn.max_pool(x, ksize=[1, 2, 2, 1], strides=[1, 2, 2, 1], padding="SAME") + return tf.nn.max_pool(x, ksize=[1, 2, 2, 1], strides=[1, 2, 2, 1], + padding="SAME") + def cnn_setup(x, y, keep_prob, lr, stddev): first_hidden = 32 @@ -49,13 +55,16 @@ def cnn_setup(x, y, keep_prob, lr, stddev): b_fc1 = bias([fc_hidden]) h_pool2_flat = tf.reshape(h_pool2, [-1, 7 * 7 * second_hidden]) h_fc1 = tf.nn.relu(tf.matmul(h_pool2_flat, W_fc1) + b_fc1) - h_fc1_drop= tf.nn.dropout(h_fc1, keep_prob) + h_fc1_drop = tf.nn.dropout(h_fc1, keep_prob) W_fc2 = weight([fc_hidden, 10], stddev) b_fc2 = bias([10]) y_conv = tf.nn.softmax(tf.matmul(h_fc1_drop, W_fc2) + b_fc2) - cross_entropy = tf.reduce_mean(-tf.reduce_sum(y * tf.log(y_conv), reduction_indices=[1])) + cross_entropy = tf.reduce_mean(-tf.reduce_sum(y * tf.log(y_conv), + reduction_indices=[1])) correct_pred = tf.equal(tf.argmax(y_conv, 1), tf.argmax(y, 1)) - return tf.train.AdamOptimizer(lr).minimize(cross_entropy), tf.reduce_mean(tf.cast(correct_pred, tf.float32)), cross_entropy + return (tf.train.AdamOptimizer(lr).minimize(cross_entropy), + tf.reduce_mean(tf.cast(correct_pred, tf.float32)), cross_entropy) + # Define a remote function that takes a set of hyperparameters as well as the # data, consructs and trains a network, and returns the validation accuracy. @@ -75,11 +84,12 @@ def train_cnn_and_compute_accuracy(params, steps, train_images, train_labels, y = tf.placeholder(tf.float32, shape=[None, 10]) keep_prob = tf.placeholder(tf.float32) # Create the network. - train_step, accuracy, loss = cnn_setup(x, y, keep_prob, learning_rate, stddev) + train_step, accuracy, loss = cnn_setup(x, y, keep_prob, learning_rate, + stddev) # Do the training and evaluation. with tf.Session() as sess: - # Use the TensorFlowVariables utility. This is only necessary if we want to - # set and get the weights. + # Use the TensorFlowVariables utility. This is only necessary if we want + # to set and get the weights. variables = ray.experimental.TensorFlowVariables(loss, sess) # Initialize the network weights. sess.run(tf.global_variables_initializer()) @@ -92,7 +102,8 @@ def train_cnn_and_compute_accuracy(params, steps, train_images, train_labels, image_batch = get_batch(train_images, i, batch_size) label_batch = get_batch(train_labels, i, batch_size) # Do one step of training. - sess.run(train_step, feed_dict={x: image_batch, y: label_batch, keep_prob: keep}) + sess.run(train_step, feed_dict={x: image_batch, y: label_batch, + keep_prob: keep}) # Training is done, so compute the validation accuracy and the current # weights and return. totalacc = accuracy.eval(feed_dict={x: validation_images, diff --git a/examples/lbfgs/driver.py b/examples/lbfgs/driver.py index 6475208c6..e81c8b52d 100644 --- a/examples/lbfgs/driver.py +++ b/examples/lbfgs/driver.py @@ -10,6 +10,7 @@ import os from tensorflow.examples.tutorials.mnist import input_data + class LinearModel(object): """Simple class for a one layer neural network. @@ -44,21 +45,27 @@ class LinearModel(object): y = tf.nn.softmax(tf.matmul(x, w) + b) y_ = tf.placeholder(tf.float32, [None, shape[1]]) self.y_ = y_ - cross_entropy = tf.reduce_mean(-tf.reduce_sum(y_ * tf.log(y), reduction_indices=[1])) + cross_entropy = tf.reduce_mean(-tf.reduce_sum(y_ * tf.log(y), + reduction_indices=[1])) self.cross_entropy = cross_entropy self.cross_entropy_grads = tf.gradients(cross_entropy, [w, b]) self.sess = tf.Session() - # In order to get and set the weights, we pass in the loss function to Ray's - # TensorFlowVariables to automatically create methods to modify the weights. - self.variables = ray.experimental.TensorFlowVariables(cross_entropy, self.sess) + # In order to get and set the weights, we pass in the loss function to + # Ray's TensorFlowVariables to automatically create methods to modify the + # weights. + self.variables = ray.experimental.TensorFlowVariables(cross_entropy, + self.sess) def loss(self, xs, ys): """Computes the loss of the network.""" - return float(self.sess.run(self.cross_entropy, feed_dict={self.x: xs, self.y_: ys})) + return float(self.sess.run(self.cross_entropy, + feed_dict={self.x: xs, self.y_: ys})) def grad(self, xs, ys): """Computes the gradients of the network.""" - return self.sess.run(self.cross_entropy_grads, feed_dict={self.x: xs, self.y_: ys}) + return self.sess.run(self.cross_entropy_grads, + feed_dict={self.x: xs, self.y_: ys}) + @ray.remote class NetActor(object): @@ -85,17 +92,21 @@ class NetActor(object): def get_flat_size(self): return self.net.variables.get_flat_size() + # Compute the loss on the entire dataset. def full_loss(theta): theta_id = ray.put(theta) loss_ids = [actor.loss.remote(theta_id) for actor in actors] return sum(ray.get(loss_ids)) + # Compute the gradient of the loss on the entire dataset. def full_grad(theta): theta_id = ray.put(theta) grad_ids = [actor.grad.remote(theta_id) for actor in actors] - return sum(ray.get(grad_ids)).astype("float64") # This conversion is necessary for use with fmin_l_bfgs_b. + # The float64 conversion is necessary for use with fmin_l_bfgs_b. + return sum(ray.get(grad_ids)).astype("float64") + if __name__ == "__main__": ray.init(redirect_output=True) @@ -124,4 +135,5 @@ if __name__ == "__main__": # Use L-BFGS to minimize the loss function. print("Running L-BFGS.") - result = scipy.optimize.fmin_l_bfgs_b(full_loss, theta_init, maxiter=10, fprime=full_grad, disp=True) + result = scipy.optimize.fmin_l_bfgs_b(full_loss, theta_init, maxiter=10, + fprime=full_grad, disp=True) diff --git a/examples/policy_gradient/examples/example.py b/examples/policy_gradient/examples/example.py index 7b9a01ab7..cbf1d338b 100644 --- a/examples/policy_gradient/examples/example.py +++ b/examples/policy_gradient/examples/example.py @@ -5,7 +5,8 @@ from __future__ import print_function import argparse import ray -from reinforce.env import NoPreprocessor, AtariRamPreprocessor, AtariPixelPreprocessor +from reinforce.env import (NoPreprocessor, AtariRamPreprocessor, + AtariPixelPreprocessor) from reinforce.agent import Agent, RemoteAgent from reinforce.rollout import collect_samples from reinforce.utils import iterate, shuffle @@ -19,8 +20,10 @@ config = {"kl_coeff": 0.2, "kl_target": 0.01, "timesteps_per_batch": 40000} + if __name__ == "__main__": - parser = argparse.ArgumentParser(description="Run the policy gradient algorithm.") + parser = argparse.ArgumentParser(description="Run the policy gradient " + "algorithm.") parser.add_argument("--environment", default="Pong-v0", type=str, help="The gym environment to use.") parser.add_argument("--redis-address", default=None, type=str, @@ -47,7 +50,8 @@ if __name__ == "__main__": preprocessor = AtariPixelPreprocessor() print("Using the environment {}.".format(mdp_name)) - agents = [RemoteAgent.remote(mdp_name, 1, preprocessor, config, False) for _ in range(5)] + agents = [RemoteAgent.remote(mdp_name, 1, preprocessor, config, False) + for _ in range(5)] agent = Agent(mdp_name, 1, preprocessor, config, True) kl_coeff = config["kl_coeff"] @@ -56,26 +60,32 @@ if __name__ == "__main__": print("== iteration", j) weights = ray.put(agent.get_weights()) [a.load_weights.remote(weights) for a in agents] - trajectory, total_reward, traj_len_mean = collect_samples(agents, config["timesteps_per_batch"], 0.995, 1.0, 2000) + trajectory, total_reward, traj_len_mean = collect_samples( + agents, config["timesteps_per_batch"], 0.995, 1.0, 2000) print("total reward is ", total_reward) print("trajectory length mean is ", traj_len_mean) print("timesteps: ", trajectory["dones"].shape[0]) - trajectory["advantages"] = (trajectory["advantages"] - trajectory["advantages"].mean()) / trajectory["advantages"].std() - print("Computing policy (optimizer='" + agent.optimizer.get_name() + "', iterations=" + str(config["num_sgd_iter"]) + ", stepsize=" + str(config["sgd_stepsize"]) + "):") + trajectory["advantages"] = ((trajectory["advantages"] - + trajectory["advantages"].mean()) / + trajectory["advantages"].std()) + print("Computing policy (optimizer='" + agent.optimizer.get_name() + + "', iterations=" + str(config["num_sgd_iter"]) + + ", stepsize=" + str(config["sgd_stepsize"]) + "):") names = ["iter", "loss", "kl", "entropy"] print(("{:>15}" * len(names)).format(*names)) trajectory = shuffle(trajectory) ppo = agent.ppo for i in range(config["num_sgd_iter"]): - # Test on current set of rollouts - loss, kl, entropy = agent.sess.run([ppo.loss, ppo.mean_kl, ppo.mean_entropy], - feed_dict={ppo.observations: trajectory["observations"], - ppo.advantages: trajectory["advantages"], - ppo.actions: trajectory["actions"].squeeze(), - ppo.prev_logits: trajectory["logprobs"], - ppo.kl_coeff: kl_coeff}) + # Test on current set of rollouts. + loss, kl, entropy = agent.sess.run( + [ppo.loss, ppo.mean_kl, ppo.mean_entropy], + feed_dict={ppo.observations: trajectory["observations"], + ppo.advantages: trajectory["advantages"], + ppo.actions: trajectory["actions"].squeeze(), + ppo.prev_logits: trajectory["logprobs"], + ppo.kl_coeff: kl_coeff}) print("{:>15}{:15.5e}{:15.5e}{:15.5e}".format(i, loss, kl, entropy)) - # Run SGD for training on current set of rollouts + # Run SGD for training on current set of rollouts. for batch in iterate(trajectory, config["sgd_batchsize"]): agent.sess.run([agent.train_op], feed_dict={ppo.observations: batch["observations"], diff --git a/examples/policy_gradient/reinforce/agent.py b/examples/policy_gradient/reinforce/agent.py index 0ae0d937d..c8f4254b9 100644 --- a/examples/policy_gradient/reinforce/agent.py +++ b/examples/policy_gradient/reinforce/agent.py @@ -12,8 +12,8 @@ from reinforce.policy import ProximalPolicyLoss from reinforce.filter import MeanStdFilter from reinforce.rollout import rollouts, add_advantage_values -class Agent(object): +class Agent(object): def __init__(self, name, batchsize, preprocessor, config, use_gpu): if not use_gpu: os.environ["CUDA_VISIBLE_DEVICES"] = "" @@ -21,10 +21,13 @@ class Agent(object): if preprocessor.shape is None: preprocessor.shape = self.env.observation_space.shape self.sess = tf.Session() - self.ppo = ProximalPolicyLoss(self.env.observation_space, self.env.action_space, preprocessor, config, self.sess) + self.ppo = ProximalPolicyLoss(self.env.observation_space, + self.env.action_space, preprocessor, config, + self.sess) self.optimizer = tf.train.AdamOptimizer(config["sgd_stepsize"]) self.train_op = self.optimizer.minimize(self.ppo.loss) - self.variables = ray.experimental.TensorFlowVariables(self.ppo.loss, self.sess) + self.variables = ray.experimental.TensorFlowVariables(self.ppo.loss, + self.sess) self.observation_filter = MeanStdFilter(preprocessor.shape, clip=None) self.reward_filter = MeanStdFilter((), clip=5.0) self.sess.run(tf.global_variables_initializer()) @@ -36,8 +39,10 @@ class Agent(object): self.variables.set_weights(weights) def compute_trajectory(self, gamma, lam, horizon): - trajectory = rollouts(self.ppo, self.env, horizon, self.observation_filter, self.reward_filter) + trajectory = rollouts(self.ppo, self.env, horizon, self.observation_filter, + self.reward_filter) add_advantage_values(trajectory, gamma, lam, self.reward_filter) return trajectory + RemoteAgent = ray.remote(Agent) diff --git a/examples/policy_gradient/reinforce/distributions.py b/examples/policy_gradient/reinforce/distributions.py index b4377c507..20e6f1741 100644 --- a/examples/policy_gradient/reinforce/distributions.py +++ b/examples/policy_gradient/reinforce/distributions.py @@ -5,54 +5,65 @@ from __future__ import print_function import tensorflow as tf import numpy as np -class Categorical(object): +class Categorical(object): def __init__(self, logits): self.logits = logits def logp(self, x): - return -tf.nn.sparse_softmax_cross_entropy_with_logits(logits=self.logits, labels=x) + return -tf.nn.sparse_softmax_cross_entropy_with_logits(logits=self.logits, + labels=x) def entropy(self): - a0 = self.logits - tf.reduce_max(self.logits, reduction_indices=[1], keep_dims=True) + a0 = self.logits - tf.reduce_max(self.logits, reduction_indices=[1], + keep_dims=True) ea0 = tf.exp(a0) z0 = tf.reduce_sum(ea0, reduction_indices=[1], keep_dims=True) p0 = ea0 / z0 return tf.reduce_sum(p0 * (tf.log(z0) - a0), reduction_indices=[1]) def kl(self, other): - a0 = self.logits - tf.reduce_max(self.logits, reduction_indices=[1], keep_dims=True) - a1 = other.logits - tf.reduce_max(other.logits, reduction_indices=[1], keep_dims=True) + a0 = self.logits - tf.reduce_max(self.logits, reduction_indices=[1], + keep_dims=True) + a1 = other.logits - tf.reduce_max(other.logits, reduction_indices=[1], + keep_dims=True) ea0 = tf.exp(a0) ea1 = tf.exp(a1) z0 = tf.reduce_sum(ea0, reduction_indices=[1], keep_dims=True) z1 = tf.reduce_sum(ea1, reduction_indices=[1], keep_dims=True) p0 = ea0 / z0 - return tf.reduce_sum(p0 * (a0 - tf.log(z0) - a1 + tf.log(z1)), reduction_indices=[1]) + return tf.reduce_sum(p0 * (a0 - tf.log(z0) - a1 + tf.log(z1)), + reduction_indices=[1]) def sample(self): return tf.multinomial(self.logits, 1) + class DiagGaussian(object): + def __init__(self, flat): + self.flat = flat + mean, logstd = tf.split(1, 2, flat) + self.mean = mean + self.logstd = logstd + self.std = tf.exp(logstd) - def __init__(self, flat): - self.flat = flat - mean, logstd = tf.split(1, 2, flat) - self.mean = mean - self.logstd = logstd - self.std = tf.exp(logstd) + def logp(self, x): + return (-0.5 * tf.reduce_sum(tf.square((x - self.mean) / self.std), + reduction_indices=[1]) - + 0.5 * np.log(2.0 * np.pi) * tf.to_float(tf.shape(x)[1]) - + tf.reduce_sum(self.logstd, reduction_indices=[1])) - def logp(self, x): - return - 0.5 * tf.reduce_sum(tf.square((x - self.mean) / self.std), reduction_indices=[1]) \ - - 0.5 * np.log(2.0 * np.pi) * tf.to_float(tf.shape(x)[1]) \ - - tf.reduce_sum(self.logstd, reduction_indices=[1]) + def kl(self, other): + assert isinstance(other, DiagGaussian) + return tf.reduce_sum(other.logstd - self.logstd + + (tf.square(self.std) + + tf.square(self.mean - other.mean)) / + (2.0 * tf.square(other.std)) - 0.5, + reduction_indices=[1]) - def kl(self, other): - assert isinstance(other, DiagGaussian) - return tf.reduce_sum(other.logstd - self.logstd + (tf.square(self.std) + tf.square(self.mean - other.mean)) / (2.0 * tf.square(other.std)) - 0.5, reduction_indices=[1]) + def entropy(self): + return tf.reduce_sum(self.logstd + .5 * np.log(2.0 * np.pi * np.e), + reduction_indices=[1]) - def entropy(self): - return tf.reduce_sum(self.logstd + .5 * np.log(2.0 * np.pi * np.e), reduction_indices=[1]) - - def sample(self): - return self.mean + self.std * tf.random_normal(tf.shape(self.mean)) + def sample(self): + return self.mean + self.std * tf.random_normal(tf.shape(self.mean)) diff --git a/examples/policy_gradient/reinforce/env.py b/examples/policy_gradient/reinforce/env.py index 2acd4f8f7..6ddb98ee3 100644 --- a/examples/policy_gradient/reinforce/env.py +++ b/examples/policy_gradient/reinforce/env.py @@ -5,34 +5,34 @@ from __future__ import print_function import gym import numpy as np -class AtariPixelPreprocessor(object): +class AtariPixelPreprocessor(object): def __init__(self): self.shape = (80, 80, 3) def __call__(self, observation): "Convert images from (210, 160, 3) to (3, 80, 80) by downsampling." - return (observation[25:-25:2,::2,:][None] - 128.0) / 128.8 + return (observation[25:-25:2, ::2, :][None] - 128) / 128 + class AtariRamPreprocessor(object): - def __init__(self): self.shape = (128,) def __call__(self, observation): - return (observation - 128.0) / 128.0 + return (observation - 128) / 128 + class NoPreprocessor(object): - def __init__(self): self.shape = None def __call__(self, observation): return observation -class BatchedEnv(object): - "A BatchedEnv holds multiple gym enviroments and performs steps on all of them." +class BatchedEnv(object): + """This holds multiple gym enviroments and performs steps on all of them.""" def __init__(self, name, batchsize, preprocessor=None): self.envs = [gym.make(name) for _ in range(batchsize)] self.observation_space = self.envs[0].observation_space @@ -54,10 +54,12 @@ class BatchedEnv(object): observations.append(np.zeros(self.shape)) rewards.append(0.0) continue - observation, reward, done, info = self.envs[i].step(action if len(action) > 1 else action[0]) + observation, reward, done, info = self.envs[i].step( + action if len(action) > 1 else action[0]) if render: self.envs[0].render() observations.append(self.preprocessor(observation)) rewards.append(reward) self.dones[i] = done - return np.vstack(observations), np.array(rewards, dtype="float32"), np.array(self.dones) + return (np.vstack(observations), np.array(rewards, dtype="float32"), + np.array(self.dones)) diff --git a/examples/policy_gradient/reinforce/filter.py b/examples/policy_gradient/reinforce/filter.py index 0b807b3d2..de3f782df 100644 --- a/examples/policy_gradient/reinforce/filter.py +++ b/examples/policy_gradient/reinforce/filter.py @@ -2,17 +2,17 @@ from __future__ import absolute_import from __future__ import division from __future__ import print_function -import warnings import numpy as np -class NoFilter(object): +class NoFilter(object): def __init__(self): pass def __call__(self, x, update=True): return np.asarray(x) + # http://www.johndcook.com/blog/standard_deviation/ class RunningStat(object): @@ -24,7 +24,8 @@ class RunningStat(object): def push(self, x): x = np.asarray(x) # Unvectorized update of the running statistics. - assert x.shape == self._M.shape, "x.shape = {}, self.shape = {}".format(x.shape, self._M.shape) + assert x.shape == self._M.shape, ("x.shape = {}, self.shape = {}" + .format(x.shape, self._M.shape)) n1 = self._n self._n += 1 if self._n == 1: @@ -56,7 +57,7 @@ class RunningStat(object): @property def var(self): - return self._S/(self._n - 1) if self._n > 1 else np.square(self._M) + return self._S / (self._n - 1) if self._n > 1 else np.square(self._M) @property def std(self): @@ -66,12 +67,8 @@ class RunningStat(object): def shape(self): return self._M.shape -class MeanStdFilter(object): - """ - y = (x-mean)/std - using running estimates of mean,std - """ +class MeanStdFilter(object): def __init__(self, shape, demean=True, destd=True, clip=10.0): self.demean = demean self.destd = destd @@ -92,7 +89,7 @@ class MeanStdFilter(object): if self.demean: x = x - self.rs.mean if self.destd: - x = x / (self.rs.std+1e-8) + x = x / (self.rs.std + 1e-8) if self.clip: if np.amin(x) < -self.clip or np.amax(x) > self.clip: print("Clipping value to " + str(self.clip)) @@ -101,7 +98,7 @@ class MeanStdFilter(object): def test_running_stat(): - for shp in ((), (3,), (3,4)): + for shp in ((), (3,), (3, 4)): li = [] rs = RunningStat(shp) for _ in range(5): @@ -113,8 +110,9 @@ def test_running_stat(): v = np.square(m) if (len(li) == 1) else np.var(li, ddof=1, axis=0) assert np.allclose(rs.var, v) + def test_combining_stat(): - for shape in [(), (3,), (3,4)]: + for shape in [(), (3,), (3, 4)]: li = [] rs1 = RunningStat(shape) rs2 = RunningStat(shape) @@ -132,5 +130,6 @@ def test_combining_stat(): assert np.allclose(rs.mean, rs1.mean) assert np.allclose(rs.std, rs1.std) + test_running_stat() test_combining_stat() diff --git a/examples/policy_gradient/reinforce/models/fcnet.py b/examples/policy_gradient/reinforce/models/fcnet.py index da0050a81..51a4bb7d7 100644 --- a/examples/policy_gradient/reinforce/models/fcnet.py +++ b/examples/policy_gradient/reinforce/models/fcnet.py @@ -7,20 +7,31 @@ import tensorflow.contrib.slim as slim import numpy as np + def normc_initializer(std=1.0): - def _initializer(shape, dtype=None, partition_info=None): #pylint: disable=W0613 - out = np.random.randn(*shape).astype(np.float32) - out *= std / np.sqrt(np.square(out).sum(axis=0, keepdims=True)) - return tf.constant(out) - return _initializer + def _initializer(shape, dtype=None, partition_info=None): + out = np.random.randn(*shape).astype(np.float32) + out *= std / np.sqrt(np.square(out).sum(axis=0, keepdims=True)) + return tf.constant(out) + return _initializer + def fc_net(inputs, num_classes=10, logstd=False): - fc1 = slim.fully_connected(inputs, 128, weights_initializer=normc_initializer(1.0), scope="fc1") - fc2 = slim.fully_connected(fc1, 128, weights_initializer=normc_initializer(1.0), scope="fc2") - fc3 = slim.fully_connected(fc2, 128, weights_initializer=normc_initializer(1.0), scope="fc3") - fc4 = slim.fully_connected(fc3, num_classes, weights_initializer=normc_initializer(0.01), activation_fn=None, scope="fc4") + fc1 = slim.fully_connected(inputs, 128, + weights_initializer=normc_initializer(1.0), + scope="fc1") + fc2 = slim.fully_connected(fc1, 128, + weights_initializer=normc_initializer(1.0), + scope="fc2") + fc3 = slim.fully_connected(fc2, 128, + weights_initializer=normc_initializer(1.0), + scope="fc3") + fc4 = slim.fully_connected(fc3, num_classes, + weights_initializer=normc_initializer(0.01), + activation_fn=None, scope="fc4") if logstd: - logstd = tf.get_variable(name="logstd", shape=[num_classes], initializer=tf.zeros_initializer) + logstd = tf.get_variable(name="logstd", shape=[num_classes], + initializer=tf.zeros_initializer) return tf.concat(1, [fc4, logstd]) else: return fc4 diff --git a/examples/policy_gradient/reinforce/models/visionnet.py b/examples/policy_gradient/reinforce/models/visionnet.py index 569eb322b..c3e240eb1 100644 --- a/examples/policy_gradient/reinforce/models/visionnet.py +++ b/examples/policy_gradient/reinforce/models/visionnet.py @@ -5,9 +5,11 @@ from __future__ import print_function import tensorflow as tf import tensorflow.contrib.slim as slim + def vision_net(inputs, num_classes=10): conv1 = slim.conv2d(inputs, 16, [8, 8], 4, scope="conv1") conv2 = slim.conv2d(conv1, 32, [4, 4], 2, scope="conv2") fc1 = slim.conv2d(conv2, 512, [10, 10], padding="VALID", scope="fc1") - fc2 = slim.conv2d(fc1, num_classes, [1, 1], activation_fn=None, normalizer_fn=None, scope="fc2") + fc2 = slim.conv2d(fc1, num_classes, [1, 1], activation_fn=None, + normalizer_fn=None, scope="fc2") return tf.squeeze(fc2, [1, 2]) diff --git a/examples/policy_gradient/reinforce/policy.py b/examples/policy_gradient/reinforce/policy.py index 11ce7f8e3..2e73a90ec 100644 --- a/examples/policy_gradient/reinforce/policy.py +++ b/examples/policy_gradient/reinforce/policy.py @@ -4,25 +4,30 @@ from __future__ import print_function import gym.spaces import tensorflow as tf -import tensorflow.contrib.slim as slim from reinforce.models.visionnet import vision_net from reinforce.models.fcnet import fc_net from reinforce.distributions import Categorical, DiagGaussian + class ProximalPolicyLoss(object): - def __init__(self, observation_space, action_space, preprocessor, config, sess): - assert isinstance(action_space, gym.spaces.Discrete) or isinstance(action_space, gym.spaces.Box) - # adapting the kl divergence + def __init__(self, observation_space, action_space, preprocessor, config, + sess): + assert (isinstance(action_space, gym.spaces.Discrete) or + isinstance(action_space, gym.spaces.Box)) + # Adapting the kl divergence. self.kl_coeff = tf.placeholder(name="newkl", shape=(), dtype=tf.float32) - self.observations = tf.placeholder(tf.float32, shape=(None,) + preprocessor.shape) + self.observations = tf.placeholder(tf.float32, + shape=(None,) + preprocessor.shape) self.advantages = tf.placeholder(tf.float32, shape=(None,)) if isinstance(action_space, gym.spaces.Box): - # First half of the dimensions are the means, the second half are the standard deviations + # The first half of the dimensions are the means, the second half are the + # standard deviations. self.action_dim = action_space.shape[0] self.logit_dim = 2 * self.action_dim - self.actions = tf.placeholder(tf.float32, shape=(None, action_space.shape[0])) + self.actions = tf.placeholder(tf.float32, + shape=(None, action_space.shape[0])) Distribution = DiagGaussian elif isinstance(action_space, gym.spaces.Discrete): self.action_dim = action_space.n @@ -30,11 +35,13 @@ class ProximalPolicyLoss(object): self.actions = tf.placeholder(tf.int64, shape=(None,)) Distribution = Categorical else: - raise NotImplemented("action space" + str(type(env.action_space)) + "currently not supported") + raise NotImplemented("action space" + str(type(action_space)) + + "currently not supported") self.prev_logits = tf.placeholder(tf.float32, shape=(None, self.logit_dim)) self.prev_dist = Distribution(self.prev_logits) if len(observation_space.shape) > 1: - self.curr_logits = vision_net(self.observations, num_classes=self.logit_dim) + self.curr_logits = vision_net(self.observations, + num_classes=self.logit_dim) else: assert len(observation_space.shape) == 1 self.curr_logits = fc_net(self.observations, num_classes=self.logit_dim) @@ -42,18 +49,22 @@ class ProximalPolicyLoss(object): self.sampler = self.curr_dist.sample() self.entropy = self.curr_dist.entropy() # Make loss functions. - self.ratio = tf.exp(self.curr_dist.logp(self.actions) - self.prev_dist.logp(self.actions)) + self.ratio = tf.exp(self.curr_dist.logp(self.actions) - + self.prev_dist.logp(self.actions)) self.kl = self.prev_dist.kl(self.curr_dist) self.mean_kl = tf.reduce_mean(self.kl) self.mean_entropy = tf.reduce_mean(self.entropy) self.surr1 = self.ratio * self.advantages - self.surr2 = tf.clip_by_value(self.ratio, 1 - config["clip_param"], 1 + config["clip_param"]) * self.advantages + self.surr2 = tf.clip_by_value(self.ratio, 1 - config["clip_param"], + 1 + config["clip_param"]) * self.advantages self.surr = tf.minimum(self.surr1, self.surr2) - self.loss = tf.reduce_mean(-self.surr + self.kl_coeff * self.kl - config["entropy_coeff"] * self.entropy) + self.loss = tf.reduce_mean(-self.surr + self.kl_coeff * self.kl - + config["entropy_coeff"] * self.entropy) self.sess = sess def compute_actions(self, observations): - return self.sess.run([self.sampler, self.curr_logits], feed_dict={self.observations: observations}) + return self.sess.run([self.sampler, self.curr_logits], + feed_dict={self.observations: observations}) def loss(self): return self.loss diff --git a/examples/policy_gradient/reinforce/rollout.py b/examples/policy_gradient/reinforce/rollout.py index 1c23d8f24..63a43c11b 100644 --- a/examples/policy_gradient/reinforce/rollout.py +++ b/examples/policy_gradient/reinforce/rollout.py @@ -8,7 +8,9 @@ import ray from reinforce.filter import NoFilter from reinforce.utils import flatten, concatenate -def rollouts(policy, env, horizon, observation_filter=NoFilter(), reward_filter=NoFilter()): + +def rollouts(policy, env, horizon, observation_filter=NoFilter(), + reward_filter=NoFilter()): """Perform a batch of rollouts of a policy in an environment. Args: @@ -23,15 +25,15 @@ def rollouts(policy, env, horizon, observation_filter=NoFilter(), reward_filter= Returns: A trajectory, which is a dictionary with keys "observations", "rewards", - "orig_rewards", "actions", "logprobs", "dones". Each value is an array of - shape (num_timesteps, env.batchsize, shape). + "orig_rewards", "actions", "logprobs", "dones". Each value is an array of + shape (num_timesteps, env.batchsize, shape). """ observation = observation_filter(env.reset()) done = np.array(env.batchsize * [False]) t = 0 observations = [] - raw_rewards = [] # Empirical rewards + raw_rewards = [] # Empirical rewards actions = [] logprobs = [] dones = [] @@ -53,6 +55,7 @@ def rollouts(policy, env, horizon, observation_filter=NoFilter(), reward_filter= "logprobs": np.vstack(logprobs), "dones": np.vstack(dones)} + def add_advantage_values(trajectory, gamma, lam, reward_filter): rewards = trajectory["raw_rewards"] dones = trajectory["dones"] @@ -60,32 +63,41 @@ def add_advantage_values(trajectory, gamma, lam, reward_filter): last_advantage = np.zeros(rewards.shape[1], dtype="float32") for t in reversed(range(len(rewards))): - delta = rewards[t,:] * (1 - dones[t,:]) + delta = rewards[t, :] * (1 - dones[t, :]) last_advantage = delta + gamma * lam * last_advantage - advantages[t,:] = last_advantage - reward_filter(advantages[t,:]) + advantages[t, :] = last_advantage + reward_filter(advantages[t, :]) trajectory["advantages"] = advantages + @ray.remote -def compute_trajectory(policy, env, gamma, lam, horizon, observation_filter, reward_filter): - trajectory = rollouts(policy, env, horizon, observation_filter, reward_filter) +def compute_trajectory(policy, env, gamma, lam, horizon, observation_filter, + reward_filter): + trajectory = rollouts(policy, env, horizon, observation_filter, + reward_filter) add_advantage_values(trajectory, gamma, lam, reward_filter) return trajectory -def collect_samples(agents, num_timesteps, gamma, lam, horizon, observation_filter=NoFilter(), reward_filter=NoFilter()): + +def collect_samples(agents, num_timesteps, gamma, lam, horizon, + observation_filter=NoFilter(), reward_filter=NoFilter()): num_timesteps_so_far = 0 trajectories = [] total_rewards = [] traj_len_means = [] while num_timesteps_so_far < num_timesteps: - trajectory_batch = ray.get([agent.compute_trajectory.remote(gamma, lam, horizon) for agent in agents]) + trajectory_batch = ray.get( + [agent.compute_trajectory.remote(gamma, lam, horizon) + for agent in agents]) trajectory = concatenate(trajectory_batch) - total_rewards.append(trajectory["raw_rewards"].sum(axis=0).mean() / len(agents)) + total_rewards.append( + trajectory["raw_rewards"].sum(axis=0).mean() / len(agents)) trajectory = flatten(trajectory) not_done = np.logical_not(trajectory["dones"]) traj_len_means.append(not_done.sum(axis=0).mean() / len(agents)) trajectory = {key: val[not_done] for key, val in trajectory.items()} num_timesteps_so_far += len(trajectory["dones"]) trajectories.append(trajectory) - return concatenate(trajectories), np.mean(total_rewards), np.mean(traj_len_means) + return (concatenate(trajectories), np.mean(total_rewards), + np.mean(traj_len_means)) diff --git a/examples/policy_gradient/reinforce/tfutils.py b/examples/policy_gradient/reinforce/tfutils.py deleted file mode 100644 index 5fed1a74e..000000000 --- a/examples/policy_gradient/reinforce/tfutils.py +++ /dev/null @@ -1,30 +0,0 @@ -from __future__ import absolute_import -from __future__ import division -from __future__ import print_function - -import tensorflow as tf -import threading - -class DataQueue(object): - - def __init__(self, placeholder_dict): - """Here, placeholder_dict is an OrderedDict.""" - placeholders = placeholder_dict.values() - shapes = [placeholder.get_shape()[1:].as_list() for placeholder in placeholders] - types = [placeholder.dtype for placeholder in placeholders] - self.queue = tf.RandomShuffleQueue(shapes=shapes, dtypes=dtypes, capacity=2000, min_after_dequeue=1000) - self.enqueue_op = self.queue.enqueue_many(placeholders) - - def thread_main(self, sess, data_iterator): - for data in data_iterator: - feed_dict = {placeholder: data[name] for (name, placeholder) in placeholder_dict} - sess.run(self.enqueue_op, feed_dict=feed_dict) - - def start_thread(self, sess, data_iterator, num_threads=1): - threads = [] - for n in range(num_thread): - t = threading.Thread(target=self.train_main, args=(sess, data_iterator)) - t.daemon = True # Thread will close when parent quits - t.start() - threads.append(t) - return threads diff --git a/examples/policy_gradient/reinforce/utils.py b/examples/policy_gradient/reinforce/utils.py index 21fe93dc5..dbc2deca6 100644 --- a/examples/policy_gradient/reinforce/utils.py +++ b/examples/policy_gradient/reinforce/utils.py @@ -4,6 +4,7 @@ from __future__ import print_function import numpy as np + def flatten(weights, start=0, stop=2): """This methods reshapes all values in a dictionary. @@ -19,6 +20,7 @@ def flatten(weights, start=0, stop=2): weights[key] = val.reshape(new_shape) return weights + def concatenate(weights_list): keys = weights_list[0].keys() result = {} @@ -26,12 +28,14 @@ def concatenate(weights_list): result[key] = np.concatenate([l[key] for l in weights_list]) return result + def shuffle(trajectory): permutation = np.random.permutation(trajectory["dones"].shape[0]) for key, val in trajectory.items(): trajectory[key] = val[permutation][permutation] return trajectory + def iterate(trajectory, batchsize): trajectory = shuffle(trajectory) curr_index = 0 @@ -39,6 +43,6 @@ def iterate(trajectory, batchsize): while curr_index + batchsize < trajectory["dones"].shape[0]: batch = dict() for key in trajectory: - batch[key] = trajectory[key][curr_index:curr_index+batchsize] + batch[key] = trajectory[key][curr_index:(curr_index + batchsize)] curr_index += batchsize yield batch diff --git a/examples/policy_gradient/tests/test.py b/examples/policy_gradient/tests/test.py index 988c15061..c1db6ddc5 100644 --- a/examples/policy_gradient/tests/test.py +++ b/examples/policy_gradient/tests/test.py @@ -10,6 +10,7 @@ from numpy.testing import assert_allclose from reinforce.distributions import Categorical from reinforce.utils import flatten, concatenate + class DistibutionsTest(unittest.TestCase): def testCategorical(self): @@ -28,6 +29,7 @@ class DistibutionsTest(unittest.TestCase): probs = np.exp(z) / np.sum(np.exp(z)) self.assertTrue(np.sum(np.abs(probs - counts / num_samples)) <= 0.01) + class UtilsTest(unittest.TestCase): def testFlatten(self): @@ -54,5 +56,6 @@ class UtilsTest(unittest.TestCase): assert_allclose(D["s"], np.array([0, 1, 4, 5])) assert_allclose(D["a"], np.array([2, 3, 6, 7])) + if __name__ == "__main__": unittest.main(verbosity=2) diff --git a/examples/resnet/cifar_input.py b/examples/resnet/cifar_input.py index 6e251c5de..3775b656e 100644 --- a/examples/resnet/cifar_input.py +++ b/examples/resnet/cifar_input.py @@ -6,9 +6,9 @@ from __future__ import absolute_import from __future__ import division from __future__ import print_function -import numpy as np import tensorflow as tf + def build_data(data_path, size, dataset): """Creates the queue and preprocessing operations for the dataset. @@ -21,14 +21,12 @@ def build_data(data_path, size, dataset): queue: A Tensorflow queue for extracting the images and labels. """ image_size = 32 - if dataset == 'cifar10': + if dataset == "cifar10": label_bytes = 1 label_offset = 0 - num_classes = 10 - elif dataset == 'cifar100': + elif dataset == "cifar100": label_bytes = 1 label_offset = 1 - num_classes = 100 depth = 3 image_bytes = image_size * image_size * depth record_bytes = label_bytes + label_offset + image_bytes @@ -50,6 +48,7 @@ def build_data(data_path, size, dataset): queue = tf.train.shuffle_batch([image, label], size, size, 0, num_threads=16) return queue + def build_input(data, batch_size, dataset, train): """Build CIFAR image and labels. @@ -69,23 +68,25 @@ def build_input(data, batch_size, dataset, train): labels_constant = tf.constant(data[1]) image_size = 32 depth = 3 - num_classes = 10 if dataset == 'cifar10' else 100 - image, label = tf.train.slice_input_producer([images_constant, labels_constant], capacity=16 * batch_size) + num_classes = 10 if dataset == "cifar10" else 100 + image, label = tf.train.slice_input_producer([images_constant, + labels_constant], + capacity=16 * batch_size) if train: - image = tf.image.resize_image_with_crop_or_pad( - image, image_size+4, image_size+4) + image = tf.image.resize_image_with_crop_or_pad(image, image_size + 4, + image_size + 4) image = tf.random_crop(image, [image_size, image_size, 3]) image = tf.image.random_flip_left_right(image) image = tf.image.per_image_standardization(image) example_queue = tf.RandomShuffleQueue( - capacity=16 * batch_size, - min_after_dequeue=8 * batch_size, - dtypes=[tf.float32, tf.int32], - shapes=[[image_size, image_size, depth], [1]]) + capacity=16 * batch_size, + min_after_dequeue=8 * batch_size, + dtypes=[tf.float32, tf.int32], + shapes=[[image_size, image_size, depth], [1]]) num_threads = 16 else: - image = tf.image.resize_image_with_crop_or_pad( - image, image_size, image_size) + image = tf.image.resize_image_with_crop_or_pad(image, image_size, + image_size) image = tf.image.per_image_standardization(image) example_queue = tf.FIFOQueue( 3 * batch_size, @@ -97,7 +98,7 @@ def build_input(data, batch_size, dataset, train): tf.train.add_queue_runner(tf.train.queue_runner.QueueRunner( example_queue, [example_enqueue_op] * num_threads)) - # Read 'batch' labels + images from the example queue. + # Read "batch" labels + images from the example queue. images, labels = example_queue.dequeue_many(batch_size) labels = tf.reshape(labels, [batch_size, 1]) indices = tf.reshape(tf.range(0, batch_size, 1), [batch_size, 1]) @@ -112,5 +113,5 @@ def build_input(data, batch_size, dataset, train): assert labels.get_shape()[0] == batch_size assert labels.get_shape()[1] == num_classes if not train: - tf.summary.image('images', images) + tf.summary.image("images", images) return images, labels diff --git a/examples/resnet/resnet_main.py b/examples/resnet/resnet_main.py index ca9de7c4a..7d29ff694 100644 --- a/examples/resnet/resnet_main.py +++ b/examples/resnet/resnet_main.py @@ -16,28 +16,38 @@ import cifar_input import resnet_model # Tensorflow must be at least version 1.0.0 for the example to work. -if int(tf.__version__.split('.')[0]) < 1: - raise Exception('Your Tensorflow version is less than 1.0.0. Please update Tensorflow to the latest version.') +if int(tf.__version__.split(".")[0]) < 1: + raise Exception("Your Tensorflow version is less than 1.0.0. Please update " + "Tensorflow to the latest version.") -parser = argparse.ArgumentParser(description="Run the hyperparameter optimization example.") -parser.add_argument("--dataset", default='cifar10', type=str, help="Dataset to use: cifar10 or cifar100.") -parser.add_argument("--train_data_path", default='cifar-10-batches-bin/data_batch*', type=str, help="Data path for the training data.") -parser.add_argument("--eval_data_path", default='cifar-10-batches-bin/test_batch.bin', type=str, help="Data path for the testing data.") -parser.add_argument("--eval_dir", default='/tmp/resnet-model/eval', type=str, help="Data path for the tensorboard logs.") -parser.add_argument("--eval_batch_count", default=50, type=int, help="Number of batches to evaluate over.") -parser.add_argument("--num_gpus", default=0, type=int, help="Number of GPUs to use for training.") +parser = argparse.ArgumentParser(description="Run the ResNet example.") +parser.add_argument("--dataset", default="cifar10", type=str, + help="Dataset to use: cifar10 or cifar100.") +parser.add_argument("--train_data_path", + default="cifar-10-batches-bin/data_batch*", type=str, + help="Data path for the training data.") +parser.add_argument("--eval_data_path", + default="cifar-10-batches-bin/test_batch.bin", type=str, + help="Data path for the testing data.") +parser.add_argument("--eval_dir", default="/tmp/resnet-model/eval", type=str, + help="Data path for the tensorboard logs.") +parser.add_argument("--eval_batch_count", default=50, type=int, + help="Number of batches to evaluate over.") +parser.add_argument("--num_gpus", default=0, type=int, + help="Number of GPUs to use for training.") FLAGS = parser.parse_args() # Determines if the actors require a gpu or not. use_gpu = 1 if int(FLAGS.num_gpus) > 0 else 0 + @ray.remote def get_data(path, size, dataset): # Retrieves all preprocessed images and labels using a tensorflow queue. # This only uses the cpu. - os.environ['CUDA_VISIBLE_DEVICES'] = '' - with tf.device('/cpu:0'): + os.environ["CUDA_VISIBLE_DEVICES"] = "" + with tf.device("/cpu:0"): queue = cifar_input.build_data(path, size, dataset) sess = tf.Session() coord = tf.train.Coordinator() @@ -47,23 +57,27 @@ def get_data(path, size, dataset): sess.close() return images, labels + @ray.remote(num_gpus=use_gpu) class ResNetTrainActor(object): def __init__(self, data, dataset, num_gpus): if num_gpus > 0: - os.environ['CUDA_VISIBLE_DEVICES'] = ','.join([str(i) for i in ray.get_gpu_ids()]) - hps = resnet_model.HParams(batch_size=128, - num_classes=100 if dataset == 'cifar100' else 10, - min_lrn_rate=0.0001, - lrn_rate=0.1, - num_residual_units=5, - use_bottleneck=False, - weight_decay_rate=0.0002, - relu_leakiness=0.1, - optimizer='mom', - num_gpus=num_gpus) + os.environ["CUDA_VISIBLE_DEVICES"] = ",".join([str(i) for i + in ray.get_gpu_ids()]) + hps = resnet_model.HParams( + batch_size=128, + num_classes=100 if dataset == "cifar100" else 10, + min_lrn_rate=0.0001, + lrn_rate=0.1, + num_residual_units=5, + use_bottleneck=False, + weight_decay_rate=0.0002, + relu_leakiness=0.1, + optimizer="mom", + num_gpus=num_gpus) - # We seed each actor differently so that each actor operates on a different subset of data. + # We seed each actor differently so that each actor operates on a different + # subset of data. if num_gpus > 0: tf.set_random_seed(ray.get_gpu_ids()[0] + 1) else: @@ -72,10 +86,11 @@ class ResNetTrainActor(object): input_images = data[0] input_labels = data[1] - with tf.device('/gpu:0' if num_gpus > 0 else '/cpu:0'): + with tf.device("/gpu:0" if num_gpus > 0 else "/cpu:0"): # Build the model. - images, labels = cifar_input.build_input([input_images, input_labels], hps.batch_size, dataset, False) - self.model = resnet_model.ResNet(hps, images, labels, 'train') + images, labels = cifar_input.build_input([input_images, input_labels], + hps.batch_size, dataset, False) + self.model = resnet_model.ResNet(hps, images, labels, "train") self.model.build_graph() config = tf.ConfigProto(allow_soft_placement=True) sess = tf.Session(config=config) @@ -87,37 +102,40 @@ class ResNetTrainActor(object): self.steps = 10 def compute_steps(self, weights): - # This method sets the weights in the network, trains the network self.steps times, - # and returns the new weights. + # This method sets the weights in the network, trains the network + # self.steps times, and returns the new weights. self.model.variables.set_weights(weights) for i in range(self.steps): self.model.variables.sess.run(self.model.train_op) return self.model.variables.get_weights() def get_weights(self): - # Note that the driver cannot directly access fields of the class, + # Note that the driver cannot directly access fields of the class, # so helper methods must be created. return self.model.variables.get_weights() + @ray.remote class ResNetTestActor(object): def __init__(self, data, dataset, eval_batch_count, eval_dir): - hps = resnet_model.HParams(batch_size=100, - num_classes=100 if dataset == 'cifar100' else 10, - min_lrn_rate=0.0001, - lrn_rate=0.1, - num_residual_units=5, - use_bottleneck=False, - weight_decay_rate=0.0002, - relu_leakiness=0.1, - optimizer='mom', - num_gpus=0) + hps = resnet_model.HParams( + batch_size=100, + num_classes=100 if dataset == "cifar100" else 10, + min_lrn_rate=0.0001, + lrn_rate=0.1, + num_residual_units=5, + use_bottleneck=False, + weight_decay_rate=0.0002, + relu_leakiness=0.1, + optimizer="mom", + num_gpus=0) input_images = data[0] input_labels = data[1] - with tf.device('/cpu:0'): + with tf.device("/cpu:0"): # Builds the testing network. - images, labels = cifar_input.build_input([input_images, input_labels], hps.batch_size, dataset, False) - self.model = resnet_model.ResNet(hps, images, labels, 'eval') + images, labels = cifar_input.build_input([input_images, input_labels], + hps.batch_size, dataset, False) + self.model = resnet_model.ResNet(hps, images, labels, "eval") self.model.build_graph() config = tf.ConfigProto(allow_soft_placement=True) sess = tf.Session(config=config) @@ -155,34 +173,40 @@ class ResNetTestActor(object): self.best_precision = max(precision, self.best_precision) precision_summ = tf.Summary() precision_summ.value.add( - tag='Precision', simple_value=precision) + tag="Precision", simple_value=precision) self.summary_writer.add_summary(precision_summ, train_step) best_precision_summ = tf.Summary() best_precision_summ.value.add( - tag='Best Precision', simple_value=self.best_precision) + tag="Best Precision", simple_value=self.best_precision) self.summary_writer.add_summary(best_precision_summ, train_step) self.summary_writer.add_summary(summaries, train_step) - tf.logging.info('loss: %.3f, precision: %.3f, best precision: %.3f' % + tf.logging.info("loss: %.3f, precision: %.3f, best precision: %.3f" % (loss, precision, self.best_precision)) self.summary_writer.flush() return precision def get_ip_addr(self): - # As above, a helper method must be created to access the field from the driver. + # As above, a helper method must be created to access the field from the + # driver. return self.ip_addr + def train(): num_gpus = FLAGS.num_gpus ray.init(num_gpus=num_gpus, redirect_output=True) train_data = get_data.remote(FLAGS.train_data_path, 50000, FLAGS.dataset) test_data = get_data.remote(FLAGS.eval_data_path, 10000, FLAGS.dataset) - # Creates an actor for each gpu, or one if only using the cpu. Each actor has its own copy of the dataset. + # Creates an actor for each gpu, or one if only using the cpu. Each actor has + # access to the dataset. if FLAGS.num_gpus > 0: - train_actors = [ResNetTrainActor.remote(train_data, FLAGS.dataset, num_gpus) for _ in range(num_gpus)] + train_actors = [ResNetTrainActor.remote(train_data, FLAGS.dataset, + num_gpus) for _ in range(num_gpus)] else: train_actors = [ResNetTrainActor.remote(train_data, FLAGS.dataset, 0)] - test_actor = ResNetTestActor.remote(test_data, FLAGS.dataset, FLAGS.eval_batch_count, FLAGS.eval_dir) - print('The log files for tensorboard are stored at ip {}.'.format(ray.get(test_actor.get_ip_addr.remote()))) + test_actor = ResNetTestActor.remote(test_data, FLAGS.dataset, + FLAGS.eval_batch_count, FLAGS.eval_dir) + print("The log files for tensorboard are stored at ip {}." + .format(ray.get(test_actor.get_ip_addr.remote()))) step = 0 weight_id = train_actors[0].get_weights.remote() acc_id = test_actor.accuracy.remote(weight_id, step) @@ -192,8 +216,11 @@ def train(): print("Starting training loop. Use Ctrl-C to exit.") try: while True: - all_weights = ray.get([actor.compute_steps.remote(weight_id) for actor in train_actors]) - mean_weights = {k: sum([weights[k] for weights in all_weights]) / num_gpus for k in all_weights[0]} + all_weights = ray.get([actor.compute_steps.remote(weight_id) + for actor in train_actors]) + mean_weights = {k: (sum([weights[k] for weights in all_weights]) / + num_gpus) + for k in all_weights[0]} weight_id = ray.put(mean_weights) step += 10 if step % 200 == 0: @@ -201,9 +228,10 @@ def train(): # testing task with the current weights every 200 steps. acc = ray.get(acc_id) acc_id = test_actor.accuracy.remote(weight_id, step) - print('Step {0}: {1:.6f}'.format(step - 200, acc)) + print("Step {0}: {1:.6f}".format(step - 200, acc)) except KeyboardInterrupt: pass -if __name__ == '__main__': + +if __name__ == "__main__": train() diff --git a/examples/resnet/resnet_model.py b/examples/resnet/resnet_model.py index 9687e064e..9671a4d36 100644 --- a/examples/resnet/resnet_model.py +++ b/examples/resnet/resnet_model.py @@ -31,7 +31,8 @@ class ResNet(object): Args: hps: Hyperparameters. - images: Batches of images of size [batch_size, image_size, image_size, 3]. + images: Batches of images of size [batch_size, image_size, image_size, + 3]. labels: Batches of labels of size [batch_size, num_classes]. mode: One of 'train' and 'eval'. """ @@ -112,12 +113,12 @@ class ResNet(object): def _build_train_op(self): """Build training specific ops for the graph.""" - rate = self.hps.lrn_rate num_gpus = self.hps.num_gpus if self.hps.num_gpus != 0 else 1 # The learning rate schedule is dependent on the number of gpus. boundaries = [int(20000 * i / np.sqrt(num_gpus)) for i in range(2, 5)] values = [0.1, 0.01, 0.001, 0.0001] - self.lrn_rate = tf.train.piecewise_constant(self.global_step, boundaries, values) + self.lrn_rate = tf.train.piecewise_constant(self.global_step, boundaries, + values) tf.summary.scalar('learning rate', self.lrn_rate) if self.hps.optimizer == 'sgd': @@ -202,7 +203,8 @@ class ResNet(object): orig_x = tf.nn.avg_pool(orig_x, stride, stride, 'VALID') orig_x = tf.pad( orig_x, [[0, 0], [0, 0], [0, 0], - [(out_filter-in_filter) // 2, (out_filter-in_filter) // 2]]) + [(out_filter - in_filter) // 2, + (out_filter - in_filter) // 2]]) x += orig_x return x @@ -227,7 +229,8 @@ class ResNet(object): with tf.variable_scope('sub2'): x = self._batch_norm('bn2', x) x = self._relu(x, self.hps.relu_leakiness) - x = self._conv('conv2', x, 3, out_filter / 4, out_filter / 4, [1, 1, 1, 1]) + x = self._conv('conv2', x, 3, out_filter / 4, out_filter / 4, + [1, 1, 1, 1]) with tf.variable_scope('sub3'): x = self._batch_norm('bn3', x) @@ -236,7 +239,8 @@ class ResNet(object): with tf.variable_scope('sub_add'): if in_filter != out_filter: - orig_x = self._conv('project', orig_x, 1, in_filter, out_filter, stride) + orig_x = self._conv('project', orig_x, 1, in_filter, out_filter, + stride) x += orig_x return x diff --git a/examples/rl_pong/driver.py b/examples/rl_pong/driver.py index 692ee8de6..4c414f417 100644 --- a/examples/rl_pong/driver.py +++ b/examples/rl_pong/driver.py @@ -26,16 +26,18 @@ decay_rate = 0.99 # The input dimensionality: 80x80 grid. D = 80 * 80 + def sigmoid(x): # Sigmoid "squashing" function to interval [0, 1]. return 1.0 / (1.0 + np.exp(-x)) + def preprocess(I): """Preprocess 210x160x3 uint8 frame into 6400 (80x80) 1D float vector.""" # Crop the image. I = I[35:195] # Downsample by factor of 2. - I = I[::2,::2,0] + I = I[::2, ::2, 0] # Erase background (background type 1). I[I == 144] = 0 # Erase background (background type 2). @@ -44,25 +46,29 @@ def preprocess(I): I[I != 0] = 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(range(0, r.size)): # Reset the sum, since this was a game boundary (pong specific!). - if r[t] != 0: running_add = 0 + if r[t] != 0: + running_add = 0 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 + h[h < 0] = 0 # ReLU nonlinearity. logp = np.dot(model["W2"], h) p = sigmoid(logp) # Return probability of taking action 2, and hidden state. return p, h + def policy_backward(eph, epx, epdlogp, model): """backward pass. (eph is array of intermediate hidden states)""" dW2 = np.dot(eph.T, epdlogp).ravel() @@ -72,6 +78,7 @@ def policy_backward(eph, epx, epdlogp, model): dW1 = np.dot(dh.T, epx) return {"W1": dW1, "W2": dW2} + @ray.remote class PongEnv(object): def __init__(self): @@ -104,7 +111,7 @@ class PongEnv(object): xs.append(x) # The hidden state. hs.append(h) - y = 1 if action == 2 else 0 # a "fake label" + y = 1 if action == 2 else 0 # A "fake label". # The gradient that encourages the action that was taken to be taken (see # http://cs231n.github.io/neural-networks-2/#losses if confused). dlogps.append(y - aprob) @@ -134,6 +141,7 @@ class PongEnv(object): epdlogp *= discounted_epr return policy_backward(eph, epx, epdlogp, model), reward_sum + if __name__ == "__main__": parser = argparse.ArgumentParser(description="Train an RL agent on Pong.") parser.add_argument("--batch-size", default=10, type=int, @@ -173,14 +181,16 @@ if __name__ == "__main__": # Accumulate the gradient over batch. for k in model: grad_buffer[k] += grad[k] - running_reward = reward_sum if running_reward is None else running_reward * 0.99 + reward_sum * 0.01 + running_reward = (reward_sum if running_reward is None + else running_reward * 0.99 + reward_sum * 0.01) end_time = time.time() print("Batch {} computed {} rollouts in {} seconds, " "running mean is {}".format(batch_num, batch_size, end_time - start_time, running_reward)) for k, v in model.items(): g = grad_buffer[k] - rmsprop_cache[k] = decay_rate * rmsprop_cache[k] + (1 - decay_rate) * g ** 2 + 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) # Reset the batch gradient buffer. grad_buffer[k] = np.zeros_like(v)