diff --git a/doc/source/conf.py b/doc/source/conf.py
index 43bd34876f55..e6123003e199 100644
--- a/doc/source/conf.py
+++ b/doc/source/conf.py
@@ -31,6 +31,8 @@ def __getattr__(cls, name):
 
 
 MOCK_MODULES = [
+    "ax",
+    "ax.service.ax_client",
     "blist",
     "gym",
     "gym.spaces",
diff --git a/doc/source/tune/api_docs/grid_random.rst b/doc/source/tune/api_docs/grid_random.rst
index 5f008fc8d279..7269507cccf4 100644
--- a/doc/source/tune/api_docs/grid_random.rst
+++ b/doc/source/tune/api_docs/grid_random.rst
@@ -164,16 +164,41 @@ tune.randn
 
 .. autofunction:: ray.tune.randn
 
+tune.qrandn
+~~~~~~~~~~~
+
+.. autofunction:: ray.tune.qrandn
+
 tune.loguniform
 ~~~~~~~~~~~~~~~
 
 .. autofunction:: ray.tune.loguniform
 
+tune.qloguniform
+~~~~~~~~~~~~~~~~
+
+.. autofunction:: ray.tune.qloguniform
+
 tune.uniform
 ~~~~~~~~~~~~
 
 .. autofunction:: ray.tune.uniform
 
+tune.quniform
+~~~~~~~~~~~~~
+
+.. autofunction:: ray.tune.quniform
+
+tune.randint
+~~~~~~~~~~~~
+
+.. autofunction:: ray.tune.randint
+
+tune.qrandint
+~~~~~~~~~~~~~
+
+.. autofunction:: ray.tune.qrandint
+
 tune.choice
 ~~~~~~~~~~~
 
@@ -182,7 +207,7 @@ tune.choice
 tune.sample_from
 ~~~~~~~~~~~~~~~~
 
-.. autoclass:: ray.tune.sample_from
+.. autofunction:: ray.tune.sample_from
 
 Grid Search API
 ---------------
diff --git a/python/ray/tune/__init__.py b/python/ray/tune/__init__.py
index 43a65c430a9d..b50d96c366cb 100644
--- a/python/ray/tune/__init__.py
+++ b/python/ray/tune/__init__.py
@@ -12,15 +12,17 @@
                               save_checkpoint, checkpoint_dir)
 from ray.tune.progress_reporter import (ProgressReporter, CLIReporter,
                                         JupyterNotebookReporter)
-from ray.tune.sample import (function, sample_from, uniform, choice, randint,
-                             randn, loguniform)
+from ray.tune.sample import (function, sample_from, uniform, quniform, choice,
+                             randint, qrandint, randn, qrandn, loguniform,
+                             qloguniform)
 
 __all__ = [
     "Trainable", "DurableTrainable", "TuneError", "grid_search",
     "register_env", "register_trainable", "run", "run_experiments", "Stopper",
     "EarlyStopping", "Experiment", "function", "sample_from", "track",
-    "uniform", "choice", "randint", "randn", "loguniform",
-    "ExperimentAnalysis", "Analysis", "CLIReporter", "JupyterNotebookReporter",
-    "ProgressReporter", "report", "get_trial_dir", "get_trial_name",
-    "get_trial_id", "make_checkpoint_dir", "save_checkpoint", "checkpoint_dir"
+    "uniform", "quniform", "choice", "randint", "qrandint", "randn", "qrandn",
+    "loguniform", "qloguniform", "ExperimentAnalysis", "Analysis",
+    "CLIReporter", "JupyterNotebookReporter", "ProgressReporter", "report",
+    "get_trial_dir", "get_trial_name", "get_trial_id", "make_checkpoint_dir",
+    "save_checkpoint", "checkpoint_dir"
 ]
diff --git a/python/ray/tune/analysis/experiment_analysis.py b/python/ray/tune/analysis/experiment_analysis.py
index 850b3fcb349d..2da4c33e8883 100644
--- a/python/ray/tune/analysis/experiment_analysis.py
+++ b/python/ray/tune/analysis/experiment_analysis.py
@@ -20,9 +20,18 @@ class Analysis:
     """Analyze all results from a directory of experiments.
 
     To use this class, the experiment must be executed with the JsonLogger.
+
+    Args:
+        experiment_dir (str): Directory of the experiment to load.
+        default_metric (str): Default metric for comparing results. Can be
+            overwritten with the ``metric`` parameter in the respective
+            functions.
+        default_mode (str): Default mode for comparing results. Has to be one
+            of [min, max]. Can be overwritten with the ``mode`` parameter
+            in the respective functions.
     """
 
-    def __init__(self, experiment_dir):
+    def __init__(self, experiment_dir, default_metric=None, default_mode=None):
         experiment_dir = os.path.expanduser(experiment_dir)
         if not os.path.isdir(experiment_dir):
             raise ValueError(
@@ -31,6 +40,12 @@ def __init__(self, experiment_dir):
         self._configs = {}
         self._trial_dataframes = {}
 
+        self.default_metric = default_metric
+        if default_mode and default_mode not in ["min", "max"]:
+            raise ValueError(
+                "`default_mode` has to be None or one of [min, max]")
+        self.default_mode = default_mode
+
         if not pd:
             logger.warning(
                 "pandas not installed. Run `pip install pandas` for "
@@ -38,6 +53,22 @@ def __init__(self, experiment_dir):
         else:
             self.fetch_trial_dataframes()
 
+    def _validate_metric(self, metric):
+        if not metric and not self.default_metric:
+            raise ValueError(
+                "No `metric` has been passed and  `default_metric` has "
+                "not been set. Please specify the `metric` parameter.")
+        return metric or self.default_metric
+
+    def _validate_mode(self, mode):
+        if not mode and not self.default_mode:
+            raise ValueError(
+                "No `mode` has been passed and  `default_mode` has "
+                "not been set. Please specify the `mode` parameter.")
+        if mode and mode not in ["min", "max"]:
+            raise ValueError("If set, `mode` has to be one of [min, max]")
+        return mode or self.default_mode
+
     def dataframe(self, metric=None, mode=None):
         """Returns a pandas.DataFrame object constructed from the trials.
 
@@ -57,13 +88,18 @@ def dataframe(self, metric=None, mode=None):
                 rows[path].update(logdir=path)
         return pd.DataFrame(list(rows.values()))
 
-    def get_best_config(self, metric, mode="max"):
+    def get_best_config(self, metric=None, mode=None):
         """Retrieve the best config corresponding to the trial.
 
         Args:
-            metric (str): Key for trial info to order on.
-            mode (str): One of [min, max].
+            metric (str): Key for trial info to order on. Defaults to
+                ``self.default_metric``.
+            mode (str): One of [min, max]. Defaults to
+                ``self.default_mode``.
         """
+        metric = self._validate_metric(metric)
+        mode = self._validate_mode(mode)
+
         rows = self._retrieve_rows(metric=metric, mode=mode)
         if not rows:
             # only nans encountered when retrieving rows
@@ -77,13 +113,17 @@ def get_best_config(self, metric, mode="max"):
         best_path = compare_op(rows, key=lambda k: rows[k][metric])
         return all_configs[best_path]
 
-    def get_best_logdir(self, metric, mode="max"):
+    def get_best_logdir(self, metric=None, mode=None):
         """Retrieve the logdir corresponding to the best trial.
 
         Args:
-            metric (str): Key for trial info to order on.
-            mode (str): One of [min, max].
+            metric (str): Key for trial info to order on. Defaults to
+                ``self.default_metric``.
+            mode (str): One of [min, max]. Defaults to ``self.default_mode``.
         """
+        metric = self._validate_metric(metric)
+        mode = self._validate_mode(mode)
+
         assert mode in ["max", "min"]
         df = self.dataframe(metric=metric, mode=mode)
         mode_idx = pd.Series.idxmax if mode == "max" else pd.Series.idxmin
@@ -140,17 +180,20 @@ def get_all_configs(self, prefix=False):
                 "Couldn't read config from {} paths".format(fail_count))
         return self._configs
 
-    def get_trial_checkpoints_paths(self, trial, metric=TRAINING_ITERATION):
+    def get_trial_checkpoints_paths(self, trial, metric=None):
         """Gets paths and metrics of all persistent checkpoints of a trial.
 
         Args:
             trial (Trial): The log directory of a trial, or a trial instance.
             metric (str): key for trial info to return, e.g. "mean_accuracy".
-                "training_iteration" is used by default.
+                "training_iteration" is used by default if no value was
+                passed to ``self.default_metric``.
 
         Returns:
             List of [path, metric] for all persistent checkpoints of the trial.
         """
+        metric = metric or self.default_metric or TRAINING_ITERATION
+
         if isinstance(trial, str):
             trial_dir = os.path.expanduser(trial)
             # Get checkpoints from logdir.
@@ -167,20 +210,22 @@ def get_trial_checkpoints_paths(self, trial, metric=TRAINING_ITERATION):
         else:
             raise ValueError("trial should be a string or a Trial instance.")
 
-    def get_best_checkpoint(self, trial, metric=TRAINING_ITERATION,
-                            mode="max"):
+    def get_best_checkpoint(self, trial, metric=None, mode=None):
         """Gets best persistent checkpoint path of provided trial.
 
         Args:
             trial (Trial): The log directory of a trial, or a trial instance.
             metric (str): key of trial info to return, e.g. "mean_accuracy".
-                "training_iteration" is used by default.
-            mode (str): Either "min" or "max".
+                "training_iteration" is used by default if no value was
+                passed to ``self.default_metric``.
+            mode (str): One of [min, max]. Defaults to ``self.default_mode``.
 
         Returns:
             Path for best checkpoint of trial determined by metric
         """
-        assert mode in ["max", "min"]
+        metric = metric or self.default_metric or TRAINING_ITERATION
+        mode = self._validate_mode(mode)
+
         checkpoint_paths = self.get_trial_checkpoints_paths(trial, metric)
         if mode == "max":
             return max(checkpoint_paths, key=lambda x: x[1])[0]
@@ -235,6 +280,12 @@ class ExperimentAnalysis(Analysis):
             Experiment.local_dir/Experiment.name/experiment_state.json
         trials (list|None): List of trials that can be accessed via
             `analysis.trials`.
+        default_metric (str): Default metric for comparing results. Can be
+            overwritten with the ``metric`` parameter in the respective
+            functions.
+        default_mode (str): Default mode for comparing results. Has to be one
+            of [min, max]. Can be overwritten with the ``mode`` parameter
+            in the respective functions.
 
     Example:
         >>> tune.run(my_trainable, name="my_exp", local_dir="~/tune_results")
@@ -242,7 +293,11 @@ class ExperimentAnalysis(Analysis):
         >>>     experiment_checkpoint_path="~/tune_results/my_exp/state.json")
     """
 
-    def __init__(self, experiment_checkpoint_path, trials=None):
+    def __init__(self,
+                 experiment_checkpoint_path,
+                 trials=None,
+                 default_metric=None,
+                 default_mode=None):
         experiment_checkpoint_path = os.path.expanduser(
             experiment_checkpoint_path)
         if not os.path.isfile(experiment_checkpoint_path):
@@ -256,17 +311,24 @@ def __init__(self, experiment_checkpoint_path, trials=None):
             raise TuneError("Experiment state invalid; no checkpoints found.")
         self._checkpoints = _experiment_state["checkpoints"]
         self.trials = trials
+
         super(ExperimentAnalysis, self).__init__(
-            os.path.dirname(experiment_checkpoint_path))
+            os.path.dirname(experiment_checkpoint_path), default_metric,
+            default_mode)
 
-    def get_best_trial(self, metric, mode="max", scope="all"):
+    def get_best_trial(self, metric=None, mode=None, scope="all"):
         """Retrieve the best trial object.
 
-        Compares all trials' scores on `metric`.
+        Compares all trials' scores on ``metric``.
+        If ``metric`` is not specified, ``self.default_metric`` will be used.
+        If `mode` is not specified, ``self.default_mode`` will be used.
+        These values are usually initialized by passing the ``metric`` and
+        ``mode`` parameters to ``tune.run()``.
 
         Args:
-            metric (str): Key for trial info to order on.
-            mode (str): One of [min, max].
+            metric (str): Key for trial info to order on. Defaults to
+                ``self.default_metric``.
+            mode (str): One of [min, max]. Defaults to ``self.default_mode``.
             scope (str): One of [all, last, avg, last-5-avg, last-10-avg].
                 If `scope=last`, only look at each trial's final step for
                 `metric`, and compare across trials based on `mode=[min,max]`.
@@ -278,16 +340,17 @@ def get_best_trial(self, metric, mode="max", scope="all"):
                 If `scope=all`, find each trial's min/max score for `metric`
                 based on `mode`, and compare trials based on `mode=[min,max]`.
         """
-        if mode not in ["max", "min"]:
-            raise ValueError(
-                "ExperimentAnalysis: attempting to get best trial for "
-                "metric {} for mode {} not in [\"max\", \"min\"]".format(
-                    metric, mode))
+        metric = self._validate_metric(metric)
+        mode = self._validate_mode(mode)
+
         if scope not in ["all", "last", "avg", "last-5-avg", "last-10-avg"]:
             raise ValueError(
                 "ExperimentAnalysis: attempting to get best trial for "
                 "metric {} for scope {} not in [\"all\", \"last\", \"avg\", "
-                "\"last-5-avg\", \"last-10-avg\"]".format(metric, scope))
+                "\"last-5-avg\", \"last-10-avg\"]. "
+                "If you didn't pass a `metric` parameter to `tune.run()`, "
+                "you have to pass one when fetching the best trial.".format(
+                    metric, scope))
         best_trial = None
         best_metric_score = None
         for trial in self.trials:
@@ -311,16 +374,25 @@ def get_best_trial(self, metric, mode="max", scope="all"):
                 best_metric_score = metric_score
                 best_trial = trial
 
+        if not best_trial:
+            logger.warning(
+                "Could not find best trial. Did you pass the correct `metric`"
+                "parameter?")
         return best_trial
 
-    def get_best_config(self, metric, mode="max", scope="all"):
+    def get_best_config(self, metric=None, mode=None, scope="all"):
         """Retrieve the best config corresponding to the trial.
 
         Compares all trials' scores on `metric`.
+        If ``metric`` is not specified, ``self.default_metric`` will be used.
+        If `mode` is not specified, ``self.default_mode`` will be used.
+        These values are usually initialized by passing the ``metric`` and
+        ``mode`` parameters to ``tune.run()``.
 
         Args:
-            metric (str): Key for trial info to order on.
-            mode (str): One of [min, max].
+            metric (str): Key for trial info to order on. Defaults to
+                ``self.default_metric``.
+            mode (str): One of [min, max]. Defaults to ``self.default_mode``.
             scope (str): One of [all, last, avg, last-5-avg, last-10-avg].
                 If `scope=last`, only look at each trial's final step for
                 `metric`, and compare across trials based on `mode=[min,max]`.
@@ -335,14 +407,19 @@ def get_best_config(self, metric, mode="max", scope="all"):
         best_trial = self.get_best_trial(metric, mode, scope)
         return best_trial.config if best_trial else None
 
-    def get_best_logdir(self, metric, mode="max", scope="all"):
+    def get_best_logdir(self, metric=None, mode=None, scope="all"):
         """Retrieve the logdir corresponding to the best trial.
 
         Compares all trials' scores on `metric`.
+        If ``metric`` is not specified, ``self.default_metric`` will be used.
+        If `mode` is not specified, ``self.default_mode`` will be used.
+        These values are usually initialized by passing the ``metric`` and
+        ``mode`` parameters to ``tune.run()``.
 
         Args:
-            metric (str): Key for trial info to order on.
-            mode (str): One of [min, max].
+            metric (str): Key for trial info to order on. Defaults to
+                ``self.default_metric``.
+            mode (str): One of [min, max]. Defaults to ``self.default_mode``.
             scope (str): One of [all, last, avg, last-5-avg, last-10-avg].
                 If `scope=last`, only look at each trial's final step for
                 `metric`, and compare across trials based on `mode=[min,max]`.
diff --git a/python/ray/tune/examples/ax_example.py b/python/ray/tune/examples/ax_example.py
index 6f5f06fb55ca..71335a094adb 100644
--- a/python/ray/tune/examples/ax_example.py
+++ b/python/ray/tune/examples/ax_example.py
@@ -106,7 +106,7 @@ def easy_objective(config):
         parameter_constraints=["x1 + x2 <= 2.0"],  # Optional.
         outcome_constraints=["l2norm <= 1.25"],  # Optional.
     )
-    algo = AxSearch(client, max_concurrent=4)
+    algo = AxSearch(ax_client=client, max_concurrent=4)
     scheduler = AsyncHyperBandScheduler(metric="hartmann6", mode="min")
     tune.run(
         easy_objective,
diff --git a/python/ray/tune/examples/logging_example.py b/python/ray/tune/examples/logging_example.py
index d7449711369f..c0bf7eda4348 100755
--- a/python/ray/tune/examples/logging_example.py
+++ b/python/ray/tune/examples/logging_example.py
@@ -3,8 +3,6 @@
 import argparse
 import json
 import os
-import random
-
 import numpy as np
 
 from ray import tune
@@ -64,7 +62,6 @@ def load_checkpoint(self, checkpoint_path):
         loggers=[TestLogger],
         stop={"training_iteration": 1 if args.smoke_test else 99999},
         config={
-            "width": tune.sample_from(
-                lambda spec: 10 + int(90 * random.random())),
-            "height": tune.sample_from(lambda spec: int(100 * random.random()))
+            "width": tune.randint(10, 100),
+            "height": tune.loguniform(10, 100)
         })
diff --git a/python/ray/tune/examples/mnist_pytorch.py b/python/ray/tune/examples/mnist_pytorch.py
index 9c8f7f0c69e1..5a2c3677079c 100644
--- a/python/ray/tune/examples/mnist_pytorch.py
+++ b/python/ray/tune/examples/mnist_pytorch.py
@@ -141,4 +141,5 @@ def train_mnist(config):
             "use_gpu": int(args.cuda)
         })
 
-    print("Best config is:", analysis.get_best_config(metric="mean_accuracy"))
+    print("Best config is:",
+          analysis.get_best_config(metric="mean_accuracy", mode="max"))
diff --git a/python/ray/tune/examples/mnist_pytorch_trainable.py b/python/ray/tune/examples/mnist_pytorch_trainable.py
index 956a9c024108..c623111daf83 100644
--- a/python/ray/tune/examples/mnist_pytorch_trainable.py
+++ b/python/ray/tune/examples/mnist_pytorch_trainable.py
@@ -86,4 +86,5 @@ def load_checkpoint(self, checkpoint_path):
             "momentum": tune.uniform(0.1, 0.9),
         })
 
-    print("Best config is:", analysis.get_best_config(metric="mean_accuracy"))
+    print("Best config is:",
+          analysis.get_best_config(metric="mean_accuracy", mode="max"))
diff --git a/python/ray/tune/examples/pbt_convnet_example.py b/python/ray/tune/examples/pbt_convnet_example.py
index c9f455ccc0fe..d996d0edda6c 100644
--- a/python/ray/tune/examples/pbt_convnet_example.py
+++ b/python/ray/tune/examples/pbt_convnet_example.py
@@ -131,8 +131,9 @@ def stop_all(self):
         })
     # __tune_end__
 
-    best_trial = analysis.get_best_trial("mean_accuracy")
-    best_checkpoint = analysis.get_best_checkpoint(best_trial, metric="mean_accuracy")
+    best_trial = analysis.get_best_trial("mean_accuracy", "max")
+    best_checkpoint = analysis.get_best_checkpoint(
+        best_trial, metric="mean_accuracy", mode="max")
     restored_trainable = PytorchTrainable()
     restored_trainable.restore(best_checkpoint)
     best_model = restored_trainable.model
diff --git a/python/ray/tune/examples/pbt_convnet_function_example.py b/python/ray/tune/examples/pbt_convnet_function_example.py
index 758148a29fba..a607bed1280a 100644
--- a/python/ray/tune/examples/pbt_convnet_function_example.py
+++ b/python/ray/tune/examples/pbt_convnet_function_example.py
@@ -116,9 +116,9 @@ def stop_all(self):
         })
     # __tune_end__
 
-    best_trial = analysis.get_best_trial("mean_accuracy")
+    best_trial = analysis.get_best_trial("mean_accuracy", mode="max")
     best_checkpoint_path = analysis.get_best_checkpoint(
-        best_trial, metric="mean_accuracy")
+        best_trial, metric="mean_accuracy", mode="max")
     best_model = ConvNet()
     best_checkpoint = torch.load(
         os.path.join(best_checkpoint_path, "checkpoint"))
diff --git a/python/ray/tune/examples/pbt_transformers/pbt_transformers.py b/python/ray/tune/examples/pbt_transformers/pbt_transformers.py
index ee144f6e3983..dc1a73e5b1e1 100644
--- a/python/ray/tune/examples/pbt_transformers/pbt_transformers.py
+++ b/python/ray/tune/examples/pbt_transformers/pbt_transformers.py
@@ -155,8 +155,8 @@ def tune_transformer(num_samples=8,
         mode="max",
         perturbation_interval=1,
         hyperparam_mutations={
-            "weight_decay": lambda: tune.uniform(0.0, 0.3).func(None),
-            "learning_rate": lambda: tune.uniform(1e-5, 5e-5).func(None),
+            "weight_decay": tune.uniform(0.0, 0.3),
+            "learning_rate": tune.uniform(1e-5, 5e-5),
             "per_gpu_train_batch_size": [16, 32, 64],
         })
 
diff --git a/python/ray/tune/experiment.py b/python/ray/tune/experiment.py
index 66410d5428ce..a312c1afeb12 100644
--- a/python/ray/tune/experiment.py
+++ b/python/ray/tune/experiment.py
@@ -8,7 +8,7 @@
 from ray.tune.function_runner import detect_checkpoint_function
 from ray.tune.registry import register_trainable, get_trainable_cls
 from ray.tune.result import DEFAULT_RESULTS_DIR
-from ray.tune.sample import sample_from
+from ray.tune.sample import Domain
 from ray.tune.stopper import FunctionStopper, Stopper
 
 logger = logging.getLogger(__name__)
@@ -235,7 +235,7 @@ def register_if_needed(cls, run_object):
 
         if isinstance(run_object, str):
             return run_object
-        elif isinstance(run_object, sample_from):
+        elif isinstance(run_object, Domain):
             logger.warning("Not registering trainable. Resolving as variant.")
             return run_object
         elif isinstance(run_object, type) or callable(run_object):
diff --git a/python/ray/tune/sample.py b/python/ray/tune/sample.py
index 4cac8572b417..c0a85a62a5f5 100644
--- a/python/ray/tune/sample.py
+++ b/python/ray/tune/sample.py
@@ -1,28 +1,312 @@
 import logging
 import random
+from copy import copy
+from inspect import signature
+from numbers import Number
+from typing import Any, Callable, Dict, List, Optional, Sequence, Union
 
 import numpy as np
 
 logger = logging.getLogger(__name__)
 
 
-class sample_from:
-    """Specify that tune should sample configuration values from this function.
+class Domain:
+    """Base class to specify a type and valid range to sample parameters from.
+
+    This base class is implemented by parameter spaces, like float ranges
+    (``Float``), integer ranges (``Integer``), or categorical variables
+    (``Categorical``). The ``Domain`` object contains information about
+    valid values (e.g. minimum and maximum values), and exposes methods that
+    allow specification of specific samplers (e.g. ``uniform()`` or
+    ``loguniform()``).
 
-    Arguments:
-        func: An callable function to draw a sample from.
     """
+    sampler = None
+    default_sampler_cls = None
+
+    def cast(self, value):
+        """Cast value to domain type"""
+        return value
+
+    def set_sampler(self, sampler, allow_override=False):
+        if self.sampler and not allow_override:
+            raise ValueError("You can only choose one sampler for parameter "
+                             "domains. Existing sampler for parameter {}: "
+                             "{}. Tried to add {}".format(
+                                 self.__class__.__name__, self.sampler,
+                                 sampler))
+        self.sampler = sampler
+
+    def get_sampler(self):
+        sampler = self.sampler
+        if not sampler:
+            sampler = self.default_sampler_cls()
+        return sampler
+
+    def sample(self, spec=None, size=1):
+        sampler = self.get_sampler()
+        return sampler.sample(self, spec=spec, size=size)
+
+    def is_grid(self):
+        return isinstance(self.sampler, Grid)
+
+    def is_function(self):
+        return False
+
+
+class Sampler:
+    def sample(self,
+               domain: Domain,
+               spec: Optional[Union[List[Dict], Dict]] = None,
+               size: int = 1):
+        raise NotImplementedError
+
+
+class BaseSampler(Sampler):
+    def __str__(self):
+        return "Base"
 
-    def __init__(self, func):
-        self.func = func
 
+class Uniform(Sampler):
     def __str__(self):
-        return "tune.sample_from({})".format(str(self.func))
+        return "Uniform"
+
+
+class LogUniform(Sampler):
+    def __init__(self, base: float = 10):
+        self.base = base
+        assert self.base > 0, "Base has to be strictly greater than 0"
+
+    def __str__(self):
+        return "LogUniform"
+
+
+class Normal(Sampler):
+    def __init__(self, mean: float = 0., sd: float = 0.):
+        self.mean = mean
+        self.sd = sd
+
+        assert self.sd > 0, "SD has to be strictly greater than 0"
+
+    def __str__(self):
+        return "Normal"
+
+
+class Grid(Sampler):
+    """Dummy sampler used for grid search"""
+
+    def sample(self,
+               domain: Domain,
+               spec: Optional[Union[List[Dict], Dict]] = None,
+               size: int = 1):
+        return RuntimeError("Do not call `sample()` on grid.")
+
+
+class Float(Domain):
+    class _Uniform(Uniform):
+        def sample(self,
+                   domain: "Float",
+                   spec: Optional[Union[List[Dict], Dict]] = None,
+                   size: int = 1):
+            assert domain.lower > float("-inf"), \
+                "Uniform needs a lower bound"
+            assert domain.upper < float("inf"), \
+                "Uniform needs a upper bound"
+            items = np.random.uniform(domain.lower, domain.upper, size=size)
+            return items if len(items) > 1 else domain.cast(items[0])
+
+    class _LogUniform(LogUniform):
+        def sample(self,
+                   domain: "Float",
+                   spec: Optional[Union[List[Dict], Dict]] = None,
+                   size: int = 1):
+            assert domain.lower > 0, \
+                "LogUniform needs a lower bound greater than 0"
+            assert 0 < domain.upper < float("inf"), \
+                "LogUniform needs a upper bound greater than 0"
+            logmin = np.log(domain.lower) / np.log(self.base)
+            logmax = np.log(domain.upper) / np.log(self.base)
+
+            items = self.base**(np.random.uniform(logmin, logmax, size=size))
+            return items if len(items) > 1 else domain.cast(items[0])
+
+    class _Normal(Normal):
+        def sample(self,
+                   domain: "Float",
+                   spec: Optional[Union[List[Dict], Dict]] = None,
+                   size: int = 1):
+            assert not domain.lower or domain.lower == float("-inf"), \
+                "Normal sampling does not allow a lower value bound."
+            assert not domain.upper or domain.upper == float("inf"), \
+                "Normal sampling does not allow a upper value bound."
+            items = np.random.normal(self.mean, self.sd, size=size)
+            return items if len(items) > 1 else domain.cast(items[0])
+
+    default_sampler_cls = _Uniform
+
+    def __init__(self, lower: Optional[float], upper: Optional[float]):
+        # Need to explicitly check for None
+        self.lower = lower if lower is not None else float("-inf")
+        self.upper = upper if upper is not None else float("inf")
+
+    def cast(self, value):
+        return float(value)
+
+    def uniform(self):
+        if not self.lower > float("-inf"):
+            raise ValueError(
+                "Uniform requires a lower bound. Make sure to set the "
+                "`lower` parameter of `Float()`.")
+        if not self.upper < float("inf"):
+            raise ValueError(
+                "Uniform requires a upper bound. Make sure to set the "
+                "`upper` parameter of `Float()`.")
+        new = copy(self)
+        new.set_sampler(self._Uniform())
+        return new
+
+    def loguniform(self, base: float = 10):
+        if not self.lower > 0:
+            raise ValueError(
+                "LogUniform requires a lower bound greater than 0."
+                f"Got: {self.lower}. Did you pass a variable that has "
+                "been log-transformed? If so, pass the non-transformed value "
+                "instead.")
+        if not 0 < self.upper < float("inf"):
+            raise ValueError(
+                "LogUniform requires a upper bound greater than 0. "
+                f"Got: {self.lower}. Did you pass a variable that has "
+                "been log-transformed? If so, pass the non-transformed value "
+                "instead.")
+        new = copy(self)
+        new.set_sampler(self._LogUniform(base))
+        return new
+
+    def normal(self, mean=0., sd=1.):
+        new = copy(self)
+        new.set_sampler(self._Normal(mean, sd))
+        return new
+
+    def quantized(self, q: Number):
+        new = copy(self)
+        new.set_sampler(Quantized(new.get_sampler(), q), allow_override=True)
+        return new
+
+
+class Integer(Domain):
+    class _Uniform(Uniform):
+        def sample(self,
+                   domain: "Integer",
+                   spec: Optional[Union[List[Dict], Dict]] = None,
+                   size: int = 1):
+            items = np.random.randint(domain.lower, domain.upper, size=size)
+            return items if len(items) > 1 else domain.cast(items[0])
+
+    default_sampler_cls = _Uniform
+
+    def __init__(self, lower, upper):
+        self.lower = lower
+        self.upper = upper
+
+    def cast(self, value):
+        return int(value)
+
+    def quantized(self, q: Number):
+        new = copy(self)
+        new.set_sampler(Quantized(new.get_sampler(), q), allow_override=True)
+        return new
+
+    def uniform(self):
+        new = copy(self)
+        new.set_sampler(self._Uniform())
+        return new
+
+
+class Categorical(Domain):
+    class _Uniform(Uniform):
+        def sample(self,
+                   domain: "Categorical",
+                   spec: Optional[Union[List[Dict], Dict]] = None,
+                   size: int = 1):
+
+            items = random.choices(domain.categories, k=size)
+            return items if len(items) > 1 else domain.cast(items[0])
+
+    default_sampler_cls = _Uniform
+
+    def __init__(self, categories: Sequence):
+        self.categories = list(categories)
+
+    def uniform(self):
+        new = copy(self)
+        new.set_sampler(self._Uniform())
+        return new
+
+    def grid(self):
+        new = copy(self)
+        new.set_sampler(Grid())
+        return new
+
+    def __len__(self):
+        return len(self.categories)
+
+    def __getitem__(self, item):
+        return self.categories[item]
+
+
+class Function(Domain):
+    class _CallSampler(BaseSampler):
+        def sample(self,
+                   domain: "Function",
+                   spec: Optional[Union[List[Dict], Dict]] = None,
+                   size: int = 1):
+            pass_spec = len(signature(domain.func).parameters) > 0
+            if pass_spec:
+                items = [
+                    domain.func(spec[i] if isinstance(spec, list) else spec)
+                    for i in range(size)
+                ]
+            else:
+                items = [domain.func() for i in range(size)]
+
+            return items if len(items) > 1 else domain.cast(items[0])
+
+    default_sampler_cls = _CallSampler
+
+    def __init__(self, func: Callable):
+        if len(signature(func).parameters) > 1:
+            raise ValueError(
+                "The function passed to a `Function` parameter must accept "
+                "either 0 or 1 parameters.")
+
+        self.func = func
+
+    def is_function(self):
+        return True
+
+
+class Quantized(Sampler):
+    def __init__(self, sampler: Sampler, q: Number):
+        self.sampler = sampler
+        self.q = q
+
+        assert self.sampler, "Quantized() expects a sampler instance"
+
+    def get_sampler(self):
+        return self.sampler
 
-    def __repr__(self):
-        return "tune.sample_from({})".format(repr(self.func))
+    def sample(self,
+               domain: Domain,
+               spec: Optional[Union[List[Dict], Dict]] = None,
+               size: int = 1):
+        values = self.sampler.sample(domain, spec, size)
+        quantized = np.round(np.divide(values, self.q)) * self.q
+        if not isinstance(quantized, np.ndarray):
+            return domain.cast(quantized)
+        return list(quantized)
 
 
+# TODO (krfricke): Remove tune.function
 def function(func):
     logger.warning(
         "DeprecationWarning: wrapping {} with tune.function() is no "
@@ -30,68 +314,126 @@ def function(func):
     return func
 
 
-class uniform(sample_from):
-    """Wraps tune.sample_from around ``np.random.uniform``.
+def sample_from(func: Callable[[Dict], Any]):
+    """Specify that tune should sample configuration values from this function.
+
+    Arguments:
+        func: An callable function to draw a sample from.
+    """
+    return Function(func)
+
 
-    ``tune.uniform(1, 10)`` is equivalent to
-    ``tune.sample_from(lambda _: np.random.uniform(1, 10))``
+def uniform(lower: float, upper: float):
+    """Sample a float value uniformly between ``lower`` and ``upper``.
+
+    Sampling from ``tune.uniform(1, 10)`` is equivalent to sampling from
+    ``np.random.uniform(1, 10))``
 
     """
+    return Float(lower, upper).uniform()
+
 
-    def __init__(self, *args, **kwargs):
-        super().__init__(lambda _: np.random.uniform(*args, **kwargs))
+def quniform(lower: float, upper: float, q: float):
+    """Sample a quantized float value uniformly between ``lower`` and ``upper``.
 
+    Sampling from ``tune.uniform(1, 10)`` is equivalent to sampling from
+    ``np.random.uniform(1, 10))``
 
-class loguniform(sample_from):
+    The value will be quantized, i.e. rounded to an integer increment of ``q``.
+    Quantization makes the upper bound inclusive.
+
+    """
+    return Float(lower, upper).uniform().quantized(q)
+
+
+def loguniform(lower: float, upper: float, base: float = 10):
     """Sugar for sampling in different orders of magnitude.
 
     Args:
-        min_bound (float): Lower boundary of the output interval (1e-4)
-        max_bound (float): Upper boundary of the output interval (1e-2)
-        base (float): Base of the log. Defaults to 10.
+        lower (float): Lower boundary of the output interval (e.g. 1e-4)
+        upper (float): Upper boundary of the output interval (e.g. 1e-2)
+        base (int): Base of the log. Defaults to 10.
+
     """
+    return Float(lower, upper).loguniform(base)
+
+
+def qloguniform(lower: float, upper: float, q: float, base: float = 10):
+    """Sugar for sampling in different orders of magnitude.
 
-    def __init__(self, min_bound, max_bound, base=10):
-        logmin = np.log(min_bound) / np.log(base)
-        logmax = np.log(max_bound) / np.log(base)
+    The value will be quantized, i.e. rounded to an integer increment of ``q``.
 
-        def apply_log(_):
-            return base**(np.random.uniform(logmin, logmax))
+    Quantization makes the upper bound inclusive.
 
-        super().__init__(apply_log)
+    Args:
+        lower (float): Lower boundary of the output interval (e.g. 1e-4)
+        upper (float): Upper boundary of the output interval (e.g. 1e-2)
+        q (float): Quantization number. The result will be rounded to an
+            integer increment of this value.
+        base (int): Base of the log. Defaults to 10.
+
+    """
+    return Float(lower, upper).loguniform(base).quantized(q)
 
 
-class choice(sample_from):
-    """Wraps tune.sample_from around ``random.choice``.
+def choice(categories: List):
+    """Sample a categorical value.
 
-    ``tune.choice([1, 2])`` is equivalent to
-    ``tune.sample_from(lambda _: random.choice([1, 2]))``
+    Sampling from ``tune.choice([1, 2])`` is equivalent to sampling from
+    ``random.choice([1, 2])``
 
     """
+    return Categorical(categories).uniform()
 
-    def __init__(self, *args, **kwargs):
-        super().__init__(lambda _: random.choice(*args, **kwargs))
 
+def randint(lower: int, upper: int):
+    """Sample an integer value uniformly between ``lower`` and ``upper``.
 
-class randint(sample_from):
-    """Wraps tune.sample_from around ``np.random.randint``.
+    ``lower`` is inclusive, ``upper`` is exclusive.
 
-    ``tune.randint(10)`` is equivalent to
-    ``tune.sample_from(lambda _: np.random.randint(10))``
+    Sampling from ``tune.randint(10)`` is equivalent to sampling from
+    ``np.random.randint(10)``
 
     """
+    return Integer(lower, upper).uniform()
+
+
+def qrandint(lower: int, upper: int, q: int = 1):
+    """Sample an integer value uniformly between ``lower`` and ``upper``.
+
+    ``lower`` is inclusive, ``upper`` is also inclusive (!).
+
+    The value will be quantized, i.e. rounded to an integer increment of ``q``.
+    Quantization makes the upper bound inclusive.
+
+    Sampling from ``tune.randint(10)`` is equivalent to sampling from
+    ``np.random.randint(10)``
 
-    def __init__(self, *args, **kwargs):
-        super().__init__(lambda _: np.random.randint(*args, **kwargs))
+    """
+    return Integer(lower, upper).uniform().quantized(q)
 
 
-class randn(sample_from):
-    """Wraps tune.sample_from around ``np.random.randn``.
+def randn(mean: float = 0., sd: float = 1.):
+    """Sample a float value normally with ``mean`` and ``sd``.
 
-    ``tune.randn(10)`` is equivalent to
-    ``tune.sample_from(lambda _: np.random.randn(10))``
+    Args:
+        mean (float): Mean of the normal distribution. Defaults to 0.
+        sd (float): SD of the normal distribution. Defaults to 1.
 
     """
+    return Float(None, None).normal(mean, sd)
+
+
+def qrandn(mean: float, sd: float, q: float):
+    """Sample a float value normally with ``mean`` and ``sd``.
 
-    def __init__(self, *args, **kwargs):
-        super().__init__(lambda _: np.random.randn(*args, **kwargs))
+    The value will be quantized, i.e. rounded to an integer increment of ``q``.
+
+    Args:
+        mean (float): Mean of the normal distribution.
+        sd (float): SD of the normal distribution.
+        q (float): Quantization number. The result will be rounded to an
+            integer increment of this value.
+
+    """
+    return Float(None, None).normal(mean, sd).quantized(q)
diff --git a/python/ray/tune/schedulers/pbt.py b/python/ray/tune/schedulers/pbt.py
index fd19ff9336c0..70137e8de3f6 100644
--- a/python/ray/tune/schedulers/pbt.py
+++ b/python/ray/tune/schedulers/pbt.py
@@ -9,7 +9,7 @@
 from ray.tune.error import TuneError
 from ray.tune.result import TRAINING_ITERATION
 from ray.tune.logger import _SafeFallbackEncoder
-from ray.tune.sample import sample_from
+from ray.tune.sample import Domain, Function
 from ray.tune.schedulers import FIFOScheduler, TrialScheduler
 from ray.tune.suggest.variant_generator import format_vars
 from ray.tune.trial import Trial, Checkpoint
@@ -68,8 +68,8 @@ def explore(config, mutations, resample_probability, custom_explore_fn):
                     distribution.index(config[key]) + 1)]
         else:
             if random.random() < resample_probability:
-                new_config[key] = distribution.func(None) if isinstance(
-                    distribution, sample_from) else distribution()
+                new_config[key] = distribution.sample(None) if isinstance(
+                    distribution, Domain) else distribution()
             elif random.random() > 0.5:
                 new_config[key] = config[key] * 1.2
             else:
@@ -96,8 +96,8 @@ def fill_config(config, attr, search_space):
     """Add attr to config by sampling from search_space."""
     if callable(search_space):
         config[attr] = search_space()
-    elif isinstance(search_space, sample_from):
-        config[attr] = search_space.func(None)
+    elif isinstance(search_space, Domain):
+        config[attr] = search_space.sample(None)
     elif isinstance(search_space, list):
         config[attr] = random.choice(search_space)
     elif isinstance(search_space, dict):
@@ -228,11 +228,11 @@ def __init__(self,
                  synch=False):
         for value in hyperparam_mutations.values():
             if not (isinstance(value,
-                               (list, dict, sample_from)) or callable(value)):
+                               (list, dict, Domain)) or callable(value)):
                 raise TypeError("`hyperparam_mutation` values must be either "
                                 "a List, Dict, a tune search space object, or "
-                                "callable.")
-            if type(value) is sample_from:
+                                "a callable.")
+            if isinstance(value, Function):
                 raise ValueError("arbitrary tune.sample_from objects are not "
                                  "supported for `hyperparam_mutation` values."
                                  "You must use other built in primitives like"
diff --git a/python/ray/tune/suggest/ax.py b/python/ray/tune/suggest/ax.py
index 9e58c4cb1f11..e25556f86296 100644
--- a/python/ray/tune/suggest/ax.py
+++ b/python/ray/tune/suggest/ax.py
@@ -1,3 +1,12 @@
+from typing import Dict
+
+from ax.service.ax_client import AxClient
+from ray.tune.sample import Categorical, Float, Integer, LogUniform, \
+    Quantized, Uniform
+from ray.tune.suggest.variant_generator import parse_spec_vars
+from ray.tune.utils import flatten_dict
+from ray.tune.utils.util import unflatten_dict
+
 try:
     import ax
 except ImportError:
@@ -24,7 +33,7 @@ class AxSearch(Searcher):
         $ pip install ax-platform sqlalchemy
 
     Parameters:
-        parameters (list[dict]): Parameters in the experiment search space.
+        space (list[dict]): Parameters in the experiment search space.
             Required elements in the dictionaries are: "name" (name of
             this parameter, string), "type" (type of the parameter: "range",
             "fixed", or "choice", string), "bounds" for range parameters
@@ -41,8 +50,11 @@ class AxSearch(Searcher):
             "x3 >= x4" or "x3 + x4 >= 2".
         outcome_constraints (list[str]): Outcome constraints of form
             "metric_name >= bound", like "m1 <= 3."
-        max_concurrent (int): Deprecated.
+        ax_client (AxClient): Optional AxClient instance. If this is set, do
+            not pass any values to these parameters: `space`, `objective_name`,
+            `parameter_constraints`, `outcome_constraints`.
         use_early_stopped_trials: Deprecated.
+        max_concurrent (int): Deprecated.
 
     .. code-block:: python
 
@@ -60,41 +72,112 @@ def easy_objective(config):
                 intermediate_result = config["x1"] + config["x2"] * i
                 tune.report(score=intermediate_result)
 
-        client = AxClient(enforce_sequential_optimization=False)
-        client.create_experiment(parameters=parameters, objective_name="score")
-        algo = AxSearch(client)
+        client = AxClient()
+        algo = AxSearch(space=parameters, objective_name="score")
         tune.run(easy_objective, search_alg=algo)
 
     """
 
     def __init__(self,
-                 ax_client,
+                 space=None,
+                 metric="episode_reward_mean",
                  mode="max",
+                 parameter_constraints=None,
+                 outcome_constraints=None,
+                 ax_client=None,
                  use_early_stopped_trials=None,
                  max_concurrent=None):
         assert ax is not None, "Ax must be installed!"
-        self._ax = ax_client
-        exp = self._ax.experiment
-        self._objective_name = exp.optimization_config.objective.metric.name
-        self.max_concurrent = max_concurrent
-        self._parameters = list(exp.parameters)
-        self._live_trial_mapping = {}
+        assert mode in ["min", "max"], "`mode` must be one of ['min', 'max']"
+
         super(AxSearch, self).__init__(
-            metric=self._objective_name,
+            metric=metric,
             mode=mode,
             max_concurrent=max_concurrent,
             use_early_stopped_trials=use_early_stopped_trials)
+
+        self._ax = ax_client
+        self._space = space
+        self._parameter_constraints = parameter_constraints
+        self._outcome_constraints = outcome_constraints
+
+        self.max_concurrent = max_concurrent
+
+        self._objective_name = metric
+        self._parameters = []
+        self._live_trial_mapping = {}
+
+        if self._ax or self._space:
+            self.setup_experiment()
+
+    def setup_experiment(self):
+        if not self._ax:
+            self._ax = AxClient()
+
+        try:
+            exp = self._ax.experiment
+            has_experiment = True
+        except ValueError:
+            has_experiment = False
+
+        if not has_experiment:
+            if not self._space:
+                raise ValueError(
+                    "You have to create an Ax experiment by calling "
+                    "`AxClient.create_experiment()`, or you should pass an "
+                    "Ax search space as the `space` parameter to `AxSearch`, "
+                    "or pass a `config` dict to `tune.run()`.")
+            self._ax.create_experiment(
+                parameters=self._space,
+                objective_name=self._metric,
+                parameter_constraints=self._parameter_constraints,
+                outcome_constraints=self._outcome_constraints,
+                minimize=self._mode != "max")
+        else:
+            if any([
+                    self._space, self._parameter_constraints,
+                    self._outcome_constraints
+            ]):
+                raise ValueError(
+                    "If you create the Ax experiment yourself, do not pass "
+                    "values for these parameters to `AxSearch`: {}.".format([
+                        "space", "parameter_constraints", "outcome_constraints"
+                    ]))
+
+        exp = self._ax.experiment
+        self._objective_name = exp.optimization_config.objective.metric.name
+        self._parameters = list(exp.parameters)
+
         if self._ax._enforce_sequential_optimization:
             logger.warning("Detected sequential enforcement. Be sure to use "
                            "a ConcurrencyLimiter.")
 
+    def set_search_properties(self, metric, mode, config):
+        if self._ax:
+            return False
+        space = self.convert_search_space(config)
+        self._space = space
+        if metric:
+            self._metric = metric
+        if mode:
+            self._mode = mode
+        self.setup_experiment()
+        return True
+
     def suggest(self, trial_id):
+        if not self._ax:
+            raise RuntimeError(
+                "Trying to sample a configuration from {}, but no search "
+                "space has been defined. Either pass the `{}` argument when "
+                "instantiating the search algorithm, or pass a `config` to "
+                "`tune.run()`.".format(self.__class__.__name__, "space"))
+
         if self.max_concurrent:
             if len(self._live_trial_mapping) >= self.max_concurrent:
                 return None
         parameters, trial_index = self._ax.get_next_trial()
         self._live_trial_mapping[trial_id] = trial_index
-        return parameters
+        return unflatten_dict(parameters)
 
     def on_trial_complete(self, trial_id, result=None, error=False):
         """Notification for the completion of trial.
@@ -117,3 +200,82 @@ def _process_result(self, trial_id, result):
         metric_dict.update({on: (result[on], 0.0) for on in outcome_names})
         self._ax.complete_trial(
             trial_index=ax_trial_index, raw_data=metric_dict)
+
+    @staticmethod
+    def convert_search_space(spec: Dict):
+        spec = flatten_dict(spec, prevent_delimiter=True)
+        resolved_vars, domain_vars, grid_vars = parse_spec_vars(spec)
+
+        if grid_vars:
+            raise ValueError(
+                "Grid search parameters cannot be automatically converted "
+                "to an Ax search space.")
+
+        def resolve_value(par, domain):
+            sampler = domain.get_sampler()
+            if isinstance(sampler, Quantized):
+                logger.warning("AxSearch does not support quantization. "
+                               "Dropped quantization.")
+                sampler = sampler.sampler
+
+            if isinstance(domain, Float):
+                if isinstance(sampler, LogUniform):
+                    return {
+                        "name": par,
+                        "type": "range",
+                        "bounds": [domain.lower, domain.upper],
+                        "value_type": "float",
+                        "log_scale": True
+                    }
+                elif isinstance(sampler, Uniform):
+                    return {
+                        "name": par,
+                        "type": "range",
+                        "bounds": [domain.lower, domain.upper],
+                        "value_type": "float",
+                        "log_scale": False
+                    }
+            elif isinstance(domain, Integer):
+                if isinstance(sampler, LogUniform):
+                    return {
+                        "name": par,
+                        "type": "range",
+                        "bounds": [domain.lower, domain.upper],
+                        "value_type": "int",
+                        "log_scale": True
+                    }
+                elif isinstance(sampler, Uniform):
+                    return {
+                        "name": par,
+                        "type": "range",
+                        "bounds": [domain.lower, domain.upper],
+                        "value_type": "int",
+                        "log_scale": False
+                    }
+            elif isinstance(domain, Categorical):
+                if isinstance(sampler, Uniform):
+                    return {
+                        "name": par,
+                        "type": "choice",
+                        "values": domain.categories
+                    }
+
+            raise ValueError("AxSearch does not support parameters of type "
+                             "`{}` with samplers of type `{}`".format(
+                                 type(domain).__name__,
+                                 type(domain.sampler).__name__))
+
+        # Fixed vars
+        fixed_values = [{
+            "name": "/".join(path),
+            "type": "fixed",
+            "value": val
+        } for path, val in resolved_vars]
+
+        # Parameter name is e.g. "a/b/c" for nested dicts
+        resolved_values = [
+            resolve_value("/".join(path), domain)
+            for path, domain in domain_vars
+        ]
+
+        return fixed_values + resolved_values
diff --git a/python/ray/tune/suggest/bayesopt.py b/python/ray/tune/suggest/bayesopt.py
index c647966fe1ae..7aa4e2bbbb5d 100644
--- a/python/ray/tune/suggest/bayesopt.py
+++ b/python/ray/tune/suggest/bayesopt.py
@@ -1,8 +1,13 @@
-import copy
 from collections import defaultdict
 import logging
 import pickle
 import json
+from typing import Dict
+
+from ray.tune.sample import Float, Quantized
+from ray.tune.suggest.variant_generator import parse_spec_vars
+from ray.tune.utils.util import unflatten_dict
+
 try:  # Python 3 only -- needed for lint test.
     import bayes_opt as byo
 except ImportError:
@@ -76,8 +81,8 @@ class BayesOptSearch(Searcher):
     optimizer = None
 
     def __init__(self,
-                 space,
-                 metric,
+                 space=None,
+                 metric="episode_reward_mean",
                  mode="max",
                  utility_kwargs=None,
                  random_state=42,
@@ -154,15 +159,45 @@ def __init__(self,
         self.random_search_trials = random_search_steps
         self._total_random_search_trials = 0
 
-        self.optimizer = byo.BayesianOptimization(
-            f=None, pbounds=space, verbose=verbose, random_state=random_state)
-
         self.utility = byo.UtilityFunction(**utility_kwargs)
 
         # Registering the provided analysis, if given
         if analysis is not None:
             self.register_analysis(analysis)
 
+        self._space = space
+        self._verbose = verbose
+        self._random_state = random_state
+
+        self.optimizer = None
+        if space:
+            self.setup_optimizer()
+
+    def setup_optimizer(self):
+        self.optimizer = byo.BayesianOptimization(
+            f=None,
+            pbounds=self._space,
+            verbose=self._verbose,
+            random_state=self._random_state)
+
+    def set_search_properties(self, metric, mode, config):
+        if self.optimizer:
+            return False
+        space = self.convert_search_space(config)
+        self._space = space
+        if metric:
+            self._metric = metric
+        if mode:
+            self._mode = mode
+
+        if self._mode == "max":
+            self._metric_op = 1.
+        elif self._mode == "min":
+            self._metric_op = -1.
+
+        self.setup_optimizer()
+        return True
+
     def suggest(self, trial_id):
         """Return new point to be explored by black box function.
 
@@ -174,6 +209,13 @@ def suggest(self, trial_id):
             Either a dictionary describing the new point to explore or
             None, when no new point is to be explored for the time being.
         """
+        if not self.optimizer:
+            raise RuntimeError(
+                "Trying to sample a configuration from {}, but no search "
+                "space has been defined. Either pass the `{}` argument when "
+                "instantiating the search algorithm, or pass a `config` to "
+                "`tune.run()`.".format(self.__class__.__name__, "space"))
+
         # If we have more active trials than the allowed maximum
         total_live_trials = len(self._live_trial_mapping)
         if self.max_concurrent and self.max_concurrent <= total_live_trials:
@@ -214,7 +256,7 @@ def suggest(self, trial_id):
         self._live_trial_mapping[trial_id] = config
 
         # Return a deep copy of the mapping
-        return copy.deepcopy(config)
+        return unflatten_dict(config)
 
     def register_analysis(self, analysis):
         """Integrate the given analysis into the gaussian process.
@@ -283,3 +325,44 @@ def restore(self, checkpoint_path):
             (self.optimizer, self._buffered_trial_results,
              self._total_random_search_trials,
              self._config_counter) = pickle.load(f)
+
+    @staticmethod
+    def convert_search_space(spec: Dict):
+        spec = flatten_dict(spec, prevent_delimiter=True)
+        resolved_vars, domain_vars, grid_vars = parse_spec_vars(spec)
+
+        if grid_vars:
+            raise ValueError(
+                "Grid search parameters cannot be automatically converted "
+                "to a BayesOpt search space.")
+
+        if resolved_vars:
+            raise ValueError(
+                "BayesOpt does not support fixed parameters. Please find a "
+                "different way to pass constants to your training function.")
+
+        def resolve_value(domain):
+            sampler = domain.get_sampler()
+            if isinstance(sampler, Quantized):
+                logger.warning(
+                    "BayesOpt search does not support quantization. "
+                    "Dropped quantization.")
+                sampler = sampler.get_sampler()
+
+            if isinstance(domain, Float):
+                if domain.sampler is not None:
+                    logger.warning(
+                        "BayesOpt does not support specific sampling methods. "
+                        "The {} sampler will be dropped.".format(sampler))
+                    return (domain.lower, domain.upper)
+
+            raise ValueError("BayesOpt does not support parameters of type "
+                             "`{}`".format(type(domain).__name__))
+
+        # Parameter name is e.g. "a/b/c" for nested dicts
+        bounds = {
+            "/".join(path): resolve_value(domain)
+            for path, domain in domain_vars
+        }
+
+        return bounds
diff --git a/python/ray/tune/suggest/hyperopt.py b/python/ray/tune/suggest/hyperopt.py
index 73e1af8704e0..6cb6d1dfe517 100644
--- a/python/ray/tune/suggest/hyperopt.py
+++ b/python/ray/tune/suggest/hyperopt.py
@@ -1,8 +1,16 @@
+from typing import Dict
+
 import numpy as np
 import copy
 import logging
 from functools import partial
 import pickle
+
+from ray.tune.sample import Categorical, Float, Integer, LogUniform, Normal, \
+    Quantized, \
+    Uniform
+from ray.tune.suggest.variant_generator import assign_value, parse_spec_vars
+
 try:
     hyperopt_logger = logging.getLogger("hyperopt")
     hyperopt_logger.setLevel(logging.WARNING)
@@ -84,7 +92,7 @@ class HyperOptSearch(Searcher):
 
     def __init__(
             self,
-            space,
+            space=None,
             metric="episode_reward_mean",
             mode="max",
             points_to_evaluate=None,
@@ -116,7 +124,6 @@ def __init__(
                 hpo.tpe.suggest, n_startup_jobs=n_initial_points)
         if gamma is not None:
             self.algo = partial(self.algo, gamma=gamma)
-        self.domain = hpo.Domain(lambda spc: spc, space)
         if points_to_evaluate is None:
             self._hpopt_trials = hpo.Trials()
             self._points_to_evaluate = 0
@@ -132,7 +139,35 @@ def __init__(
         else:
             self.rstate = np.random.RandomState(random_state_seed)
 
+        self.domain = None
+        if space:
+            self.domain = hpo.Domain(lambda spc: spc, space)
+
+    def set_search_properties(self, metric, mode, config):
+        if self.domain:
+            return False
+        space = self.convert_search_space(config)
+        self.domain = hpo.Domain(lambda spc: spc, space)
+
+        if metric:
+            self._metric = metric
+        if mode:
+            self._mode = mode
+
+        if self._mode == "max":
+            self.metric_op = -1.
+        elif self._mode == "min":
+            self.metric_op = 1.
+
+        return True
+
     def suggest(self, trial_id):
+        if not self.domain:
+            raise RuntimeError(
+                "Trying to sample a configuration from {}, but no search "
+                "space has been defined. Either pass the `{}` argument when "
+                "instantiating the search algorithm, or pass a `config` to "
+                "`tune.run()`.".format(self.__class__.__name__, "space"))
         if self.max_concurrent:
             if len(self._live_trial_mapping) >= self.max_concurrent:
                 return None
@@ -235,3 +270,75 @@ def restore(self, checkpoint_path):
             self.rstate.set_state(trials_object[1])
         else:
             self.set_state(trials_object)
+
+    @staticmethod
+    def convert_search_space(spec: Dict):
+        spec = copy.deepcopy(spec)
+        resolved_vars, domain_vars, grid_vars = parse_spec_vars(spec)
+
+        if not domain_vars and not grid_vars:
+            return []
+
+        if grid_vars:
+            raise ValueError(
+                "Grid search parameters cannot be automatically converted "
+                "to a HyperOpt search space.")
+
+        def resolve_value(par, domain):
+            quantize = None
+
+            sampler = domain.get_sampler()
+            if isinstance(sampler, Quantized):
+                quantize = sampler.q
+                sampler = sampler.sampler
+
+            if isinstance(domain, Float):
+                if isinstance(sampler, LogUniform):
+                    if quantize:
+                        return hpo.hp.qloguniform(par, domain.lower,
+                                                  domain.upper, quantize)
+                    return hpo.hp.loguniform(par, np.log(domain.lower),
+                                             np.log(domain.upper))
+                elif isinstance(sampler, Uniform):
+                    if quantize:
+                        return hpo.hp.quniform(par, domain.lower, domain.upper,
+                                               quantize)
+                    return hpo.hp.uniform(par, domain.lower, domain.upper)
+                elif isinstance(sampler, Normal):
+                    if quantize:
+                        return hpo.hp.qnormal(par, sampler.mean, sampler.sd,
+                                              quantize)
+                    return hpo.hp.normal(par, sampler.mean, sampler.sd)
+
+            elif isinstance(domain, Integer):
+                if isinstance(sampler, Uniform):
+                    if quantize:
+                        logger.warning(
+                            "HyperOpt does not support quantization for "
+                            "integer values. Reverting back to 'randint'.")
+                    if domain.lower != 0:
+                        raise ValueError(
+                            "HyperOpt only allows integer sampling with "
+                            f"lower bound 0. Got: {domain.lower}.")
+                    if domain.upper < 1:
+                        raise ValueError(
+                            "HyperOpt does not support integer sampling "
+                            "of values lower than 0. Set your maximum range "
+                            "to something above 0 (currently {})".format(
+                                domain.upper))
+                    return hpo.hp.randint(par, domain.upper)
+            elif isinstance(domain, Categorical):
+                if isinstance(sampler, Uniform):
+                    return hpo.hp.choice(par, domain.categories)
+
+            raise ValueError("HyperOpt does not support parameters of type "
+                             "`{}` with samplers of type `{}`".format(
+                                 type(domain).__name__,
+                                 type(domain.sampler).__name__))
+
+        for path, domain in domain_vars:
+            par = "/".join(path)
+            value = resolve_value(par, domain)
+            assign_value(spec, path, value)
+
+        return spec
diff --git a/python/ray/tune/suggest/nevergrad.py b/python/ray/tune/suggest/nevergrad.py
index 17e4defaeba8..6208657225a8 100644
--- a/python/ray/tune/suggest/nevergrad.py
+++ b/python/ray/tune/suggest/nevergrad.py
@@ -115,7 +115,6 @@ def suggest(self, trial_id):
         # in v0.2.0+, output of ask() is a Candidate,
         # with fields args and kwargs
         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
diff --git a/python/ray/tune/suggest/optuna.py b/python/ray/tune/suggest/optuna.py
index 51d3efc14ee7..2d184f78d0a7 100644
--- a/python/ray/tune/suggest/optuna.py
+++ b/python/ray/tune/suggest/optuna.py
@@ -1,7 +1,13 @@
 import logging
 import pickle
+from typing import Dict
 
 from ray.tune.result import TRAINING_ITERATION
+from ray.tune.sample import Categorical, Float, Integer, LogUniform, \
+    Quantized, Uniform
+from ray.tune.suggest.variant_generator import parse_spec_vars
+from ray.tune.utils import flatten_dict
+from ray.tune.utils.util import unflatten_dict
 
 try:
     import optuna as ot
@@ -74,13 +80,11 @@ class OptunaSearch(Searcher):
 
     """
 
-    def __init__(
-            self,
-            space,
-            metric="episode_reward_mean",
-            mode="max",
-            sampler=None,
-    ):
+    def __init__(self,
+                 space=None,
+                 metric="episode_reward_mean",
+                 mode="max",
+                 sampler=None):
         assert ot is not None, (
             "Optuna must be installed! Run `pip install optuna`.")
         super(OptunaSearch, self).__init__(
@@ -101,6 +105,11 @@ def __init__(
         self._storage = ot.storages.InMemoryStorage()
 
         self._ot_trials = {}
+        self._ot_study = None
+        if self._space:
+            self.setup_study(mode)
+
+    def setup_study(self, mode):
         self._ot_study = ot.study.create_study(
             storage=self._storage,
             sampler=self._sampler,
@@ -109,18 +118,40 @@ def __init__(
             direction="minimize" if mode == "min" else "maximize",
             load_if_exists=True)
 
+    def set_search_properties(self, metric, mode, config):
+        if self._space:
+            return False
+        space = self.convert_search_space(config)
+        self._space = space
+        if metric:
+            self._metric = metric
+        if mode:
+            self._mode = mode
+        self.setup_study(mode)
+        return True
+
     def suggest(self, trial_id):
+        if not self._space:
+            raise RuntimeError(
+                "Trying to sample a configuration from {}, but no search "
+                "space has been defined. Either pass the `{}` argument when "
+                "instantiating the search algorithm, or pass a `config` to "
+                "`tune.run()`.".format(self.__class__.__name__, "space"))
+
         if trial_id not in self._ot_trials:
             ot_trial_id = self._storage.create_new_trial(
                 self._ot_study._study_id)
             self._ot_trials[trial_id] = ot.trial.Trial(self._ot_study,
                                                        ot_trial_id)
         ot_trial = self._ot_trials[trial_id]
-        params = {}
-        for (fn, args, kwargs) in self._space:
-            param_name = args[0] if len(args) > 0 else kwargs["name"]
-            params[param_name] = getattr(ot_trial, fn)(*args, **kwargs)
-        return params
+
+        # getattr will fetch the trial.suggest_ function on Optuna trials
+        params = {
+            args[0] if len(args) > 0 else kwargs["name"]: getattr(
+                ot_trial, fn)(*args, **kwargs)
+            for (fn, args, kwargs) in self._space
+        }
+        return unflatten_dict(params)
 
     def on_trial_result(self, trial_id, result):
         metric = result[self.metric]
@@ -147,3 +178,67 @@ def restore(self, checkpoint_path):
             save_object = pickle.load(inputFile)
         self._storage, self._pruner, self._sampler, \
             self._ot_trials, self._ot_study = save_object
+
+    @staticmethod
+    def convert_search_space(spec: Dict):
+        spec = flatten_dict(spec, prevent_delimiter=True)
+        resolved_vars, domain_vars, grid_vars = parse_spec_vars(spec)
+
+        if not domain_vars and not grid_vars:
+            return []
+
+        if grid_vars:
+            raise ValueError(
+                "Grid search parameters cannot be automatically converted "
+                "to an Optuna search space.")
+
+        def resolve_value(par, domain):
+            quantize = None
+
+            sampler = domain.get_sampler()
+            if isinstance(sampler, Quantized):
+                quantize = sampler.q
+                sampler = sampler.sampler
+
+            if isinstance(domain, Float):
+                if isinstance(sampler, LogUniform):
+                    if quantize:
+                        logger.warning(
+                            "Optuna does not support both quantization and "
+                            "sampling from LogUniform. Dropped quantization.")
+                    return param.suggest_loguniform(par, domain.lower,
+                                                    domain.upper)
+                elif isinstance(sampler, Uniform):
+                    if quantize:
+                        return param.suggest_discrete_uniform(
+                            par, domain.lower, domain.upper, quantize)
+                    return param.suggest_uniform(par, domain.lower,
+                                                 domain.upper)
+            elif isinstance(domain, Integer):
+                if isinstance(sampler, LogUniform):
+                    if quantize:
+                        logger.warning(
+                            "Optuna does not support both quantization and "
+                            "sampling from LogUniform. Dropped quantization.")
+                    return param.suggest_int(
+                        par, domain.lower, domain.upper, log=True)
+                elif isinstance(sampler, Uniform):
+                    return param.suggest_int(
+                        par, domain.lower, domain.upper, step=quantize or 1)
+            elif isinstance(domain, Categorical):
+                if isinstance(sampler, Uniform):
+                    return param.suggest_categorical(par, domain.categories)
+
+            raise ValueError(
+                "Optuna search does not support parameters of type "
+                "`{}` with samplers of type `{}`".format(
+                    type(domain).__name__,
+                    type(domain.sampler).__name__))
+
+        # Parameter name is e.g. "a/b/c" for nested dicts
+        values = [
+            resolve_value("/".join(path), domain)
+            for path, domain in domain_vars
+        ]
+
+        return values
diff --git a/python/ray/tune/suggest/search.py b/python/ray/tune/suggest/search.py
index 64562753384e..a878410a95f9 100644
--- a/python/ray/tune/suggest/search.py
+++ b/python/ray/tune/suggest/search.py
@@ -12,6 +12,23 @@ class SearchAlgorithm:
     """
     _finished = False
 
+    def set_search_properties(self, metric, mode, config):
+        """Pass search properties to search algorithm.
+
+        This method acts as an alternative to instantiating search algorithms
+        with their own specific search spaces. Instead they can accept a
+        Tune config through this method.
+
+        The search algorithm will usually pass this method to their
+        ``Searcher`` instance.
+
+        Args:
+            metric (str): Metric to optimize
+            mode (str): One of ["min", "max"]. Direction to optimize.
+            config (dict): Tune config dict.
+        """
+        return True
+
     def add_configurations(self, experiments):
         """Tracks given experiment specifications.
 
diff --git a/python/ray/tune/suggest/search_generator.py b/python/ray/tune/suggest/search_generator.py
index bc23247d904a..c5990c63c5c1 100644
--- a/python/ray/tune/suggest/search_generator.py
+++ b/python/ray/tune/suggest/search_generator.py
@@ -69,6 +69,9 @@ def __init__(self, searcher):
         self._total_samples = None  # int: total samples to evaluate.
         self._finished = False
 
+    def set_search_properties(self, metric, mode, config):
+        return self.searcher.set_search_properties(metric, mode, config)
+
     def add_configurations(self, experiments):
         """Registers experiment specifications.
 
diff --git a/python/ray/tune/suggest/suggestion.py b/python/ray/tune/suggest/suggestion.py
index 403cfbfb0fb5..633fe33718e1 100644
--- a/python/ray/tune/suggest/suggestion.py
+++ b/python/ray/tune/suggest/suggestion.py
@@ -86,6 +86,22 @@ def __init__(self,
         self._metric = metric
         self._mode = mode
 
+    def set_search_properties(self, metric, mode, config):
+        """Pass search properties to searcher.
+
+        This method acts as an alternative to instantiating search algorithms
+        with their own specific search spaces. Instead they can accept a
+        Tune config through this method. A searcher should return ``True``
+        if setting the config was successful, or ``False`` if it was
+        unsuccessful, e.g. when the search space has already been set.
+
+        Args:
+            metric (str): Metric to optimize
+            mode (str): One of ["min", "max"]. Direction to optimize.
+            config (dict): Tune config dict.
+        """
+        return False
+
     def on_trial_result(self, trial_id, result):
         """Optional notification for result during training.
 
diff --git a/python/ray/tune/suggest/variant_generator.py b/python/ray/tune/suggest/variant_generator.py
index e772ffb4933c..a539c7b2c481 100644
--- a/python/ray/tune/suggest/variant_generator.py
+++ b/python/ray/tune/suggest/variant_generator.py
@@ -4,7 +4,7 @@
 import random
 
 from ray.tune import TuneError
-from ray.tune.sample import sample_from
+from ray.tune.sample import Categorical, Domain, Function
 
 logger = logging.getLogger(__name__)
 
@@ -115,25 +115,36 @@ def _clean_value(value):
         return str(value).replace("/", "_")
 
 
-def _generate_variants(spec):
-    spec = copy.deepcopy(spec)
-    unresolved = _unresolved_values(spec)
+def parse_spec_vars(spec):
+    resolved, unresolved = _split_resolved_unresolved_values(spec)
+    resolved_vars = list(resolved.items())
+
     if not unresolved:
-        yield {}, spec
-        return
+        return resolved_vars, [], []
 
     grid_vars = []
-    lambda_vars = []
+    domain_vars = []
     for path, value in unresolved.items():
-        if callable(value):
-            lambda_vars.append((path, value))
-        else:
+        if value.is_grid():
             grid_vars.append((path, value))
+        else:
+            domain_vars.append((path, value))
     grid_vars.sort()
 
+    return resolved_vars, domain_vars, grid_vars
+
+
+def _generate_variants(spec):
+    spec = copy.deepcopy(spec)
+    _, domain_vars, grid_vars = parse_spec_vars(spec)
+
+    if not domain_vars and not grid_vars:
+        yield {}, spec
+        return
+
     grid_search = _grid_search_generator(spec, grid_vars)
     for resolved_spec in grid_search:
-        resolved_vars = _resolve_lambda_vars(resolved_spec, lambda_vars)
+        resolved_vars = _resolve_domain_vars(resolved_spec, domain_vars)
         for resolved, spec in _generate_variants(resolved_spec):
             for path, value in grid_vars:
                 resolved_vars[path] = _get_value(spec, path)
@@ -148,7 +159,7 @@ def _generate_variants(spec):
             yield resolved_vars, spec
 
 
-def _assign_value(spec, path, value):
+def assign_value(spec, path, value):
     for k in path[:-1]:
         spec = spec[k]
     spec[path[-1]] = value
@@ -160,23 +171,26 @@ def _get_value(spec, path):
     return spec
 
 
-def _resolve_lambda_vars(spec, lambda_vars):
+def _resolve_domain_vars(spec, domain_vars):
     resolved = {}
     error = True
     num_passes = 0
     while error and num_passes < _MAX_RESOLUTION_PASSES:
         num_passes += 1
         error = False
-        for path, fn in lambda_vars:
+        for path, domain in domain_vars:
+            if path in resolved:
+                continue
             try:
-                value = fn(_UnresolvedAccessGuard(spec))
+                value = domain.sample(_UnresolvedAccessGuard(spec))
             except RecursiveDependencyError as e:
                 error = e
             except Exception:
                 raise ValueError(
-                    "Failed to evaluate expression: {}: {}".format(path, fn))
+                    "Failed to evaluate expression: {}: {}".format(
+                        path, domain))
             else:
-                _assign_value(spec, path, value)
+                assign_value(spec, path, value)
                 resolved[path] = value
     if error:
         raise error
@@ -203,7 +217,7 @@ def increment(i):
     while value_indices[-1] < len(grid_vars[-1][1]):
         spec = copy.deepcopy(unresolved_spec)
         for i, (path, values) in enumerate(grid_vars):
-            _assign_value(spec, path, values[value_indices[i]])
+            assign_value(spec, path, values[value_indices[i]])
         yield spec
         if grid_vars:
             done = increment(0)
@@ -217,13 +231,13 @@ def _is_resolved(v):
 
 
 def _try_resolve(v):
-    if isinstance(v, sample_from):
-        # Function to sample from
-        return False, v.func
+    if isinstance(v, Domain):
+        # Domain to sample from
+        return False, v
     elif isinstance(v, dict) and len(v) == 1 and "eval" in v:
         # Lambda function in eval syntax
-        return False, lambda spec: eval(
-            v["eval"], _STANDARD_IMPORTS, {"spec": spec})
+        return False, Function(
+            lambda spec: eval(v["eval"], _STANDARD_IMPORTS, {"spec": spec}))
     elif isinstance(v, dict) and len(v) == 1 and "grid_search" in v:
         # Grid search values
         grid_values = v["grid_search"]
@@ -231,26 +245,45 @@ def _try_resolve(v):
             raise TuneError(
                 "Grid search expected list of values, got: {}".format(
                     grid_values))
-        return False, grid_values
+        return False, Categorical(grid_values).grid()
     return True, v
 
 
-def _unresolved_values(spec):
-    found = {}
+def _split_resolved_unresolved_values(spec):
+    resolved_vars = {}
+    unresolved_vars = {}
     for k, v in spec.items():
         resolved, v = _try_resolve(v)
         if not resolved:
-            found[(k, )] = v
+            unresolved_vars[(k, )] = v
         elif isinstance(v, dict):
             # Recurse into a dict
-            for (path, value) in _unresolved_values(v).items():
-                found[(k, ) + path] = value
+            _resolved_children, _unresolved_children = \
+                _split_resolved_unresolved_values(v)
+            for (path, value) in _resolved_children.items():
+                resolved_vars[(k, ) + path] = value
+            for (path, value) in _unresolved_children.items():
+                unresolved_vars[(k, ) + path] = value
         elif isinstance(v, list):
             # Recurse into a list
             for i, elem in enumerate(v):
-                for (path, value) in _unresolved_values({i: elem}).items():
-                    found[(k, ) + path] = value
-    return found
+                _resolved_children, _unresolved_children = \
+                    _split_resolved_unresolved_values({i: elem})
+                for (path, value) in _resolved_children.items():
+                    resolved_vars[(k, ) + path] = value
+                for (path, value) in _unresolved_children.items():
+                    unresolved_vars[(k, ) + path] = value
+        else:
+            resolved_vars[(k, )] = v
+    return resolved_vars, unresolved_vars
+
+
+def _unresolved_values(spec):
+    return _split_resolved_unresolved_values(spec)[1]
+
+
+def has_unresolved_values(spec):
+    return True if _unresolved_values(spec) else False
 
 
 class _UnresolvedAccessGuard(dict):
diff --git a/python/ray/tune/tests/example.py b/python/ray/tune/tests/example.py
index be0bd2d17148..69d1f854b577 100644
--- a/python/ray/tune/tests/example.py
+++ b/python/ray/tune/tests/example.py
@@ -35,7 +35,8 @@ def training_function(config):
         "beta": tune.choice([1, 2, 3])
     })
 
-print("Best config: ", analysis.get_best_config(metric="mean_loss"))
+print("Best config: ", analysis.get_best_config(
+    metric="mean_loss", mode="min"))
 
 # Get a dataframe for analyzing trial results.
 df = analysis.dataframe()
diff --git a/python/ray/tune/tests/test_experiment_analysis.py b/python/ray/tune/tests/test_experiment_analysis.py
index 5b1f4c7bb22c..2aef0db1b0d1 100644
--- a/python/ray/tune/tests/test_experiment_analysis.py
+++ b/python/ray/tune/tests/test_experiment_analysis.py
@@ -73,18 +73,18 @@ def testTrialDataframe(self):
         self.assertEqual(trial_df.shape[0], 1)
 
     def testBestConfig(self):
-        best_config = self.ea.get_best_config(self.metric)
+        best_config = self.ea.get_best_config(self.metric, mode="max")
         self.assertTrue(isinstance(best_config, dict))
         self.assertTrue("width" in best_config)
         self.assertTrue("height" in best_config)
 
     def testBestConfigNan(self):
         nan_ea = self.nan_test_exp()
-        best_config = nan_ea.get_best_config(self.metric)
+        best_config = nan_ea.get_best_config(self.metric, mode="max")
         self.assertIsNone(best_config)
 
     def testBestLogdir(self):
-        logdir = self.ea.get_best_logdir(self.metric)
+        logdir = self.ea.get_best_logdir(self.metric, mode="max")
         self.assertTrue(logdir.startswith(self.test_path))
         logdir2 = self.ea.get_best_logdir(self.metric, mode="min")
         self.assertTrue(logdir2.startswith(self.test_path))
@@ -92,45 +92,46 @@ def testBestLogdir(self):
 
     def testBestLogdirNan(self):
         nan_ea = self.nan_test_exp()
-        logdir = nan_ea.get_best_logdir(self.metric)
+        logdir = nan_ea.get_best_logdir(self.metric, mode="max")
         self.assertIsNone(logdir)
 
     def testGetTrialCheckpointsPathsByTrial(self):
-        best_trial = self.ea.get_best_trial(self.metric)
+        best_trial = self.ea.get_best_trial(self.metric, mode="max")
         checkpoints_metrics = self.ea.get_trial_checkpoints_paths(best_trial)
-        logdir = self.ea.get_best_logdir(self.metric)
+        logdir = self.ea.get_best_logdir(self.metric, mode="max")
         expected_path = os.path.join(logdir, "checkpoint_1", "checkpoint")
         assert checkpoints_metrics[0][0] == expected_path
         assert checkpoints_metrics[0][1] == 1
 
     def testGetTrialCheckpointsPathsByPath(self):
-        logdir = self.ea.get_best_logdir(self.metric)
+        logdir = self.ea.get_best_logdir(self.metric, mode="max")
         checkpoints_metrics = self.ea.get_trial_checkpoints_paths(logdir)
         expected_path = os.path.join(logdir, "checkpoint_1/", "checkpoint")
         assert checkpoints_metrics[0][0] == expected_path
         assert checkpoints_metrics[0][1] == 1
 
     def testGetTrialCheckpointsPathsWithMetricByTrial(self):
-        best_trial = self.ea.get_best_trial(self.metric)
+        best_trial = self.ea.get_best_trial(self.metric, mode="max")
         paths = self.ea.get_trial_checkpoints_paths(best_trial, self.metric)
-        logdir = self.ea.get_best_logdir(self.metric)
+        logdir = self.ea.get_best_logdir(self.metric, mode="max")
         expected_path = os.path.join(logdir, "checkpoint_1", "checkpoint")
         assert paths[0][0] == expected_path
         assert paths[0][1] == best_trial.metric_analysis[self.metric]["last"]
 
     def testGetTrialCheckpointsPathsWithMetricByPath(self):
-        best_trial = self.ea.get_best_trial(self.metric)
-        logdir = self.ea.get_best_logdir(self.metric)
+        best_trial = self.ea.get_best_trial(self.metric, mode="max")
+        logdir = self.ea.get_best_logdir(self.metric, mode="max")
         paths = self.ea.get_trial_checkpoints_paths(best_trial, self.metric)
         expected_path = os.path.join(logdir, "checkpoint_1", "checkpoint")
         assert paths[0][0] == expected_path
         assert paths[0][1] == best_trial.metric_analysis[self.metric]["last"]
 
     def testGetBestCheckpoint(self):
-        best_trial = self.ea.get_best_trial(self.metric)
+        best_trial = self.ea.get_best_trial(self.metric, mode="max")
         checkpoints_metrics = self.ea.get_trial_checkpoints_paths(best_trial)
         expected_path = max(checkpoints_metrics, key=lambda x: x[1])[0]
-        best_checkpoint = self.ea.get_best_checkpoint(best_trial, self.metric)
+        best_checkpoint = self.ea.get_best_checkpoint(
+            best_trial, self.metric, mode="max")
         assert expected_path == best_checkpoint
 
     def testAllDataframes(self):
diff --git a/python/ray/tune/tests/test_experiment_analysis_mem.py b/python/ray/tune/tests/test_experiment_analysis_mem.py
index 94e544b766ec..65e5320251c6 100644
--- a/python/ray/tune/tests/test_experiment_analysis_mem.py
+++ b/python/ray/tune/tests/test_experiment_analysis_mem.py
@@ -156,7 +156,7 @@ def testDataframe(self):
 
     def testBestLogdir(self):
         analysis = Analysis(self.test_dir)
-        logdir = analysis.get_best_logdir(self.metric)
+        logdir = analysis.get_best_logdir(self.metric, mode="max")
         self.assertTrue(logdir.startswith(self.test_dir))
         logdir2 = analysis.get_best_logdir(self.metric, mode="min")
         self.assertTrue(logdir2.startswith(self.test_dir))
diff --git a/python/ray/tune/tests/test_sample.py b/python/ray/tune/tests/test_sample.py
new file mode 100644
index 000000000000..63080f06b234
--- /dev/null
+++ b/python/ray/tune/tests/test_sample.py
@@ -0,0 +1,334 @@
+import numpy as np
+import unittest
+
+from ray import tune
+from ray.tune.suggest.variant_generator import generate_variants
+
+
+def _mock_objective(config):
+    tune.report(**config)
+
+
+class SearchSpaceTest(unittest.TestCase):
+    def setUp(self):
+        pass
+
+    def tearDown(self):
+        pass
+
+    def testTuneSampleAPI(self):
+        config = {
+            "func": tune.sample_from(lambda spec: spec.config.uniform * 0.01),
+            "uniform": tune.uniform(-5, -1),
+            "quniform": tune.quniform(3.2, 5.4, 0.2),
+            "loguniform": tune.loguniform(1e-4, 1e-2),
+            "qloguniform": tune.qloguniform(1e-4, 1e-1, 5e-4),
+            "choice": tune.choice([2, 3, 4]),
+            "randint": tune.randint(-9, 15),
+            "qrandint": tune.qrandint(-21, 12, 3),
+            "randn": tune.randn(10, 2),
+            "qrandn": tune.qrandn(10, 2, 0.2),
+        }
+        for _, (_, generated) in zip(
+                range(10), generate_variants({
+                    "config": config
+                })):
+            out = generated["config"]
+
+            self.assertAlmostEqual(out["func"], out["uniform"] * 0.01)
+
+            self.assertGreater(out["uniform"], -5)
+            self.assertLess(out["uniform"], -1)
+
+            self.assertGreater(out["quniform"], 3.2)
+            self.assertLessEqual(out["quniform"], 5.4)
+            self.assertAlmostEqual(out["quniform"] / 0.2,
+                                   round(out["quniform"] / 0.2))
+
+            self.assertGreater(out["loguniform"], 1e-4)
+            self.assertLess(out["loguniform"], 1e-2)
+
+            self.assertGreater(out["qloguniform"], 1e-4)
+            self.assertLessEqual(out["qloguniform"], 1e-1)
+            self.assertAlmostEqual(out["qloguniform"] / 5e-4,
+                                   round(out["qloguniform"] / 5e-4))
+
+            self.assertIn(out["choice"], [2, 3, 4])
+
+            self.assertGreater(out["randint"], -9)
+            self.assertLess(out["randint"], 15)
+
+            self.assertGreater(out["qrandint"], -21)
+            self.assertLessEqual(out["qrandint"], 12)
+            self.assertEqual(out["qrandint"] % 3, 0)
+
+            # Very improbable
+            self.assertGreater(out["randn"], 0)
+            self.assertLess(out["randn"], 20)
+
+            self.assertGreater(out["qrandn"], 0)
+            self.assertLess(out["qrandn"], 20)
+            self.assertAlmostEqual(out["qrandn"] / 0.2,
+                                   round(out["qrandn"] / 0.2))
+
+    def testBoundedFloat(self):
+        bounded = tune.sample.Float(-4.2, 8.3)
+
+        # Don't allow to specify more than one sampler
+        with self.assertRaises(ValueError):
+            bounded.normal().uniform()
+
+        # Uniform
+        samples = bounded.uniform().sample(size=1000)
+        self.assertTrue(any(-4.2 < s < 8.3 for s in samples))
+        self.assertFalse(np.mean(samples) < -2)
+
+        # Loguniform
+        with self.assertRaises(ValueError):
+            bounded.loguniform().sample(size=1000)
+
+        bounded_positive = tune.sample.Float(1e-4, 1e-1)
+        samples = bounded_positive.loguniform().sample(size=1000)
+        self.assertTrue(any(1e-4 < s < 1e-1 for s in samples))
+
+    def testUnboundedFloat(self):
+        unbounded = tune.sample.Float(None, None)
+
+        # Require min and max bounds for loguniform
+        with self.assertRaises(ValueError):
+            unbounded.loguniform()
+
+        # Normal
+        samples = tune.sample.Float(None, None).normal().sample(size=1000)
+        self.assertTrue(any(-5 < s < 5 for s in samples))
+        self.assertTrue(-1 < np.mean(samples) < 1)
+
+    def testBoundedInt(self):
+        bounded = tune.sample.Integer(-3, 12)
+
+        samples = bounded.uniform().sample(size=1000)
+        self.assertTrue(any(-3 <= s < 12 for s in samples))
+        self.assertFalse(np.mean(samples) < 2)
+
+    def testCategorical(self):
+        categories = [-2, -1, 0, 1, 2]
+        cat = tune.sample.Categorical(categories)
+
+        samples = cat.uniform().sample(size=1000)
+        self.assertTrue(any(-2 <= s <= 2 for s in samples))
+        self.assertTrue(all(c in samples for c in categories))
+
+    def testFunction(self):
+        def sample(spec):
+            return np.random.uniform(-4, 4)
+
+        fnc = tune.sample.Function(sample)
+
+        samples = fnc.sample(size=1000)
+        self.assertTrue(any(-4 < s < 4 for s in samples))
+        self.assertTrue(-2 < np.mean(samples) < 2)
+
+    def testQuantized(self):
+        bounded_positive = tune.sample.Float(1e-4, 1e-1)
+        samples = bounded_positive.loguniform().quantized(5e-4).sample(size=10)
+
+        for sample in samples:
+            factor = sample / 5e-4
+            self.assertAlmostEqual(factor, round(factor), places=10)
+
+    def testConvertAx(self):
+        from ray.tune.suggest.ax import AxSearch
+        from ax.service.ax_client import AxClient
+
+        config = {
+            "a": tune.sample.Categorical([2, 3, 4]).uniform(),
+            "b": {
+                "x": tune.sample.Integer(0, 5).quantized(2),
+                "y": 4,
+                "z": tune.sample.Float(1e-4, 1e-2).loguniform()
+            }
+        }
+        converted_config = AxSearch.convert_search_space(config)
+        ax_config = [
+            {
+                "name": "a",
+                "type": "choice",
+                "values": [2, 3, 4]
+            },
+            {
+                "name": "b/x",
+                "type": "range",
+                "bounds": [0, 5],
+                "value_type": "int"
+            },
+            {
+                "name": "b/y",
+                "type": "fixed",
+                "value": 4
+            },
+            {
+                "name": "b/z",
+                "type": "range",
+                "bounds": [1e-4, 1e-2],
+                "value_type": "float",
+                "log_scale": True
+            },
+        ]
+
+        client1 = AxClient(random_seed=1234)
+        client1.create_experiment(parameters=converted_config)
+        searcher1 = AxSearch(ax_client=client1)
+
+        client2 = AxClient(random_seed=1234)
+        client2.create_experiment(parameters=ax_config)
+        searcher2 = AxSearch(ax_client=client2)
+
+        config1 = searcher1.suggest("0")
+        config2 = searcher2.suggest("0")
+
+        self.assertEqual(config1, config2)
+        self.assertIn(config1["a"], [2, 3, 4])
+        self.assertIn(config1["b"]["x"], list(range(5)))
+        self.assertEqual(config1["b"]["y"], 4)
+        self.assertLess(1e-4, config1["b"]["z"])
+        self.assertLess(config1["b"]["z"], 1e-2)
+
+        searcher = AxSearch(metric="a", mode="max")
+        analysis = tune.run(
+            _mock_objective, config=config, search_alg=searcher, num_samples=1)
+        trial = analysis.trials[0]
+        assert trial.config["a"] in [2, 3, 4]
+
+    def testConvertBayesOpt(self):
+        from ray.tune.suggest.bayesopt import BayesOptSearch
+
+        config = {
+            "a": tune.sample.Categorical([2, 3, 4]).uniform(),
+            "b": {
+                "x": tune.sample.Integer(0, 5).quantized(2),
+                "y": 4,
+                "z": tune.sample.Float(1e-4, 1e-2).loguniform()
+            }
+        }
+        with self.assertRaises(ValueError):
+            converted_config = BayesOptSearch.convert_search_space(config)
+
+        config = {"b": {"z": tune.sample.Float(1e-4, 1e-2).loguniform()}}
+        bayesopt_config = {"b/z": (1e-4, 1e-2)}
+        converted_config = BayesOptSearch.convert_search_space(config)
+
+        searcher1 = BayesOptSearch(space=converted_config, metric="none")
+        searcher2 = BayesOptSearch(space=bayesopt_config, metric="none")
+
+        config1 = searcher1.suggest("0")
+        config2 = searcher2.suggest("0")
+
+        self.assertEqual(config1, config2)
+        self.assertLess(1e-4, config1["b"]["z"])
+        self.assertLess(config1["b"]["z"], 1e-2)
+
+        searcher = BayesOptSearch()
+        invalid_config = {"a/b": tune.uniform(4.0, 8.0)}
+        with self.assertRaises(ValueError):
+            searcher.set_search_properties("none", "max", invalid_config)
+        invalid_config = {"a": {"b/c": tune.uniform(4.0, 8.0)}}
+        with self.assertRaises(ValueError):
+            searcher.set_search_properties("none", "max", invalid_config)
+
+        searcher = BayesOptSearch(metric="a", mode="max")
+        analysis = tune.run(
+            _mock_objective, config=config, search_alg=searcher, num_samples=1)
+        trial = analysis.trials[0]
+        self.assertLess(trial.config["b"]["z"], 1e-2)
+
+    def testConvertHyperOpt(self):
+        from ray.tune.suggest.hyperopt import HyperOptSearch
+        from hyperopt import hp
+
+        config = {
+            "a": tune.sample.Categorical([2, 3, 4]).uniform(),
+            "b": {
+                "x": tune.sample.Integer(0, 5).quantized(2),
+                "y": 4,
+                "z": tune.sample.Float(1e-4, 1e-2).loguniform()
+            }
+        }
+        converted_config = HyperOptSearch.convert_search_space(config)
+        hyperopt_config = {
+            "a": hp.choice("a", [2, 3, 4]),
+            "b": {
+                "x": hp.randint("x", 5),
+                "y": 4,
+                "z": hp.loguniform("z", np.log(1e-4), np.log(1e-2))
+            }
+        }
+
+        searcher1 = HyperOptSearch(
+            space=converted_config, random_state_seed=1234)
+        searcher2 = HyperOptSearch(
+            space=hyperopt_config, random_state_seed=1234)
+
+        config1 = searcher1.suggest("0")
+        config2 = searcher2.suggest("0")
+
+        self.assertEqual(config1, config2)
+        self.assertIn(config1["a"], [2, 3, 4])
+        self.assertIn(config1["b"]["x"], list(range(5)))
+        self.assertEqual(config1["b"]["y"], 4)
+        self.assertLess(1e-4, config1["b"]["z"])
+        self.assertLess(config1["b"]["z"], 1e-2)
+
+        searcher = HyperOptSearch(metric="a", mode="max")
+        analysis = tune.run(
+            _mock_objective, config=config, search_alg=searcher, num_samples=1)
+        trial = analysis.trials[0]
+        assert trial.config["a"] in [2, 3, 4]
+
+    def testConvertOptuna(self):
+        from ray.tune.suggest.optuna import OptunaSearch, param
+        from optuna.samplers import RandomSampler
+
+        config = {
+            "a": tune.sample.Categorical([2, 3, 4]).uniform(),
+            "b": {
+                "x": tune.sample.Integer(0, 5).quantized(2),
+                "y": 4,
+                "z": tune.sample.Float(1e-4, 1e-2).loguniform()
+            }
+        }
+        converted_config = OptunaSearch.convert_search_space(config)
+        optuna_config = [
+            param.suggest_categorical("a", [2, 3, 4]),
+            param.suggest_int("b/x", 0, 5, 2),
+            param.suggest_loguniform("b/z", 1e-4, 1e-2)
+        ]
+
+        sampler1 = RandomSampler(seed=1234)
+        searcher1 = OptunaSearch(
+            space=converted_config, sampler=sampler1, base_config=config)
+
+        sampler2 = RandomSampler(seed=1234)
+        searcher2 = OptunaSearch(
+            space=optuna_config, sampler=sampler2, base_config=config)
+
+        config1 = searcher1.suggest("0")
+        config2 = searcher2.suggest("0")
+
+        self.assertEqual(config1, config2)
+        self.assertIn(config1["a"], [2, 3, 4])
+        self.assertIn(config1["b"]["x"], list(range(5)))
+        self.assertEqual(config1["b"]["y"], 4)
+        self.assertLess(1e-4, config1["b"]["z"])
+        self.assertLess(config1["b"]["z"], 1e-2)
+
+        searcher = OptunaSearch(metric="a", mode="max")
+        analysis = tune.run(
+            _mock_objective, config=config, search_alg=searcher, num_samples=1)
+        trial = analysis.trials[0]
+        assert trial.config["a"] in [2, 3, 4]
+
+
+if __name__ == "__main__":
+    import pytest
+    import sys
+    sys.exit(pytest.main(["-v", __file__]))
diff --git a/python/ray/tune/tests/test_var.py b/python/ray/tune/tests/test_var.py
index 3df8e7f16edb..082a9fe0345c 100644
--- a/python/ray/tune/tests/test_var.py
+++ b/python/ray/tune/tests/test_var.py
@@ -263,13 +263,13 @@ def testNestedValues(self):
         })
 
     def testLogUniform(self):
-        sampler = tune.loguniform(1e-10, 1e-1).func
-        results = [sampler(None) for i in range(1000)]
+        sampler = tune.loguniform(1e-10, 1e-1)
+        results = sampler.sample(None, 1000)
         assert abs(np.log(min(results)) / np.log(10) - -10) < 0.1
         assert abs(np.log(max(results)) / np.log(10) - -1) < 0.1
 
-        sampler_e = tune.loguniform(np.e**-4, np.e, base=np.e).func
-        results_e = [sampler_e(None) for i in range(1000)]
+        sampler_e = tune.loguniform(np.e**-4, np.e, base=np.e)
+        results_e = sampler_e.sample(None, 1000)
         assert abs(np.log(min(results_e)) - -4) < 0.1
         assert abs(np.log(max(results_e)) - 1) < 0.1
 
diff --git a/python/ray/tune/tune.py b/python/ray/tune/tune.py
index 543de99a51c5..fba669a1b3b6 100644
--- a/python/ray/tune/tune.py
+++ b/python/ray/tune/tune.py
@@ -5,6 +5,7 @@
 from ray.tune.analysis import ExperimentAnalysis
 from ray.tune.suggest import BasicVariantGenerator, SearchGenerator
 from ray.tune.suggest.suggestion import Searcher
+from ray.tune.suggest.variant_generator import has_unresolved_values
 from ray.tune.trial import Trial
 from ray.tune.trainable import Trainable
 from ray.tune.ray_trial_executor import RayTrialExecutor
@@ -328,6 +329,16 @@ def run(run_or_experiment,
     if not search_alg:
         search_alg = BasicVariantGenerator()
 
+    # TODO (krfricke): Introduce metric/mode as top level API
+    if config and not search_alg.set_search_properties(None, None, config):
+        if has_unresolved_values(config):
+            raise ValueError(
+                "You passed a `config` parameter to `tune.run()` with "
+                "unresolved parameters, but the search algorithm was already "
+                "instantiated with a search space. Make sure that `config` "
+                "does not contain any more parameter definitions - include "
+                "them in the search algorithm's search space if necessary.")
+
     runner = TrialRunner(
         search_alg=search_alg,
         scheduler=scheduler or FIFOScheduler(),
@@ -404,7 +415,11 @@ def run(run_or_experiment,
     trials = runner.get_trials()
     if return_trials:
         return trials
-    return ExperimentAnalysis(runner.checkpoint_file, trials=trials)
+    return ExperimentAnalysis(
+        runner.checkpoint_file,
+        trials=trials,
+        default_metric=None,
+        default_mode=None)
 
 
 def run_experiments(experiments,
diff --git a/python/ray/tune/utils/util.py b/python/ray/tune/utils/util.py
index 14af9d998df1..ff5e7b15dc79 100644
--- a/python/ray/tune/utils/util.py
+++ b/python/ray/tune/utils/util.py
@@ -215,14 +215,25 @@ def deep_update(original,
     return original
 
 
-def flatten_dict(dt, delimiter="/"):
+def flatten_dict(dt, delimiter="/", prevent_delimiter=False):
     dt = copy.deepcopy(dt)
+    if prevent_delimiter and any(delimiter in key for key in dt):
+        # Raise if delimiter is any of the keys
+        raise ValueError(
+            "Found delimiter `{}` in key when trying to flatten array."
+            "Please avoid using the delimiter in your specification.")
     while any(isinstance(v, dict) for v in dt.values()):
         remove = []
         add = {}
         for key, value in dt.items():
             if isinstance(value, dict):
                 for subkey, v in value.items():
+                    if prevent_delimiter and delimiter in subkey:
+                        # Raise  if delimiter is in any of the subkeys
+                        raise ValueError(
+                            "Found delimiter `{}` in key when trying to "
+                            "flatten array. Please avoid using the delimiter "
+                            "in your specification.")
                     add[delimiter.join([key, subkey])] = v
                 remove.append(key)
         dt.update(add)
@@ -231,6 +242,18 @@ def flatten_dict(dt, delimiter="/"):
     return dt
 
 
+def unflatten_dict(dt, delimiter="/"):
+    """Unflatten dict. Does not support unflattening lists."""
+    out = defaultdict(dict)
+    for key, val in dt.items():
+        path = key.split(delimiter)
+        item = out
+        for k in path[:-1]:
+            item = item[k]
+        item[path[-1]] = val
+    return dict(out)
+
+
 def unflattened_lookup(flat_key, lookup, delimiter="/", **kwargs):
     """
     Unflatten `flat_key` and iteratively look up in `lookup`. E.g.
diff --git a/python/ray/util/sgd/tf/examples/tensorflow_train_example.py b/python/ray/util/sgd/tf/examples/tensorflow_train_example.py
index 5982f7b8595f..41f4505ea864 100644
--- a/python/ray/util/sgd/tf/examples/tensorflow_train_example.py
+++ b/python/ray/util/sgd/tf/examples/tensorflow_train_example.py
@@ -95,8 +95,8 @@ def train_example(num_replicas=1, batch_size=128, use_gpu=False):
 
 def tune_example(num_replicas=1, use_gpu=False):
     config = {
-        "model_creator": tune.function(simple_model),
-        "data_creator": tune.function(simple_dataset),
+        "model_creator": simple_model,
+        "data_creator": simple_dataset,
         "num_replicas": num_replicas,
         "use_gpu": use_gpu,
         "trainer_config": create_config(batch_size=128)