[tune] Dragonfly Optimizer (#5955)

* Add sample example

* Copy relevant lines of ask from inherited Optimizer

* Ignore strategy

* Additional changes

* Add DragonflySearch for tune connector for Dragonfly

* Add example and fix small errors

* lint

* Remove skopt references

* Update example based off of Dragonfly changes

* Edit example for final Dragonfly edits

* Formatting and documentation edits

* Add documentation and add to test pipeline

* Address PR comments

* Fix Jenkins test

* Adjust Dragonfly to PR#7366

* Lint

* fix_tests

Co-authored-by: Richard Liaw <rliaw@berkeley.edu>

python/ray/tune/examples/dragonfly_example.py

@@ -0,0 +1,83 @@
+"""This test checks that Dragonfly is functional.
+
+It also checks that it is usable with a separate scheduler.
+"""
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+
+import ray
+from ray.tune import run
+from ray.tune.schedulers import AsyncHyperBandScheduler
+from ray.tune.suggest.dragonfly import DragonflySearch
+
+
+def objective(config, reporter):
+    import numpy as np
+    import time
+    time.sleep(0.2)
+    for i in range(config["iterations"]):
+        vol1 = config["point"][0]  # LiNO3
+        vol2 = config["point"][1]  # Li2SO4
+        vol3 = config["point"][2]  # NaClO4
+        vol4 = 10 - (vol1 + vol2 + vol3)  # Water
+        # Synthetic functions
+        conductivity = vol1 + 0.1 * (vol2 + vol3)**2 + 2.3 * vol4 * (vol1**1.5)
+        # Add Gaussian noise to simulate experimental noise
+        conductivity += np.random.normal() * 0.01
+        reporter(timesteps_total=i, objective=conductivity)
+        time.sleep(0.02)
+
+
+if __name__ == "__main__":
+    import argparse
+    from dragonfly.opt.gp_bandit import EuclideanGPBandit
+    from dragonfly.exd.experiment_caller import EuclideanFunctionCaller
+    from dragonfly import load_config
+
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--smoke-test", action="store_true", help="Finish quickly for testing")
+    args, _ = parser.parse_known_args()
+    ray.init()
+
+    config = {
+        "num_samples": 10 if args.smoke_test else 50,
+        "config": {
+            "iterations": 100,
+        },
+        "stop": {
+            "timesteps_total": 100
+        },
+    }
+
+    domain_vars = [{
+        "name": "LiNO3_vol",
+        "type": "float",
+        "min": 0,
+        "max": 7
+    }, {
+        "name": "Li2SO4_vol",
+        "type": "float",
+        "min": 0,
+        "max": 7
+    }, {
+        "name": "NaClO4_vol",
+        "type": "float",
+        "min": 0,
+        "max": 7
+    }]
+
+    domain_config = load_config({"domain": domain_vars})
+
+    func_caller = EuclideanFunctionCaller(
+        None, domain_config.domain.list_of_domains[0])
+    optimizer = EuclideanGPBandit(func_caller, ask_tell_mode=True)
+    algo = DragonflySearch(
+        optimizer, max_concurrent=4, metric="objective", mode="max")
+    scheduler = AsyncHyperBandScheduler(metric="objective", mode="max")
+    run(objective,
+        name="dragonfly_search",
+        search_alg=algo,
+        scheduler=scheduler,
+        **config)

python/ray/tune/suggest/dragonfly.py

@@ -0,0 +1,150 @@
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+
+import logging
+import pickle
+
+try:  # Python 3 only -- needed for lint test.
+    import dragonfly
+except ImportError:
+    dragonfly = None
+
+from ray.tune.suggest.suggestion import SuggestionAlgorithm
+
+logger = logging.getLogger(__name__)
+
+
+class DragonflySearch(SuggestionAlgorithm):
+    """A wrapper around Dragonfly to provide trial suggestions.
+
+    Requires Dragonfly to be installed.
+
+    Parameters:
+        optimizer (dragonfly.opt.BlackboxOptimiser): Optimizer provided
+            from dragonfly. Choose an optimiser that extends BlackboxOptimiser.
+        max_concurrent (int): Number of maximum concurrent trials. Defaults
+            to 10.
+        metric (str): The training result objective value attribute.
+        mode (str): One of {min, max}. Determines whether objective is
+            minimizing or maximizing the metric attribute.
+        points_to_evaluate (list of lists): A list of points you'd like to run
+            first before sampling from the optimiser, e.g. these could be
+            parameter configurations you already know work well to help
+            the optimiser select good values. Each point is a list of the
+            parameters using the order definition given by parameter_names.
+        evaluated_rewards (list): If you have previously evaluated the
+            parameters passed in as points_to_evaluate you can avoid
+            re-running those trials by passing in the reward attributes
+            as a list so the optimiser can be told the results without
+            needing to re-compute the trial. Must be the same length as
+            points_to_evaluate.
+
+    Example:
+        >>> from dragonfly.opt.gp_bandit import EuclideanGPBandit
+        >>> from dragonfly.exd.experiment_caller import EuclideanFunctionCaller
+        >>> from dragonfly import load_config
+        >>> domain_vars = [{
+                "name": "LiNO3_vol",
+                "type": "float",
+                "min": 0,
+                "max": 7
+            }, {
+                "name": "Li2SO4_vol",
+                "type": "float",
+                "min": 0,
+                "max": 7
+            }, {
+                "name": "NaClO4_vol",
+                "type": "float",
+                "min": 0,
+                "max": 7
+            }]
+
+        >>> domain_config = load_config({"domain": domain_vars})
+        >>> func_caller = EuclideanFunctionCaller(None,
+                domain_config.domain.list_of_domains[0])
+        >>> optimizer = EuclideanGPBandit(func_caller, ask_tell_mode=True)
+        >>> algo = DragonflySearch(optimizer, max_concurrent=4,
+                metric="objective", mode="max")
+    """
+
+    def __init__(self,
+                 optimizer,
+                 max_concurrent=10,
+                 reward_attr=None,
+                 metric="episode_reward_mean",
+                 mode="max",
+                 points_to_evaluate=None,
+                 evaluated_rewards=None,
+                 **kwargs):
+        assert dragonfly is not None, """dragonfly must be installed!
+            You can install Dragonfly with the command:
+            `pip install dragonfly`."""
+        assert type(max_concurrent) is int and max_concurrent > 0
+        assert mode in ["min", "max"], "`mode` must be 'min' or 'max'!"
+
+        if reward_attr is not None:
+            mode = "max"
+            metric = reward_attr
+            logger.warning(
+                "`reward_attr` is deprecated and will be removed in a future "
+                "version of Tune. "
+                "Setting `metric={}` and `mode=max`.".format(reward_attr))
+
+        self._initial_points = []
+        if points_to_evaluate and evaluated_rewards:
+            optimizer.tell(points_to_evaluate, evaluated_rewards)
+        elif points_to_evaluate:
+            self._initial_points = points_to_evaluate
+        self._max_concurrent = max_concurrent
+        self._metric = metric
+        # Dragonfly internally maximizes, so "min" => -1
+        if mode == "min":
+            self._metric_op = -1.
+        elif mode == "max":
+            self._metric_op = 1.
+        self._opt = optimizer
+        self._opt.initialise()
+        self._live_trial_mapping = {}
+        super(DragonflySearch, self).__init__(
+            metric=self._metric, mode=mode, **kwargs)
+
+    def suggest(self, trial_id):
+        if self._num_live_trials() >= self._max_concurrent:
+            return None
+        if self._initial_points:
+            suggested_config = self._initial_points[0]
+            del self._initial_points[0]
+        else:
+            suggested_config = self._opt.ask()
+        self._live_trial_mapping[trial_id] = suggested_config
+        return {"point": suggested_config}
+
+    def on_trial_result(self, trial_id, result):
+        pass
+
+    def on_trial_complete(self,
+                          trial_id,
+                          result=None,
+                          error=False,
+                          early_terminated=False):
+        """Passes result to Dragonfly unless early terminated or errored."""
+        trial_info = self._live_trial_mapping.pop(trial_id)
+        if result:
+            self._opt.tell([(trial_info,
+                             self._metric_op * result[self._metric])])
+
+    def _num_live_trials(self):
+        return len(self._live_trial_mapping)
+
+    def save(self, checkpoint_dir):
+        trials_object = (self._initial_points, self._opt)
+        with open(checkpoint_dir, "wb") as outputFile:
+            pickle.dump(trials_object, outputFile)
+
+    def restore(self, checkpoint_dir):
+        with open(checkpoint_dir, "rb") as inputFile:
+            trials_object = pickle.load(inputFile)
+        self._initial_points = trials_object[0]
+        self._opt = trials_object[1]

python/ray/tune/suggest/nevergrad.py

@@ -116,13 +116,11 @@ class NevergradSearch(SuggestionAlgorithm):
         self._live_trial_mapping[trial_id] = suggested_config
         # in v0.2.0+, output of ask() is a Candidate,
         # with fields args and kwargs
-        if hasattr(self._nevergrad_opt, "instrumentation"):
-            if not suggested_config.kwargs:
-                return dict(zip(self._parameters, suggested_config.args[0]))
-            else:
-                return suggested_config.kwargs
-        # legacy: output of ask() is a np.ndarray
-        return dict(zip(self._parameters, suggested_config))
+        if not suggested_config.kwargs:
+            print(suggested_config.args, suggested_config.kwargs)
+            return dict(zip(self._parameters, suggested_config.args[0]))
+        else:
+            return suggested_config.kwargs
 
     def on_trial_result(self, trial_id, result):
         pass

python/ray/tune/tests/test_tune_restore.py

@@ -144,7 +144,7 @@ class AbstractWarmStartTest:
     def run_exp_1(self):
         np.random.seed(162)
         search_alg, cost = self.set_basic_conf()
-        results_exp_1 = tune.run(cost, num_samples=15, search_alg=search_alg)
+        results_exp_1 = tune.run(cost, num_samples=5, search_alg=search_alg)
         self.log_dir = os.path.join(self.tmpdir, "warmStartTest.pkl")
         search_alg.save(self.log_dir)
         return results_exp_1
@@ -152,12 +152,12 @@ class AbstractWarmStartTest:
     def run_exp_2(self):
         search_alg2, cost = self.set_basic_conf()
         search_alg2.restore(self.log_dir)
-        return tune.run(cost, num_samples=15, search_alg=search_alg2)
+        return tune.run(cost, num_samples=5, search_alg=search_alg2)
 
     def run_exp_3(self):
         np.random.seed(162)
         search_alg3, cost = self.set_basic_conf()
-        return tune.run(cost, num_samples=30, search_alg=search_alg3)
+        return tune.run(cost, num_samples=10, search_alg=search_alg3)
 
     def testWarmStart(self):
         results_exp_1 = self.run_exp_1()