from __future__ import annotations
from typing import Any
import copy
import numpy as np
from smac.acquisition.function.abstract_acquisition_function import (
AbstractAcquisitionFunction,
)
from smac.model.abstract_model import AbstractModel
from smac.utils.logging import get_logger
__copyright__ = "Copyright 2022, automl.org"
__license__ = "3-clause BSD"
logger = get_logger(__name__)
[docs]class IntegratedAcquisitionFunction(AbstractAcquisitionFunction):
r"""Compute the integrated acquisition function by marginalizing over model hyperparameters
See "Practical Bayesian Optimization of Machine Learning Algorithms" by Jasper Snoek et al.
(https://papers.nips.cc/paper/4522-practical-bayesian-optimization-of-machine-learning-algorithms.pdf)
for further details.
Parameters
----------
acquisition_function : AbstractAcquisitionFunction
Acquisition function to be integrated.
Attributes
----------
_acquisition_function : AbstractAcquisitionFunction
Acquisition function to be integrated.
_functions: list[AbstractAcquisitionFunction]
Holds n (n = number of models) copies of the acquisition function.
_eta : float
Current incumbent function value.
"""
def __init__(self, acquisition_function: AbstractAcquisitionFunction) -> None:
super().__init__()
self._acquisition_function: AbstractAcquisitionFunction = acquisition_function
self._functions: list[AbstractAcquisitionFunction] = []
self._eta: float | None = None
@property
def name(self) -> str: # noqa: D102
return f"Integrated 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})
return meta
def _update(self, **kwargs: Any) -> None:
"""Update the acquisition functions values.
This method will be called if the model is updated. For example, entropy search uses it to update its
approximation of P(x=x_min) and EI uses it to update the current fmin.
This implementation creates an acquisition function object for each model to integrate over and sets the
respective attributes for each acquisition function object.
Parameters
----------
kwargs : Any
Keyword arguments for the model.
Raises
------
ValueError
If the number of models is zero.
"""
model = self.model
models: list[AbstractModel] | None = None
if hasattr(model, "models"):
models = model.models # type: ignore
if models is None or len(models) == 0:
raise ValueError("IntegratedAcquisitionFunction requires at least one model to integrate!")
if len(self._functions) == 0 or len(self._functions) != len(models):
self._functions = [copy.deepcopy(self._acquisition_function) for _ in models]
for submodel, func in zip(models, self._functions):
func.update(model=submodel, **kwargs)
def _compute(self, X: np.ndarray) -> np.ndarray:
"""Compute integrated acquisition values
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]
Acquisition function values wrt X.
Raises
------
ValueError
If `update` has not been called first (`_functions` not up to date).
"""
if self._functions is None:
raise ValueError("Need to call `update` first!")
return np.array([func._compute(X) for func in self._functions]).mean(axis=0)