# 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.
r"""Gaussian Process Regression models with fully Bayesian inference.
We use a lightweight PyTorch implementation of a Matern-5/2 kernel as there are some
performance issues with running NUTS on top of standard GPyTorch models. The resulting
hyperparameter samples are loaded into a batched GPyTorch model after fitting.
References:
.. [Eriksson2021saasbo]
D. Eriksson, M. Jankowiak. High-Dimensional Bayesian Optimization
with Sparse Axis-Aligned Subspaces. Proceedings of the Thirty-
Seventh Conference on Uncertainty in Artificial Intelligence, 2021.
"""
import math
from abc import abstractmethod
from typing import Any, Dict, List, Optional, Tuple, Union
import pyro
import torch
from botorch.acquisition.objective import PosteriorTransform
from botorch.models.gp_regression import FixedNoiseGP, SingleTaskGP
from botorch.models.transforms.input import InputTransform
from botorch.models.transforms.outcome import OutcomeTransform
from botorch.models.utils import validate_input_scaling
from botorch.posteriors.fully_bayesian import FullyBayesianPosterior, MCMC_DIM
from botorch.sampling.samplers import MCSampler
from botorch.utils.containers import TrainingData
from gpytorch.constraints import GreaterThan
from gpytorch.distributions.multivariate_normal import MultivariateNormal
from gpytorch.kernels import MaternKernel, ScaleKernel
from gpytorch.kernels.kernel import Distance, Kernel
from gpytorch.likelihoods.gaussian_likelihood import (
FixedNoiseGaussianLikelihood,
GaussianLikelihood,
)
from gpytorch.likelihoods.likelihood import Likelihood
from gpytorch.means.constant_mean import ConstantMean
from gpytorch.means.mean import Mean
from torch import Tensor
MIN_INFERRED_NOISE_LEVEL = 1e-6
[docs]def matern52_kernel(X: Tensor, lengthscale: Tensor) -> Tensor:
"""Matern-5/2 kernel."""
nu = 5 / 2
dist = compute_dists(X=X, lengthscale=lengthscale)
exp_component = torch.exp(-math.sqrt(nu * 2) * dist)
constant_component = (math.sqrt(5) * dist).add(1).add(5.0 / 3.0 * (dist ** 2))
return constant_component * exp_component
[docs]def compute_dists(X: Tensor, lengthscale: Tensor) -> Tensor:
"""Compute kernel distances."""
return Distance()._dist(
X / lengthscale, X / lengthscale, postprocess=False, x1_eq_x2=True
)
[docs]def reshape_and_detach(target: Tensor, new_value: Tensor) -> None:
"""Detach and reshape `new_value` to match `target`."""
return new_value.detach().clone().view(target.shape).to(target)
[docs]class PyroModel:
r"""Abstract base class for a Pyro model."""
[docs] @abstractmethod
def sample(self) -> None:
r"""Sample from the model."""
pass # pragma: no cover
[docs] @abstractmethod
def postprocess_mcmc_samples(
self, mcmc_samples: Dict[str, Tensor], **kwargs: Any
) -> Dict[str, Tensor]:
"""Post-process the final MCMC samples."""
pass # pragma: no cover
[docs] @abstractmethod
def load_mcmc_samples(
self, mcmc_samples: Dict[str, Tensor]
) -> Tuple[Mean, Kernel, Likelihood]:
pass # pragma: no cover
[docs]class SaasPyroModel(PyroModel):
r"""Implementation of the sparse axis-aligned subspace priors (SAAS) model.
The SAAS model uses sparsity-inducing priors to identift the most important
parameters. This model is suitable for high-dimensional BO with potentially
hundreds of tunable parameters. See [Eriksson2021saasbo]_ for more details.
"""
[docs] def sample(self) -> None:
r"""Sample from the SAAS model.
This samples the mean, noise variance, outputscale, and lengthscales according
to the SAAS prior.
"""
tkwargs = {"dtype": self.train_X.dtype, "device": self.train_X.device}
outputscale = self.sample_outputscale(concentration=2.0, rate=0.15, **tkwargs)
mean = self.sample_mean(**tkwargs)
noise = self.sample_noise(**tkwargs)
lengthscale = self.sample_lengthscale(dim=self.train_X.shape[-1], **tkwargs)
k = matern52_kernel(X=self.train_X, lengthscale=lengthscale)
k = outputscale * k + noise * torch.eye(self.train_X.shape[0], **tkwargs)
pyro.sample(
"Y",
pyro.distributions.MultivariateNormal(
loc=mean.view(-1).expand(self.train_X.shape[0]),
covariance_matrix=k,
),
obs=self.train_Y.squeeze(-1),
)
[docs] def sample_outputscale(
self, concentration: float = 2.0, rate: float = 0.15, **tkwargs: Any
) -> Tensor:
r"""Sample the outputscale."""
return pyro.sample(
"outputscale",
pyro.distributions.Gamma(
torch.tensor(concentration, **tkwargs),
torch.tensor(rate, **tkwargs),
),
)
[docs] def sample_mean(self, **tkwargs: Any) -> Tensor:
r"""Sample the mean constant."""
return pyro.sample(
"mean",
pyro.distributions.Normal(
torch.tensor(0.0, **tkwargs),
torch.tensor(1.0, **tkwargs),
),
)
[docs] def sample_noise(self, **tkwargs: Any) -> Tensor:
r"""Sample the noise variance."""
if self.train_Yvar is None:
return pyro.sample(
"noise",
pyro.distributions.Gamma(
torch.tensor(0.9, **tkwargs),
torch.tensor(10.0, **tkwargs),
),
)
else:
return self.train_Yvar
[docs] def sample_lengthscale(
self, dim: int, alpha: float = 0.1, **tkwargs: Any
) -> Tensor:
r"""Sample the lengthscale."""
tausq = pyro.sample(
"kernel_tausq",
pyro.distributions.HalfCauchy(torch.tensor(alpha, **tkwargs)),
)
inv_length_sq = pyro.sample(
"_kernel_inv_length_sq",
pyro.distributions.HalfCauchy(torch.ones(dim, **tkwargs)),
)
inv_length_sq = pyro.deterministic(
"kernel_inv_length_sq", tausq * inv_length_sq
)
lengthscale = pyro.deterministic(
"lengthscale",
(1.0 / inv_length_sq).sqrt(),
)
return lengthscale
[docs] def postprocess_mcmc_samples(
self, mcmc_samples: Dict[str, Tensor]
) -> Dict[str, Tensor]:
r"""Post-process the MCMC samples.
This computes the true lengthscales and removes the inverse lengthscales and
tausq (global shrinkage).
"""
inv_length_sq = (
mcmc_samples["kernel_tausq"].unsqueeze(-1)
* mcmc_samples["_kernel_inv_length_sq"]
)
mcmc_samples["lengthscale"] = (1.0 / inv_length_sq).sqrt()
# Delete `kernel_tausq` and `_kernel_inv_length_sq` since they aren't loaded
# into the final model.
del mcmc_samples["kernel_tausq"], mcmc_samples["_kernel_inv_length_sq"]
return mcmc_samples
[docs] def load_mcmc_samples(
self, mcmc_samples: Dict[str, Tensor]
) -> Tuple[Mean, Kernel, Likelihood]:
r"""Load the MCMC samples into the mean_module, covar_module, and likelihood."""
tkwargs = {"device": self.train_X.device, "dtype": self.train_X.dtype}
num_mcmc_samples = len(mcmc_samples["mean"])
batch_shape = torch.Size([num_mcmc_samples])
mean_module = ConstantMean(batch_shape=batch_shape).to(**tkwargs)
covar_module = ScaleKernel(
base_kernel=MaternKernel(
ard_num_dims=self.train_X.shape[-1],
batch_shape=batch_shape,
),
batch_shape=batch_shape,
).to(**tkwargs)
if self.train_Yvar is not None:
likelihood = FixedNoiseGaussianLikelihood(
# Reshape to shape `num_mcmc_samples x N`
noise=self.train_Yvar.squeeze(-1).expand(
num_mcmc_samples, len(self.train_Yvar)
),
batch_shape=batch_shape,
).to(**tkwargs)
else:
likelihood = GaussianLikelihood(
batch_shape=batch_shape,
noise_constraint=GreaterThan(MIN_INFERRED_NOISE_LEVEL),
).to(**tkwargs)
likelihood.noise_covar.noise = reshape_and_detach(
target=likelihood.noise_covar.noise,
new_value=mcmc_samples["noise"].clamp_min(MIN_INFERRED_NOISE_LEVEL),
)
covar_module.base_kernel.lengthscale = reshape_and_detach(
target=covar_module.base_kernel.lengthscale,
new_value=mcmc_samples["lengthscale"],
)
covar_module.outputscale = reshape_and_detach(
target=covar_module.outputscale,
new_value=mcmc_samples["outputscale"],
)
mean_module.constant.data = reshape_and_detach(
target=mean_module.constant.data,
new_value=mcmc_samples["mean"],
)
return mean_module, covar_module, likelihood
[docs]class SaasFullyBayesianSingleTaskGP(SingleTaskGP):
r"""A fully Bayesian single-task GP model with the SAAS prior.
This model assumes that the inputs have been normalized to [0, 1]^d and that
the output has been standardized to have zero mean and unit variance. You can
either normalize and standardize the data before constructing the model or use
an `input_transform` and `outcome_transform`. The SAAS model [Eriksson2021saasbo]_
with a Matern-5/2 kernel is used by default.
You are expected to use `fit_fully_bayesian_model_nuts` to fit this model as it
isn't compatible with `fit_gpytorch_model`.
Example:
>>> saas_gp = SaasFullyBayesianSingleTaskGP(train_X, train_Y)
>>> fit_fully_bayesian_model_nuts(saas_gp)
>>> posterior = saas_gp.posterior(test_X)
"""
def __init__(
self,
train_X: Tensor,
train_Y: Tensor,
train_Yvar: Optional[Tensor] = None,
outcome_transform: Optional[OutcomeTransform] = None,
input_transform: Optional[InputTransform] = None,
pyro_model: Optional[PyroModel] = None,
) -> None:
r"""Initialize the fully Bayesian single-task GP model.
Args:
train_X: Training inputs (n x d)
train_Y: Training targets (n x 1)
train_Yvar: Observed noise variance (n x 1). Inferred if None.
outcome_transform: An outcome transform that is applied to the
training data during instantiation and to the posterior during
inference (that is, the `Posterior` obtained by calling
`.posterior` on the model will be on the original scale).
input_transform: An input transform that is applied in the model's
forward pass.
pyro_model: Optional `PyroModel`, defaults to `SaasPyroModel`.
"""
if not (
train_X.ndim == train_Y.ndim == 2
and len(train_X) == len(train_Y)
and train_Y.shape[-1] == 1
):
raise ValueError(
"Expected train_X to have shape n x d and train_Y to have shape n x 1"
)
if train_Yvar is not None:
if train_Y.shape != train_Yvar.shape:
raise ValueError(
"Expected train_Yvar to be None or have the same shape as train_Y"
)
with torch.no_grad():
transformed_X = self.transform_inputs(
X=train_X, input_transform=input_transform
)
if outcome_transform is not None:
train_Y, train_Yvar = outcome_transform(train_Y, train_Yvar)
self._validate_tensor_args(X=transformed_X, Y=train_Y)
validate_input_scaling(
train_X=transformed_X, train_Y=train_Y, train_Yvar=train_Yvar
)
self._set_dimensions(train_X=train_X, train_Y=train_Y)
if train_Yvar is not None: # Clamp after transforming
train_Yvar = train_Yvar.clamp(MIN_INFERRED_NOISE_LEVEL)
super().__init__(train_X, train_Y)
self.mean_module = None
self.covar_module = None
self.likelihood = None
if pyro_model is None:
pyro_model = SaasPyroModel()
pyro_model.set_inputs(
train_X=transformed_X, train_Y=train_Y, train_Yvar=train_Yvar
)
self.pyro_model = pyro_model
if outcome_transform is not None:
self.outcome_transform = outcome_transform
if input_transform is not None:
self.input_transform = input_transform
def _check_if_fitted(self):
r"""Raise an exception if the model hasn't been fitted."""
if self.covar_module is None:
raise RuntimeError(
"Model has not been fitted. You need to call "
"`fit_fully_bayesian_model_nuts` to fit the model."
)
@property
def median_lengthscale(self) -> Tensor:
r"""Median lengthscales across the MCMC samples."""
self._check_if_fitted()
lengthscale = self.covar_module.base_kernel.lengthscale.clone()
return lengthscale.median(0).values.squeeze(0)
@property
def num_mcmc_samples(self) -> int:
r"""Number of MCMC samples in the model."""
self._check_if_fitted()
return len(self.covar_module.outputscale)
[docs] def fantasize(
self,
X: Tensor,
sampler: MCSampler,
observation_noise: Union[bool, Tensor] = True,
**kwargs: Any,
) -> FixedNoiseGP:
raise NotImplementedError("Fantasize is not implemented!")
[docs] def train(self, mode: bool = True) -> None:
r"""Puts the model in `train` mode."""
super().train(mode=mode)
if mode:
self.mean_module = None
self.covar_module = None
self.likelihood = None
[docs] def load_mcmc_samples(self, mcmc_samples: Dict[str, Tensor]) -> None:
r"""Load the MCMC hyperparameter samples into the model.
This method will be called by `fit_fully_bayesian_model_nuts` when the model
has been fitted in order to create a batched SingleTaskGP model.
"""
(
self.mean_module,
self.covar_module,
self.likelihood,
) = self.pyro_model.load_mcmc_samples(mcmc_samples=mcmc_samples)
[docs] def forward(self, X: Tensor) -> MultivariateNormal:
self._check_if_fitted()
return super().forward(X.unsqueeze(MCMC_DIM))
[docs] def posterior(
self,
X: Tensor,
output_indices: Optional[List[int]] = None,
observation_noise: bool = False,
posterior_transform: Optional[PosteriorTransform] = None,
**kwargs: Any,
) -> FullyBayesianPosterior:
r"""Computes the posterior over model outputs at the provided points.
Args:
X: A `(batch_shape) x q x d`-dim Tensor, where `d` is the dimension
of the feature space and `q` is the number of points considered
jointly.
output_indices: A list of indices, corresponding to the outputs over
which to compute the posterior (if the model is multi-output).
Can be used to speed up computation if only a subset of the
model's outputs are required for optimization. If omitted,
computes the posterior over all model outputs.
observation_noise: If True, add the observation noise from the
likelihood to the posterior. If a Tensor, use it directly as the
observation noise (must be of shape `(batch_shape) x q x m`).
posterior_transform: An optional PosteriorTransform.
Returns:
A `FullyBayesianPosterior` object. Includes observation noise if specified.
"""
self._check_if_fitted()
posterior = super().posterior(
X=X,
output_indices=output_indices,
observation_noise=observation_noise,
posterior_transform=posterior_transform,
**kwargs,
)
posterior = FullyBayesianPosterior(mvn=posterior.mvn)
return posterior
