from __future__ import annotations
from typing import Any
import numpy as np
from ConfigSpace import Configuration
from ConfigSpace.hyperparameters import FloatHyperparameter
from smac.acquisition.function.abstract_acquisition_function import (
AbstractAcquisitionFunction,
)
from smac.acquisition.function.confidence_bound import LCB
from smac.acquisition.function.integrated_acquisition_function import (
IntegratedAcquisitionFunction,
)
from smac.acquisition.function.thompson import TS
from smac.model.abstract_model import AbstractModel
from smac.model.random_forest.abstract_random_forest import AbstractRandomForest
from smac.utils.logging import get_logger
__copyright__ = "Copyright 2022, automl.org"
__license__ = "3-clause BSD"
logger = get_logger(__name__)
[docs]
class PriorAcquisitionFunction(AbstractAcquisitionFunction):
r"""Weight the acquisition function with a user-defined prior over the optimum.
See "piBO: Augmenting Acquisition Functions with User Beliefs for Bayesian Optimization" by Carl
Hvarfner et al. [HSSL22]_ for further details.
Parameters
----------
decay_beta: float
Decay factor on the user prior. A solid default value for decay_beta (empirically founded) is
``scenario.n_trials`` / 10.
prior_floor : float, defaults to 1e-12
Lowest possible value of the prior to ensure non-negativity for all values in the search space.
discretize : bool, defaults to False
Whether to discretize (bin) the densities for continous parameters. Triggered for Random Forest models and
continous hyperparameters to avoid a pathological case where all Random Forest randomness is removed
(RF surrogates require piecewise constant acquisition functions to be well-behaved).
discrete_bins_factor : float, defaults to 10.0
If discretizing, the multiple on the number of allowed bins for each parameter.
"""
def __init__(
self,
acquisition_function: AbstractAcquisitionFunction,
decay_beta: float,
prior_floor: float = 1e-12,
discretize: bool = False,
discrete_bins_factor: float = 10.0,
):
super().__init__()
self._acquisition_function: AbstractAcquisitionFunction = acquisition_function
self._functions: list[AbstractAcquisitionFunction] = []
self._eta: float | None = None
self._hyperparameters: dict[Any, Configuration] | None = None
self._decay_beta = decay_beta
self._prior_floor = prior_floor
self._discretize = discretize
self._discrete_bins_factor = discrete_bins_factor
# check if the acquisition function is LCB or TS - then the acquisition function values
# need to be rescaled to assure positiveness & correct magnitude
if isinstance(self._acquisition_function, IntegratedAcquisitionFunction):
acquisition_type = self._acquisition_function._acquisition_function
else:
acquisition_type = self._acquisition_function
self._rescale = isinstance(acquisition_type, (LCB, TS))
self._iteration_number = 0
@property
def name(self) -> str: # noqa: D102
return f"Prior Acquisition Function ({self._acquisition_function.__class__.__name__})"
@property
def meta(self) -> dict[str, Any]: # noqa: D102
meta = super().meta
meta.update(
{
"acquisition_function": self._acquisition_function.meta,
"decay_beta": self._decay_beta,
"prior_floor": self._prior_floor,
"discretize": self._discretize,
"discrete_bins_factor": self._discrete_bins_factor,
}
)
return meta
@property
def model(self) -> AbstractModel | None: # noqa: D102
return self._model
@model.setter
def model(self, model: AbstractModel) -> None:
self._model = model
self._hyperparameters = model._configspace.get_hyperparameters_dict()
if isinstance(model, AbstractRandomForest):
if not self._discretize:
logger.warning("Discretizing the prior for random forest models.")
self._discretize = True
def _update(self, **kwargs: Any) -> None:
"""Update the acquisition function attributes required for calculation.
Parameters
----------
eta : float
Current incumbent value.
"""
assert "eta" in kwargs
self._iteration_number += 1
self._eta = kwargs["eta"]
assert self.model is not None
self._acquisition_function.update(model=self.model, **kwargs)
def _compute_prior(self, X: np.ndarray) -> np.ndarray:
"""Compute the prior-weighted acquisition function values, where the prior on each
parameter is multiplied by a decay factor controlled by the parameter decay_beta and
the iteration number. Multivariate priors are not supported, for now.
Parameters
----------
X: np.ndarray [N, D]
The input points where the user-specified prior should be evaluated. The dimensionality of X is (N, D),
with N as the number of points to evaluate at and D is the number of dimensions of one X.
Returns
-------
np.ndarray [N, 1]
The user prior over the optimum for values of X.
"""
assert self._hyperparameters is not None
prior_values = np.ones((len(X), 1))
# iterate over the hyperparmeters (alphabetically sorted) and the columns, which come
# in the same order
for parameter, X_col in zip(self._hyperparameters.values(), X.T):
if self._discretize and isinstance(parameter, FloatHyperparameter):
assert self._discrete_bins_factor is not None
number_of_bins = int(np.ceil(self._discrete_bins_factor * self._decay_beta / self._iteration_number))
prior_values *= self._compute_discretized_pdf(parameter, X_col, number_of_bins) + self._prior_floor
else:
prior_values *= parameter._pdf(X_col[:, np.newaxis])
return prior_values
def _compute_discretized_pdf(
self,
hyperparameter: FloatHyperparameter,
X_col: np.ndarray,
number_of_bins: int,
) -> np.ndarray:
"""Discretize (bins) prior values on continous a specific continous parameter
to an increasingly coarse discretization determined by the prior decay parameter.
Parameters
----------
hyperparameter : FloatHyperparameter
A float hyperparameter that, due to using a random forest surrogate, must have its prior discretized.
X_col: np.ndarray [N, ]
The input points where the acquisition function should be evaluated. The dimensionality of X is (N, ),
with N as the number of points to evaluate for the specific hyperparameter.
number_of_bins : int
The number of unique values allowed on the discretized version of the pdf.
Returns
-------
np.ndarray [N, 1]
The user prior over the optimum for the parameter at hand.
"""
# Evaluates the actual pdf on all the relevant points
pdf_values = hyperparameter._pdf(X_col[:, np.newaxis])
# Retrieves the largest value of the pdf in the domain
lower, upper = (0, hyperparameter.get_max_density())
# Creates the bins (the possible discrete options of the pdf)
bin_values = np.linspace(lower, upper, number_of_bins)
# Generates an index (bin) for each evaluated point
bin_indices = np.clip(
np.round((pdf_values - lower) * number_of_bins / (upper - lower)), 0, number_of_bins - 1
).astype(int)
# Gets the actual value for each point
prior_values = bin_values[bin_indices]
return prior_values
def _compute(self, X: np.ndarray) -> np.ndarray:
"""Compute the prior-weighted acquisition function values, where the prior on each
parameter is multiplied by a decay factor controlled by the parameter decay_beta and
the iteration number. Multivariate priors are not supported, for now.
Parameters
----------
X: np.ndarray [N, D]
The input points where the acquisition function should be evaluated. The dimensionality of X is (N, D),
with N as the number of points to evaluate at and D is the number of dimensions of one X.
Returns
-------
np.ndarray [N, 1]
Prior-weighted acquisition function values of X
"""
if self._rescale:
# for TS and UCB, we need to scale the function values to not run into issues
# of negative values or issues of varying magnitudes (here, they are both)
# negative by design and just flipping the sign leads to picking the worst point)
acq_values = np.clip(self._acquisition_function._compute(X) + self._eta, 0, np.inf)
else:
acq_values = self._acquisition_function._compute(X)
prior_values = self._compute_prior(X) + self._prior_floor
decayed_prior_values = np.power(prior_values, self._decay_beta / self._iteration_number)
return acq_values * decayed_prior_values