#!/usr/bin/env python3
# Copyright (c) Meta Platforms, Inc. and affiliates.
#
# This source code is licensed under the MIT license found in the
# LICENSE file in the root directory of this source tree.
# pyre-strict
import enum
from collections.abc import Sequence
from copy import deepcopy
from dataclasses import dataclass
from logging import Logger
from typing import Any, Union
from ax.core.arm import Arm
from ax.core.auxiliary import AuxiliaryExperiment, AuxiliaryExperimentPurpose
from ax.core.experiment import DataType, Experiment
from ax.core.metric import Metric
from ax.core.multi_type_experiment import MultiTypeExperiment
from ax.core.objective import MultiObjective, Objective
from ax.core.observation import ObservationFeatures
from ax.core.optimization_config import (
MultiObjectiveOptimizationConfig,
ObjectiveThreshold,
OptimizationConfig,
)
from ax.core.outcome_constraint import OutcomeConstraint
from ax.core.parameter import (
ChoiceParameter,
FixedParameter,
Parameter,
PARAMETER_PYTHON_TYPE_MAP,
ParameterType,
RangeParameter,
TParameterType,
)
from ax.core.parameter_constraint import (
OrderConstraint,
ParameterConstraint,
validate_constraint_parameters,
)
from ax.core.runner import Runner
from ax.core.search_space import HierarchicalSearchSpace, SearchSpace
from ax.core.types import ComparisonOp, TParameterization, TParamValue
from ax.exceptions.core import UnsupportedError
from ax.utils.common.constants import Keys
from ax.utils.common.logger import get_logger
from ax.utils.common.typeutils import (
checked_cast,
checked_cast_optional,
checked_cast_to_tuple,
)
from pyre_extensions import none_throws
DEFAULT_OBJECTIVE_NAME = "objective"
logger: Logger = get_logger(__name__)
"""Utilities for RESTful-like instantiation of Ax classes needed in AxClient."""
TParameterRepresentation = dict[
str, Union[TParamValue, Sequence[TParamValue], dict[str, list[str]]]
]
PARAM_CLASSES = ["range", "choice", "fixed"]
PARAM_TYPES = {"int": int, "float": float, "bool": bool, "str": str}
COMPARISON_OPS: dict[str, ComparisonOp] = {
"<=": ComparisonOp.LEQ,
">=": ComparisonOp.GEQ,
}
EXPECTED_KEYS_IN_PARAM_REPR = {
"name",
"type",
"values",
"bounds",
"value",
"value_type",
"log_scale",
"target_value",
"is_fidelity",
"sort_values",
"is_ordered",
"is_task",
"digits",
"dependents",
}
[docs]
class MetricObjective(enum.Enum):
MINIMIZE = enum.auto()
MAXIMIZE = enum.auto()
[docs]
@dataclass
class ObjectiveProperties:
r"""Class that holds properties of objective functions. Can be used to define an
the `objectives` argument of ax_client.create_experiment, e.g.:
ax_client.create_experiment(
name="moo_experiment",
parameters=[...],
objectives={
# `threshold` arguments are optional
"a": ObjectiveProperties(minimize=False, threshold=ref_point[0]),
"b": ObjectiveProperties(minimize=False, threshold=ref_point[1]),
},
)
Args:
- minimize: Boolean indicating whether the objective is to be minimized
or maximized.
- threshold: Optional `float` representing the smallest objective value
(resp. largest if minimize=True) that is considered valuable in the context
of multi-objective optimization. In BoTorch and in the literature, this is
also known as an element of the reference point vector that defines the
hyper-volume of the Pareto front.
"""
minimize: bool
threshold: float | None = None
[docs]
@dataclass(frozen=True)
class FixedFeatures:
"""Class for representing fixed features via the Service API."""
parameters: TParameterization
trial_index: int | None = None
[docs]
class InstantiationBase:
"""
This is a lightweight stateless class that bundles together instantiation utils.
It is used both on its own and as a mixin to AxClient, with the intent that
these methods can be overridden by its subclasses for specific use cases.
"""
@staticmethod
def _get_deserialized_metric_kwargs(
metric_class: type[Metric],
name: str,
metric_definitions: dict[str, dict[str, Any]] | None,
) -> dict[str, Any]:
"""Get metric kwargs from metric_definitions if available and deserialize
if so. Deserialization is necessary because they were serialized on creation"""
# deepcopy is used because of subsequent modifications to the dict
metric_kwargs = deepcopy((metric_definitions or {}).get(name, {}))
metric_class = metric_kwargs.pop("metric_class", metric_class)
# this is necessary before deserialization because name will be required
metric_kwargs["name"] = metric_kwargs.get("name", name)
metric_kwargs = metric_class.deserialize_init_args(metric_kwargs)
return metric_kwargs
@classmethod
def _make_metric(
cls,
name: str,
lower_is_better: bool | None = None,
metric_class: type[Metric] = Metric,
for_opt_config: bool = False,
metric_definitions: dict[str, dict[str, Any]] | None = None,
) -> Metric:
if " " in name:
raise ValueError(
"Metric names cannot contain spaces when used with AxClient. Got "
f"{name!r}."
)
kwargs = cls._get_deserialized_metric_kwargs(
name=name,
metric_definitions=metric_definitions,
metric_class=metric_class,
)
# avoid conflict is lower_is_better is specified in kwargs
kwargs["lower_is_better"] = kwargs.get("lower_is_better", lower_is_better)
return metric_class(
**kwargs,
)
@staticmethod
def _get_parameter_type(python_type: TParameterType) -> ParameterType:
for param_type, py_type in PARAMETER_PYTHON_TYPE_MAP.items():
if py_type is python_type:
return param_type
raise ValueError(f"No AE parameter type corresponding to {python_type}.")
@classmethod
def _to_parameter_type(
cls,
vals: list[TParamValue],
typ: str | None,
param_name: str,
field_name: str,
) -> ParameterType:
if typ is None:
typ = type(none_throws(vals[0]))
parameter_type = cls._get_parameter_type(typ) # pyre-ignore[6]
assert all(isinstance(x, typ) for x in vals), (
f"Values in `{field_name}` not of the same type and no "
"`value_type` was explicitly specified; cannot infer "
f"value type for parameter {param_name}."
)
logger.info(
f"Inferred value type of {parameter_type} for parameter {param_name}. "
"If that is not the expected value type, you can explicitly specify "
"'value_type' ('int', 'float', 'bool' or 'str') in parameter dict."
)
return parameter_type
return cls._get_parameter_type(PARAM_TYPES[typ]) # pyre-ignore[6]
@classmethod
def _make_range_param(
cls,
name: str,
representation: TParameterRepresentation,
parameter_type: str | None,
) -> RangeParameter:
assert "bounds" in representation, "Bounds are required for range parameters."
bounds = representation["bounds"]
assert isinstance(bounds, list) and len(bounds) == 2, (
f"Cannot parse parameter {name}: for range parameters, json representation "
"should include a list of two values, lower and upper bounds of the range."
)
return RangeParameter(
name=name,
parameter_type=cls._to_parameter_type(
bounds, parameter_type, name, "bounds"
),
lower=checked_cast_to_tuple((float, int), bounds[0]),
upper=checked_cast_to_tuple((float, int), bounds[1]),
log_scale=checked_cast(bool, representation.get("log_scale", False)),
digits=representation.get("digits", None), # pyre-ignore[6]
is_fidelity=checked_cast(bool, representation.get("is_fidelity", False)),
# pyre-ignore[6]: Expected `Union[None, bool, float, int, str]`
# for 8th param but got `Union[None, List[
# Union[None, bool, float, int, str]], bool, float, int, str]`.
target_value=representation.get("target_value", None),
)
@classmethod
def _make_choice_param(
cls,
name: str,
representation: TParameterRepresentation,
parameter_type: str | None,
) -> ChoiceParameter:
values = representation["values"]
assert isinstance(values, list) and len(values) > 1, (
f"Cannot parse parameter {name}: for choice parameters, json representation"
" should include a list of two or more values."
)
return ChoiceParameter(
name=name,
parameter_type=cls._to_parameter_type(
values, parameter_type, name, "values"
),
values=values,
is_ordered=checked_cast_optional(bool, representation.get("is_ordered")),
is_fidelity=checked_cast(bool, representation.get("is_fidelity", False)),
is_task=checked_cast(bool, representation.get("is_task", False)),
target_value=representation.get("target_value", None), # pyre-ignore[6]
sort_values=checked_cast_optional(
bool, representation.get("sort_values", None)
),
dependents=checked_cast_optional(
dict, representation.get("dependents", None)
),
)
@classmethod
def _make_fixed_param(
cls,
name: str,
representation: TParameterRepresentation,
parameter_type: str | None,
) -> FixedParameter:
assert "value" in representation, "Value is required for fixed parameters."
value = representation["value"]
assert type(value) in PARAM_TYPES.values(), (
f"Cannot parse fixed parameter {name}: for fixed parameters, json "
"representation should include a single value."
)
return FixedParameter(
name=name,
parameter_type=(
cls._get_parameter_type(type(value)) # pyre-ignore[6]
if parameter_type is None
# pyre-ignore[6]
else cls._get_parameter_type(PARAM_TYPES[parameter_type])
),
value=value, # pyre-ignore[6]
is_fidelity=checked_cast(bool, representation.get("is_fidelity", False)),
target_value=representation.get("target_value", None), # pyre-ignore[6]
dependents=representation.get("dependents", None), # pyre-ignore[6]
)
[docs]
@classmethod
def parameter_from_json(
cls,
representation: TParameterRepresentation,
) -> Parameter:
"""Instantiate a parameter from JSON representation."""
if "parameter_type" in representation:
raise ValueError(
"'parameter_type' is not an expected key in parameter dictionary. "
"If you are looking to specify the type of values that this "
"parameter should take, use 'value_type' (expects 'int', 'float', "
"'str' or 'bool')."
)
unexpected_keys = set(representation.keys()) - EXPECTED_KEYS_IN_PARAM_REPR
if unexpected_keys:
raise ValueError(
f"Unexpected keys {unexpected_keys} in parameter representation."
f"Exhaustive set of expected keys: {EXPECTED_KEYS_IN_PARAM_REPR}."
)
name = representation["name"]
assert isinstance(name, str), "Parameter name must be a string."
parameter_class = representation["type"]
assert isinstance(parameter_class, str) and parameter_class in PARAM_CLASSES, (
"Type in parameter JSON representation must be "
"`range`, `choice`, or `fixed`."
)
parameter_type = representation.get("value_type", None)
if parameter_type is not None:
assert isinstance(parameter_type, str) and parameter_type in PARAM_TYPES, (
"Value type in parameter JSON representation must be 'int', 'float', "
"'bool' or 'str'."
)
if " " in name:
raise ValueError(
"Parameter names cannot contain spaces when used with AxClient. Got "
f"{name!r}."
)
if parameter_class == "range":
return cls._make_range_param(
name=name,
representation=representation,
parameter_type=parameter_type,
)
if parameter_class == "choice":
assert (
"values" in representation
), "Values are required for choice parameters."
values = representation["values"]
if isinstance(values, list) and len(values) == 1:
logger.info(
f"Choice parameter {name} contains only one value, converting to a"
+ " fixed parameter instead."
)
# update the representation to a fixed parameter class
parameter_class = "fixed"
representation["type"] = parameter_class
representation["value"] = values[0]
del representation["values"]
else:
return cls._make_choice_param(
name=name,
representation=representation,
parameter_type=parameter_type,
)
if parameter_class == "fixed":
assert not any(isinstance(val, list) for val in representation.values())
return cls._make_fixed_param(
name=name,
representation=representation,
parameter_type=parameter_type,
)
else:
raise ValueError(f"Unrecognized parameter type {parameter_class}.")
[docs]
@staticmethod
def constraint_from_str(
representation: str, parameters: dict[str, Parameter]
) -> ParameterConstraint:
"""Parse string representation of a parameter constraint."""
tokens = representation.split()
parameter_names = parameters.keys()
try:
float(tokens[-1])
last_token_is_numeric = True
except ValueError:
last_token_is_numeric = False
order_const = len(tokens) == 3 and tokens[1] in COMPARISON_OPS
sum_const = (
len(tokens) >= 5 and len(tokens) % 2 == 1 and tokens[-2] in COMPARISON_OPS
)
if not (order_const or sum_const):
raise ValueError(
"Parameter constraint should be of form <parameter_name> >= "
"<other_parameter_name> for order constraints or `<parameter_name> "
"+ <other_parameter_name> >= x, where any number of terms can be "
"added and `x` is a float bound. Acceptable comparison operators "
'are ">=" and "<=".'
)
# Case "x1 >= x2" => order constraint.
if len(tokens) == 3 and not last_token_is_numeric:
left, right = tokens[0], tokens[2]
assert (
left in parameter_names
), f"Parameter {left} not in {parameter_names}."
assert (
right in parameter_names
), f"Parameter {right} not in {parameter_names}."
validate_constraint_parameters(
parameters=[parameters[left], parameters[right]]
)
return (
OrderConstraint(
lower_parameter=parameters[left], upper_parameter=parameters[right]
)
if COMPARISON_OPS[tokens[1]] is ComparisonOp.LEQ
else OrderConstraint(
lower_parameter=parameters[right], upper_parameter=parameters[left]
)
)
if not last_token_is_numeric:
raise ValueError(
f"Bound for the constraint must be a number; got {tokens[-1]}"
)
bound = float(tokens[-1])
if any(token[0] == "*" or token[-1] == "*" for token in tokens):
raise ValueError(
"A linear constraint should be the form a*x + b*y - c*z <= d"
", where a,b,c,d are float constants and x,y,z are parameters. "
"There should be no space in each term around the operator * while "
"there should be a single space around each operator +, -, <= and >=."
)
parameter_weight = {}
comparison_multiplier = (
1.0 if COMPARISON_OPS[tokens[-2]] is ComparisonOp.LEQ else -1.0
)
operator_sign = 1.0 # Determines whether the operator is + or -
# tokens are alternating monomials and operators
for idx, token in enumerate(tokens[:-2]):
# for monomials
if idx % 2 == 0:
split_token = token.split("*")
parameter = "" # Initializing the parameter
multiplier = 1.0 # Initializing the multiplier
if len(split_token) == 2: # There is a non-unit multiplier
try:
multiplier = float(split_token[0])
except ValueError:
raise ValueError(
f"Multiplier should be float; got {split_token[0]}"
)
parameter = split_token[1]
elif len(split_token) == 1: # The multiplier is either -1 or 1
parameter = split_token[0]
if parameter[0] == "-": # The multiplier is -1
parameter = parameter[1:]
multiplier = -1.0
else:
multiplier = 1.0
assert (
parameter in parameter_names
), f"Parameter {parameter} not in {parameter_names}."
validate_constraint_parameters(parameters=[parameters[parameter]])
parameter_weight[parameter] = operator_sign * multiplier
# for operators
else:
assert (
token == "+" or token == "-"
), f"Expected a mixed constraint, found operator {token}."
operator_sign = 1.0 if token == "+" else -1.0
return ParameterConstraint(
constraint_dict={
p: comparison_multiplier * parameter_weight[p] for p in parameter_weight
},
bound=comparison_multiplier * bound,
)
[docs]
@classmethod
def outcome_constraint_from_str(
cls,
representation: str,
metric_definitions: dict[str, dict[str, Any]] | None = None,
) -> OutcomeConstraint:
"""Parse string representation of an outcome constraint."""
tokens = representation.split()
assert len(tokens) == 3 and tokens[1] in COMPARISON_OPS, (
f"Outcome constraint '{representation}' should be of "
"form `metric_name >= x`, where x is a "
"float bound and comparison operator is >= or <=."
)
op = COMPARISON_OPS[tokens[1]]
rel = False
try:
bound_repr = tokens[2]
if bound_repr[-1] == "%":
rel = True
bound_repr = bound_repr[:-1]
bound = float(bound_repr)
except ValueError:
raise ValueError(
f"Outcome constraint bound should be a float for '{representation}'."
)
return OutcomeConstraint(
cls._make_metric(
name=tokens[0],
for_opt_config=True,
metric_definitions=metric_definitions,
lower_is_better=op is ComparisonOp.LEQ,
),
op=op,
bound=bound,
relative=rel,
)
[docs]
@classmethod
def objective_threshold_constraint_from_str(
cls,
representation: str,
metric_definitions: dict[str, dict[str, Any]] | None = None,
) -> ObjectiveThreshold:
oc = cls.outcome_constraint_from_str(
representation, metric_definitions=metric_definitions
)
return ObjectiveThreshold(
metric=oc.metric.clone(),
bound=oc.bound,
relative=oc.relative,
op=oc.op,
)
[docs]
@classmethod
def make_objectives(
cls,
objectives: dict[str, str],
metric_definitions: dict[str, dict[str, Any]] | None = None,
) -> list[Objective]:
try:
output_objectives = []
for metric_name, min_or_max in objectives.items():
minimize = (
MetricObjective[min_or_max.upper()] == MetricObjective.MINIMIZE
)
objective = Objective(
metric=cls._make_metric(
name=metric_name,
for_opt_config=True,
lower_is_better=minimize,
metric_definitions=metric_definitions,
),
minimize=minimize,
)
output_objectives.append(objective)
return output_objectives
except KeyError as k:
raise ValueError(
"Objective values should specify "
f"'{MetricObjective.MINIMIZE.name.lower()}' or "
f"'{MetricObjective.MAXIMIZE.name.lower()}', got {k} in"
f" objectives({objectives})"
)
[docs]
@classmethod
def make_outcome_constraints(
cls,
outcome_constraints: list[str],
status_quo_defined: bool,
metric_definitions: dict[str, dict[str, Any]] | None = None,
) -> list[OutcomeConstraint]:
typed_outcome_constraints = [
cls.outcome_constraint_from_str(c, metric_definitions=metric_definitions)
for c in outcome_constraints
]
if status_quo_defined is False and any(
oc.relative for oc in typed_outcome_constraints
):
raise ValueError(
"Must set status_quo to have relative outcome constraints."
)
return typed_outcome_constraints
[docs]
@classmethod
def make_objective_thresholds(
cls,
objective_thresholds: list[str],
status_quo_defined: bool,
metric_definitions: dict[str, dict[str, Any]] | None = None,
) -> list[ObjectiveThreshold]:
typed_objective_thresholds = (
[
cls.objective_threshold_constraint_from_str(
c, metric_definitions=metric_definitions
)
for c in objective_thresholds
]
if objective_thresholds is not None
else []
)
if status_quo_defined is False and any(
oc.relative for oc in typed_objective_thresholds
):
raise ValueError(
"Must set status_quo to have relative objective thresholds."
)
return typed_objective_thresholds
[docs]
@staticmethod
def optimization_config_from_objectives(
objectives: list[Objective],
objective_thresholds: list[ObjectiveThreshold],
outcome_constraints: list[OutcomeConstraint],
) -> OptimizationConfig:
"""Parse objectives and constraints to define optimization config.
The resulting optimization config will be regular single-objective config
if `objectives` is a list of one element and a multi-objective config
otherwise.
NOTE: If passing in multiple objectives, `objective_thresholds` must be a
non-empty list defining constraints for each objective.
"""
if len(objectives) == 1:
if objective_thresholds:
raise ValueError(
"Single-objective optimizations must not specify objective "
"thresholds."
)
return OptimizationConfig(
objective=objectives[0],
outcome_constraints=outcome_constraints,
)
if not objective_thresholds:
logger.info(
"Due to non-specification, we will use the heuristic for selecting "
"objective thresholds."
)
return MultiObjectiveOptimizationConfig(
objective=MultiObjective(objectives=objectives),
outcome_constraints=outcome_constraints,
objective_thresholds=objective_thresholds,
)
[docs]
@classmethod
def make_optimization_config(
cls,
objectives: dict[str, str],
objective_thresholds: list[str],
outcome_constraints: list[str],
status_quo_defined: bool,
metric_definitions: dict[str, dict[str, Any]] | None = None,
) -> OptimizationConfig:
return cls.optimization_config_from_objectives(
cls.make_objectives(objectives, metric_definitions=metric_definitions),
cls.make_objective_thresholds(
objective_thresholds,
status_quo_defined,
metric_definitions=metric_definitions,
),
cls.make_outcome_constraints(
outcome_constraints,
status_quo_defined,
metric_definitions=metric_definitions,
),
)
[docs]
@classmethod
def make_optimization_config_from_properties(
cls,
objectives: dict[str, ObjectiveProperties] | None = None,
outcome_constraints: list[str] | None = None,
metric_definitions: dict[str, dict[str, Any]] | None = None,
status_quo_defined: bool = False,
) -> OptimizationConfig | None:
"""Makes optimization config based on ObjectiveProperties objects
Args:
objectives: Mapping from an objective name to object containing:
minimize: Whether this experiment represents a minimization problem.
threshold: The bound in the objective's threshold constraint.
outcome_constraints: List of string representation of outcome
constraints of form "metric_name >= bound", like "m1 <= 3."
status_quo_defined: bool for whether the experiment has a status quo
metric_definitions: A mapping of metric names to extra kwargs to pass
to that metric
"""
if objectives is not None:
objective_thresholds = (
cls.build_objective_thresholds(objectives)
if objectives is not None
else []
)
simple_objectives = {
objective: ("minimize" if properties.minimize else "maximize")
for objective, properties in objectives.items()
}
return cls.make_optimization_config(
objectives=simple_objectives,
objective_thresholds=objective_thresholds,
outcome_constraints=outcome_constraints or [],
status_quo_defined=status_quo_defined,
metric_definitions=metric_definitions,
)
return None
[docs]
@classmethod
def make_search_space(
cls,
parameters: list[TParameterRepresentation],
parameter_constraints: list[str] | None,
) -> SearchSpace:
parameter_constraints = (
parameter_constraints if parameter_constraints is not None else []
)
typed_parameters = [cls.parameter_from_json(p) for p in parameters]
is_hss = any(p.is_hierarchical for p in typed_parameters)
search_space_cls = HierarchicalSearchSpace if is_hss else SearchSpace
parameter_map = {p.name: p for p in typed_parameters}
typed_parameter_constraints = [
cls.constraint_from_str(c, parameter_map) for c in parameter_constraints
]
if any(
any(
isinstance(parameter_map[parameter], ChoiceParameter)
for parameter in constraint.constraint_dict
)
for constraint in typed_parameter_constraints
):
raise UnsupportedError(
"Constraints on ChoiceParameters are not allowed. Try absorbing "
"this constraint into the associated range parameter's bounds."
)
if any(
any(
isinstance(parameter_map[parameter], FixedParameter)
for parameter in constraint.constraint_dict
)
for constraint in typed_parameter_constraints
):
raise UnsupportedError(
"Constraints on FixedParameters are not allowed. Try absorbing "
"this constraint into the associated range parameter's bounds."
)
ss = search_space_cls(
parameters=typed_parameters,
parameter_constraints=typed_parameter_constraints,
)
logger.info(f"Created search space: {ss}.")
if is_hss:
hss = checked_cast(HierarchicalSearchSpace, ss)
logger.info(
"Hieararchical structure of the search space: \n"
f"{hss.hierarchical_structure_str(parameter_names_only=True)}"
)
return search_space_cls(
parameters=typed_parameters,
parameter_constraints=typed_parameter_constraints,
)
@classmethod
def _get_default_objectives(cls) -> dict[str, str] | None:
"""Get the default objective and its optimization direction.
The return type is optional since some subclasses may not wish to
use any optimization config by default.
"""
return {DEFAULT_OBJECTIVE_NAME: "maximize"}
[docs]
@classmethod
def make_experiment(
cls,
parameters: list[TParameterRepresentation],
name: str | None = None,
description: str | None = None,
owners: list[str] | None = None,
parameter_constraints: list[str] | None = None,
outcome_constraints: list[str] | None = None,
status_quo: TParameterization | None = None,
experiment_type: str | None = None,
tracking_metric_names: list[str] | None = None,
metric_definitions: dict[str, dict[str, Any]] | None = None,
objectives: dict[str, str] | None = None,
objective_thresholds: list[str] | None = None,
support_intermediate_data: bool = False,
immutable_search_space_and_opt_config: bool = True,
auxiliary_experiments_by_purpose: None
| (dict[AuxiliaryExperimentPurpose, list[AuxiliaryExperiment]]) = None,
default_trial_type: str | None = None,
default_runner: Runner | None = None,
is_test: bool = False,
) -> Experiment:
"""Instantiation wrapper that allows for Ax `Experiment` creation
without importing or instantiating any Ax classes.
Args:
parameters: List of dictionaries representing parameters in the
experiment search space.
Required elements in the dictionaries are:
1. "name" (name of parameter, string),
2. "type" (type of parameter: "range", "fixed", or "choice", string),
and one of the following:
3a. "bounds" for range parameters (list of two values, lower bound
first),
3b. "values" for choice parameters (list of values), or
3c. "value" for fixed parameters (single value).
Optional elements are:
1. "log_scale" (for float-valued range parameters, bool),
2. "value_type" (to specify type that values of this parameter should
take; expects "float", "int", "bool" or "str"),
3. "is_fidelity" (bool) and "target_value" (float) for fidelity
parameters,
4. "is_ordered" (bool) for choice parameters,
5. "is_task" (bool) for task parameters, and
6. "digits" (int) for float-valued range parameters.
name: Name of the experiment to be created.
parameter_constraints: List of string representation of parameter
constraints, such as "x3 >= x4" or "-x3 + 2*x4 - 3.5*x5 >= 2". For
the latter constraints, any number of arguments is accepted, and
acceptable operators are "<=" and ">=".
outcome_constraints: List of string representation of outcome
constraints of form "metric_name >= bound", like "m1 <= 3."
status_quo: Parameterization of the current state of the system.
If set, this will be added to each trial to be evaluated alongside
test configurations.
experiment_type: String indicating type of the experiment (e.g. name of
a product in which it is used), if any.
tracking_metric_names: Names of additional tracking metrics not used for
optimization.
metric_definitions: A mapping of metric names to extra kwargs to pass
to that metric
objectives: Mapping from an objective name to "minimize" or "maximize"
representing the direction for that objective.
objective_thresholds: A list of objective threshold constraints for multi-
objective optimization, in the same string format as
`outcome_constraints` argument.
support_intermediate_data: Whether trials may report metrics results for
incomplete runs.
immutable_search_space_and_opt_config: Whether it's possible to update the
search space and optimization config on this experiment after creation.
Defaults to True. If set to True, we won't store or load copies of the
search space and optimization config on each generator run, which will
improve storage performance.
auxiliary_experiments_by_purpose: Dictionary of auxiliary experiments for
different use cases (e.g., transfer learning).
default_trial_type: The default trial type if multiple
trial types are intended to be used in the experiment. If specified,
a MultiTypeExperiment will be created. Otherwise, a single-type
Experiment will be created.
default_runner: The default runner in this experiment.
This only applies to MultiTypeExperiment (when default_trial_type
is specified).
is_test: Whether this experiment will be a test experiment (useful for
marking test experiments in storage etc). Defaults to False.
"""
if (default_trial_type is None) != (default_runner is None):
raise ValueError(
"Must specify both default_trial_type and default_runner if "
"using a MultiTypeExperiment."
)
status_quo_arm = None if status_quo is None else Arm(parameters=status_quo)
objectives = objectives or cls._get_default_objectives()
if objectives:
optimization_config = cls.make_optimization_config(
objectives=objectives,
objective_thresholds=objective_thresholds or [],
outcome_constraints=outcome_constraints or [],
status_quo_defined=status_quo_arm is not None,
metric_definitions=metric_definitions,
)
else:
optimization_config = None
tracking_metrics = (
None
if tracking_metric_names is None
else [
cls._make_metric(
name=metric_name, metric_definitions=metric_definitions
)
for metric_name in tracking_metric_names
]
)
default_data_type = (
DataType.MAP_DATA if support_intermediate_data else DataType.DATA
)
properties: dict[str, Any] = {}
if immutable_search_space_and_opt_config:
properties[Keys.IMMUTABLE_SEARCH_SPACE_AND_OPT_CONF] = (
immutable_search_space_and_opt_config
)
if owners is not None:
properties["owners"] = owners
if default_trial_type is not None:
return MultiTypeExperiment(
name=none_throws(name),
search_space=cls.make_search_space(parameters, parameter_constraints),
default_trial_type=none_throws(default_trial_type),
default_runner=none_throws(default_runner),
optimization_config=optimization_config,
tracking_metrics=tracking_metrics,
status_quo=status_quo_arm,
description=description,
is_test=is_test,
experiment_type=experiment_type,
properties=properties,
default_data_type=default_data_type,
)
return Experiment(
name=name,
description=description,
search_space=cls.make_search_space(parameters, parameter_constraints),
optimization_config=optimization_config,
status_quo=status_quo_arm,
experiment_type=experiment_type,
tracking_metrics=tracking_metrics,
default_data_type=default_data_type,
properties=properties,
auxiliary_experiments_by_purpose=auxiliary_experiments_by_purpose,
is_test=is_test,
)
[docs]
@classmethod
def build_objective_thresholds(
cls, objectives: dict[str, ObjectiveProperties]
) -> list[str]:
"""Construct a list of constraint string for an objective thresholds
interpretable by `make_experiment()`
Args:
objectives: Mapping of name of the objective to Object containing:
minimize: Whether this experiment represents a minimization problem.
threshold: The bound in the objective's threshold constraint.
"""
return [
cls.build_objective_threshold(objective, properties)
for objective, properties in objectives.items()
if properties.threshold is not None
]
[docs]
@staticmethod
def build_objective_threshold(
objective: str, objective_properties: ObjectiveProperties
) -> str:
"""
Constructs constraint string for an objective threshold interpretable
by `make_experiment()`
Args:
objective: Name of the objective
objective_properties: Object containing:
minimize: Whether this experiment represents a minimization problem.
threshold: The bound in the objective's threshold constraint.
"""
operator = "<=" if objective_properties.minimize else ">="
return f"{objective} {operator} {objective_properties.threshold}"
[docs]
@staticmethod
def make_fixed_observation_features(
fixed_features: FixedFeatures,
) -> ObservationFeatures:
"""Construct ObservationFeatures from FixedFeatures.
Args:
fixed_features: The fixed features for generation.
Returns:
The new ObservationFeatures object.
"""
return ObservationFeatures(
parameters=fixed_features.parameters,
trial_index=(
None
if fixed_features.trial_index is None
else fixed_features.trial_index
),
)
