[RLLib] DDPG

python/ray/rllib/__init__.py

@@ -8,8 +8,8 @@
 
 
 def _register_all():
-    for key in ["PPO", "ES", "DQN", "APEX", "A3C", "BC", "PG", "__fake",
-                "__sigmoid_fake_data", "__parameter_tuning"]:
+    for key in ["PPO", "ES", "DQN", "APEX", "A3C", "BC", "PG", "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/a3c/shared_model.py

@@ -38,7 +38,7 @@ def _setup_graph(self, ob_space, ac_space):
     def compute_gradients(self, samples):
         info = {}
         feed_dict = {
-            self.x: samples["observations"],
+            self.x: samples["obs"],
             self.ac: samples["actions"],
             self.adv: samples["advantages"],
             self.r: samples["value_targets"],

python/ray/rllib/a3c/shared_model_lstm.py

@@ -57,7 +57,7 @@ def compute_gradients(self, samples):
         """
         features = samples["features"][0]
         feed_dict = {
-            self.x: samples["observations"],
+            self.x: samples["obs"],
             self.ac: samples["actions"],
             self.adv: samples["advantages"],
             self.r: samples["value_targets"],

python/ray/rllib/agent.py

@@ -243,6 +243,9 @@ 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/__init__.py

@@ -0,0 +1,3 @@
+from ray.rllib.ddpg.ddpg import DDPGAgent, DEFAULT_CONFIG
+
+__all__ = ["DDPGAgent", "DEFAULT_CONFIG"]

python/ray/rllib/ddpg/ddpg.py

@@ -0,0 +1,112 @@
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+
+import numpy as np
+
+import ray
+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,
+
+    # 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,
+    },
+
+    # 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 DDPGAgent(Agent):
+    _agent_name = "DDPG"
+    _default_config = DEFAULT_CONFIG
+
+    def _init(self):
+        self.local_evaluator = DDPGEvaluator(
+            self.registry, self.env_creator, self.config)
+        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,
+            self.remote_evaluators)
+
+    def _train(self):
+        for _ in range(self.config["train_steps"]):
+            self.optimizer.step()
+            # update target
+            if self.optimizer.num_steps_trained > 0:
+                self.local_evaluator.update_target()
+
+        # generate training result
+        return self._fetch_metrics()
+
+    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:
+            metrics = self.local_evaluator.get_completed_rollout_metrics()
+            for episode in metrics:
+                episode_lengths.append(episode.episode_length)
+                episode_rewards.append(episode.episode_reward)
+
+        avg_reward = (np.mean(episode_rewards))
+        avg_length = (np.mean(episode_lengths))
+        timesteps = np.sum(episode_lengths)
+
+        result = TrainingResult(
+            episode_reward_mean=avg_reward,
+            episode_len_mean=avg_length,
+            timesteps_this_iter=timesteps,
+            info={})
+
+        return result

python/ray/rllib/ddpg/ddpg_evaluator.py

@@ -0,0 +1,75 @@
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+
+import numpy as np
+
+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
+
+
+class DDPGEvaluator(PolicyEvaluator):
+
+    def __init__(self, registry, env_creator, config):
+        self.env = ModelCatalog.get_preprocessor_as_wrapper(
+            registry, env_creator(config["env_config"]))
+
+        # contains model, target_model
+        self.model = DDPGModel(registry, self.env, config)
+
+        self.sampler = SyncSampler(
+                        self.env, self.model.model, NoFilter(),
+                        config["num_local_steps"], horizon=config["horizon"])
+
+    def sample(self):
+        """Returns a batch of samples."""
+
+        rollout = self.sampler.get_data()
+        rollout.data["weights"] = np.ones_like(rollout.data["rewards"])
+
+        # 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)
+
+        return samples
+
+    def update_target(self):
+        """Updates target critic and target actor."""
+        self.model.update_target()
+
+    def compute_gradients(self, samples):
+        """Returns critic, actor gradients."""
+        return self.model.compute_gradients(samples)
+
+    def apply_gradients(self, grads):
+        """Applies gradients to evaluator weights."""
+        self.model.apply_gradients(grads)
+
+    def compute_apply(self, samples):
+        grads, _ = self.compute_gradients(samples)
+        self.apply_gradients(grads)
+
+    def get_weights(self):
+        """Returns model weights."""
+        return self.model.get_weights()
+
+    def set_weights(self, weights):
+        """Sets model weights."""
+        self.model.set_weights(weights)
+
+    def get_completed_rollout_metrics(self):
+        """Returns metrics on previously completed rollouts.
+
+        Calling this clears the queue of completed rollout metrics.
+        """
+        return self.sampler.get_metrics()
+
+
+RemoteDDPGEvaluator = ray.remote(DDPGEvaluator)