#!/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 Mapping, Sequence
from copy import deepcopy
from dataclasses import dataclass
from logging import Logger
from typing import Any, cast, Union
from ax.core.arm import Arm
from ax.core.auxiliary import AuxiliaryExperiment, AuxiliaryExperimentPurpose
from ax.core.experiment import 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,
OptimizationConfig,
)
from ax.core.outcome_constraint import ObjectiveThreshold, OutcomeConstraint
from ax.core.parameter import (
ChoiceParameter,
DerivedParameter,
FixedParameter,
Parameter,
PARAMETER_PYTHON_TYPE_MAP,
ParameterType,
RangeParameter,
TParameterType,
)
from ax.core.parameter_constraint import (
ParameterConstraint,
validate_constraint_parameters,
)
from ax.core.runner import Runner
from ax.core.search_space import 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 (
assert_is_instance_of_tuple,
assert_is_instance_optional,
)
from pyre_extensions import assert_is_instance, none_throws
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", "derived"]
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",
"expression_str",
}
COMPARISON_OPS_WITH_EQ: set[str] = {"<=", ">=", "=="}
INVALID_CONSTRAINT_ERROR_MSG = (
"Received invalid parameter constraint format: `{}`. "
"Please use one of the following forms:\n"
"* Ordered constraints: `<p1> >= <p2>` or `<p1> <= <p2>`, where `<p1>` and `<p2>` "
"are parameter names.\n"
"* Unweighted linear constraints: `<p1> >= <b>` or `<p1> + <p2> <= <b>`, where you "
"can add one or more parameters on the left side, and `<b>` is a numeric value.\n"
"* Weighted linear constraints: `<w1>*<p1> >= <b>` or "
"`<w1>*<p1> + <w2>*<p2> <= <b>`, where you can add one or more weighted terms on "
"the left side, and there should be no spaces between weights and parameter "
"names.\n"
"* Equality constraints: `<p1> + <p2> == <b>` or `<w1>*<p1> + <w2>*<p2> == <b>`, "
"same as linear constraints but using `==` instead of `<=` or `>=`.\n"
'Acceptable comparison operators are ">=", "<=", and "==".'
)
[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,
) -> tuple[type[Metric], 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_class, 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:
metric_definitions = metric_definitions or {}
metric_class, 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:
inferred_type = type(none_throws(vals[0]))
parameter_type = cls._get_parameter_type(
cast(TParameterType, inferred_type)
)
assert all(isinstance(x, inferred_type) 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(cast(TParameterType, PARAM_TYPES[typ]))
@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=assert_is_instance_of_tuple(bounds[0], (float, int)),
upper=assert_is_instance_of_tuple(bounds[1], (float, int)),
log_scale=assert_is_instance(representation.get("log_scale", False), bool),
digits=assert_is_instance_optional(representation.get("digits", None), int),
is_fidelity=assert_is_instance(
representation.get("is_fidelity", False), bool
),
target_value=cast(TParamValue, 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=assert_is_instance_optional(
representation.get("is_ordered"), bool
),
is_fidelity=assert_is_instance(
representation.get("is_fidelity", False), bool
),
is_task=assert_is_instance(representation.get("is_task", False), bool),
target_value=cast(TParamValue, representation.get("target_value", None)),
sort_values=assert_is_instance_optional(
representation.get("sort_values", None), bool
),
dependents=assert_is_instance_optional(
representation.get("dependents", None), dict
),
)
@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(cast(TParameterType, type(value)))
if parameter_type is None
else cls._get_parameter_type(
cast(TParameterType, PARAM_TYPES[parameter_type])
)
),
value=cast(TParamValue, value),
is_fidelity=assert_is_instance(
representation.get("is_fidelity", False), bool
),
target_value=cast(TParamValue, representation.get("target_value", None)),
dependents=cast(
dict[TParamValue, list[str]] | None,
representation.get("dependents", None),
),
)
@classmethod
def _make_derived_param(
cls,
name: str,
representation: TParameterRepresentation,
parameter_type: str | None,
) -> DerivedParameter:
assert "expression_str" in representation, (
"expression_str is required for derived parameters."
)
msg = "parameter_type is required for derived parameters."
return DerivedParameter(
name=name,
parameter_type=cls._get_parameter_type(
cast(TParameterType, PARAM_TYPES[none_throws(parameter_type, msg)])
),
expression_str=assert_is_instance(representation["expression_str"], str),
)
[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`, `fixed` or `derived`."
)
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 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,
)
if parameter_class == "derived":
return cls._make_derived_param(
name=name,
representation=representation,
parameter_type=parameter_type,
)
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.
Args:
representation: String representation of the parameter constraint.
parameters: Dictionary of parameter names to parameter objects or configs.
Returns:
An instantiated ParameterConstraint, either an OrderConstraint or a
ParameterConstraint, representing a linear constraint.
"""
for param_name in parameters:
if " " in param_name:
raise ValueError(
"Parameter names cannot contain spaces when used in "
f"constraint strings. Got {param_name!r}."
)
tokens = representation.split()
try:
float(tokens[-1])
last_token_is_numeric = True
except ValueError:
last_token_is_numeric = False
# Identify the comparison operator (second-to-last for linear, middle
# for order constraints).
is_order_constraint = (
len(tokens) == 3
and tokens[1] in COMPARISON_OPS_WITH_EQ
and not last_token_is_numeric
)
is_linear_constraint = (
# if len == 3, then this is a single parameter bound constraint,
# otherwise it corresponds to a numerical bound on a sum of
# parameters
len(tokens) >= 3
and len(tokens) % 2 == 1
and tokens[-2] in COMPARISON_OPS_WITH_EQ
and last_token_is_numeric
)
if is_order_constraint: # e.g. "x1 >= x2"
if tokens[1] == "==":
raise ValueError(
"Equality order constraints (e.g. 'x1 == x2') are not "
"supported. Use a DerivedParameter to express that two "
"parameters must be equal."
)
return _process_order_constraint(
tokens=tokens,
parameters=parameters,
)
if is_linear_constraint: # e.g. "x1 + x2 >= 3" or "x1 + x2 == 3"
if tokens[-2] == "==":
return _process_equality_constraint(
tokens=tokens,
parameters=parameters,
)
return _process_linear_constraint(
tokens=tokens,
parameters=parameters,
)
raise ValueError(INVALID_CONSTRAINT_ERROR_MSG.format(representation))
[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()
if len(tokens) != 3 or tokens[1] not in COMPARISON_OPS:
raise ValueError(
f"Outcome constraint '{representation}' should be of "
"form `metric_name >= x`, where x is a float bound and "
"comparison operator is >= or <=. Note that metric names "
"cannot contain spaces in constraint strings."
)
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(
metric=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
)
# Create an ObjectiveThreshold that shares the same expression string
# as the OutcomeConstraint, bypassing the deprecated __init__.
ot = ObjectiveThreshold.__new__(ObjectiveThreshold)
ot._expression_str = oc._expression_str
ot._metric_name_to_signature = oc._metric_name_to_signature
return ot
[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[OutcomeConstraint]:
# pyre-ignore[9]: ObjectiveThreshold is a subclass of OutcomeConstraint;
# list invariance prevents direct assignment.
typed_objective_thresholds: list[OutcomeConstraint] = (
[
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[OutcomeConstraint],
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.warning(
"Objective thresholds were not set. They will be selected using a "
"heuristic, but should be specified on the objective for best "
"performance."
)
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]
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], 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 = SearchSpace(
parameters=typed_parameters,
parameter_constraints=typed_parameter_constraints,
)
logger.debug(f"Created search space: {ss}.")
if ss.is_hierarchical:
logger.debug(
"Hierarchical structure of the search space: \n"
f"{ss.hierarchical_structure_str(parameter_names_only=True)}"
)
return SearchSpace(
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 {Keys.DEFAULT_OBJECTIVE_NAME.value: "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", "choice", or
"derived", 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).
3d. "expression_str" for derived parameters (string containing
the definition of the derived parameter).
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
]
)
# Also create properly-typed metrics for optimization config metric
# names so they are registered with proper types and properties
# before the auto-registration in Experiment.__init__ fires.
if optimization_config is not None:
# Build lower_is_better map from the optimization config so
# that pre-registered metrics carry the correct directionality.
lower_is_better_map: dict[str, bool] = {}
obj = optimization_config.objective
obj_names = obj.metric_names
obj_weights = [w for _, w in obj.metric_weights]
for mn, weight in zip(obj_names, obj_weights):
lower_is_better_map[mn] = weight < 0
for constraint in optimization_config.outcome_constraints:
for mn in constraint.metric_names:
if mn not in lower_is_better_map:
lower_is_better_map[mn] = constraint.op is ComparisonOp.LEQ
tracking_names = {m.name for m in (tracking_metrics or [])}
for metric_name in optimization_config.metric_names:
if metric_name not in tracking_names:
opt_metric = cls._make_metric(
name=metric_name,
metric_definitions=metric_definitions,
lower_is_better=lower_is_better_map.get(metric_name),
)
if tracking_metrics is None:
tracking_metrics = []
tracking_metrics.append(opt_metric)
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=parameters, parameter_constraints=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,
)
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,
properties=properties,
auxiliary_experiments_by_purpose=auxiliary_experiments_by_purpose,
is_test=is_test,
runner=default_runner,
)
[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
),
)
# Helpers for parsing parameter constraints
def _process_order_constraint(
tokens: Sequence[str],
parameters: Mapping[str, Parameter],
) -> ParameterConstraint:
"""Processes an order constraint, e.g. "x1 <= x2".
Args:
tokens: A list of tokens in the constraint string.
parameters: A mapping from parameter names to their definitions.
Returns:
An OrderConstraint object representing the order constraint.
"""
left, right = tokens[0], tokens[2]
parameter_names = parameters.keys()
if missing_tokens := ({left, right} - set(parameter_names)):
missing_tokens = list(missing_tokens)
missing_tokens.sort() # making error message deterministic
raise ValueError(
f"OrderConstraint token(s) {missing_tokens} are not present in "
f"parameters list {list(parameter_names)}."
)
validate_constraint_parameters(parameters=[parameters[left], parameters[right]])
# tokens[1] is checked to be either LEQ or GEQ above if order_const is True
if COMPARISON_OPS[tokens[1]] is ComparisonOp.LEQ:
lower_parameter = parameters[left]
upper_parameter = parameters[right]
else:
lower_parameter = parameters[right]
upper_parameter = parameters[left]
return ParameterConstraint(
inequality=f"{lower_parameter.name} <= {upper_parameter.name}",
)
def _parse_linear_constraint_tokens(
tokens: Sequence[str],
parameters: Mapping[str, Parameter],
operator_str: str,
) -> tuple[dict[str, float], float]:
"""Parse tokens of a linear constraint into parameter weights and bound.
Shared helper for ``_process_linear_constraint`` and
``_process_equality_constraint``. Validates ``*`` placement, processes
alternating monomials / operators, and returns the raw
``parameter_weights`` dict and ``bound``.
Args:
tokens: A list of tokens in the constraint string.
parameters: A mapping from parameter names to their definitions.
operator_str: The comparison operator string (e.g. ``"<="``/``">="``
/``"=="``), used only for error messages.
Returns:
A tuple of (parameter_weights, bound).
"""
parameter_names = parameters.keys()
bound = float(tokens[-1])
if any(token[0] == "*" or token[-1] == "*" for token in tokens):
raise ValueError(
f"A linear constraint should be the form "
f"a*x + b*y - c*z {operator_str} 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 `*`, and there should be a "
f"single space around each operator +, -, "
f"and {operator_str}."
)
parameter_weights: dict[str, float] = {}
current_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:
multiplier, parameter_name = _process_monomial(monomial_str=token)
if parameter_name not in parameter_names:
raise ValueError(
f"Constraint parameter '{parameter_name}' is not present in this "
f"experiment's search space parameters: {list(parameter_names)}."
)
validate_constraint_parameters(parameters=[parameters[parameter_name]])
parameter_weights[parameter_name] = current_sign * multiplier
# for operators
else:
if token == "+":
current_sign = 1.0
elif token == "-":
current_sign = -1.0
else:
raise ValueError(
f"Expected a mixed constraint, found operator `{token}`."
)
return parameter_weights, bound
def _process_linear_constraint(
tokens: Sequence[str],
parameters: Mapping[str, Parameter],
) -> ParameterConstraint:
"""Processes a linear constraint, e.g. "x1 + x2 <= 3". The last token is expected
to be a numeric constant, and the other tokens are expected to be parameters, their
multiplicative coefficients (e.g."2.5*x1") and "+" or "-" operators (e.g. "+").
Args:
tokens: A list of tokens in the constraint string.
parameters: A mapping from parameter names to their definitions.
Returns:
A ParameterConstraint object representing the linear constraint.
"""
parameter_weights, bound = _parse_linear_constraint_tokens(
tokens=tokens, parameters=parameters, operator_str="<= or >="
)
# tokens[-2] is checked to be either LEQ or GEQ if sum_const is True
comparison_multiplier = (
1.0 if COMPARISON_OPS[tokens[-2]] is ComparisonOp.LEQ else -1.0
)
constraint_dict = {
p: comparison_multiplier * parameter_weights[p] for p in parameter_weights
}
expr = " + ".join(f"{coeff} * {param}" for param, coeff in constraint_dict.items())
return ParameterConstraint(
inequality=f"{expr} <= {comparison_multiplier * bound}",
)
def _process_equality_constraint(
tokens: Sequence[str],
parameters: Mapping[str, Parameter],
) -> ParameterConstraint:
"""Processes a linear equality constraint, e.g. "x1 + x2 == 3".
The last token is expected to be a numeric constant, the second-to-last
is ``"=="``, and the other tokens are parameters, their multiplicative
coefficients (e.g. ``"2.5*x1"``) and ``"+"`` or ``"-"`` operators.
Args:
tokens: A list of tokens in the constraint string.
parameters: A mapping from parameter names to their definitions.
Returns:
A ParameterConstraint with ``equality=...``.
"""
parameter_weights, bound = _parse_linear_constraint_tokens(
tokens=tokens, parameters=parameters, operator_str="=="
)
# Reject equality constraints that equate two parameters
# (e.g. "x1 - x2 == 0"). DerivedParameter is the correct tool.
if (
bound == 0.0
and len(parameter_weights) == 2
and set(parameter_weights.values()) == {1.0, -1.0}
):
params = list(parameter_weights.keys())
raise ValueError(
f"Equality constraint '{' '.join(tokens)}' is equivalent to "
f"'{params[0]} == {params[1]}'. Use a DerivedParameter to "
"express that two parameters must be equal."
)
expr = " + ".join(
f"{coeff} * {param}" for param, coeff in parameter_weights.items()
)
return ParameterConstraint(equality=f"{expr} == {bound}")
def _process_monomial(monomial_str: str) -> tuple[float, str]:
"""Process a monomial in a linear constraint.
Args:
monomial_str: A string representation of a monomial in a linear constraint.
Returns:
A tuple of the multiplier and the parameter name.
"""
split_token = monomial_str.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]} for "
f"parameter {split_token[1]}."
)
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
else:
raise ValueError(
"Monomial format does not match `multiplier*parameter_name`. "
f"Got `{monomial_str}`."
)
return multiplier, parameter