BoTorch Base Optimizer

This optimizer is a basic Gaussian Process based Bayesian optimization implementation leveraging BoTorch in a way that is compatible with the blackboxopt interface. While this is a functional optimizer, it is more intended as a basis for other BoTorch based optimizer implementations.

Reference

SingleObjectiveBOTorchOptimizer (SingleObjectiveOptimizer)

Source code in blackboxopt/optimizers/botorch_base.py

class SingleObjectiveBOTorchOptimizer(SingleObjectiveOptimizer):
    def __init__(
        self,
        search_space: ps.ParameterSpace,
        objective: Objective,
        model: Model,
        acquisition_function_factory: Callable[[Model], AcquisitionFunction],
        af_optimizer_kwargs=None,
        num_initial_random_samples: int = 1,
        max_pending_evaluations: Optional[int] = 1,
        batch_shape: torch.Size = torch.Size(),
        logger: Optional[logging.Logger] = None,
        seed: Optional[int] = None,
        torch_dtype: torch.dtype = torch.float64,
    ):
        """Single objective BO optimizer that uses as a surrogate model the `model`
        object provided by user.

        The `model` is expected to be extended from BoTorch base model `Model` class,
        and does not require to be a GP model.

        Args:
            search_space: The space in which to optimize.
            objective: The objective to optimize.
            model: Surrogate model of `Model` type.
            acquisition_function_factory: Callable that produces an acquisition function
                instance, could also be a compatible acquisition function class.
                Only acquisition functions to be minimized are supported.
                Providing a partially initialized class is possible with, e.g.
                `functools.partial(UpperConfidenceBound, beta=6.0, maximize=False)`.
            af_optimizer_kwargs: Settings for acquisition function optimizer,
                see `botorch.optim.optimize_acqf` and in case the whole search space
                is discrete: `botorch.optim.optimize_acqf_discrete`. The former can be
                enforced by providing `raw_samples` or `num_restarts`, the latter by
                providing `num_random_choices`.
            num_initial_random_samples: Size of the initial space-filling design that
                is used before starting BO. The points are sampled randomly in the
                search space. If no random sampling is required, set it to 0.
                When random sampling is enabled, but evaluations with missing objective
                values are reported, more specifications are sampled until
                `num_initial_random_samples` many valid evaluations were reported.
            max_pending_evaluations: Maximum number of parallel evaluations. For
                sequential BO use the default value of 1. If no limit is required,
                set it to None.
            batch_shape: Batch dimension(s) used for batched models.
            logger: Custom logger.
            seed: A seed to make the optimization reproducible.
            torch_dtype: Torch data type used for storing the data. This needs to match
                the dtype of the model used
        """
        super().__init__(search_space=search_space, objective=objective, seed=seed)
        self.num_initial_random = num_initial_random_samples
        self.max_pending_evaluations = max_pending_evaluations
        self.batch_shape = batch_shape
        self.logger = logger or logging.getLogger("blackboxopt")

        self.torch_dtype = torch_dtype
        self.X = torch.empty(
            (*self.batch_shape, 0, len(search_space)), dtype=torch_dtype
        )
        self.losses = torch.empty((*self.batch_shape, 0, 1), dtype=torch_dtype)
        self.pending_specifications: Dict[int, EvaluationSpecification] = {}
        if seed is not None:
            torch.manual_seed(seed=seed)

        self.model = model
        self.acquisition_function_factory = acquisition_function_factory
        self.af_optimizer_kwargs = af_optimizer_kwargs

    def _create_fantasy_model(self, model: Model) -> Model:
        """Create model with the pending specifications and model based
        outcomes added to the training data."""

        if not self.pending_specifications:
            # nothing to do when there are no pending specs
            return model

        pending_X = torch.tensor(
            np.array(
                [
                    self.search_space.to_numerical(e.configuration)
                    for e in self.pending_specifications.values()
                ]
            ),
            dtype=self.torch_dtype,
        )

        model = model.fantasize(pending_X, IIDNormalSampler(1), observation_noise=False)

        if isinstance(model, ExactGP):
            # ExactGP.fantasize extends model's X and Y with batch_size, even if
            # originally not given -> need to reshape these to their original
            # representation
            n_samples = model.train_targets.size(-1)
            n_features = len(self.search_space)
            model.train_inputs[0] = model.train_inputs[0].reshape(
                torch.Size((*self.batch_shape, n_samples, n_features))
            )
            model.train_targets = model.train_targets.reshape(
                torch.Size((*self.batch_shape, n_samples, 1))
            )
        return model

    def _generate_evaluation_specification(self):
        """Optimize acquisition on fantasy model to pick next point."""
        fantasy_model = self._create_fantasy_model(self.model)
        fantasy_model.eval()

        af = self.acquisition_function_factory(fantasy_model)
        if getattr(af, "maximize", False):
            raise ValueError(
                "Only acquisition functions that need to be minimized are supported. "
                f"The given {af.__class__.__name__} has maximize=True. "
                "One potential fix is using functools.partial("
                f"{af.__class__.__name__}, maximize=False) as the "
                "acquisition_function_factory init argument."
            )

        acquisition_function_optimizer = _acquisition_function_optimizer_factory(
            search_space=self.search_space,
            af_opt_kwargs=self.af_optimizer_kwargs,
            torch_dtype=self.torch_dtype,
        )
        configuration, _ = acquisition_function_optimizer(af)

        return EvaluationSpecification(
            configuration=self.search_space.from_numerical(configuration[0]),
        )

    def generate_evaluation_specification(self) -> EvaluationSpecification:
        """Call the optimizer specific function and append a unique integer id
        to the specification.

        Please refer to the docstring of
        `blackboxopt.base.SingleObjectiveOptimizer.generate_evaluation_specification`
        for a description of the method.
        """
        if (
            self.max_pending_evaluations
            and len(self.pending_specifications) == self.max_pending_evaluations
        ):
            raise OptimizerNotReady

        # Generate random samples until there are enough samples where at least one of
        # the objective values is available
        if self.num_initial_random > 0 and (
            sum(~torch.any(self.losses.isnan(), dim=1)) < self.num_initial_random
        ):
            eval_spec = EvaluationSpecification(
                configuration=self.search_space.sample(),
                optimizer_info={"model_based_pick": False},
            )
        else:
            eval_spec = self._generate_evaluation_specification()
            eval_spec.optimizer_info["model_based_pick"] = True

        eval_id = self.X.size(-2) + len(self.pending_specifications)
        eval_spec.optimizer_info["evaluation_id"] = eval_id
        self.pending_specifications[eval_id] = eval_spec
        return eval_spec

    def _remove_pending_specifications(
        self, evaluations: Union[Evaluation, Iterable[Evaluation]]
    ):
        """Find and remove the corresponding entries in `self.pending_specifications`.

        Args:
            evaluations: List of completed evaluations.
        Raises:
            ValueError: If an evaluation is reported with an ID that was not issued
            by the optimizer, the method will fail.
        """
        _evals = [evaluations] if isinstance(evaluations, Evaluation) else evaluations

        for e in _evals:
            if "evaluation_id" not in e.optimizer_info:
                self.logger.debug("User provided EvaluationSpecification received.")
                continue

            if e.optimizer_info["evaluation_id"] not in self.pending_specifications:
                msg = (
                    "Unknown evaluation_id reported. This could indicate that the "
                    "evaluation has been reported before!"
                )
                self.logger.error(msg)
                raise ValueError(msg)

            del self.pending_specifications[e.optimizer_info["evaluation_id"]]

    def _append_evaluations_to_data(
        self, evaluations: Union[Evaluation, Iterable[Evaluation]]
    ):
        """Convert the reported evaluation into its numerical representation
        and append it to the training data.

        Args:
            evaluations: List of completed evaluations.
        """
        _evals = [evaluations] if isinstance(evaluations, Evaluation) else evaluations

        X, Y = to_numerical(
            _evals,
            self.search_space,
            objectives=[self.objective],
            batch_shape=self.batch_shape,
            torch_dtype=self.torch_dtype,
        )

        # fill in NaNs originating from inactive parameters (conditional spaces support)
        # botorch expect numerical representation of inputs to be within the unit
        # hypercube, thus we can't use the default c=-1.0
        X = impute_nans_with_constant(X, c=0.0)

        self.logger.debug(f"Next training configuration(s):{X}, {Y}")

        self.X = torch.cat([self.X, X], dim=-2)
        self.losses = torch.cat([self.losses, Y], dim=-2)

    def _update_internal_evaluation_data(
        self, evaluations: Iterable[Evaluation]
    ) -> None:
        """Check validity of the evaluations and do optimizer agnostic bookkeeping."""
        call_functions_with_evaluations_and_collect_errors(
            [
                functools.partial(validate_objectives, objectives=[self.objective]),
                self._remove_pending_specifications,
                self._append_evaluations_to_data,
            ],
            sort_evaluations(evaluations),
        )

    def report(self, evaluations: Union[Evaluation, Iterable[Evaluation]]) -> None:
        """A simple report method that conditions the model on data.
        This likely needs to be overridden for more specific BO implementations.
        """
        _evals = [evaluations] if isinstance(evaluations, Evaluation) else evaluations
        self._update_internal_evaluation_data(_evals)
        # Just for populating all relevant caches
        self.model.posterior(self.X)

        x_filtered, y_filtered = filter_y_nans(self.X, self.losses)

        # The actual model update
        # Ignore BotorchTensorDimensionWarning which is always reported to make the user
        # aware that they are reponsible for the right input Tensors dimensionality.
        with warnings.catch_warnings():
            warnings.simplefilter(
                action="ignore", category=BotorchTensorDimensionWarning
            )
            self.model = self.model.condition_on_observations(x_filtered, y_filtered)

    def predict_model_based_best(self) -> Optional[Evaluation]:
        """Get the current configuration that is estimated to be the best (in terms of
        optimal objective value) without waiting for a reported evaluation of that
        configuration. Instead, the objective value estimation relies on BO's
        underlying model.

        This might return `None` in case there is no successfully evaluated
        configuration yet (thus, the optimizer has not been given training data yet).

        Returns:
            blackboxopt.evaluation.Evaluation
                The evaluated specification containing the estimated best configuration
                or `None` in case no evaluations have been reported yet.
        """
        return predict_model_based_best(
            model=self.model,
            objective=self.objective,
            search_space=self.search_space,
            torch_dtype=self.torch_dtype,
        )

__init__(self, search_space, objective, model, acquisition_function_factory, af_optimizer_kwargs=None, num_initial_random_samples=1, max_pending_evaluations=1, batch_shape=torch.Size([]), logger=None, seed=None, torch_dtype=torch.float64) special

Single objective BO optimizer that uses as a surrogate model the model object provided by user.

The model is expected to be extended from BoTorch base model Model class, and does not require to be a GP model.

Parameters:

Name	Type	Description	Default
search_space	ParameterSpace	The space in which to optimize.	required
objective	Objective	The objective to optimize.	required
model	Model	Surrogate model of Model type.	required
acquisition_function_factory	Callable[[botorch.models.model.Model], botorch.acquisition.acquisition.AcquisitionFunction]	Callable that produces an acquisition function instance, could also be a compatible acquisition function class. Only acquisition functions to be minimized are supported. Providing a partially initialized class is possible with, e.g. functools.partial(UpperConfidenceBound, beta=6.0, maximize=False).	required
af_optimizer_kwargs		Settings for acquisition function optimizer, see botorch.optim.optimize_acqf and in case the whole search space is discrete: botorch.optim.optimize_acqf_discrete. The former can be enforced by providing raw_samples or num_restarts, the latter by providing num_random_choices.	None
num_initial_random_samples	int	Size of the initial space-filling design that is used before starting BO. The points are sampled randomly in the search space. If no random sampling is required, set it to 0. When random sampling is enabled, but evaluations with missing objective values are reported, more specifications are sampled until num_initial_random_samples many valid evaluations were reported.	1
max_pending_evaluations	Optional[int]	Maximum number of parallel evaluations. For sequential BO use the default value of 1. If no limit is required, set it to None.	1
batch_shape	Size	Batch dimension(s) used for batched models.	torch.Size([])
logger	Optional[logging.Logger]	Custom logger.	None
seed	Optional[int]	A seed to make the optimization reproducible.	None
torch_dtype	dtype	Torch data type used for storing the data. This needs to match the dtype of the model used	torch.float64

Source code in blackboxopt/optimizers/botorch_base.py

def __init__(
    self,
    search_space: ps.ParameterSpace,
    objective: Objective,
    model: Model,
    acquisition_function_factory: Callable[[Model], AcquisitionFunction],
    af_optimizer_kwargs=None,
    num_initial_random_samples: int = 1,
    max_pending_evaluations: Optional[int] = 1,
    batch_shape: torch.Size = torch.Size(),
    logger: Optional[logging.Logger] = None,
    seed: Optional[int] = None,
    torch_dtype: torch.dtype = torch.float64,
):
    """Single objective BO optimizer that uses as a surrogate model the `model`
    object provided by user.

    The `model` is expected to be extended from BoTorch base model `Model` class,
    and does not require to be a GP model.

    Args:
        search_space: The space in which to optimize.
        objective: The objective to optimize.
        model: Surrogate model of `Model` type.
        acquisition_function_factory: Callable that produces an acquisition function
            instance, could also be a compatible acquisition function class.
            Only acquisition functions to be minimized are supported.
            Providing a partially initialized class is possible with, e.g.
            `functools.partial(UpperConfidenceBound, beta=6.0, maximize=False)`.
        af_optimizer_kwargs: Settings for acquisition function optimizer,
            see `botorch.optim.optimize_acqf` and in case the whole search space
            is discrete: `botorch.optim.optimize_acqf_discrete`. The former can be
            enforced by providing `raw_samples` or `num_restarts`, the latter by
            providing `num_random_choices`.
        num_initial_random_samples: Size of the initial space-filling design that
            is used before starting BO. The points are sampled randomly in the
            search space. If no random sampling is required, set it to 0.
            When random sampling is enabled, but evaluations with missing objective
            values are reported, more specifications are sampled until
            `num_initial_random_samples` many valid evaluations were reported.
        max_pending_evaluations: Maximum number of parallel evaluations. For
            sequential BO use the default value of 1. If no limit is required,
            set it to None.
        batch_shape: Batch dimension(s) used for batched models.
        logger: Custom logger.
        seed: A seed to make the optimization reproducible.
        torch_dtype: Torch data type used for storing the data. This needs to match
            the dtype of the model used
    """
    super().__init__(search_space=search_space, objective=objective, seed=seed)
    self.num_initial_random = num_initial_random_samples
    self.max_pending_evaluations = max_pending_evaluations
    self.batch_shape = batch_shape
    self.logger = logger or logging.getLogger("blackboxopt")

    self.torch_dtype = torch_dtype
    self.X = torch.empty(
        (*self.batch_shape, 0, len(search_space)), dtype=torch_dtype
    )
    self.losses = torch.empty((*self.batch_shape, 0, 1), dtype=torch_dtype)
    self.pending_specifications: Dict[int, EvaluationSpecification] = {}
    if seed is not None:
        torch.manual_seed(seed=seed)

    self.model = model
    self.acquisition_function_factory = acquisition_function_factory
    self.af_optimizer_kwargs = af_optimizer_kwargs

generate_evaluation_specification(self)

Call the optimizer specific function and append a unique integer id to the specification.

Please refer to the docstring of blackboxopt.base.SingleObjectiveOptimizer.generate_evaluation_specification for a description of the method.

Source code in blackboxopt/optimizers/botorch_base.py

def generate_evaluation_specification(self) -> EvaluationSpecification:
    """Call the optimizer specific function and append a unique integer id
    to the specification.

    Please refer to the docstring of
    `blackboxopt.base.SingleObjectiveOptimizer.generate_evaluation_specification`
    for a description of the method.
    """
    if (
        self.max_pending_evaluations
        and len(self.pending_specifications) == self.max_pending_evaluations
    ):
        raise OptimizerNotReady

    # Generate random samples until there are enough samples where at least one of
    # the objective values is available
    if self.num_initial_random > 0 and (
        sum(~torch.any(self.losses.isnan(), dim=1)) < self.num_initial_random
    ):
        eval_spec = EvaluationSpecification(
            configuration=self.search_space.sample(),
            optimizer_info={"model_based_pick": False},
        )
    else:
        eval_spec = self._generate_evaluation_specification()
        eval_spec.optimizer_info["model_based_pick"] = True

    eval_id = self.X.size(-2) + len(self.pending_specifications)
    eval_spec.optimizer_info["evaluation_id"] = eval_id
    self.pending_specifications[eval_id] = eval_spec
    return eval_spec

predict_model_based_best(self)

Get the current configuration that is estimated to be the best (in terms of optimal objective value) without waiting for a reported evaluation of that configuration. Instead, the objective value estimation relies on BO's underlying model.

This might return None in case there is no successfully evaluated configuration yet (thus, the optimizer has not been given training data yet).

Returns:

Type	Description
Optional[blackboxopt.evaluation.Evaluation]	blackboxopt.evaluation.Evaluation The evaluated specification containing the estimated best configuration or None in case no evaluations have been reported yet.

Source code in blackboxopt/optimizers/botorch_base.py

def predict_model_based_best(self) -> Optional[Evaluation]:
    """Get the current configuration that is estimated to be the best (in terms of
    optimal objective value) without waiting for a reported evaluation of that
    configuration. Instead, the objective value estimation relies on BO's
    underlying model.

    This might return `None` in case there is no successfully evaluated
    configuration yet (thus, the optimizer has not been given training data yet).

    Returns:
        blackboxopt.evaluation.Evaluation
            The evaluated specification containing the estimated best configuration
            or `None` in case no evaluations have been reported yet.
    """
    return predict_model_based_best(
        model=self.model,
        objective=self.objective,
        search_space=self.search_space,
        torch_dtype=self.torch_dtype,
    )

report(self, evaluations)

A simple report method that conditions the model on data. This likely needs to be overridden for more specific BO implementations.

Source code in blackboxopt/optimizers/botorch_base.py

def report(self, evaluations: Union[Evaluation, Iterable[Evaluation]]) -> None:
    """A simple report method that conditions the model on data.
    This likely needs to be overridden for more specific BO implementations.
    """
    _evals = [evaluations] if isinstance(evaluations, Evaluation) else evaluations
    self._update_internal_evaluation_data(_evals)
    # Just for populating all relevant caches
    self.model.posterior(self.X)

    x_filtered, y_filtered = filter_y_nans(self.X, self.losses)

    # The actual model update
    # Ignore BotorchTensorDimensionWarning which is always reported to make the user
    # aware that they are reponsible for the right input Tensors dimensionality.
    with warnings.catch_warnings():
        warnings.simplefilter(
            action="ignore", category=BotorchTensorDimensionWarning
        )
        self.model = self.model.condition_on_observations(x_filtered, y_filtered)