Source code for ax.models.torch.botorch_modular.model

#!/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 dataclasses
import warnings
from collections import OrderedDict
from collections.abc import Mapping, Sequence
from typing import Any

import numpy.typing as npt
import torch
from ax.core.search_space import SearchSpaceDigest
from ax.core.types import TCandidateMetadata, TGenMetadata
from ax.exceptions.core import UnsupportedError, UserInputError
from ax.models.torch.botorch import (
    get_feature_importances_from_botorch_model,
    get_rounding_func,
)
from ax.models.torch.botorch_modular.acquisition import Acquisition
from ax.models.torch.botorch_modular.surrogate import Surrogate, SurrogateSpec
from ax.models.torch.botorch_modular.utils import (
    check_outcome_dataset_match,
    choose_botorch_acqf_class,
    construct_acquisition_and_optimizer_options,
    ModelConfig,
)
from ax.models.torch.utils import _to_inequality_constraints
from ax.models.torch_base import TorchGenResults, TorchModel, TorchOptConfig
from ax.utils.common.base import Base
from ax.utils.common.constants import Keys
from ax.utils.common.docutils import copy_doc
from ax.utils.common.typeutils import checked_cast
from botorch.acquisition.acquisition import AcquisitionFunction
from botorch.models.deterministic import FixedSingleSampleModel
from botorch.utils.datasets import SupervisedDataset
from torch import Tensor


[docs] class BoTorchModel(TorchModel, Base): """**All classes in 'botorch_modular' directory are under construction, incomplete, and should be treated as alpha versions only.** Modular ``Model`` class for combining BoTorch subcomponents in Ax. Specified via ``Surrogate`` and ``Acquisition``, which wrap BoTorch ``Model`` and ``AcquisitionFunction``, respectively, for convenient use in Ax. Args: acquisition_class: Type of ``Acquisition`` to be used in this model, auto-selected based on experiment and data if not specified. acquisition_options: Optional dict of kwargs, passed to the constructor of BoTorch ``AcquisitionFunction``. botorch_acqf_class: Type of ``AcquisitionFunction`` to be used in this model, auto-selected based on experiment and data if not specified. surrogate_spec: An optional ``SurrogateSpec`` object specifying how to construct the ``Surrogate`` and the underlying BoTorch ``Model``. surrogate_specs: DEPRECATED. Please use ``surrogate_spec`` instead. surrogate: In lieu of ``SurrogateSpec``, an instance of ``Surrogate`` may be provided. In most cases, ``surrogate_spec`` should be used instead. refit_on_cv: Whether to reoptimize model parameters during call to ``BoTorchmodel.cross_validate``. warm_start_refit: Whether to load parameters from either the provided state dict or the state dict of the current BoTorch ``Model`` during refitting. If False, model parameters will be reoptimized from scratch on refit. NOTE: This setting is ignored during ``cross_validate`` if ``refit_on_cv`` is False. """ acquisition_class: type[Acquisition] acquisition_options: dict[str, Any] surrogate_spec: SurrogateSpec | None _surrogate: Surrogate | None _botorch_acqf_class: type[AcquisitionFunction] | None _search_space_digest: SearchSpaceDigest | None = None _supports_robust_optimization: bool = True def __init__( self, surrogate_spec: SurrogateSpec | None = None, surrogate_specs: Mapping[str, SurrogateSpec] | None = None, surrogate: Surrogate | None = None, acquisition_class: type[Acquisition] | None = None, acquisition_options: dict[str, Any] | None = None, botorch_acqf_class: type[AcquisitionFunction] | None = None, refit_on_cv: bool = False, warm_start_refit: bool = True, ) -> None: # Check that only one surrogate related option is provided. if bool(surrogate_spec) + bool(surrogate_specs) + bool(surrogate) > 1: raise UserInputError( "Only one of `surrogate_spec`, `surrogate_specs`, and `surrogate` " "can be specified. Please use `surrogate_spec`." ) if surrogate_specs is not None: if len(surrogate_specs) > 1: raise DeprecationWarning( "Support for multiple `Surrogate`s has been deprecated. " "Please use the `surrogate_spec` input in the future to " "specify a single `Surrogate`." ) warnings.warn( "The `surrogate_specs` argument is deprecated in favor of " "`surrogate_spec`, which accepts a single `SurrogateSpec` object. " "Please use `surrogate_spec` in the future.", DeprecationWarning, stacklevel=2, ) surrogate_spec = next(iter(surrogate_specs.values())) self.surrogate_spec = surrogate_spec self._surrogate = surrogate self.acquisition_class = acquisition_class or Acquisition self.acquisition_options = acquisition_options or {} self._botorch_acqf_class = botorch_acqf_class self.refit_on_cv = refit_on_cv self.warm_start_refit = warm_start_refit @property def surrogate(self) -> Surrogate: """Returns the ``Surrogate``, if it has been constructed.""" if self._surrogate is None: raise ValueError("Surrogate has not yet been constructed.") return self._surrogate @property def Xs(self) -> list[Tensor]: """A list of tensors, each of shape ``batch_shape x n_i x d``, where `n_i` is the number of training inputs for the i-th model. NOTE: This is an accessor for ``self.surrogate.Xs`` and returns it unchanged. """ return self.surrogate.Xs @property def botorch_acqf_class(self) -> type[AcquisitionFunction]: """BoTorch ``AcquisitionFunction`` class, associated with this model. Raises an error if one is not yet set. """ if not self._botorch_acqf_class: raise ValueError("BoTorch `AcquisitionFunction` has not yet been set.") return self._botorch_acqf_class
[docs] def fit( self, datasets: Sequence[SupervisedDataset], search_space_digest: SearchSpaceDigest, candidate_metadata: list[list[TCandidateMetadata]] | None = None, state_dict: OrderedDict[str, Tensor] | None = None, refit: bool = True, **additional_model_inputs: Any, ) -> None: """Fit model to m outcomes. Args: datasets: A list of ``SupervisedDataset`` containers, each corresponding to the data of one or more outcomes. search_space_digest: A ``SearchSpaceDigest`` object containing metadata on the features in the datasets. candidate_metadata: Model-produced metadata for candidates, in the order corresponding to the Xs. state_dict: An optional model statedict for the underlying ``Surrogate``. Primarily used in ``BoTorchModel.cross_validate``. refit: Whether to re-optimize model parameters. additional_model_inputs: Additional kwargs to pass to the model input constructor in ``Surrogate.fit``. """ outcome_names = sum((ds.outcome_names for ds in datasets), []) check_outcome_dataset_match( outcome_names=outcome_names, datasets=datasets, exact_match=True ) # Checks for duplicate outcome names # Store search space info for later use (e.g. during generation) self._search_space_digest = search_space_digest # If a surrogate has not been constructed, construct it. if self._surrogate is None: surrogate_spec = ( SurrogateSpec(model_configs=[ModelConfig(name="default")]) if self.surrogate_spec is None else self.surrogate_spec ) self._surrogate = Surrogate( surrogate_spec=surrogate_spec, refit_on_cv=self.refit_on_cv ) # Fit the surrogate. for config in self.surrogate.surrogate_spec.model_configs: config.model_options.update(additional_model_inputs) for ( config_list ) in self.surrogate.surrogate_spec.metric_to_model_configs.values(): for config in config_list: config.model_options.update(additional_model_inputs) self.surrogate.fit( datasets=datasets, search_space_digest=search_space_digest, candidate_metadata=candidate_metadata, state_dict=state_dict, refit=refit, )
[docs] def predict( self, X: Tensor, use_posterior_predictive: bool = False ) -> tuple[Tensor, Tensor]: """Predicts, potentially from multiple surrogates. Args: X: (n x d) Tensor of input locations. use_posterior_predictive: A boolean indicating if the predictions should be from the posterior predictive (i.e. including observation noise). Returns: Tuple of tensors: (n x m) mean, (n x m x m) covariance. """ return self.surrogate.predict( X=X, use_posterior_predictive=use_posterior_predictive )
[docs] @copy_doc(TorchModel.gen) def gen( self, n: int, search_space_digest: SearchSpaceDigest, torch_opt_config: TorchOptConfig, ) -> TorchGenResults: acq_options, opt_options = construct_acquisition_and_optimizer_options( acqf_options=self.acquisition_options, model_gen_options=torch_opt_config.model_gen_options, ) # update bounds / target values search_space_digest = dataclasses.replace( self.search_space_digest, bounds=search_space_digest.bounds, target_values=search_space_digest.target_values or {}, ) acqf = self._instantiate_acquisition( search_space_digest=search_space_digest, torch_opt_config=torch_opt_config, acq_options=acq_options, ) botorch_rounding_func = get_rounding_func(torch_opt_config.rounding_func) candidates, expected_acquisition_value, weights = acqf.optimize( n=n, search_space_digest=search_space_digest, inequality_constraints=_to_inequality_constraints( linear_constraints=torch_opt_config.linear_constraints ), fixed_features=torch_opt_config.fixed_features, rounding_func=botorch_rounding_func, optimizer_options=checked_cast(dict, opt_options), ) gen_metadata = self._get_gen_metadata_from_acqf( acqf=acqf, torch_opt_config=torch_opt_config, expected_acquisition_value=expected_acquisition_value, ) # log what model was used metric_to_model_config_name = { metric_name: model_config.name or str(model_config) for metric_name_tuple, model_config in ( self.surrogate.metric_to_best_model_config.items() ) for metric_name in metric_name_tuple } gen_metadata["metric_to_model_config_name"] = metric_to_model_config_name return TorchGenResults( points=candidates.detach().cpu(), weights=weights, gen_metadata=gen_metadata, )
def _get_gen_metadata_from_acqf( self, acqf: Acquisition, torch_opt_config: TorchOptConfig, expected_acquisition_value: Tensor, ) -> TGenMetadata: gen_metadata: TGenMetadata = { Keys.EXPECTED_ACQF_VAL: expected_acquisition_value.tolist() } if torch_opt_config.objective_weights.nonzero().numel() > 1: gen_metadata["objective_thresholds"] = acqf.objective_thresholds gen_metadata["objective_weights"] = acqf.objective_weights if hasattr(acqf.acqf, "outcome_model"): outcome_model = acqf.acqf.outcome_model if isinstance( outcome_model, FixedSingleSampleModel, ): gen_metadata["outcome_model_fixed_draw_weights"] = outcome_model.w return gen_metadata
[docs] @copy_doc(TorchModel.best_point) def best_point( self, search_space_digest: SearchSpaceDigest, torch_opt_config: TorchOptConfig, ) -> Tensor | None: try: return self.surrogate.best_in_sample_point( search_space_digest=search_space_digest, torch_opt_config=torch_opt_config, )[0] except ValueError: return None
[docs] @copy_doc(TorchModel.evaluate_acquisition_function) def evaluate_acquisition_function( self, X: Tensor, search_space_digest: SearchSpaceDigest, torch_opt_config: TorchOptConfig, acq_options: dict[str, Any] | None = None, ) -> Tensor: acqf = self._instantiate_acquisition( search_space_digest=search_space_digest, torch_opt_config=torch_opt_config, acq_options=acq_options, ) return acqf.evaluate(X=X)
[docs] @copy_doc(TorchModel.cross_validate) def cross_validate( self, datasets: Sequence[SupervisedDataset], X_test: Tensor, search_space_digest: SearchSpaceDigest, use_posterior_predictive: bool = False, **additional_model_inputs: Any, ) -> tuple[Tensor, Tensor]: current_surrogate = self.surrogate # If we should be refitting but not warm-starting the refit, set # `state_dict` to None to avoid loading it. state_dict = ( None if self.refit_on_cv and not self.warm_start_refit else current_surrogate.model.state_dict() ) # Temporarily set `_surrogate` to cloned surrogate to set # the training data on cloned surrogate to train set and # use it to predict the test point. self._surrogate = current_surrogate.clone_reset() # Remove the `robust_digest` since we do not want to use perturbations here. search_space_digest = dataclasses.replace( search_space_digest, robust_digest=None, ) try: self.fit( datasets=datasets, search_space_digest=search_space_digest, # pyre-fixme [6]: state_dict() has a generic dict[str, Any] return type # but it is actually an OrderedDict[str, Tensor]. state_dict=state_dict, refit=self.refit_on_cv, **additional_model_inputs, ) X_test_prediction = self.predict( X=X_test, use_posterior_predictive=use_posterior_predictive, ) finally: # Reset the surrogates back to this model's surrogate, make # sure the cloned surrogate doesn't stay around if fit or # predict fail. self._surrogate = current_surrogate return X_test_prediction
@property def dtype(self) -> torch.dtype: """Torch data type of the tensors in the training data used in the model, of which this ``Acquisition`` is a subcomponent. """ return self.surrogate.dtype @property def device(self) -> torch.device: """Torch device type of the tensors in the training data used in the model, of which this ``Acquisition`` is a subcomponent. """ return self.surrogate.device def _instantiate_acquisition( self, search_space_digest: SearchSpaceDigest, torch_opt_config: TorchOptConfig, acq_options: dict[str, Any] | None = None, ) -> Acquisition: """Set a BoTorch acquisition function class for this model if needed and instantiate it. Returns: A BoTorch ``AcquisitionFunction`` instance. """ if not self._botorch_acqf_class: if torch_opt_config.risk_measure is not None: raise UnsupportedError( "Automated selection of `botorch_acqf_class` is not supported " "for robust optimization with risk measures. Please specify " "`botorch_acqf_class` as part of `model_kwargs`." ) self._botorch_acqf_class = choose_botorch_acqf_class( torch_opt_config=torch_opt_config ) return self.acquisition_class( surrogate=self.surrogate, botorch_acqf_class=self.botorch_acqf_class, search_space_digest=search_space_digest, torch_opt_config=torch_opt_config, options=acq_options, )
[docs] def feature_importances(self) -> npt.NDArray: """Compute feature importances from the model. This assumes that we can get model lengthscales from either ``covar_module.base_kernel.lengthscale`` or ``covar_module.lengthscale``. Returns: The feature importances as a numpy array of size len(metrics) x 1 x dim where each row sums to 1. """ return get_feature_importances_from_botorch_model(model=self.surrogate.model)
@property def search_space_digest(self) -> SearchSpaceDigest: if self._search_space_digest is None: raise RuntimeError( "`search_space_digest` is not initialized. Must `fit` the model first." ) return self._search_space_digest @search_space_digest.setter def search_space_digest(self, value: SearchSpaceDigest) -> None: raise RuntimeError("Setting search_space_digest manually is disallowed.")