diff --git a/doc/source/policy-optimizers.rst b/doc/source/policy-optimizers.rst
index 3a3c60bf2..8753c2932 100644
--- a/doc/source/policy-optimizers.rst
+++ b/doc/source/policy-optimizers.rst
@@ -20,6 +20,8 @@ Example of constructing and using a policy optimizer `(link to full example) `__.
+
Here are the steps for using a RLlib policy optimizer with an existing algorithm.
1. Implement the `Policy evaluator interface `__.
diff --git a/doc/source/rllib.rst b/doc/source/rllib.rst
index a9bee7daf..9ff199eaf 100644
--- a/doc/source/rllib.rst
+++ b/doc/source/rllib.rst
@@ -7,19 +7,19 @@ You can find the code for RLlib `here on GitHub `__ which
- is a proximal variant of `TRPO `__.
+- Proximal Policy Optimization (`PPO `__) which is a proximal variant of `TRPO `__.
-- `The Asynchronous Advantage Actor-Critic (A3C) `__.
+- Policy Gradients (`PG `__).
-- `Deep Q Networks (DQN) `__.
+- Asynchronous Advantage Actor-Critic (`A3C `__).
-- `Ape-X Distributed Prioritized Experience Replay `__.
+- Deep Q Networks (`DQN `__).
-- Evolution Strategies, as described in `this
- paper `__. Our implementation
- is adapted from
- `here `__.
+- Deep Deterministic Policy Gradients (`DDPG `__, `DDPG2 `__).
+
+- Ape-X Distributed Prioritized Experience Replay, including both `DQN `__ and `DDPG `__ variants.
+
+- Evolution Strategies (`ES `__), as described in `this paper `__.
These algorithms can be run on any `OpenAI Gym MDP `__,
including custom ones written and registered by the user.
@@ -76,7 +76,7 @@ The ``train.py`` script has a number of options you can show by running
The most important options are for choosing the environment
with ``--env`` (any OpenAI gym environment including ones registered by the user
can be used) and for choosing the algorithm with ``--run``
-(available options are ``PPO``, ``A3C``, ``ES``, ``DQN`` and ``APEX``).
+(available options are ``PPO``, ``PG``, ``A3C``, ``ES``, ``DDPG``, ``DDPG2``, ``DQN``, ``APEX``, and ``APEX_DDPG2``).
Specifying Parameters
~~~~~~~~~~~~~~~~~~~~~
@@ -84,10 +84,14 @@ Specifying Parameters
Each algorithm has specific hyperparameters that can be set with ``--config`` - see the
``DEFAULT_CONFIG`` variable in
`PPO `__,
+`PG `__,
`A3C `__,
`ES `__,
-`DQN `__ and
-`APEX `__.
+`DQN `__,
+`DDPG `__,
+`DDPG2 `__,
+`APEX `__, and
+`APEX_DDPG2 `__.
In an example below, we train A3C by specifying 8 workers through the config flag.
function that creates the env to refer to it by name. The contents of the env_config agent config field will be passed to that function to allow the environment to be configured. The return type should be an OpenAI gym.Env. For example:
diff --git a/python/ray/rllib/README.rst b/python/ray/rllib/README.rst
index 29b31e625..8b2cdf44f 100644
--- a/python/ray/rllib/README.rst
+++ b/python/ray/rllib/README.rst
@@ -5,18 +5,19 @@ Ray RLlib is an RL execution toolkit built on the Ray distributed execution fram
RLlib includes the following reference algorithms:
-- `Proximal Policy Optimization (PPO) `__ which
- is a proximal variant of `TRPO `__.
+- Proximal Policy Optimization (`PPO `__) which is a proximal variant of `TRPO `__.
-- `The Asynchronous Advantage Actor-Critic (A3C) `__.
+- Policy Gradients (`PG `__).
-- `Deep Q Networks (DQN) `__.
+- Asynchronous Advantage Actor-Critic (`A3C `__).
-- `Ape-X Distributed Prioritized Experience Replay `__.
+- Deep Q Networks (`DQN `__).
-- Evolution Strategies, as described in `this
- paper `__. Our implementation
- is adapted from
- `here `__.
+- Deep Deterministic Policy Gradients (`DDPG `__, `DDPG2 `__).
+
+- Ape-X Distributed Prioritized Experience Replay, including both `DQN `__ and `DDPG `__ variants.
+
+- Evolution Strategies (`ES `__), as described in `this
+ paper `__.
These algorithms can be run on any OpenAI Gym MDP, including custom ones written and registered by the user.
diff --git a/python/ray/rllib/__init__.py b/python/ray/rllib/__init__.py
index 7e2145c2c..a2441f0b5 100644
--- a/python/ray/rllib/__init__.py
+++ b/python/ray/rllib/__init__.py
@@ -9,7 +9,7 @@ from ray.tune.registry import register_trainable
def _register_all():
for key in ["PPO", "ES", "DQN", "APEX", "A3C", "BC", "PG", "DDPG",
- "DDPG2", "APEX_DDPG2", "__fake", "__sigmoid_fake_data",
+ "DDPG2", "APEX_DDPG", "__fake", "__sigmoid_fake_data",
"__parameter_tuning"]:
from ray.rllib.agent import get_agent_class
register_trainable(key, get_agent_class(key))
diff --git a/python/ray/rllib/agent.py b/python/ray/rllib/agent.py
index fd32edf56..5699022b2 100644
--- a/python/ray/rllib/agent.py
+++ b/python/ray/rllib/agent.py
@@ -234,9 +234,12 @@ def get_agent_class(alg):
if alg == "DDPG2":
from ray.rllib import ddpg2
return ddpg2.DDPG2Agent
- elif alg == "APEX_DDPG2":
- from ray.rllib import ddpg2
- return ddpg2.ApexDDPG2Agent
+ elif alg == "DDPG":
+ from ray.rllib import ddpg
+ return ddpg.DDPGAgent
+ elif alg == "APEX_DDPG":
+ from ray.rllib import ddpg
+ return ddpg.ApexDDPGAgent
elif alg == "PPO":
from ray.rllib import ppo
return ppo.PPOAgent
@@ -258,9 +261,6 @@ def get_agent_class(alg):
elif alg == "PG":
from ray.rllib import pg
return pg.PGAgent
- elif alg == "DDPG":
- from ray.rllib import ddpg
- return ddpg.DDPGAgent
elif alg == "script":
from ray.tune import script_runner
return script_runner.ScriptRunner
diff --git a/python/ray/rllib/ddpg/README.md b/python/ray/rllib/ddpg/README.md
new file mode 100644
index 000000000..93c32b0a2
--- /dev/null
+++ b/python/ray/rllib/ddpg/README.md
@@ -0,0 +1 @@
+Implementation of deep deterministic policy gradients (https://arxiv.org/abs/1509.02971), including an Ape-X variant.
diff --git a/python/ray/rllib/ddpg/__init__.py b/python/ray/rllib/ddpg/__init__.py
index 004e0f128..932b9f0c8 100644
--- a/python/ray/rllib/ddpg/__init__.py
+++ b/python/ray/rllib/ddpg/__init__.py
@@ -1,3 +1,8 @@
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+
+from ray.rllib.ddpg.apex import ApexDDPGAgent
from ray.rllib.ddpg.ddpg import DDPGAgent, DEFAULT_CONFIG
-__all__ = ["DDPGAgent", "DEFAULT_CONFIG"]
+__all__ = ["DDPGAgent", "ApexDDPGAgent", "DEFAULT_CONFIG"]
diff --git a/python/ray/rllib/ddpg2/apex.py b/python/ray/rllib/ddpg/apex.py
similarity index 95%
rename from python/ray/rllib/ddpg2/apex.py
rename to python/ray/rllib/ddpg/apex.py
index 9ace851b5..c670198c3 100644
--- a/python/ray/rllib/ddpg2/apex.py
+++ b/python/ray/rllib/ddpg/apex.py
@@ -2,7 +2,7 @@ from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
-from ray.rllib.ddpg2.ddpg import DDPG2Agent, DEFAULT_CONFIG as DDPG_CONFIG
+from ray.rllib.ddpg.ddpg import DDPGAgent, DEFAULT_CONFIG as DDPG_CONFIG
APEX_DDPG_DEFAULT_CONFIG = dict(DDPG_CONFIG,
**dict(
@@ -28,7 +28,7 @@ APEX_DDPG_DEFAULT_CONFIG = dict(DDPG_CONFIG,
))
-class ApexDDPG2Agent(DDPG2Agent):
+class ApexDDPGAgent(DDPGAgent):
"""DDPG variant that uses the Ape-X distributed policy optimizer.
By default, this is configured for a large single node (32 cores). For
diff --git a/python/ray/rllib/ddpg2/common/__init__.py b/python/ray/rllib/ddpg/common/__init__.py
similarity index 100%
rename from python/ray/rllib/ddpg2/common/__init__.py
rename to python/ray/rllib/ddpg/common/__init__.py
diff --git a/python/ray/rllib/ddpg/ddpg.py b/python/ray/rllib/ddpg/ddpg.py
index fc7901383..343b32394 100644
--- a/python/ray/rllib/ddpg/ddpg.py
+++ b/python/ray/rllib/ddpg/ddpg.py
@@ -2,111 +2,267 @@ from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
+import pickle
+import os
+
import numpy as np
+import tensorflow as tf
import ray
+from ray.rllib import optimizers
+from ray.rllib.ddpg.ddpg_evaluator import DDPGEvaluator
from ray.rllib.agent import Agent
-from ray.rllib.ddpg.ddpg_evaluator import DDPGEvaluator, RemoteDDPGEvaluator
-from ray.rllib.optimizers import LocalSyncReplayOptimizer
from ray.tune.result import TrainingResult
-DEFAULT_CONFIG = {
- # Actor learning rate
- "actor_lr": 0.0001,
- # Critic learning rate
- "critic_lr": 0.001,
- # Arguments to pass in to env creator
- "env_config": {},
- # MDP Discount factor
- "gamma": 0.99,
- # Number of steps after which the rollout gets cut
- "horizon": 500,
+OPTIMIZER_SHARED_CONFIGS = [
+ "buffer_size", "prioritized_replay", "prioritized_replay_alpha",
+ "prioritized_replay_beta", "prioritized_replay_eps", "sample_batch_size",
+ "train_batch_size", "learning_starts", "clip_rewards"
+]
- # Whether to include parameter noise
- "noise_add": True,
- # Linear decay of exploration policy
- "noise_epsilon": 0.0002,
- # Parameters for noise process
- "noise_parameters": {
- "mu": 0,
- "sigma": 0.2,
- "theta": 0.15,
+DEFAULT_CONFIG = dict(
+ # === Model ===
+ # Hidden layer sizes of the policy networks
+ actor_hiddens=[64, 64],
+ # Hidden layer sizes of the policy networks
+ critic_hiddens=[64, 64],
+ # N-step Q learning
+ n_step=1,
+ # Config options to pass to the model constructor
+ model={},
+ # Discount factor for the MDP
+ gamma=0.99,
+ # Arguments to pass to the env creator
+ env_config={},
+
+ # === Exploration ===
+ # Max num timesteps for annealing schedules. Exploration is annealed from
+ # 1.0 to exploration_fraction over this number of timesteps scaled by
+ # exploration_fraction
+ schedule_max_timesteps=100000,
+ # Number of env steps to optimize for before returning
+ timesteps_per_iteration=1000,
+ # Fraction of entire training period over which the exploration rate is
+ # annealed
+ exploration_fraction=0.1,
+ # Final value of random action probability
+ exploration_final_eps=0.02,
+ # OU-noise scale
+ noise_scale=0.1,
+ # theta
+ exploration_theta=0.15,
+ # sigma
+ exploration_sigma=0.2,
+ # Update the target network every `target_network_update_freq` steps.
+ target_network_update_freq=0,
+ # Update the target by \tau * policy + (1-\tau) * target_policy
+ tau=0.002,
+ # Whether to start with random actions instead of noops.
+ random_starts=True,
+
+ # === Replay buffer ===
+ # Size of the replay buffer. Note that if async_updates is set, then
+ # each worker will have a replay buffer of this size.
+ buffer_size=50000,
+ # If True prioritized replay buffer will be used.
+ prioritized_replay=True,
+ # Alpha parameter for prioritized replay buffer.
+ prioritized_replay_alpha=0.6,
+ # Beta parameter for sampling from prioritized replay buffer.
+ prioritized_replay_beta=0.4,
+ # Epsilon to add to the TD errors when updating priorities.
+ prioritized_replay_eps=1e-6,
+ # Whether to clip rewards to [-1, 1] prior to adding to the replay buffer.
+ clip_rewards=True,
+
+ # === Optimization ===
+ # Learning rate for adam optimizer
+ actor_lr=1e-4,
+ critic_lr=1e-3,
+ # If True, use huber loss instead of squared loss for critic network
+ # Conventionally, no need to clip gradients if using a huber loss
+ use_huber=False,
+ # Threshold of a huber loss
+ huber_threshold=1.0,
+ # Weights for L2 regularization
+ l2_reg=1e-6,
+ # If not None, clip gradients during optimization at this value
+ grad_norm_clipping=None,
+ # How many steps of the model to sample before learning starts.
+ learning_starts=1500,
+ # Update the replay buffer with this many samples at once. Note that this
+ # setting applies per-worker if num_workers > 1.
+ sample_batch_size=1,
+ # Size of a batched sampled from replay buffer for training. Note that
+ # if async_updates is set, then each worker returns gradients for a
+ # batch of this size.
+ train_batch_size=256,
+ # Smooth the current average reward over this many previous episodes.
+ smoothing_num_episodes=100,
+
+ # === Tensorflow ===
+ # Arguments to pass to tensorflow
+ tf_session_args={
+ "device_count": {
+ "CPU": 2
+ },
+ "log_device_placement": False,
+ "allow_soft_placement": True,
+ "gpu_options": {
+ "allow_growth": True
+ },
+ "inter_op_parallelism_threads": 1,
+ "intra_op_parallelism_threads": 1,
},
- # Number of local steps taken for each call to sample
- "num_local_steps": 1,
- # Number of workers (excluding master)
- "num_workers": 0,
-
- "optimizer": {
- # Replay buffer size
- "buffer_size": 10000,
- # Number of steps in warm-up phase before learning starts
- "learning_starts": 500,
- # Whether to clip rewards
- "clip_rewards": False,
- # Whether to use prioritized replay
- "prioritized_replay": False,
- # Size of batch sampled from replay buffer
- "train_batch_size": 64,
- },
-
- # Controls how fast target networks move
- "tau": 0.001,
- # Number of steps taken per training iteration
- "train_steps": 600,
-}
+ # === Parallelism ===
+ # Number of workers for collecting samples with. This only makes sense
+ # to increase if your environment is particularly slow to sample, or if
+ # you're using the Async or Ape-X optimizers.
+ num_workers=0,
+ # Whether to allocate GPUs for workers (if > 0).
+ num_gpus_per_worker=0,
+ # Optimizer class to use.
+ optimizer_class="LocalSyncReplayOptimizer",
+ # Config to pass to the optimizer.
+ optimizer_config=dict(),
+ # Whether to use a distribution of epsilons across workers for exploration.
+ per_worker_exploration=False,
+ # Whether to compute priorities on workers.
+ worker_side_prioritization=False)
class DDPGAgent(Agent):
_agent_name = "DDPG"
+ _allow_unknown_subkeys = [
+ "model", "optimizer", "tf_session_args", "env_config"
+ ]
_default_config = DEFAULT_CONFIG
def _init(self):
- self.local_evaluator = DDPGEvaluator(
- self.registry, self.env_creator, self.config)
+ self.local_evaluator = DDPGEvaluator(self.registry, self.env_creator,
+ self.config, self.logdir, 0)
+ remote_cls = ray.remote(
+ num_cpus=1,
+ num_gpus=self.config["num_gpus_per_worker"])(DDPGEvaluator)
self.remote_evaluators = [
- RemoteDDPGEvaluator.remote(
- self.registry, self.env_creator, self.config)
- for _ in range(self.config["num_workers"])]
- self.optimizer = LocalSyncReplayOptimizer(
- self.config["optimizer"], self.local_evaluator,
+ remote_cls.remote(self.registry, self.env_creator, self.config,
+ self.logdir, i)
+ for i in range(self.config["num_workers"])
+ ]
+
+ for k in OPTIMIZER_SHARED_CONFIGS:
+ if k not in self.config["optimizer_config"]:
+ self.config["optimizer_config"][k] = self.config[k]
+
+ self.optimizer = getattr(optimizers, self.config["optimizer_class"])(
+ self.config["optimizer_config"], self.local_evaluator,
self.remote_evaluators)
+ self.saver = tf.train.Saver(max_to_keep=None)
+ self.last_target_update_ts = 0
+ self.num_target_updates = 0
+
+ @property
+ def global_timestep(self):
+ return self.optimizer.num_steps_sampled
+
+ def update_target_if_needed(self):
+ if self.global_timestep - self.last_target_update_ts > \
+ self.config["target_network_update_freq"]:
+ self.local_evaluator.update_target()
+ self.last_target_update_ts = self.global_timestep
+ self.num_target_updates += 1
+
def _train(self):
- for _ in range(self.config["train_steps"]):
+ start_timestep = self.global_timestep
+
+ while (self.global_timestep - start_timestep <
+ self.config["timesteps_per_iteration"]):
+
self.optimizer.step()
- # update target
- if self.optimizer.num_steps_trained > 0:
- self.local_evaluator.update_target()
+ self.update_target_if_needed()
- # generate training result
- return self._fetch_metrics()
+ self.local_evaluator.set_global_timestep(self.global_timestep)
+ for e in self.remote_evaluators:
+ e.set_global_timestep.remote(self.global_timestep)
- def _fetch_metrics(self):
- episode_rewards = []
- episode_lengths = []
- if self.config["num_workers"] > 0:
- metric_lists = [a.get_completed_rollout_metrics.remote()
- for a in self.remote_evaluators]
- for metrics in metric_lists:
- for episode in ray.get(metrics):
- episode_lengths.append(episode.episode_length)
- episode_rewards.append(episode.episode_reward)
+ return self._train_stats(start_timestep)
+
+ def _train_stats(self, start_timestep):
+ if self.remote_evaluators:
+ stats = ray.get([e.stats.remote() for e in self.remote_evaluators])
else:
- metrics = self.local_evaluator.get_completed_rollout_metrics()
- for episode in metrics:
- episode_lengths.append(episode.episode_length)
- episode_rewards.append(episode.episode_reward)
+ stats = self.local_evaluator.stats()
+ if not isinstance(stats, list):
+ stats = [stats]
- avg_reward = (np.mean(episode_rewards))
- avg_length = (np.mean(episode_lengths))
- timesteps = np.sum(episode_lengths)
+ mean_100ep_reward = 0.0
+ mean_100ep_length = 0.0
+ num_episodes = 0
+ explorations = []
+
+ if self.config["per_worker_exploration"]:
+ # Return stats from workers with the lowest 20% of exploration
+ test_stats = stats[-int(max(1, len(stats) * 0.2)):]
+ else:
+ test_stats = stats
+
+ for s in test_stats:
+ mean_100ep_reward += s["mean_100ep_reward"] / len(test_stats)
+ mean_100ep_length += s["mean_100ep_length"] / len(test_stats)
+
+ for s in stats:
+ num_episodes += s["num_episodes"]
+ explorations.append(s["exploration"])
+
+ opt_stats = self.optimizer.stats()
result = TrainingResult(
- episode_reward_mean=avg_reward,
- episode_len_mean=avg_length,
- timesteps_this_iter=timesteps,
- info={})
+ episode_reward_mean=mean_100ep_reward,
+ episode_len_mean=mean_100ep_length,
+ episodes_total=num_episodes,
+ timesteps_this_iter=self.global_timestep - start_timestep,
+ info=dict({
+ "min_exploration": min(explorations),
+ "max_exploration": max(explorations),
+ "num_target_updates": self.num_target_updates,
+ }, **opt_stats))
return result
+
+ def _stop(self):
+ # workaround for https://github.com/ray-project/ray/issues/1516
+ for ev in self.remote_evaluators:
+ ev.__ray_terminate__.remote(ev._ray_actor_id.id())
+
+ def _save(self, checkpoint_dir):
+ checkpoint_path = self.saver.save(
+ self.local_evaluator.sess,
+ os.path.join(checkpoint_dir, "checkpoint"),
+ global_step=self.iteration)
+ extra_data = [
+ self.local_evaluator.save(),
+ ray.get([e.save.remote() for e in self.remote_evaluators]),
+ self.optimizer.save(), self.num_target_updates,
+ self.last_target_update_ts
+ ]
+ pickle.dump(extra_data, open(checkpoint_path + ".extra_data", "wb"))
+ return checkpoint_path
+
+ def _restore(self, checkpoint_path):
+ self.saver.restore(self.local_evaluator.sess, checkpoint_path)
+ extra_data = pickle.load(open(checkpoint_path + ".extra_data", "rb"))
+ self.local_evaluator.restore(extra_data[0])
+ ray.get([
+ e.restore.remote(d)
+ for (d, e) in zip(extra_data[1], self.remote_evaluators)
+ ])
+ self.optimizer.restore(extra_data[2])
+ self.num_target_updates = extra_data[3]
+ self.last_target_update_ts = extra_data[4]
+
+ def compute_action(self, observation):
+ return self.local_evaluator.ddpg_graph.act(self.local_evaluator.sess,
+ np.array(observation)[None],
+ 0.0)[0]
diff --git a/python/ray/rllib/ddpg/ddpg_evaluator.py b/python/ray/rllib/ddpg/ddpg_evaluator.py
index dda3c3479..5a68c4b58 100644
--- a/python/ray/rllib/ddpg/ddpg_evaluator.py
+++ b/python/ray/rllib/ddpg/ddpg_evaluator.py
@@ -2,74 +2,185 @@ from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
+from gym.spaces import Box
import numpy as np
+import tensorflow as tf
import ray
-from ray.rllib.ddpg.models import DDPGModel
-from ray.rllib.models.catalog import ModelCatalog
-from ray.rllib.optimizers import PolicyEvaluator
-from ray.rllib.utils.filter import NoFilter
-from ray.rllib.utils.process_rollout import process_rollout
-from ray.rllib.utils.sampler import SyncSampler
+from ray.rllib.utils.error import UnsupportedSpaceException
+from ray.rllib.ddpg import models
+from ray.rllib.dqn.common.schedules import ConstantSchedule, LinearSchedule
+from ray.rllib.optimizers import SampleBatch, PolicyEvaluator
+from ray.rllib.utils.compression import pack
+from ray.rllib.dqn.dqn_evaluator import adjust_nstep
+from ray.rllib.dqn.common.wrappers import wrap_dqn
class DDPGEvaluator(PolicyEvaluator):
+ """The base DDPG Evaluator."""
- def __init__(self, registry, env_creator, config):
- self.env = ModelCatalog.get_preprocessor_as_wrapper(
- registry, env_creator(config["env_config"]))
+ def __init__(self, registry, env_creator, config, logdir, worker_index):
+ env = env_creator(config["env_config"])
+ env = wrap_dqn(registry, env, config["model"], config["random_starts"])
+ self.env = env
+ self.config = config
- # contains model, target_model
- self.model = DDPGModel(registry, self.env, config)
+ # when env.action_space is of Box type, e.g., Pendulum-v0
+ # action_space.low is [-2.0], high is [2.0]
+ # take action by calling, e.g., env.step([3.5])
+ if not isinstance(env.action_space, Box):
+ raise UnsupportedSpaceException(
+ "Action space {} is not supported for DDPG.".format(
+ env.action_space))
- self.sampler = SyncSampler(
- self.env, self.model.model, NoFilter(),
- config["num_local_steps"], horizon=config["horizon"])
+ tf_config = tf.ConfigProto(**config["tf_session_args"])
+ self.sess = tf.Session(config=tf_config)
+ self.ddpg_graph = models.DDPGGraph(registry, env, config, logdir)
- def sample(self):
- """Returns a batch of samples."""
+ # Use either a different `eps` per worker, or a linear schedule.
+ if config["per_worker_exploration"]:
+ assert config["num_workers"] > 1, "This requires multiple workers"
+ self.exploration = ConstantSchedule(
+ config["noise_scale"] * 0.4 **
+ (1 + worker_index / float(config["num_workers"] - 1) * 7))
+ else:
+ self.exploration = LinearSchedule(
+ schedule_timesteps=int(config["exploration_fraction"] *
+ config["schedule_max_timesteps"]),
+ initial_p=config["noise_scale"] * 1.0,
+ final_p=config["noise_scale"] *
+ config["exploration_final_eps"])
- rollout = self.sampler.get_data()
- rollout.data["weights"] = np.ones_like(rollout.data["rewards"])
+ # Initialize the parameters and copy them to the target network.
+ self.sess.run(tf.global_variables_initializer())
+ # hard instead of soft
+ self.ddpg_graph.update_target(self.sess, 1.0)
+ self.global_timestep = 0
+ self.local_timestep = 0
- # since each sample is one step, no discounting needs to be applied;
- # this does not involve config["gamma"]
- samples = process_rollout(
- rollout, NoFilter(),
- gamma=1.0, use_gae=False)
+ # Note that this encompasses both the policy and Q-value networks and
+ # their corresponding target networks
+ self.variables = ray.experimental.TensorFlowVariables(
+ tf.group(self.ddpg_graph.q_tp0, self.ddpg_graph.q_tp1), self.sess)
- return samples
+ self.episode_rewards = [0.0]
+ self.episode_lengths = [0.0]
+ self.saved_mean_reward = None
+
+ self.obs = self.env.reset()
+
+ def set_global_timestep(self, global_timestep):
+ self.global_timestep = global_timestep
def update_target(self):
- """Updates target critic and target actor."""
- self.model.update_target()
+ self.ddpg_graph.update_target(self.sess)
+
+ def sample(self):
+ obs, actions, rewards, new_obs, dones = [], [], [], [], []
+ for _ in range(
+ self.config["sample_batch_size"] + self.config["n_step"] - 1):
+ ob, act, rew, ob1, done = self._step(self.global_timestep)
+ obs.append(ob)
+ actions.append(act)
+ rewards.append(rew)
+ new_obs.append(ob1)
+ dones.append(done)
+
+ # N-step Q adjustments
+ if self.config["n_step"] > 1:
+ # Adjust for steps lost from truncation
+ self.local_timestep -= (self.config["n_step"] - 1)
+ adjust_nstep(self.config["n_step"], self.config["gamma"], obs,
+ actions, rewards, new_obs, dones)
+
+ batch = SampleBatch({
+ "obs": [pack(np.array(o)) for o in obs],
+ "actions": actions,
+ "rewards": rewards,
+ "new_obs": [pack(np.array(o)) for o in new_obs],
+ "dones": dones,
+ "weights": np.ones_like(rewards)
+ })
+ assert (batch.count == self.config["sample_batch_size"])
+
+ # Prioritize on the worker side
+ if self.config["worker_side_prioritization"]:
+ td_errors = self.ddpg_graph.compute_td_error(
+ self.sess, obs, batch["actions"], batch["rewards"], new_obs,
+ batch["dones"], batch["weights"])
+ new_priorities = (
+ np.abs(td_errors) + self.config["prioritized_replay_eps"])
+ batch.data["weights"] = new_priorities
+
+ return batch
def compute_gradients(self, samples):
- """Returns critic, actor gradients."""
- return self.model.compute_gradients(samples)
+ td_err, grads = self.ddpg_graph.compute_gradients(
+ self.sess, samples["obs"], samples["actions"], samples["rewards"],
+ samples["new_obs"], samples["dones"], samples["weights"])
+ return grads, {"td_error": td_err}
def apply_gradients(self, grads):
- """Applies gradients to evaluator weights."""
- self.model.apply_gradients(grads)
+ self.ddpg_graph.apply_gradients(self.sess, grads)
def compute_apply(self, samples):
- grads, _ = self.compute_gradients(samples)
- self.apply_gradients(grads)
+ td_error = self.ddpg_graph.compute_apply(
+ self.sess, samples["obs"], samples["actions"], samples["rewards"],
+ samples["new_obs"], samples["dones"], samples["weights"])
+ return {"td_error": td_error}
def get_weights(self):
- """Returns model weights."""
- return self.model.get_weights()
+ return self.variables.get_weights()
def set_weights(self, weights):
- """Sets model weights."""
- self.model.set_weights(weights)
+ self.variables.set_weights(weights)
- def get_completed_rollout_metrics(self):
- """Returns metrics on previously completed rollouts.
+ def _step(self, global_timestep):
+ """Takes a single step, and returns the result of the step."""
+ action = self.ddpg_graph.act(
+ self.sess,
+ np.array(self.obs)[None],
+ self.exploration.value(global_timestep))[0]
+ new_obs, rew, done, _ = self.env.step(action)
+ ret = (self.obs, action, rew, new_obs, float(done))
+ self.obs = new_obs
+ self.episode_rewards[-1] += rew
+ self.episode_lengths[-1] += 1
+ if done:
+ self.obs = self.env.reset()
+ self.episode_rewards.append(0.0)
+ self.episode_lengths.append(0.0)
+ # reset UO noise for each episode
+ self.ddpg_graph.reset_noise(self.sess)
- Calling this clears the queue of completed rollout metrics.
- """
- return self.sampler.get_metrics()
+ self.local_timestep += 1
+ return ret
+ def stats(self):
+ n = self.config["smoothing_num_episodes"] + 1
+ mean_100ep_reward = round(np.mean(self.episode_rewards[-n:-1]), 5)
+ mean_100ep_length = round(np.mean(self.episode_lengths[-n:-1]), 5)
+ exploration = self.exploration.value(self.global_timestep)
+ return {
+ "mean_100ep_reward": mean_100ep_reward,
+ "mean_100ep_length": mean_100ep_length,
+ "num_episodes": len(self.episode_rewards),
+ "exploration": exploration,
+ "local_timestep": self.local_timestep,
+ }
-RemoteDDPGEvaluator = ray.remote(DDPGEvaluator)
+ def save(self):
+ return [
+ self.exploration, self.episode_rewards, self.episode_lengths,
+ self.saved_mean_reward, self.obs, self.global_timestep,
+ self.local_timestep
+ ]
+
+ def restore(self, data):
+ self.exploration = data[0]
+ self.episode_rewards = data[1]
+ self.episode_lengths = data[2]
+ self.saved_mean_reward = data[3]
+ self.obs = data[4]
+ self.global_timestep = data[5]
+ self.local_timestep = data[6]
diff --git a/python/ray/rllib/ddpg/models.py b/python/ray/rllib/ddpg/models.py
index 20a661a97..d58f37dc6 100644
--- a/python/ray/rllib/ddpg/models.py
+++ b/python/ray/rllib/ddpg/models.py
@@ -3,239 +3,389 @@ from __future__ import division
from __future__ import print_function
import numpy as np
+
import tensorflow as tf
+import tensorflow.contrib.layers as layers
-from ray.experimental.tfutils import TensorFlowVariables
-from ray.rllib.models.ddpgnet import DDPGActor, DDPGCritic
-from ray.rllib.ddpg.random_process import OrnsteinUhlenbeckProcess
+from ray.rllib.models import ModelCatalog
-class DDPGModel():
- def __init__(self, registry, env, config):
- self.config = config
- self.sess = tf.Session()
+def _build_p_network(registry, inputs, dim_actions, config):
+ """
+ map an observation (i.e., state) to an action where
+ each entry takes value from (0, 1) due to the sigmoid function
+ """
+ frontend = ModelCatalog.get_model(registry, inputs, 1, config["model"])
- with tf.variable_scope("model"):
- self.model = DDPGActorCritic(
- registry, env, self.config, self.sess)
- with tf.variable_scope("target_model"):
- self.target_model = DDPGActorCritic(
- registry, env, self.config, self.sess)
- self._setup_gradients()
- self._setup_target_updates()
+ hiddens = config["actor_hiddens"]
+ action_out = frontend.last_layer
+ for hidden in hiddens:
+ action_out = layers.fully_connected(
+ action_out, num_outputs=hidden, activation_fn=tf.nn.relu)
+ # Use sigmoid layer to bound values within (0, 1)
+ # shape of action_scores is [batch_size, dim_actions]
+ action_scores = layers.fully_connected(
+ action_out, num_outputs=dim_actions, activation_fn=tf.nn.sigmoid)
- self.initialize()
- self._initialize_target_weights()
-
- def initialize(self):
- self.sess.run(tf.global_variables_initializer())
-
- def _initialize_target_weights(self):
- """Set initial target weights to match model weights."""
- a_updates = []
- for var, target_var in zip(
- self.model.actor_var_list, self.target_model.actor_var_list):
- a_updates.append(tf.assign(target_var, var))
- actor_updates = tf.group(*a_updates)
-
- c_updates = []
- for var, target_var in zip(
- self.model.critic_var_list, self.target_model.critic_var_list):
- c_updates.append(tf.assign(target_var, var))
- critic_updates = tf.group(*c_updates)
- self.sess.run([actor_updates, critic_updates])
-
- def _setup_gradients(self):
- """Setup critic and actor gradients."""
- self.critic_grads = tf.gradients(
- self.model.critic_loss, self.model.critic_var_list)
- c_grads_and_vars = list(zip(
- self.critic_grads, self.model.critic_var_list))
- c_opt = tf.train.AdamOptimizer(self.config["critic_lr"])
- self._apply_c_gradients = c_opt.apply_gradients(c_grads_and_vars)
-
- self.actor_grads = tf.gradients(
- -self.model.cn_for_loss, self.model.actor_var_list)
- a_grads_and_vars = list(zip(
- self.actor_grads, self.model.actor_var_list))
- a_opt = tf.train.AdamOptimizer(self.config["actor_lr"])
- self._apply_a_gradients = a_opt.apply_gradients(a_grads_and_vars)
-
- def compute_gradients(self, samples):
- """ Returns gradient w.r.t. samples."""
- # actor gradients
- actor_actions = self.sess.run(
- self.model.output_action,
- feed_dict={self.model.obs: samples["obs"]}
- )
-
- actor_feed_dict = {
- self.model.obs: samples["obs"],
- self.model.output_action: actor_actions,
- }
- self.actor_grads = [g for g in self.actor_grads if g is not None]
- actor_grad = self.sess.run(self.actor_grads, feed_dict=actor_feed_dict)
-
- # feed samples into target actor
- target_Q_act = self.sess.run(
- self.target_model.output_action,
- feed_dict={self.target_model.obs: samples["new_obs"]}
- )
- target_Q_dict = {
- self.target_model.obs: samples["new_obs"],
- self.target_model.act: target_Q_act,
- }
-
- target_Q = self.sess.run(
- self.target_model.critic_eval, feed_dict=target_Q_dict)
-
- # critic gradients
- critic_feed_dict = {
- self.model.obs: samples["obs"],
- self.model.act: samples["actions"],
- self.model.reward: samples["rewards"],
- self.model.target_Q: target_Q,
- }
- self.critic_grads = [g for g in self.critic_grads if g is not None]
- critic_grad = self.sess.run(
- self.critic_grads, feed_dict=critic_feed_dict)
- return (critic_grad, actor_grad), {}
-
- def apply_gradients(self, grads):
- """Applies gradients to evaluator weights."""
- c_grads, a_grads = grads
- critic_feed_dict = dict(zip(self.critic_grads, c_grads))
- self.sess.run(self._apply_c_gradients, feed_dict=critic_feed_dict)
- actor_feed_dict = dict(zip(self.actor_grads, a_grads))
- self.sess.run(self._apply_a_gradients, feed_dict=actor_feed_dict)
-
- def get_weights(self):
- """Returns model weights, target model weights."""
- return self.model.get_weights(), self.target_model.get_weights()
-
- def set_weights(self, weights):
- """Sets model and target model weights."""
- model_weights, target_model_weights = weights
- self.model.set_weights(model_weights)
- self.target_model.set_weights(target_model_weights)
-
- def _setup_target_updates(self):
- """Set up target actor and critic updates."""
- a_updates = []
- tau = self.config["tau"]
- for var, target_var in zip(
- self.model.actor_var_list, self.target_model.actor_var_list):
- a_updates.append(tf.assign(
- target_var, tau * var + (1. - tau) * target_var))
- actor_updates = tf.group(*a_updates)
-
- c_updates = []
- for var, target_var in zip(
- self.model.critic_var_list, self.target_model.critic_var_list):
- c_updates.append(tf.assign(
- target_var, tau * var + (1. - tau) * target_var))
- critic_updates = tf.group(*c_updates)
- self.target_updates = [actor_updates, critic_updates]
-
- def update_target(self):
- """Updates target critic and target actor."""
- self.sess.run(self.target_updates)
+ return action_scores
-class DDPGActorCritic():
- other_output = []
- is_recurrent = False
+# As a stochastic policy for inference, but a deterministic policy for training
+# thus ignore batch_size issue when constructing a stochastic action
+def _build_action_network(p_values, low_action, high_action, stochastic, eps,
+ theta, sigma):
+ # shape is [None, dim_action]
+ deterministic_actions = (high_action - low_action) * p_values + low_action
- def __init__(self, registry, env, config, sess):
- self.config = config
- self.sess = sess
+ exploration_sample = tf.get_variable(
+ name="ornstein_uhlenbeck",
+ dtype=tf.float32,
+ initializer=low_action.size * [.0],
+ trainable=False)
+ normal_sample = tf.random_normal(
+ shape=[low_action.size], mean=0.0, stddev=1.0)
+ exploration_value = tf.assign_add(
+ exploration_sample,
+ theta * (.0 - exploration_sample) + sigma * normal_sample)
+ stochastic_actions = deterministic_actions + eps * (
+ high_action - low_action) * exploration_value
- obs_space = env.observation_space
- ac_space = env.action_space
+ return tf.cond(stochastic, lambda: stochastic_actions,
+ lambda: deterministic_actions)
- self.obs_size = int(np.prod(obs_space.shape))
- self.obs = tf.placeholder(tf.float32, [None, self.obs_size])
- self.ac_size = int(np.prod(ac_space.shape))
- self.act = tf.placeholder(tf.float32, [None, self.ac_size])
- self.action_bound = env.action_space.high
- # TODO: change action_bound to make more general
- self._setup_actor_network(obs_space, ac_space)
- self._setup_critic_network(obs_space, ac_space)
- self._setup_critic_loss(ac_space)
+def _build_q_network(registry, inputs, action_inputs, config):
+ frontend = ModelCatalog.get_model(registry, inputs, 1, config["model"])
- with tf.variable_scope("critic"):
- self.critic_var_list = tf.get_collection(
- tf.GraphKeys.TRAINABLE_VARIABLES,
- tf.get_variable_scope().name
- )
- self.critic_vars = TensorFlowVariables(self.critic_loss,
- self.sess)
+ hiddens = config["critic_hiddens"]
- with tf.variable_scope("actor"):
- self.actor_var_list = tf.get_collection(
- tf.GraphKeys.TRAINABLE_VARIABLES,
- tf.get_variable_scope().name
- )
- self.actor_vars = TensorFlowVariables(self.output_action,
- self.sess)
+ q_out = tf.concat([frontend.last_layer, action_inputs], axis=1)
+ for hidden in hiddens:
+ q_out = layers.fully_connected(
+ q_out, num_outputs=hidden, activation_fn=tf.nn.relu)
+ q_scores = layers.fully_connected(q_out, num_outputs=1, activation_fn=None)
- if (self.config["noise_add"]):
- params = self.config["noise_parameters"]
- self.rand_process = OrnsteinUhlenbeckProcess(size=self.ac_size,
- theta=params["theta"],
- mu=params["mu"],
- sigma=params["sigma"])
- self.epsilon = 1.0
+ return q_scores
- def _setup_critic_loss(self, action_space):
- """Sets up critic loss."""
- self.target_Q = tf.placeholder(tf.float32, [None, 1], name="target_q")
- # compare critic eval to critic_target (squared loss)
- self.reward = tf.placeholder(tf.float32, [None], name="reward")
- self.critic_target = tf.expand_dims(self.reward, 1) + \
- self.config['gamma'] * self.target_Q
- self.critic_loss = tf.reduce_mean(tf.square(
- self.critic_target - self.critic_eval))
+def _huber_loss(x, delta=1.0):
+ """Reference: https://en.wikipedia.org/wiki/Huber_loss"""
+ return tf.where(
+ tf.abs(x) < delta,
+ tf.square(x) * 0.5, delta * (tf.abs(x) - 0.5 * delta))
- def _setup_critic_network(self, obs_space, ac_space):
- """Sets up Q network."""
- with tf.variable_scope("critic", reuse=tf.AUTO_REUSE):
- self.critic_network = DDPGCritic((self.obs, self.act), 1, {})
- self.critic_eval = self.critic_network.outputs
- with tf.variable_scope("critic", reuse=True):
- self.cn_for_loss = DDPGCritic(
- (self.obs, self.output_action), 1, {}).outputs
+def _minimize_and_clip(optimizer, objective, var_list, clip_val=10):
+ """Minimized `objective` using `optimizer` w.r.t. variables in
+ `var_list` while ensure the norm of the gradients for each
+ variable is clipped to `clip_val`
+ """
+ gradients = optimizer.compute_gradients(objective, var_list=var_list)
+ for i, (grad, var) in enumerate(gradients):
+ if grad is not None:
+ gradients[i] = (tf.clip_by_norm(grad, clip_val), var)
+ return gradients
- def _setup_actor_network(self, obs_space, ac_space):
- """Sets up actor network."""
- with tf.variable_scope("actor", reuse=tf.AUTO_REUSE):
- self.actor_network = DDPGActor(
- self.obs, self.ac_size,
- options={"action_bound": self.action_bound})
- self.output_action = self.actor_network.outputs
- def get_weights(self):
- """Returns critic weights, actor weights."""
- return self.critic_vars.get_weights(), self.actor_vars.get_weights()
+def _scope_vars(scope, trainable_only=False):
+ """
+ Get variables inside a scope
+ The scope can be specified as a string
- def set_weights(self, weights):
- """Sets critic and actor weights."""
- critic_weights, actor_weights = weights
- self.critic_vars.set_weights(critic_weights)
- self.actor_vars.set_weights(actor_weights)
+ Parameters
+ ----------
+ scope: str or VariableScope
+ scope in which the variables reside.
+ trainable_only: bool
+ whether or not to return only the variables that were marked as
+ trainable.
- def compute(self, ob):
- """Returns action, given state."""
- flattened_ob = np.reshape(ob, [-1, np.prod(ob.shape)])
- action = self.sess.run(self.output_action, {self.obs: flattened_ob})
- if (self.config["noise_add"]):
- action += self.epsilon * self.rand_process.sample()
- if (self.epsilon > 0):
- self.epsilon -= self.config["noise_epsilon"]
- return action[0], {}
+ Returns
+ -------
+ vars: [tf.Variable]
+ list of variables in `scope`.
+ """
+ return tf.get_collection(
+ tf.GraphKeys.TRAINABLE_VARIABLES
+ if trainable_only else tf.GraphKeys.VARIABLES,
+ scope=scope if isinstance(scope, str) else scope.name)
- def value(self, *args):
- return 0
+
+class ModelAndLoss(object):
+ """Holds the model and loss function.
+
+ Both graphs are necessary in order for the multi-gpu SGD implementation
+ to create towers on each device.
+ """
+
+ def __init__(self, registry, dim_actions, low_action, high_action, config,
+ obs_t, act_t, rew_t, obs_tp1, done_mask, importance_weights):
+ # p network evaluation
+ with tf.variable_scope("p_func", reuse=True) as scope:
+ self.p_t = _build_p_network(registry, obs_t, dim_actions, config)
+
+ # target p network evaluation
+ with tf.variable_scope("target_p_func") as scope:
+ self.p_tp1 = _build_p_network(registry, obs_tp1, dim_actions,
+ config)
+ self.target_p_func_vars = _scope_vars(scope.name)
+
+ # Action outputs
+ with tf.variable_scope("a_func", reuse=True):
+ deterministic_flag = tf.constant(value=False, dtype=tf.bool)
+ zero_eps = tf.constant(value=.0, dtype=tf.float32)
+ output_actions = _build_action_network(
+ self.p_t, low_action, high_action, deterministic_flag,
+ zero_eps, config["exploration_theta"],
+ config["exploration_sigma"])
+
+ output_actions_estimated = _build_action_network(
+ self.p_tp1, low_action, high_action, deterministic_flag,
+ zero_eps, config["exploration_theta"],
+ config["exploration_sigma"])
+
+ # q network evaluation
+ with tf.variable_scope("q_func") as scope:
+ self.q_t = _build_q_network(registry, obs_t, act_t, config)
+ self.q_func_vars = _scope_vars(scope.name)
+ with tf.variable_scope("q_func", reuse=True):
+ self.q_tp0 = _build_q_network(registry, obs_t, output_actions,
+ config)
+
+ # target q network evalution
+ with tf.variable_scope("target_q_func") as scope:
+ self.q_tp1 = _build_q_network(registry, obs_tp1,
+ output_actions_estimated, config)
+ self.target_q_func_vars = _scope_vars(scope.name)
+
+ q_t_selected = tf.squeeze(self.q_t, axis=len(self.q_t.shape) - 1)
+
+ q_tp1_best = tf.squeeze(
+ input=self.q_tp1, axis=len(self.q_tp1.shape) - 1)
+ q_tp1_best_masked = (1.0 - done_mask) * q_tp1_best
+
+ # compute RHS of bellman equation
+ q_t_selected_target = (
+ rew_t + config["gamma"]**config["n_step"] * q_tp1_best_masked)
+
+ # compute the error (potentially clipped)
+ self.td_error = q_t_selected - tf.stop_gradient(q_t_selected_target)
+ if config.get("use_huber"):
+ errors = _huber_loss(self.td_error, config.get("huber_threshold"))
+ else:
+ errors = 0.5 * tf.square(self.td_error)
+
+ weighted_error = tf.reduce_mean(importance_weights * errors)
+
+ self.loss = weighted_error
+
+ # for policy gradient
+ self.actor_loss = -1.0 * tf.reduce_mean(self.q_tp0)
+
+
+class DDPGGraph(object):
+ def __init__(self, registry, env, config, logdir):
+ self.env = env
+ dim_actions = env.action_space.shape[0]
+ low_action = env.action_space.low
+ high_action = env.action_space.high
+ actor_optimizer = tf.train.AdamOptimizer(
+ learning_rate=config["actor_lr"])
+ critic_optimizer = tf.train.AdamOptimizer(
+ learning_rate=config["critic_lr"])
+
+ # Action inputs
+ self.stochastic = tf.placeholder(tf.bool, (), name="stochastic")
+ self.eps = tf.placeholder(tf.float32, (), name="eps")
+ self.cur_observations = tf.placeholder(
+ tf.float32, shape=(None, ) + env.observation_space.shape)
+
+ # Actor: P (policy) network
+ p_scope_name = "p_func"
+ with tf.variable_scope(p_scope_name) as scope:
+ p_values = _build_p_network(registry, self.cur_observations,
+ dim_actions, config)
+ p_func_vars = _scope_vars(scope.name)
+
+ # Action outputs
+ a_scope_name = "a_func"
+ with tf.variable_scope(a_scope_name):
+ self.output_actions = _build_action_network(
+ p_values, low_action, high_action, self.stochastic, self.eps,
+ config["exploration_theta"], config["exploration_sigma"])
+
+ with tf.variable_scope(a_scope_name, reuse=True):
+ exploration_sample = tf.get_variable(name="ornstein_uhlenbeck")
+ self.reset_noise_op = tf.assign(exploration_sample,
+ dim_actions * [.0])
+
+ # Replay inputs
+ self.obs_t = tf.placeholder(
+ tf.float32,
+ shape=(None, ) + env.observation_space.shape,
+ name="observation")
+ self.act_t = tf.placeholder(
+ tf.float32, shape=(None, ) + env.action_space.shape, name="action")
+ self.rew_t = tf.placeholder(tf.float32, [None], name="reward")
+ self.obs_tp1 = tf.placeholder(
+ tf.float32, shape=(None, ) + env.observation_space.shape)
+ self.done_mask = tf.placeholder(tf.float32, [None], name="done")
+ self.importance_weights = tf.placeholder(
+ tf.float32, [None], name="weight")
+
+ def build_loss(obs_t, act_t, rew_t, obs_tp1, done_mask,
+ importance_weights):
+ return ModelAndLoss(registry, dim_actions, low_action, high_action,
+ config, obs_t, act_t, rew_t, obs_tp1,
+ done_mask, importance_weights)
+
+ self.loss_inputs = [
+ ("obs", self.obs_t),
+ ("actions", self.act_t),
+ ("rewards", self.rew_t),
+ ("new_obs", self.obs_tp1),
+ ("dones", self.done_mask),
+ ("weights", self.importance_weights),
+ ]
+
+ loss_obj = build_loss(self.obs_t, self.act_t, self.rew_t, self.obs_tp1,
+ self.done_mask, self.importance_weights)
+
+ self.build_loss = build_loss
+
+ actor_loss = loss_obj.actor_loss
+ weighted_error = loss_obj.loss
+ q_func_vars = loss_obj.q_func_vars
+ target_p_func_vars = loss_obj.target_p_func_vars
+ target_q_func_vars = loss_obj.target_q_func_vars
+ self.p_t = loss_obj.p_t
+ self.q_t = loss_obj.q_t
+ self.q_tp0 = loss_obj.q_tp0
+ self.q_tp1 = loss_obj.q_tp1
+ self.td_error = loss_obj.td_error
+
+ if config["l2_reg"] is not None:
+ for var in p_func_vars:
+ if "bias" not in var.name:
+ actor_loss += config["l2_reg"] * 0.5 * tf.nn.l2_loss(var)
+ for var in q_func_vars:
+ if "bias" not in var.name:
+ weighted_error += config["l2_reg"] * 0.5 * tf.nn.l2_loss(
+ var)
+
+ # compute optimization op (potentially with gradient clipping)
+ if config["grad_norm_clipping"] is not None:
+ self.actor_grads_and_vars = _minimize_and_clip(
+ actor_optimizer,
+ actor_loss,
+ var_list=p_func_vars,
+ clip_val=config["grad_norm_clipping"])
+ self.critic_grads_and_vars = _minimize_and_clip(
+ critic_optimizer,
+ weighted_error,
+ var_list=q_func_vars,
+ clip_val=config["grad_norm_clipping"])
+ else:
+ self.actor_grads_and_vars = actor_optimizer.compute_gradients(
+ actor_loss, var_list=p_func_vars)
+ self.critic_grads_and_vars = critic_optimizer.compute_gradients(
+ weighted_error, var_list=q_func_vars)
+ self.actor_grads_and_vars = [(g, v)
+ for (g, v) in self.actor_grads_and_vars
+ if g is not None]
+ self.critic_grads_and_vars = [(g, v)
+ for (g, v) in self.critic_grads_and_vars
+ if g is not None]
+ self.grads_and_vars = (
+ self.actor_grads_and_vars + self.critic_grads_and_vars)
+ self.grads = [g for (g, v) in self.grads_and_vars]
+ self.actor_train_expr = actor_optimizer.apply_gradients(
+ self.actor_grads_and_vars)
+ self.critic_train_expr = critic_optimizer.apply_gradients(
+ self.critic_grads_and_vars)
+
+ # update_target_fn will be called periodically to copy Q network to
+ # target Q network
+ self.tau_value = config.get("tau")
+ self.tau = tf.placeholder(tf.float32, (), name="tau")
+ update_target_expr = []
+ for var, var_target in zip(
+ sorted(q_func_vars, key=lambda v: v.name),
+ sorted(target_q_func_vars, key=lambda v: v.name)):
+ update_target_expr.append(
+ var_target.assign(self.tau * var +
+ (1.0 - self.tau) * var_target))
+ for var, var_target in zip(
+ sorted(p_func_vars, key=lambda v: v.name),
+ sorted(target_p_func_vars, key=lambda v: v.name)):
+ update_target_expr.append(
+ var_target.assign(self.tau * var +
+ (1.0 - self.tau) * var_target))
+ self.update_target_expr = tf.group(*update_target_expr)
+
+ # support both hard and soft sync
+ def update_target(self, sess, tau=None):
+ return sess.run(
+ self.update_target_expr,
+ feed_dict={self.tau: tau or self.tau_value})
+
+ def act(self, sess, obs, eps, stochastic=True):
+ return sess.run(
+ self.output_actions,
+ feed_dict={
+ self.cur_observations: obs,
+ self.stochastic: stochastic,
+ self.eps: eps
+ })
+
+ def compute_gradients(self, sess, obs_t, act_t, rew_t, obs_tp1, done_mask,
+ importance_weights):
+ td_err, grads = sess.run(
+ [self.td_error, self.grads],
+ feed_dict={
+ self.obs_t: obs_t,
+ self.act_t: act_t,
+ self.rew_t: rew_t,
+ self.obs_tp1: obs_tp1,
+ self.done_mask: done_mask,
+ self.importance_weights: importance_weights
+ })
+ return td_err, grads
+
+ def compute_td_error(self, sess, obs_t, act_t, rew_t, obs_tp1, done_mask,
+ importance_weights):
+ td_err = sess.run(
+ self.td_error,
+ feed_dict={
+ self.obs_t: [np.array(ob) for ob in obs_t],
+ self.act_t: act_t,
+ self.rew_t: rew_t,
+ self.obs_tp1: [np.array(ob) for ob in obs_tp1],
+ self.done_mask: done_mask,
+ self.importance_weights: importance_weights
+ })
+ return td_err
+
+ def apply_gradients(self, sess, grads):
+ assert len(grads) == len(self.grads_and_vars)
+ feed_dict = {ph: g for (g, ph) in zip(grads, self.grads)}
+ sess.run(
+ [self.critic_train_expr, self.actor_train_expr],
+ feed_dict=feed_dict)
+
+ def compute_apply(self, sess, obs_t, act_t, rew_t, obs_tp1, done_mask,
+ importance_weights):
+ td_err, _, _ = sess.run(
+ [self.td_error, self.critic_train_expr, self.actor_train_expr],
+ feed_dict={
+ self.obs_t: obs_t,
+ self.act_t: act_t,
+ self.rew_t: rew_t,
+ self.obs_tp1: obs_tp1,
+ self.done_mask: done_mask,
+ self.importance_weights: importance_weights
+ })
+ return td_err
+
+ def reset_noise(self, sess):
+ sess.run(self.reset_noise_op)
diff --git a/python/ray/rllib/ddpg2/README.md b/python/ray/rllib/ddpg2/README.md
index af64c1530..54dc3996b 100644
--- a/python/ray/rllib/ddpg2/README.md
+++ b/python/ray/rllib/ddpg2/README.md
@@ -1 +1 @@
-Code in this package follows the style of dqn.
+Alternate DDPG implementation. See also https://github.com/ray-project/ray/tree/master/python/ray/rllib/ddpg.
diff --git a/python/ray/rllib/ddpg2/__init__.py b/python/ray/rllib/ddpg2/__init__.py
index ece9c54f3..a7ace46c1 100644
--- a/python/ray/rllib/ddpg2/__init__.py
+++ b/python/ray/rllib/ddpg2/__init__.py
@@ -1,8 +1,3 @@
-from __future__ import absolute_import
-from __future__ import division
-from __future__ import print_function
-
-from ray.rllib.ddpg2.apex import ApexDDPG2Agent
from ray.rllib.ddpg2.ddpg import DDPG2Agent, DEFAULT_CONFIG
-__all__ = ["DDPG2Agent", "ApexDDPG2Agent", "DEFAULT_CONFIG"]
+__all__ = ["DDPG2Agent", "DEFAULT_CONFIG"]
diff --git a/python/ray/rllib/ddpg2/ddpg.py b/python/ray/rllib/ddpg2/ddpg.py
index c3bee0cbd..0de2a865f 100644
--- a/python/ray/rllib/ddpg2/ddpg.py
+++ b/python/ray/rllib/ddpg2/ddpg.py
@@ -2,267 +2,111 @@ from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
-import pickle
-import os
-
import numpy as np
-import tensorflow as tf
import ray
-from ray.rllib import optimizers
-from ray.rllib.ddpg2.ddpg_evaluator import DDPGEvaluator
from ray.rllib.agent import Agent
+from ray.rllib.ddpg2.ddpg_evaluator import DDPGEvaluator, RemoteDDPGEvaluator
+from ray.rllib.optimizers import LocalSyncReplayOptimizer
from ray.tune.result import TrainingResult
-OPTIMIZER_SHARED_CONFIGS = [
- "buffer_size", "prioritized_replay", "prioritized_replay_alpha",
- "prioritized_replay_beta", "prioritized_replay_eps", "sample_batch_size",
- "train_batch_size", "learning_starts", "clip_rewards"
-]
+DEFAULT_CONFIG = {
+ # Actor learning rate
+ "actor_lr": 0.0001,
+ # Critic learning rate
+ "critic_lr": 0.001,
+ # Arguments to pass in to env creator
+ "env_config": {},
+ # MDP Discount factor
+ "gamma": 0.99,
+ # Number of steps after which the rollout gets cut
+ "horizon": 500,
-DEFAULT_CONFIG = dict(
- # === Model ===
- # Hidden layer sizes of the policy networks
- actor_hiddens=[64, 64],
- # Hidden layer sizes of the policy networks
- critic_hiddens=[64, 64],
- # N-step Q learning
- n_step=1,
- # Config options to pass to the model constructor
- model={},
- # Discount factor for the MDP
- gamma=0.99,
- # Arguments to pass to the env creator
- env_config={},
-
- # === Exploration ===
- # Max num timesteps for annealing schedules. Exploration is annealed from
- # 1.0 to exploration_fraction over this number of timesteps scaled by
- # exploration_fraction
- schedule_max_timesteps=100000,
- # Number of env steps to optimize for before returning
- timesteps_per_iteration=1000,
- # Fraction of entire training period over which the exploration rate is
- # annealed
- exploration_fraction=0.1,
- # Final value of random action probability
- exploration_final_eps=0.02,
- # OU-noise scale
- noise_scale=0.1,
- # theta
- exploration_theta=0.15,
- # sigma
- exploration_sigma=0.2,
- # Update the target network every `target_network_update_freq` steps.
- target_network_update_freq=0,
- # Update the target by \tau * policy + (1-\tau) * target_policy
- tau=0.002,
- # Whether to start with random actions instead of noops.
- random_starts=True,
-
- # === Replay buffer ===
- # Size of the replay buffer. Note that if async_updates is set, then
- # each worker will have a replay buffer of this size.
- buffer_size=50000,
- # If True prioritized replay buffer will be used.
- prioritized_replay=True,
- # Alpha parameter for prioritized replay buffer.
- prioritized_replay_alpha=0.6,
- # Beta parameter for sampling from prioritized replay buffer.
- prioritized_replay_beta=0.4,
- # Epsilon to add to the TD errors when updating priorities.
- prioritized_replay_eps=1e-6,
- # Whether to clip rewards to [-1, 1] prior to adding to the replay buffer.
- clip_rewards=True,
-
- # === Optimization ===
- # Learning rate for adam optimizer
- actor_lr=1e-4,
- critic_lr=1e-3,
- # If True, use huber loss instead of squared loss for critic network
- # Conventionally, no need to clip gradients if using a huber loss
- use_huber=False,
- # Threshold of a huber loss
- huber_threshold=1.0,
- # Weights for L2 regularization
- l2_reg=1e-6,
- # If not None, clip gradients during optimization at this value
- grad_norm_clipping=None,
- # How many steps of the model to sample before learning starts.
- learning_starts=1500,
- # Update the replay buffer with this many samples at once. Note that this
- # setting applies per-worker if num_workers > 1.
- sample_batch_size=1,
- # Size of a batched sampled from replay buffer for training. Note that
- # if async_updates is set, then each worker returns gradients for a
- # batch of this size.
- train_batch_size=256,
- # Smooth the current average reward over this many previous episodes.
- smoothing_num_episodes=100,
-
- # === Tensorflow ===
- # Arguments to pass to tensorflow
- tf_session_args={
- "device_count": {
- "CPU": 2
- },
- "log_device_placement": False,
- "allow_soft_placement": True,
- "gpu_options": {
- "allow_growth": True
- },
- "inter_op_parallelism_threads": 1,
- "intra_op_parallelism_threads": 1,
+ # Whether to include parameter noise
+ "noise_add": True,
+ # Linear decay of exploration policy
+ "noise_epsilon": 0.0002,
+ # Parameters for noise process
+ "noise_parameters": {
+ "mu": 0,
+ "sigma": 0.2,
+ "theta": 0.15,
},
- # === Parallelism ===
- # Number of workers for collecting samples with. This only makes sense
- # to increase if your environment is particularly slow to sample, or if
- # you're using the Async or Ape-X optimizers.
- num_workers=0,
- # Whether to allocate GPUs for workers (if > 0).
- num_gpus_per_worker=0,
- # Optimizer class to use.
- optimizer_class="LocalSyncReplayOptimizer",
- # Config to pass to the optimizer.
- optimizer_config=dict(),
- # Whether to use a distribution of epsilons across workers for exploration.
- per_worker_exploration=False,
- # Whether to compute priorities on workers.
- worker_side_prioritization=False)
+ # Number of local steps taken for each call to sample
+ "num_local_steps": 1,
+ # Number of workers (excluding master)
+ "num_workers": 0,
+
+ "optimizer": {
+ # Replay buffer size
+ "buffer_size": 10000,
+ # Number of steps in warm-up phase before learning starts
+ "learning_starts": 500,
+ # Whether to clip rewards
+ "clip_rewards": False,
+ # Whether to use prioritized replay
+ "prioritized_replay": False,
+ # Size of batch sampled from replay buffer
+ "train_batch_size": 64,
+ },
+
+ # Controls how fast target networks move
+ "tau": 0.001,
+ # Number of steps taken per training iteration
+ "train_steps": 600,
+}
class DDPG2Agent(Agent):
_agent_name = "DDPG2"
- _allow_unknown_subkeys = [
- "model", "optimizer", "tf_session_args", "env_config"
- ]
_default_config = DEFAULT_CONFIG
def _init(self):
- self.local_evaluator = DDPGEvaluator(self.registry, self.env_creator,
- self.config, self.logdir, 0)
- remote_cls = ray.remote(
- num_cpus=1,
- num_gpus=self.config["num_gpus_per_worker"])(DDPGEvaluator)
+ self.local_evaluator = DDPGEvaluator(
+ self.registry, self.env_creator, self.config)
self.remote_evaluators = [
- remote_cls.remote(self.registry, self.env_creator, self.config,
- self.logdir, i)
- for i in range(self.config["num_workers"])
- ]
-
- for k in OPTIMIZER_SHARED_CONFIGS:
- if k not in self.config["optimizer_config"]:
- self.config["optimizer_config"][k] = self.config[k]
-
- self.optimizer = getattr(optimizers, self.config["optimizer_class"])(
- self.config["optimizer_config"], self.local_evaluator,
+ RemoteDDPGEvaluator.remote(
+ self.registry, self.env_creator, self.config)
+ for _ in range(self.config["num_workers"])]
+ self.optimizer = LocalSyncReplayOptimizer(
+ self.config["optimizer"], self.local_evaluator,
self.remote_evaluators)
- self.saver = tf.train.Saver(max_to_keep=None)
- self.last_target_update_ts = 0
- self.num_target_updates = 0
-
- @property
- def global_timestep(self):
- return self.optimizer.num_steps_sampled
-
- def update_target_if_needed(self):
- if self.global_timestep - self.last_target_update_ts > \
- self.config["target_network_update_freq"]:
- self.local_evaluator.update_target()
- self.last_target_update_ts = self.global_timestep
- self.num_target_updates += 1
-
def _train(self):
- start_timestep = self.global_timestep
-
- while (self.global_timestep - start_timestep <
- self.config["timesteps_per_iteration"]):
-
+ for _ in range(self.config["train_steps"]):
self.optimizer.step()
- self.update_target_if_needed()
+ # update target
+ if self.optimizer.num_steps_trained > 0:
+ self.local_evaluator.update_target()
- self.local_evaluator.set_global_timestep(self.global_timestep)
- for e in self.remote_evaluators:
- e.set_global_timestep.remote(self.global_timestep)
+ # generate training result
+ return self._fetch_metrics()
- return self._train_stats(start_timestep)
-
- def _train_stats(self, start_timestep):
- if self.remote_evaluators:
- stats = ray.get([e.stats.remote() for e in self.remote_evaluators])
+ def _fetch_metrics(self):
+ episode_rewards = []
+ episode_lengths = []
+ if self.config["num_workers"] > 0:
+ metric_lists = [a.get_completed_rollout_metrics.remote()
+ for a in self.remote_evaluators]
+ for metrics in metric_lists:
+ for episode in ray.get(metrics):
+ episode_lengths.append(episode.episode_length)
+ episode_rewards.append(episode.episode_reward)
else:
- stats = self.local_evaluator.stats()
- if not isinstance(stats, list):
- stats = [stats]
+ metrics = self.local_evaluator.get_completed_rollout_metrics()
+ for episode in metrics:
+ episode_lengths.append(episode.episode_length)
+ episode_rewards.append(episode.episode_reward)
- mean_100ep_reward = 0.0
- mean_100ep_length = 0.0
- num_episodes = 0
- explorations = []
-
- if self.config["per_worker_exploration"]:
- # Return stats from workers with the lowest 20% of exploration
- test_stats = stats[-int(max(1, len(stats) * 0.2)):]
- else:
- test_stats = stats
-
- for s in test_stats:
- mean_100ep_reward += s["mean_100ep_reward"] / len(test_stats)
- mean_100ep_length += s["mean_100ep_length"] / len(test_stats)
-
- for s in stats:
- num_episodes += s["num_episodes"]
- explorations.append(s["exploration"])
-
- opt_stats = self.optimizer.stats()
+ avg_reward = (np.mean(episode_rewards))
+ avg_length = (np.mean(episode_lengths))
+ timesteps = np.sum(episode_lengths)
result = TrainingResult(
- episode_reward_mean=mean_100ep_reward,
- episode_len_mean=mean_100ep_length,
- episodes_total=num_episodes,
- timesteps_this_iter=self.global_timestep - start_timestep,
- info=dict({
- "min_exploration": min(explorations),
- "max_exploration": max(explorations),
- "num_target_updates": self.num_target_updates,
- }, **opt_stats))
+ episode_reward_mean=avg_reward,
+ episode_len_mean=avg_length,
+ timesteps_this_iter=timesteps,
+ info={})
return result
-
- def _stop(self):
- # workaround for https://github.com/ray-project/ray/issues/1516
- for ev in self.remote_evaluators:
- ev.__ray_terminate__.remote(ev._ray_actor_id.id())
-
- def _save(self, checkpoint_dir):
- checkpoint_path = self.saver.save(
- self.local_evaluator.sess,
- os.path.join(checkpoint_dir, "checkpoint"),
- global_step=self.iteration)
- extra_data = [
- self.local_evaluator.save(),
- ray.get([e.save.remote() for e in self.remote_evaluators]),
- self.optimizer.save(), self.num_target_updates,
- self.last_target_update_ts
- ]
- pickle.dump(extra_data, open(checkpoint_path + ".extra_data", "wb"))
- return checkpoint_path
-
- def _restore(self, checkpoint_path):
- self.saver.restore(self.local_evaluator.sess, checkpoint_path)
- extra_data = pickle.load(open(checkpoint_path + ".extra_data", "rb"))
- self.local_evaluator.restore(extra_data[0])
- ray.get([
- e.restore.remote(d)
- for (d, e) in zip(extra_data[1], self.remote_evaluators)
- ])
- self.optimizer.restore(extra_data[2])
- self.num_target_updates = extra_data[3]
- self.last_target_update_ts = extra_data[4]
-
- def compute_action(self, observation):
- return self.local_evaluator.ddpg_graph.act(self.local_evaluator.sess,
- np.array(observation)[None],
- 0.0)[0]
diff --git a/python/ray/rllib/ddpg2/ddpg_evaluator.py b/python/ray/rllib/ddpg2/ddpg_evaluator.py
index e177a37a1..8a5ab5ed3 100644
--- a/python/ray/rllib/ddpg2/ddpg_evaluator.py
+++ b/python/ray/rllib/ddpg2/ddpg_evaluator.py
@@ -2,185 +2,74 @@ from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
-from gym.spaces import Box
import numpy as np
-import tensorflow as tf
import ray
-from ray.rllib.utils.error import UnsupportedSpaceException
-from ray.rllib.ddpg2 import models
-from ray.rllib.dqn.common.schedules import ConstantSchedule, LinearSchedule
-from ray.rllib.optimizers import SampleBatch, PolicyEvaluator
-from ray.rllib.utils.compression import pack
-from ray.rllib.dqn.dqn_evaluator import adjust_nstep
-from ray.rllib.dqn.common.wrappers import wrap_dqn
+from ray.rllib.ddpg2.models import DDPGModel
+from ray.rllib.models.catalog import ModelCatalog
+from ray.rllib.optimizers import PolicyEvaluator
+from ray.rllib.utils.filter import NoFilter
+from ray.rllib.utils.process_rollout import process_rollout
+from ray.rllib.utils.sampler import SyncSampler
class DDPGEvaluator(PolicyEvaluator):
- """The base DDPG Evaluator."""
- def __init__(self, registry, env_creator, config, logdir, worker_index):
- env = env_creator(config["env_config"])
- env = wrap_dqn(registry, env, config["model"], config["random_starts"])
- self.env = env
- self.config = config
+ def __init__(self, registry, env_creator, config):
+ self.env = ModelCatalog.get_preprocessor_as_wrapper(
+ registry, env_creator(config["env_config"]))
- # when env.action_space is of Box type, e.g., Pendulum-v0
- # action_space.low is [-2.0], high is [2.0]
- # take action by calling, e.g., env.step([3.5])
- if not isinstance(env.action_space, Box):
- raise UnsupportedSpaceException(
- "Action space {} is not supported for DDPG.".format(
- env.action_space))
+ # contains model, target_model
+ self.model = DDPGModel(registry, self.env, config)
- tf_config = tf.ConfigProto(**config["tf_session_args"])
- self.sess = tf.Session(config=tf_config)
- self.ddpg_graph = models.DDPGGraph(registry, env, config, logdir)
-
- # Use either a different `eps` per worker, or a linear schedule.
- if config["per_worker_exploration"]:
- assert config["num_workers"] > 1, "This requires multiple workers"
- self.exploration = ConstantSchedule(
- config["noise_scale"] * 0.4 **
- (1 + worker_index / float(config["num_workers"] - 1) * 7))
- else:
- self.exploration = LinearSchedule(
- schedule_timesteps=int(config["exploration_fraction"] *
- config["schedule_max_timesteps"]),
- initial_p=config["noise_scale"] * 1.0,
- final_p=config["noise_scale"] *
- config["exploration_final_eps"])
-
- # Initialize the parameters and copy them to the target network.
- self.sess.run(tf.global_variables_initializer())
- # hard instead of soft
- self.ddpg_graph.update_target(self.sess, 1.0)
- self.global_timestep = 0
- self.local_timestep = 0
-
- # Note that this encompasses both the policy and Q-value networks and
- # their corresponding target networks
- self.variables = ray.experimental.TensorFlowVariables(
- tf.group(self.ddpg_graph.q_tp0, self.ddpg_graph.q_tp1), self.sess)
-
- self.episode_rewards = [0.0]
- self.episode_lengths = [0.0]
- self.saved_mean_reward = None
-
- self.obs = self.env.reset()
-
- def set_global_timestep(self, global_timestep):
- self.global_timestep = global_timestep
-
- def update_target(self):
- self.ddpg_graph.update_target(self.sess)
+ self.sampler = SyncSampler(
+ self.env, self.model.model, NoFilter(),
+ config["num_local_steps"], horizon=config["horizon"])
def sample(self):
- obs, actions, rewards, new_obs, dones = [], [], [], [], []
- for _ in range(
- self.config["sample_batch_size"] + self.config["n_step"] - 1):
- ob, act, rew, ob1, done = self._step(self.global_timestep)
- obs.append(ob)
- actions.append(act)
- rewards.append(rew)
- new_obs.append(ob1)
- dones.append(done)
+ """Returns a batch of samples."""
- # N-step Q adjustments
- if self.config["n_step"] > 1:
- # Adjust for steps lost from truncation
- self.local_timestep -= (self.config["n_step"] - 1)
- adjust_nstep(self.config["n_step"], self.config["gamma"], obs,
- actions, rewards, new_obs, dones)
+ rollout = self.sampler.get_data()
+ rollout.data["weights"] = np.ones_like(rollout.data["rewards"])
- batch = SampleBatch({
- "obs": [pack(np.array(o)) for o in obs],
- "actions": actions,
- "rewards": rewards,
- "new_obs": [pack(np.array(o)) for o in new_obs],
- "dones": dones,
- "weights": np.ones_like(rewards)
- })
- assert (batch.count == self.config["sample_batch_size"])
+ # since each sample is one step, no discounting needs to be applied;
+ # this does not involve config["gamma"]
+ samples = process_rollout(
+ rollout, NoFilter(),
+ gamma=1.0, use_gae=False)
- # Prioritize on the worker side
- if self.config["worker_side_prioritization"]:
- td_errors = self.ddpg_graph.compute_td_error(
- self.sess, obs, batch["actions"], batch["rewards"], new_obs,
- batch["dones"], batch["weights"])
- new_priorities = (
- np.abs(td_errors) + self.config["prioritized_replay_eps"])
- batch.data["weights"] = new_priorities
+ return samples
- return batch
+ def update_target(self):
+ """Updates target critic and target actor."""
+ self.model.update_target()
def compute_gradients(self, samples):
- td_err, grads = self.ddpg_graph.compute_gradients(
- self.sess, samples["obs"], samples["actions"], samples["rewards"],
- samples["new_obs"], samples["dones"], samples["weights"])
- return grads, {"td_error": td_err}
+ """Returns critic, actor gradients."""
+ return self.model.compute_gradients(samples)
def apply_gradients(self, grads):
- self.ddpg_graph.apply_gradients(self.sess, grads)
+ """Applies gradients to evaluator weights."""
+ self.model.apply_gradients(grads)
def compute_apply(self, samples):
- td_error = self.ddpg_graph.compute_apply(
- self.sess, samples["obs"], samples["actions"], samples["rewards"],
- samples["new_obs"], samples["dones"], samples["weights"])
- return {"td_error": td_error}
+ grads, _ = self.compute_gradients(samples)
+ self.apply_gradients(grads)
def get_weights(self):
- return self.variables.get_weights()
+ """Returns model weights."""
+ return self.model.get_weights()
def set_weights(self, weights):
- self.variables.set_weights(weights)
+ """Sets model weights."""
+ self.model.set_weights(weights)
- def _step(self, global_timestep):
- """Takes a single step, and returns the result of the step."""
- action = self.ddpg_graph.act(
- self.sess,
- np.array(self.obs)[None],
- self.exploration.value(global_timestep))[0]
- new_obs, rew, done, _ = self.env.step(action)
- ret = (self.obs, action, rew, new_obs, float(done))
- self.obs = new_obs
- self.episode_rewards[-1] += rew
- self.episode_lengths[-1] += 1
- if done:
- self.obs = self.env.reset()
- self.episode_rewards.append(0.0)
- self.episode_lengths.append(0.0)
- # reset UO noise for each episode
- self.ddpg_graph.reset_noise(self.sess)
+ def get_completed_rollout_metrics(self):
+ """Returns metrics on previously completed rollouts.
- self.local_timestep += 1
- return ret
+ Calling this clears the queue of completed rollout metrics.
+ """
+ return self.sampler.get_metrics()
- def stats(self):
- n = self.config["smoothing_num_episodes"] + 1
- mean_100ep_reward = round(np.mean(self.episode_rewards[-n:-1]), 5)
- mean_100ep_length = round(np.mean(self.episode_lengths[-n:-1]), 5)
- exploration = self.exploration.value(self.global_timestep)
- return {
- "mean_100ep_reward": mean_100ep_reward,
- "mean_100ep_length": mean_100ep_length,
- "num_episodes": len(self.episode_rewards),
- "exploration": exploration,
- "local_timestep": self.local_timestep,
- }
- def save(self):
- return [
- self.exploration, self.episode_rewards, self.episode_lengths,
- self.saved_mean_reward, self.obs, self.global_timestep,
- self.local_timestep
- ]
-
- def restore(self, data):
- self.exploration = data[0]
- self.episode_rewards = data[1]
- self.episode_lengths = data[2]
- self.saved_mean_reward = data[3]
- self.obs = data[4]
- self.global_timestep = data[5]
- self.local_timestep = data[6]
+RemoteDDPGEvaluator = ray.remote(DDPGEvaluator)
diff --git a/python/ray/rllib/ddpg2/models.py b/python/ray/rllib/ddpg2/models.py
index d58f37dc6..e785f518f 100644
--- a/python/ray/rllib/ddpg2/models.py
+++ b/python/ray/rllib/ddpg2/models.py
@@ -3,389 +3,239 @@ from __future__ import division
from __future__ import print_function
import numpy as np
-
import tensorflow as tf
-import tensorflow.contrib.layers as layers
-from ray.rllib.models import ModelCatalog
+from ray.experimental.tfutils import TensorFlowVariables
+from ray.rllib.models.ddpgnet import DDPGActor, DDPGCritic
+from ray.rllib.ddpg2.random_process import OrnsteinUhlenbeckProcess
-def _build_p_network(registry, inputs, dim_actions, config):
- """
- map an observation (i.e., state) to an action where
- each entry takes value from (0, 1) due to the sigmoid function
- """
- frontend = ModelCatalog.get_model(registry, inputs, 1, config["model"])
+class DDPGModel():
+ def __init__(self, registry, env, config):
+ self.config = config
+ self.sess = tf.Session()
- hiddens = config["actor_hiddens"]
- action_out = frontend.last_layer
- for hidden in hiddens:
- action_out = layers.fully_connected(
- action_out, num_outputs=hidden, activation_fn=tf.nn.relu)
- # Use sigmoid layer to bound values within (0, 1)
- # shape of action_scores is [batch_size, dim_actions]
- action_scores = layers.fully_connected(
- action_out, num_outputs=dim_actions, activation_fn=tf.nn.sigmoid)
+ with tf.variable_scope("model"):
+ self.model = DDPGActorCritic(
+ registry, env, self.config, self.sess)
+ with tf.variable_scope("target_model"):
+ self.target_model = DDPGActorCritic(
+ registry, env, self.config, self.sess)
+ self._setup_gradients()
+ self._setup_target_updates()
- return action_scores
+ self.initialize()
+ self._initialize_target_weights()
+
+ def initialize(self):
+ self.sess.run(tf.global_variables_initializer())
+
+ def _initialize_target_weights(self):
+ """Set initial target weights to match model weights."""
+ a_updates = []
+ for var, target_var in zip(
+ self.model.actor_var_list, self.target_model.actor_var_list):
+ a_updates.append(tf.assign(target_var, var))
+ actor_updates = tf.group(*a_updates)
+
+ c_updates = []
+ for var, target_var in zip(
+ self.model.critic_var_list, self.target_model.critic_var_list):
+ c_updates.append(tf.assign(target_var, var))
+ critic_updates = tf.group(*c_updates)
+ self.sess.run([actor_updates, critic_updates])
+
+ def _setup_gradients(self):
+ """Setup critic and actor gradients."""
+ self.critic_grads = tf.gradients(
+ self.model.critic_loss, self.model.critic_var_list)
+ c_grads_and_vars = list(zip(
+ self.critic_grads, self.model.critic_var_list))
+ c_opt = tf.train.AdamOptimizer(self.config["critic_lr"])
+ self._apply_c_gradients = c_opt.apply_gradients(c_grads_and_vars)
+
+ self.actor_grads = tf.gradients(
+ -self.model.cn_for_loss, self.model.actor_var_list)
+ a_grads_and_vars = list(zip(
+ self.actor_grads, self.model.actor_var_list))
+ a_opt = tf.train.AdamOptimizer(self.config["actor_lr"])
+ self._apply_a_gradients = a_opt.apply_gradients(a_grads_and_vars)
+
+ def compute_gradients(self, samples):
+ """ Returns gradient w.r.t. samples."""
+ # actor gradients
+ actor_actions = self.sess.run(
+ self.model.output_action,
+ feed_dict={self.model.obs: samples["obs"]}
+ )
+
+ actor_feed_dict = {
+ self.model.obs: samples["obs"],
+ self.model.output_action: actor_actions,
+ }
+ self.actor_grads = [g for g in self.actor_grads if g is not None]
+ actor_grad = self.sess.run(self.actor_grads, feed_dict=actor_feed_dict)
+
+ # feed samples into target actor
+ target_Q_act = self.sess.run(
+ self.target_model.output_action,
+ feed_dict={self.target_model.obs: samples["new_obs"]}
+ )
+ target_Q_dict = {
+ self.target_model.obs: samples["new_obs"],
+ self.target_model.act: target_Q_act,
+ }
+
+ target_Q = self.sess.run(
+ self.target_model.critic_eval, feed_dict=target_Q_dict)
+
+ # critic gradients
+ critic_feed_dict = {
+ self.model.obs: samples["obs"],
+ self.model.act: samples["actions"],
+ self.model.reward: samples["rewards"],
+ self.model.target_Q: target_Q,
+ }
+ self.critic_grads = [g for g in self.critic_grads if g is not None]
+ critic_grad = self.sess.run(
+ self.critic_grads, feed_dict=critic_feed_dict)
+ return (critic_grad, actor_grad), {}
+
+ def apply_gradients(self, grads):
+ """Applies gradients to evaluator weights."""
+ c_grads, a_grads = grads
+ critic_feed_dict = dict(zip(self.critic_grads, c_grads))
+ self.sess.run(self._apply_c_gradients, feed_dict=critic_feed_dict)
+ actor_feed_dict = dict(zip(self.actor_grads, a_grads))
+ self.sess.run(self._apply_a_gradients, feed_dict=actor_feed_dict)
+
+ def get_weights(self):
+ """Returns model weights, target model weights."""
+ return self.model.get_weights(), self.target_model.get_weights()
+
+ def set_weights(self, weights):
+ """Sets model and target model weights."""
+ model_weights, target_model_weights = weights
+ self.model.set_weights(model_weights)
+ self.target_model.set_weights(target_model_weights)
+
+ def _setup_target_updates(self):
+ """Set up target actor and critic updates."""
+ a_updates = []
+ tau = self.config["tau"]
+ for var, target_var in zip(
+ self.model.actor_var_list, self.target_model.actor_var_list):
+ a_updates.append(tf.assign(
+ target_var, tau * var + (1. - tau) * target_var))
+ actor_updates = tf.group(*a_updates)
+
+ c_updates = []
+ for var, target_var in zip(
+ self.model.critic_var_list, self.target_model.critic_var_list):
+ c_updates.append(tf.assign(
+ target_var, tau * var + (1. - tau) * target_var))
+ critic_updates = tf.group(*c_updates)
+ self.target_updates = [actor_updates, critic_updates]
+
+ def update_target(self):
+ """Updates target critic and target actor."""
+ self.sess.run(self.target_updates)
-# As a stochastic policy for inference, but a deterministic policy for training
-# thus ignore batch_size issue when constructing a stochastic action
-def _build_action_network(p_values, low_action, high_action, stochastic, eps,
- theta, sigma):
- # shape is [None, dim_action]
- deterministic_actions = (high_action - low_action) * p_values + low_action
+class DDPGActorCritic():
+ other_output = []
+ is_recurrent = False
- exploration_sample = tf.get_variable(
- name="ornstein_uhlenbeck",
- dtype=tf.float32,
- initializer=low_action.size * [.0],
- trainable=False)
- normal_sample = tf.random_normal(
- shape=[low_action.size], mean=0.0, stddev=1.0)
- exploration_value = tf.assign_add(
- exploration_sample,
- theta * (.0 - exploration_sample) + sigma * normal_sample)
- stochastic_actions = deterministic_actions + eps * (
- high_action - low_action) * exploration_value
+ def __init__(self, registry, env, config, sess):
+ self.config = config
+ self.sess = sess
- return tf.cond(stochastic, lambda: stochastic_actions,
- lambda: deterministic_actions)
+ obs_space = env.observation_space
+ ac_space = env.action_space
+ self.obs_size = int(np.prod(obs_space.shape))
+ self.obs = tf.placeholder(tf.float32, [None, self.obs_size])
+ self.ac_size = int(np.prod(ac_space.shape))
+ self.act = tf.placeholder(tf.float32, [None, self.ac_size])
+ self.action_bound = env.action_space.high
+ # TODO: change action_bound to make more general
-def _build_q_network(registry, inputs, action_inputs, config):
- frontend = ModelCatalog.get_model(registry, inputs, 1, config["model"])
+ self._setup_actor_network(obs_space, ac_space)
+ self._setup_critic_network(obs_space, ac_space)
+ self._setup_critic_loss(ac_space)
- hiddens = config["critic_hiddens"]
+ with tf.variable_scope("critic"):
+ self.critic_var_list = tf.get_collection(
+ tf.GraphKeys.TRAINABLE_VARIABLES,
+ tf.get_variable_scope().name
+ )
+ self.critic_vars = TensorFlowVariables(self.critic_loss,
+ self.sess)
- q_out = tf.concat([frontend.last_layer, action_inputs], axis=1)
- for hidden in hiddens:
- q_out = layers.fully_connected(
- q_out, num_outputs=hidden, activation_fn=tf.nn.relu)
- q_scores = layers.fully_connected(q_out, num_outputs=1, activation_fn=None)
+ with tf.variable_scope("actor"):
+ self.actor_var_list = tf.get_collection(
+ tf.GraphKeys.TRAINABLE_VARIABLES,
+ tf.get_variable_scope().name
+ )
+ self.actor_vars = TensorFlowVariables(self.output_action,
+ self.sess)
- return q_scores
+ if (self.config["noise_add"]):
+ params = self.config["noise_parameters"]
+ self.rand_process = OrnsteinUhlenbeckProcess(size=self.ac_size,
+ theta=params["theta"],
+ mu=params["mu"],
+ sigma=params["sigma"])
+ self.epsilon = 1.0
+ def _setup_critic_loss(self, action_space):
+ """Sets up critic loss."""
+ self.target_Q = tf.placeholder(tf.float32, [None, 1], name="target_q")
-def _huber_loss(x, delta=1.0):
- """Reference: https://en.wikipedia.org/wiki/Huber_loss"""
- return tf.where(
- tf.abs(x) < delta,
- tf.square(x) * 0.5, delta * (tf.abs(x) - 0.5 * delta))
+ # compare critic eval to critic_target (squared loss)
+ self.reward = tf.placeholder(tf.float32, [None], name="reward")
+ self.critic_target = tf.expand_dims(self.reward, 1) + \
+ self.config['gamma'] * self.target_Q
+ self.critic_loss = tf.reduce_mean(tf.square(
+ self.critic_target - self.critic_eval))
+ def _setup_critic_network(self, obs_space, ac_space):
+ """Sets up Q network."""
+ with tf.variable_scope("critic", reuse=tf.AUTO_REUSE):
+ self.critic_network = DDPGCritic((self.obs, self.act), 1, {})
+ self.critic_eval = self.critic_network.outputs
-def _minimize_and_clip(optimizer, objective, var_list, clip_val=10):
- """Minimized `objective` using `optimizer` w.r.t. variables in
- `var_list` while ensure the norm of the gradients for each
- variable is clipped to `clip_val`
- """
- gradients = optimizer.compute_gradients(objective, var_list=var_list)
- for i, (grad, var) in enumerate(gradients):
- if grad is not None:
- gradients[i] = (tf.clip_by_norm(grad, clip_val), var)
- return gradients
+ with tf.variable_scope("critic", reuse=True):
+ self.cn_for_loss = DDPGCritic(
+ (self.obs, self.output_action), 1, {}).outputs
+ def _setup_actor_network(self, obs_space, ac_space):
+ """Sets up actor network."""
+ with tf.variable_scope("actor", reuse=tf.AUTO_REUSE):
+ self.actor_network = DDPGActor(
+ self.obs, self.ac_size,
+ options={"action_bound": self.action_bound})
+ self.output_action = self.actor_network.outputs
-def _scope_vars(scope, trainable_only=False):
- """
- Get variables inside a scope
- The scope can be specified as a string
+ def get_weights(self):
+ """Returns critic weights, actor weights."""
+ return self.critic_vars.get_weights(), self.actor_vars.get_weights()
- Parameters
- ----------
- scope: str or VariableScope
- scope in which the variables reside.
- trainable_only: bool
- whether or not to return only the variables that were marked as
- trainable.
+ def set_weights(self, weights):
+ """Sets critic and actor weights."""
+ critic_weights, actor_weights = weights
+ self.critic_vars.set_weights(critic_weights)
+ self.actor_vars.set_weights(actor_weights)
- Returns
- -------
- vars: [tf.Variable]
- list of variables in `scope`.
- """
- return tf.get_collection(
- tf.GraphKeys.TRAINABLE_VARIABLES
- if trainable_only else tf.GraphKeys.VARIABLES,
- scope=scope if isinstance(scope, str) else scope.name)
+ def compute(self, ob):
+ """Returns action, given state."""
+ flattened_ob = np.reshape(ob, [-1, np.prod(ob.shape)])
+ action = self.sess.run(self.output_action, {self.obs: flattened_ob})
+ if (self.config["noise_add"]):
+ action += self.epsilon * self.rand_process.sample()
+ if (self.epsilon > 0):
+ self.epsilon -= self.config["noise_epsilon"]
+ return action[0], {}
-
-class ModelAndLoss(object):
- """Holds the model and loss function.
-
- Both graphs are necessary in order for the multi-gpu SGD implementation
- to create towers on each device.
- """
-
- def __init__(self, registry, dim_actions, low_action, high_action, config,
- obs_t, act_t, rew_t, obs_tp1, done_mask, importance_weights):
- # p network evaluation
- with tf.variable_scope("p_func", reuse=True) as scope:
- self.p_t = _build_p_network(registry, obs_t, dim_actions, config)
-
- # target p network evaluation
- with tf.variable_scope("target_p_func") as scope:
- self.p_tp1 = _build_p_network(registry, obs_tp1, dim_actions,
- config)
- self.target_p_func_vars = _scope_vars(scope.name)
-
- # Action outputs
- with tf.variable_scope("a_func", reuse=True):
- deterministic_flag = tf.constant(value=False, dtype=tf.bool)
- zero_eps = tf.constant(value=.0, dtype=tf.float32)
- output_actions = _build_action_network(
- self.p_t, low_action, high_action, deterministic_flag,
- zero_eps, config["exploration_theta"],
- config["exploration_sigma"])
-
- output_actions_estimated = _build_action_network(
- self.p_tp1, low_action, high_action, deterministic_flag,
- zero_eps, config["exploration_theta"],
- config["exploration_sigma"])
-
- # q network evaluation
- with tf.variable_scope("q_func") as scope:
- self.q_t = _build_q_network(registry, obs_t, act_t, config)
- self.q_func_vars = _scope_vars(scope.name)
- with tf.variable_scope("q_func", reuse=True):
- self.q_tp0 = _build_q_network(registry, obs_t, output_actions,
- config)
-
- # target q network evalution
- with tf.variable_scope("target_q_func") as scope:
- self.q_tp1 = _build_q_network(registry, obs_tp1,
- output_actions_estimated, config)
- self.target_q_func_vars = _scope_vars(scope.name)
-
- q_t_selected = tf.squeeze(self.q_t, axis=len(self.q_t.shape) - 1)
-
- q_tp1_best = tf.squeeze(
- input=self.q_tp1, axis=len(self.q_tp1.shape) - 1)
- q_tp1_best_masked = (1.0 - done_mask) * q_tp1_best
-
- # compute RHS of bellman equation
- q_t_selected_target = (
- rew_t + config["gamma"]**config["n_step"] * q_tp1_best_masked)
-
- # compute the error (potentially clipped)
- self.td_error = q_t_selected - tf.stop_gradient(q_t_selected_target)
- if config.get("use_huber"):
- errors = _huber_loss(self.td_error, config.get("huber_threshold"))
- else:
- errors = 0.5 * tf.square(self.td_error)
-
- weighted_error = tf.reduce_mean(importance_weights * errors)
-
- self.loss = weighted_error
-
- # for policy gradient
- self.actor_loss = -1.0 * tf.reduce_mean(self.q_tp0)
-
-
-class DDPGGraph(object):
- def __init__(self, registry, env, config, logdir):
- self.env = env
- dim_actions = env.action_space.shape[0]
- low_action = env.action_space.low
- high_action = env.action_space.high
- actor_optimizer = tf.train.AdamOptimizer(
- learning_rate=config["actor_lr"])
- critic_optimizer = tf.train.AdamOptimizer(
- learning_rate=config["critic_lr"])
-
- # Action inputs
- self.stochastic = tf.placeholder(tf.bool, (), name="stochastic")
- self.eps = tf.placeholder(tf.float32, (), name="eps")
- self.cur_observations = tf.placeholder(
- tf.float32, shape=(None, ) + env.observation_space.shape)
-
- # Actor: P (policy) network
- p_scope_name = "p_func"
- with tf.variable_scope(p_scope_name) as scope:
- p_values = _build_p_network(registry, self.cur_observations,
- dim_actions, config)
- p_func_vars = _scope_vars(scope.name)
-
- # Action outputs
- a_scope_name = "a_func"
- with tf.variable_scope(a_scope_name):
- self.output_actions = _build_action_network(
- p_values, low_action, high_action, self.stochastic, self.eps,
- config["exploration_theta"], config["exploration_sigma"])
-
- with tf.variable_scope(a_scope_name, reuse=True):
- exploration_sample = tf.get_variable(name="ornstein_uhlenbeck")
- self.reset_noise_op = tf.assign(exploration_sample,
- dim_actions * [.0])
-
- # Replay inputs
- self.obs_t = tf.placeholder(
- tf.float32,
- shape=(None, ) + env.observation_space.shape,
- name="observation")
- self.act_t = tf.placeholder(
- tf.float32, shape=(None, ) + env.action_space.shape, name="action")
- self.rew_t = tf.placeholder(tf.float32, [None], name="reward")
- self.obs_tp1 = tf.placeholder(
- tf.float32, shape=(None, ) + env.observation_space.shape)
- self.done_mask = tf.placeholder(tf.float32, [None], name="done")
- self.importance_weights = tf.placeholder(
- tf.float32, [None], name="weight")
-
- def build_loss(obs_t, act_t, rew_t, obs_tp1, done_mask,
- importance_weights):
- return ModelAndLoss(registry, dim_actions, low_action, high_action,
- config, obs_t, act_t, rew_t, obs_tp1,
- done_mask, importance_weights)
-
- self.loss_inputs = [
- ("obs", self.obs_t),
- ("actions", self.act_t),
- ("rewards", self.rew_t),
- ("new_obs", self.obs_tp1),
- ("dones", self.done_mask),
- ("weights", self.importance_weights),
- ]
-
- loss_obj = build_loss(self.obs_t, self.act_t, self.rew_t, self.obs_tp1,
- self.done_mask, self.importance_weights)
-
- self.build_loss = build_loss
-
- actor_loss = loss_obj.actor_loss
- weighted_error = loss_obj.loss
- q_func_vars = loss_obj.q_func_vars
- target_p_func_vars = loss_obj.target_p_func_vars
- target_q_func_vars = loss_obj.target_q_func_vars
- self.p_t = loss_obj.p_t
- self.q_t = loss_obj.q_t
- self.q_tp0 = loss_obj.q_tp0
- self.q_tp1 = loss_obj.q_tp1
- self.td_error = loss_obj.td_error
-
- if config["l2_reg"] is not None:
- for var in p_func_vars:
- if "bias" not in var.name:
- actor_loss += config["l2_reg"] * 0.5 * tf.nn.l2_loss(var)
- for var in q_func_vars:
- if "bias" not in var.name:
- weighted_error += config["l2_reg"] * 0.5 * tf.nn.l2_loss(
- var)
-
- # compute optimization op (potentially with gradient clipping)
- if config["grad_norm_clipping"] is not None:
- self.actor_grads_and_vars = _minimize_and_clip(
- actor_optimizer,
- actor_loss,
- var_list=p_func_vars,
- clip_val=config["grad_norm_clipping"])
- self.critic_grads_and_vars = _minimize_and_clip(
- critic_optimizer,
- weighted_error,
- var_list=q_func_vars,
- clip_val=config["grad_norm_clipping"])
- else:
- self.actor_grads_and_vars = actor_optimizer.compute_gradients(
- actor_loss, var_list=p_func_vars)
- self.critic_grads_and_vars = critic_optimizer.compute_gradients(
- weighted_error, var_list=q_func_vars)
- self.actor_grads_and_vars = [(g, v)
- for (g, v) in self.actor_grads_and_vars
- if g is not None]
- self.critic_grads_and_vars = [(g, v)
- for (g, v) in self.critic_grads_and_vars
- if g is not None]
- self.grads_and_vars = (
- self.actor_grads_and_vars + self.critic_grads_and_vars)
- self.grads = [g for (g, v) in self.grads_and_vars]
- self.actor_train_expr = actor_optimizer.apply_gradients(
- self.actor_grads_and_vars)
- self.critic_train_expr = critic_optimizer.apply_gradients(
- self.critic_grads_and_vars)
-
- # update_target_fn will be called periodically to copy Q network to
- # target Q network
- self.tau_value = config.get("tau")
- self.tau = tf.placeholder(tf.float32, (), name="tau")
- update_target_expr = []
- for var, var_target in zip(
- sorted(q_func_vars, key=lambda v: v.name),
- sorted(target_q_func_vars, key=lambda v: v.name)):
- update_target_expr.append(
- var_target.assign(self.tau * var +
- (1.0 - self.tau) * var_target))
- for var, var_target in zip(
- sorted(p_func_vars, key=lambda v: v.name),
- sorted(target_p_func_vars, key=lambda v: v.name)):
- update_target_expr.append(
- var_target.assign(self.tau * var +
- (1.0 - self.tau) * var_target))
- self.update_target_expr = tf.group(*update_target_expr)
-
- # support both hard and soft sync
- def update_target(self, sess, tau=None):
- return sess.run(
- self.update_target_expr,
- feed_dict={self.tau: tau or self.tau_value})
-
- def act(self, sess, obs, eps, stochastic=True):
- return sess.run(
- self.output_actions,
- feed_dict={
- self.cur_observations: obs,
- self.stochastic: stochastic,
- self.eps: eps
- })
-
- def compute_gradients(self, sess, obs_t, act_t, rew_t, obs_tp1, done_mask,
- importance_weights):
- td_err, grads = sess.run(
- [self.td_error, self.grads],
- feed_dict={
- self.obs_t: obs_t,
- self.act_t: act_t,
- self.rew_t: rew_t,
- self.obs_tp1: obs_tp1,
- self.done_mask: done_mask,
- self.importance_weights: importance_weights
- })
- return td_err, grads
-
- def compute_td_error(self, sess, obs_t, act_t, rew_t, obs_tp1, done_mask,
- importance_weights):
- td_err = sess.run(
- self.td_error,
- feed_dict={
- self.obs_t: [np.array(ob) for ob in obs_t],
- self.act_t: act_t,
- self.rew_t: rew_t,
- self.obs_tp1: [np.array(ob) for ob in obs_tp1],
- self.done_mask: done_mask,
- self.importance_weights: importance_weights
- })
- return td_err
-
- def apply_gradients(self, sess, grads):
- assert len(grads) == len(self.grads_and_vars)
- feed_dict = {ph: g for (g, ph) in zip(grads, self.grads)}
- sess.run(
- [self.critic_train_expr, self.actor_train_expr],
- feed_dict=feed_dict)
-
- def compute_apply(self, sess, obs_t, act_t, rew_t, obs_tp1, done_mask,
- importance_weights):
- td_err, _, _ = sess.run(
- [self.td_error, self.critic_train_expr, self.actor_train_expr],
- feed_dict={
- self.obs_t: obs_t,
- self.act_t: act_t,
- self.rew_t: rew_t,
- self.obs_tp1: obs_tp1,
- self.done_mask: done_mask,
- self.importance_weights: importance_weights
- })
- return td_err
-
- def reset_noise(self, sess):
- sess.run(self.reset_noise_op)
+ def value(self, *args):
+ return 0
diff --git a/python/ray/rllib/ddpg/random_process.py b/python/ray/rllib/ddpg2/random_process.py
similarity index 100%
rename from python/ray/rllib/ddpg/random_process.py
rename to python/ray/rllib/ddpg2/random_process.py
diff --git a/python/ray/rllib/optimizers/local_sync_replay.py b/python/ray/rllib/optimizers/local_sync_replay.py
index 5ba8b6f9a..ac430c6a1 100644
--- a/python/ray/rllib/optimizers/local_sync_replay.py
+++ b/python/ray/rllib/optimizers/local_sync_replay.py
@@ -9,6 +9,7 @@ from ray.rllib.optimizers.replay_buffer import ReplayBuffer, \
PrioritizedReplayBuffer
from ray.rllib.optimizers.policy_optimizer import PolicyOptimizer
from ray.rllib.optimizers.sample_batch import SampleBatch
+from ray.rllib.utils.compression import pack_if_needed
from ray.rllib.utils.filter import RunningStat
from ray.rllib.utils.timer import TimerStat
@@ -64,7 +65,8 @@ class LocalSyncReplayOptimizer(PolicyOptimizer):
batch = self.local_evaluator.sample()
for row in batch.rows():
self.replay_buffer.add(
- row["obs"], row["actions"], row["rewards"], row["new_obs"],
+ pack_if_needed(row["obs"]), row["actions"], row["rewards"],
+ pack_if_needed(row["new_obs"]),
row["dones"], row["weights"])
if len(self.replay_buffer) >= self.replay_starts:
diff --git a/python/ray/rllib/test/test_checkpoint_restore.py b/python/ray/rllib/test/test_checkpoint_restore.py
index f19eecd2c..9e583c877 100644
--- a/python/ray/rllib/test/test_checkpoint_restore.py
+++ b/python/ray/rllib/test/test_checkpoint_restore.py
@@ -22,7 +22,7 @@ ray.init()
CONFIGS = {
"ES": {"episodes_per_batch": 10, "timesteps_per_batch": 100},
"DQN": {},
- "DDPG2": {"noise_scale": 0.0},
+ "DDPG": {"noise_scale": 0.0},
"PPO": {"num_sgd_iter": 5, "timesteps_per_batch": 1000},
"A3C": {"use_lstm": False},
}
@@ -30,7 +30,7 @@ CONFIGS = {
def test(use_object_store, alg_name):
cls = get_agent_class(alg_name)
- if alg_name == "DDPG2":
+ if alg_name == "DDPG":
alg1 = cls(config=CONFIGS[name], env="Pendulum-v0")
alg2 = cls(config=CONFIGS[name], env="Pendulum-v0")
else:
@@ -48,7 +48,7 @@ def test(use_object_store, alg_name):
alg2.restore(alg1.save())
for _ in range(10):
- if alg_name == "DDPG2":
+ if alg_name == "DDPG":
obs = np.random.uniform(size=3)
else:
obs = np.random.uniform(size=4)
@@ -59,9 +59,8 @@ def test(use_object_store, alg_name):
if __name__ == "__main__":
- # https://github.com/ray-project/ray/issues/1062 for enabling ES test too
for use_object_store in [False, True]:
- for name in ["ES", "DQN", "DDPG2", "PPO", "A3C"]:
+ for name in ["ES", "DQN", "DDPG", "PPO", "A3C"]:
test(use_object_store, name)
print("All checkpoint restore tests passed!")
diff --git a/python/ray/rllib/test/test_supported_spaces.py b/python/ray/rllib/test/test_supported_spaces.py
index 5fa46f559..2e41c85a0 100644
--- a/python/ray/rllib/test/test_supported_spaces.py
+++ b/python/ray/rllib/test/test_supported_spaces.py
@@ -114,7 +114,7 @@ class ModelSupportedSpaces(unittest.TestCase):
def testAll(self):
ray.init()
stats = {}
- check_support("DDPG2", {"timesteps_per_iteration": 1}, stats)
+ check_support("DDPG", {"timesteps_per_iteration": 1}, stats)
check_support("DQN", {"timesteps_per_iteration": 1}, stats)
check_support(
"A3C", {"num_workers": 1, "optimizer": {"grads_per_step": 1}},
diff --git a/python/ray/rllib/tuned_examples/mountaincarcontinuous-apex-ddpg2.yaml b/python/ray/rllib/tuned_examples/mountaincarcontinuous-apex-ddpg.yaml
similarity index 66%
rename from python/ray/rllib/tuned_examples/mountaincarcontinuous-apex-ddpg2.yaml
rename to python/ray/rllib/tuned_examples/mountaincarcontinuous-apex-ddpg.yaml
index 7e5af40ab..82947d872 100644
--- a/python/ray/rllib/tuned_examples/mountaincarcontinuous-apex-ddpg2.yaml
+++ b/python/ray/rllib/tuned_examples/mountaincarcontinuous-apex-ddpg.yaml
@@ -1,12 +1,7 @@
# This can be expected to reach 90 reward within ~1.5-2.5m timesteps / ~150-250 seconds on a K40 GPU
-mountaincarcontinuous-apex-ddpg-2:
+mountaincarcontinuous-apex-ddpg:
env: MountainCarContinuous-v0
- run: APEX_DDPG2
- trial_resources:
- cpu: 1
- gpu: 1
- extra_cpu:
- eval: 4 + spec.config.num_workers
+ run: APEX_DDPG
stop:
episode_reward_mean: 90
config:
diff --git a/python/ray/rllib/tuned_examples/mountaincarcontinuous-ddpg2.yaml b/python/ray/rllib/tuned_examples/mountaincarcontinuous-ddpg.yaml
similarity index 85%
rename from python/ray/rllib/tuned_examples/mountaincarcontinuous-ddpg2.yaml
rename to python/ray/rllib/tuned_examples/mountaincarcontinuous-ddpg.yaml
index 157e7e3a7..0a330bb5b 100644
--- a/python/ray/rllib/tuned_examples/mountaincarcontinuous-ddpg2.yaml
+++ b/python/ray/rllib/tuned_examples/mountaincarcontinuous-ddpg.yaml
@@ -1,9 +1,7 @@
# can expect improvement to 90 reward in ~12-24k timesteps
-mountaincarcontinuous-ddpg-2:
+mountaincarcontinuous-ddpg:
env: MountainCarContinuous-v0
- run: DDPG2
- trial_resources:
- cpu: 6
+ run: DDPG
stop:
episode_reward_mean: 90
config:
diff --git a/python/ray/rllib/tuned_examples/pendulum-apex-ddpg2.yaml b/python/ray/rllib/tuned_examples/pendulum-apex-ddpg.yaml
similarity index 67%
rename from python/ray/rllib/tuned_examples/pendulum-apex-ddpg2.yaml
rename to python/ray/rllib/tuned_examples/pendulum-apex-ddpg.yaml
index ec4361e7a..f7a7c71f6 100644
--- a/python/ray/rllib/tuned_examples/pendulum-apex-ddpg2.yaml
+++ b/python/ray/rllib/tuned_examples/pendulum-apex-ddpg.yaml
@@ -1,12 +1,7 @@
# This can be expected to reach -160 reward within 2.5 timesteps / ~250 seconds on a K40 GPU
-pendulum-apex-ddpg-2:
+pendulum-apex-ddpg:
env: Pendulum-v0
- run: APEX_DDPG2
- trial_resources:
- cpu: 1
- gpu: 1
- extra_cpu:
- eval: 4 + spec.config.num_workers
+ run: APEX_DDPG
stop:
episode_reward_mean: -160
config:
diff --git a/python/ray/rllib/tuned_examples/pendulum-ddpg.yaml b/python/ray/rllib/tuned_examples/pendulum-ddpg.yaml
new file mode 100644
index 000000000..2166989d8
--- /dev/null
+++ b/python/ray/rllib/tuned_examples/pendulum-ddpg.yaml
@@ -0,0 +1,11 @@
+# can expect improvement to -160 reward in ~30k timesteps
+pendulum-ddpg:
+ env: Pendulum-v0
+ run: DDPG
+ stop:
+ episode_reward_mean: -160
+ config:
+ use_huber: True
+ random_starts: False
+ clip_rewards: False
+ exploration_fraction: 0.1
diff --git a/python/ray/rllib/tuned_examples/pendulum-ddpg2.yaml b/python/ray/rllib/tuned_examples/pendulum-ddpg2.yaml
deleted file mode 100644
index 43327705c..000000000
--- a/python/ray/rllib/tuned_examples/pendulum-ddpg2.yaml
+++ /dev/null
@@ -1,16 +0,0 @@
-# can expect improvement to -160 reward in ~30-40k timesteps
-pendulum-ddpg-2:
- env: Pendulum-v0
- run: DDPG2
- trial_resources:
- cpu: 6
- gpu: 1
- stop:
- episode_reward_mean: -160
- config:
- use_huber: True
- random_starts: False
- clip_rewards: False
- exploration_fraction: 0.4
- model:
- fcnet_hiddens: []
diff --git a/python/ray/rllib/tuned_examples/regression_tests/pendulum-ddpg.yaml b/python/ray/rllib/tuned_examples/regression_tests/pendulum-ddpg.yaml
index b25180ff0..840f6d963 100644
--- a/python/ray/rllib/tuned_examples/regression_tests/pendulum-ddpg.yaml
+++ b/python/ray/rllib/tuned_examples/regression_tests/pendulum-ddpg.yaml
@@ -2,9 +2,11 @@ pendulum-ddpg:
env: Pendulum-v0
run: DDPG
stop:
- episode_reward_mean: -100
- time_total_s: 600
- trial_resources:
- cpu: 1
+ episode_reward_mean: -160
+ time_total_s: 900
config:
- num_workers: 1
+ use_huber: True
+ random_starts: False
+ clip_rewards: False
+ exploration_fraction: 0.1
+ smoothing_num_episodes: 10
diff --git a/python/ray/rllib/tuned_examples/regression_tests/pendulum-ddpg2.yaml b/python/ray/rllib/tuned_examples/regression_tests/pendulum-ddpg2.yaml
index c60d09872..eaf33eb7e 100644
--- a/python/ray/rllib/tuned_examples/regression_tests/pendulum-ddpg2.yaml
+++ b/python/ray/rllib/tuned_examples/regression_tests/pendulum-ddpg2.yaml
@@ -1,16 +1,8 @@
-pendulum-ddpg-2:
+pendulum-ddpg2:
env: Pendulum-v0
run: DDPG2
- trial_resources:
- cpu: 2
stop:
- episode_reward_mean: -160
- time_total_s: 900
+ episode_reward_mean: -100
+ time_total_s: 600
config:
- use_huber: True
- random_starts: False
- clip_rewards: False
- exploration_fraction: 0.4
- model:
- fcnet_hiddens: []
- smoothing_num_episodes: 10
+ num_workers: 1
diff --git a/python/ray/rllib/utils/compression.py b/python/ray/rllib/utils/compression.py
index 24176285b..dee8d875d 100644
--- a/python/ray/rllib/utils/compression.py
+++ b/python/ray/rllib/utils/compression.py
@@ -28,6 +28,12 @@ def pack(data):
return data
+def pack_if_needed(data):
+ if isinstance(data, np.ndarray):
+ data = pack(data)
+ return data
+
+
def unpack(data):
if LZ4_ENABLED:
data = base64.b64decode(data)