wassname/ray
1
0
Fork 0
mirror of https://github.com/wassname/ray.git synced 2026-07-09 06:06:26 +08:00
Code Issues Packages Projects Releases Wiki Activity

[rllib] Add DDPG documentation, rename DDPG2 <=> DDPG (#1946)

Browse Source 
* updates

* updates

* updates

* updates

* updates

* updates

* Update rllib.rst

* Update policy-optimizers.rst
This commit is contained in:
Eric Liang
2018-04-30 00:18:15 -07:00
committed by GitHub
parent 9ad94e33d6
commit 47bc4c3009
29 changed files with 1171 additions and 1179 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/ray/rllib/README.rst
+10 -9
View File
@@ -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) <https://arxiv.org/abs/1707.06347>`__ which
is a proximal variant of `TRPO <https://arxiv.org/abs/1502.05477>`__.
- Proximal Policy Optimization (`PPO <https://github.com/ray-project/ray/tree/master/python/ray/rllib/ppo>`__) which is a proximal variant of `TRPO <https://arxiv.org/abs/1502.05477>`__.
- `The Asynchronous Advantage Actor-Critic (A3C) <https://arxiv.org/abs/1602.01783>`__.
- Policy Gradients (`PG <https://github.com/ray-project/ray/tree/master/python/ray/rllib/pg>`__).
- `Deep Q Networks (DQN) <https://arxiv.org/abs/1312.5602>`__.
- Asynchronous Advantage Actor-Critic (`A3C <https://github.com/ray-project/ray/tree/master/python/ray/rllib/a3c>`__).
- `Ape-X Distributed Prioritized Experience Replay <https://arxiv.org/abs/1803.00933>`__.
- Deep Q Networks (`DQN <https://github.com/ray-project/ray/tree/master/python/ray/rllib/dqn>`__).
- Evolution Strategies, as described in `this
paper <https://arxiv.org/abs/1703.03864>`__. Our implementation
is adapted from
`here <https://github.com/openai/evolution-strategies-starter>`__.
- Deep Deterministic Policy Gradients (`DDPG <https://github.com/ray-project/ray/tree/master/python/ray/rllib/ddpg>`__, `DDPG2 <https://github.com/ray-project/ray/tree/master/python/ray/rllib/ddpg2>`__).
- Ape-X Distributed Prioritized Experience Replay, including both `DQN <https://github.com/ray-project/ray/blob/master/python/ray/rllib/dqn/apex.py>`__ and `DDPG <https://github.com/ray-project/ray/blob/master/python/ray/rllib/ddpg2/apex.py>`__ variants.
- Evolution Strategies (`ES <https://github.com/ray-project/ray/tree/master/python/ray/rllib/es>`__), as described in `this
paper <https://arxiv.org/abs/1703.03864>`__.
These algorithms can be run on any OpenAI Gym MDP, including custom ones written and registered by the user.
python/ray/rllib/__init__.py
+1 -1
View File
@@ -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))
python/ray/rllib/agent.py
+6 -6
View File
@@ -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
python/ray/rllib/ddpg/README.md
+1
View File
@@ -0,0 +1 @@
Implementation of deep deterministic policy gradients (https://arxiv.org/abs/1509.02971), including an Ape-X variant.
python/ray/rllib/ddpg/__init__.py
+6 -1
View File
@@ -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"]
python/ray/rllib/ddpg2/apex.py → python/ray/rllib/ddpg/apex.py
+2 -2
View File
@@ -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
python/ray/rllib/ddpg2/common/__init__.py → python/ray/rllib/ddpg/common/__init__.py
View File
python/ray/rllib/ddpg/ddpg.py
+235 -79
View File
@@ -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]
python/ray/rllib/ddpg/ddpg_evaluator.py
+153 -42
View File
@@ -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]
python/ray/rllib/ddpg/models.py
+362 -212
View File
@@ -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)
python/ray/rllib/ddpg2/README.md
+1 -1
View File
@@ -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.
python/ray/rllib/ddpg2/__init__.py
+1 -6
View File
@@ -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"]
python/ray/rllib/ddpg2/ddpg.py
+79 -235
View File
@@ -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]
python/ray/rllib/ddpg2/ddpg_evaluator.py
+42 -153
View File
@@ -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)
python/ray/rllib/ddpg2/models.py
+212 -362
View File
@@ -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
python/ray/rllib/ddpg/random_process.py → python/ray/rllib/ddpg2/random_process.py
View File
python/ray/rllib/optimizers/local_sync_replay.py
+3 -1
View File
@@ -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:
python/ray/rllib/test/test_checkpoint_restore.py
+4 -5
View File
@@ -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!")
python/ray/rllib/test/test_supported_spaces.py
+1 -1
View File
@@ -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}},
python/ray/rllib/tuned_examples/mountaincarcontinuous-apex-ddpg2.yaml → python/ray/rllib/tuned_examples/mountaincarcontinuous-apex-ddpg.yaml
+2 -7
View File
@@ -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:
python/ray/rllib/tuned_examples/mountaincarcontinuous-ddpg2.yaml → python/ray/rllib/tuned_examples/mountaincarcontinuous-ddpg.yaml
+2 -4
View File
@@ -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:
python/ray/rllib/tuned_examples/pendulum-apex-ddpg2.yaml → python/ray/rllib/tuned_examples/pendulum-apex-ddpg.yaml
+2 -7
View File
@@ -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:
python/ray/rllib/tuned_examples/pendulum-ddpg.yaml
+11
View File
@@ -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
python/ray/rllib/tuned_examples/pendulum-ddpg2.yaml
-16
View File
@@ -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: []
python/ray/rllib/tuned_examples/regression_tests/pendulum-ddpg.yaml
+7 -5
View File
@@ -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
python/ray/rllib/tuned_examples/regression_tests/pendulum-ddpg2.yaml
+4 -12
View File
@@ -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
python/ray/rllib/utils/compression.py
+6
View File
@@ -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)