Source code for botorch.acquisition.input_constructors

#!/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.

r"""
A registry of helpers for generating inputs to acquisition function
constructors programmatically from a consistent input format.
"""

from __future__ import annotations

import warnings
from typing import Any, Callable, Dict, List, Optional, Tuple, Type, Union

import torch
from botorch.acquisition.acquisition import AcquisitionFunction
from botorch.acquisition.analytic import (
    ConstrainedExpectedImprovement,
    ExpectedImprovement,
    NoisyExpectedImprovement,
    PosteriorMean,
    ProbabilityOfImprovement,
    UpperConfidenceBound,
)
from botorch.acquisition.cost_aware import InverseCostWeightedUtility
from botorch.acquisition.fixed_feature import FixedFeatureAcquisitionFunction
from botorch.acquisition.knowledge_gradient import (
    qKnowledgeGradient,
    qMultiFidelityKnowledgeGradient,
)
from botorch.acquisition.max_value_entropy_search import (
    qMaxValueEntropy,
    qMultiFidelityMaxValueEntropy,
)
from botorch.acquisition.monte_carlo import (
    qExpectedImprovement,
    qNoisyExpectedImprovement,
    qProbabilityOfImprovement,
    qSimpleRegret,
    qUpperConfidenceBound,
)
from botorch.acquisition.multi_objective import (
    ExpectedHypervolumeImprovement,
    MCMultiOutputObjective,
    qExpectedHypervolumeImprovement,
    qNoisyExpectedHypervolumeImprovement,
)
from botorch.acquisition.multi_objective.objective import (
    AnalyticMultiOutputObjective,
    IdentityAnalyticMultiOutputObjective,
    IdentityMCMultiOutputObjective,
)
from botorch.acquisition.multi_objective.utils import get_default_partitioning_alpha
from botorch.acquisition.objective import (
    AcquisitionObjective,
    IdentityMCObjective,
    MCAcquisitionObjective,
    PosteriorTransform,
    ScalarizedObjective,
    ScalarizedPosteriorTransform,
)
from botorch.acquisition.utils import (
    expand_trace_observations,
    project_to_target_fidelity,
)
from botorch.exceptions.errors import UnsupportedError
from botorch.models.cost import AffineFidelityCostModel
from botorch.models.model import Model
from botorch.optim.optimize import optimize_acqf
from botorch.sampling.samplers import IIDNormalSampler, MCSampler, SobolQMCNormalSampler
from botorch.utils.constraints import get_outcome_constraint_transforms
from botorch.utils.containers import TrainingData
from botorch.utils.multi_objective.box_decompositions.non_dominated import (
    FastNondominatedPartitioning,
    NondominatedPartitioning,
)
from torch import Tensor


ACQF_INPUT_CONSTRUCTOR_REGISTRY = {}


[docs]def get_acqf_input_constructor( acqf_cls: Type[AcquisitionFunction], ) -> Callable[..., Dict[str, Any]]: r"""Get acqusition function input constructor from registry. Args: acqf_cls: The AcquisitionFunction class (not instance) for which to retrieve the input constructor. Returns: The input constructor associated with `acqf_cls`. """ if acqf_cls not in ACQF_INPUT_CONSTRUCTOR_REGISTRY: raise RuntimeError( f"Input constructor for acquisition class `{acqf_cls.__name__}` not " "registered. Use the `@acqf_input_constructor` decorator to register " "a new method." ) return ACQF_INPUT_CONSTRUCTOR_REGISTRY[acqf_cls]
[docs]def acqf_input_constructor( *acqf_cls: Type[AcquisitionFunction], ) -> Callable[..., AcquisitionFunction]: r"""Decorator for registering acquisition function input constructors. Args: acqf_cls: The AcquisitionFunction classes (not instances) for which to register the input constructor. """ for acqf_cls_ in acqf_cls: if acqf_cls_ in ACQF_INPUT_CONSTRUCTOR_REGISTRY: raise ValueError( "Cannot register duplicate arg constructor for acquisition " f"class `{acqf_cls_.__name__}`" ) def decorator(method): for acqf_cls_ in acqf_cls: _register_acqf_input_constructor( acqf_cls=acqf_cls_, input_constructor=method ) ACQF_INPUT_CONSTRUCTOR_REGISTRY[acqf_cls_] = method return method return decorator
def _register_acqf_input_constructor( acqf_cls: Type[AcquisitionFunction], input_constructor: Callable[..., Dict[str, Any]], ) -> None: ACQF_INPUT_CONSTRUCTOR_REGISTRY[acqf_cls] = input_constructor # ------------------------- Deprecation Helpers ------------------------- # def _deprecate_objective_arg( posterior_transform: Optional[PosteriorTransform] = None, objective: Optional[AcquisitionObjective] = None, ) -> Optional[PosteriorTransform]: if posterior_transform is not None: if objective is None: return posterior_transform else: raise RuntimeError( "Got both a non-MC objective (DEPRECATED) and a posterior " "transform. Use only a posterior transform instead." ) elif objective is not None: warnings.warn( "The `objective` argument to AnalyticAcquisitionFunctions is deprecated " "and will be removed in the next version. Use `posterior_transform` " "instead.", DeprecationWarning, ) if not isinstance(objective, ScalarizedObjective): raise UnsupportedError( "Analytic acquisition functions only support ScalarizedObjective " "(DEPRECATED) type objectives." ) return ScalarizedPosteriorTransform( weights=objective.weights, offset=objective.offset ) else: return None # --------------------- Input argument constructors --------------------- #
[docs]@acqf_input_constructor(PosteriorMean) def construct_inputs_analytic_base( model: Model, training_data: TrainingData, posterior_transform: Optional[PosteriorTransform] = None, **kwargs: Any, ) -> Dict[str, Any]: r"""Construct kwargs for basic analytic acquisition functions. Args: model: The model to be used in the acquisition function. training_data: A TrainingData object contraining the model's training data. `best_f` is extracted from here. posterior_transform: The posterior transform to be used in the acquisition function. Returns: A dict mapping kwarg names of the constructor to values. """ return { "model": model, "posterior_transform": _deprecate_objective_arg( posterior_transform=posterior_transform, objective=kwargs.get("objective"), ), }
[docs]@acqf_input_constructor(ExpectedImprovement, ProbabilityOfImprovement) def construct_inputs_best_f( model: Model, training_data: TrainingData, posterior_transform: Optional[PosteriorTransform] = None, maximize: bool = True, **kwargs: Any, ) -> Dict[str, Any]: r"""Construct kwargs for the acquisition functions requiring `best_f`. Args: model: The model to be used in the acquisition function. training_data: A TrainingData object contraining the model's training data. `best_f` is extracted from here. posterior_transform: The posterior transform to be used in the acquisition function. maximize: If True, consider the problem a maximization problem. Returns: A dict mapping kwarg names of the constructor to values. """ base_inputs = construct_inputs_analytic_base( model=model, training_data=training_data, posterior_transform=posterior_transform, **kwargs, ) best_f = kwargs.get( "best_f", get_best_f_analytic( training_data=training_data, objective=kwargs.get("objective"), posterior_transform=posterior_transform, ), ) return {**base_inputs, "best_f": best_f, "maximize": maximize}
[docs]@acqf_input_constructor(UpperConfidenceBound) def construct_inputs_ucb( model: Model, training_data: TrainingData, posterior_transform: Optional[PosteriorTransform] = None, beta: Union[float, Tensor] = 0.2, maximize: bool = True, **kwargs: Any, ) -> Dict[str, Any]: r"""Construct kwargs for `UpperConfidenceBound`. Args: model: The model to be used in the acquisition function. training_data: A TrainingData object contraining the model's training data. `best_f` is extracted from here. posterior_transform: The posterior transform to be used in the acquisition function. beta: Either a scalar or a one-dim tensor with `b` elements (batch mode) representing the trade-off parameter between mean and covariance maximize: If True, consider the problem a maximization problem. Returns: A dict mapping kwarg names of the constructor to values. """ base_inputs = construct_inputs_analytic_base( model=model, training_data=training_data, posterior_transform=posterior_transform, **kwargs, ) return {**base_inputs, "beta": beta, "maximize": maximize}
[docs]@acqf_input_constructor(ConstrainedExpectedImprovement) def construct_inputs_constrained_ei( model: Model, training_data: TrainingData, objective_index: int, constraints: Dict[int, Tuple[Optional[float], Optional[float]]], maximize: bool = True, **kwargs: Any, ) -> Dict[str, Any]: r"""Construct kwargs for `ConstrainedExpectedImprovement`. Args: model: The model to be used in the acquisition function. training_data: A TrainingData object contraining the model's training data. `best_f` is extracted from here. objective_index: The index of the objective. constraints: A dictionary of the form `{i: [lower, upper]}`, where `i` is the output index, and `lower` and `upper` are lower and upper bounds on that output (resp. interpreted as -Inf / Inf if None) maximize: If True, consider the problem a maximization problem. Returns: A dict mapping kwarg names of the constructor to values. """ # TODO: Implement best point computation from training data # best_f = # return { # "model": model, # "best_f": best_f, # "objective_index": objective_index, # "constraints": constraints, # "maximize": maximize, # } raise NotImplementedError # pragma: nocover
[docs]@acqf_input_constructor(NoisyExpectedImprovement) def construct_inputs_noisy_ei( model: Model, training_data: TrainingData, num_fantasies: int = 20, maximize: bool = True, **kwargs: Any, ) -> Dict[str, Any]: r"""Construct kwargs for `NoisyExpectedImprovement`. Args: model: The model to be used in the acquisition function. training_data: A TrainingData object contraining the model's training data. `best_f` is extracted from here. num_fantasies: The number of fantasies to generate. The higher this number the more accurate the model (at the expense of model complexity and performance). maximize: If True, consider the problem a maximization problem. Returns: A dict mapping kwarg names of the constructor to values. """ # TODO: Add prune_baseline functionality as for qNEI if not training_data.is_block_design: raise NotImplementedError("Currently only block designs are supported") return { "model": model, "X_observed": training_data.X, "num_fantasies": num_fantasies, "maximize": maximize, }
[docs]@acqf_input_constructor(qSimpleRegret) def construct_inputs_mc_base( model: Model, training_data: TrainingData, objective: Optional[MCAcquisitionObjective] = None, posterior_transform: Optional[PosteriorTransform] = None, X_pending: Optional[Tensor] = None, sampler: Optional[MCSampler] = None, **kwargs: Any, ) -> Dict[str, Any]: r"""Construct kwargs for basic MC acquisition functions. Args: model: The model to be used in the acquisition function. training_data: A TrainingData object contraining the model's training data. Used e.g. to extract inputs such as `best_f` for expected improvement acquisition functions. objective: The objective to be used in the acquisition function. posterior_transform: The posterior transform to be used in the acquisition function. X_pending: A `batch_shape, m x d`-dim Tensor of `m` design points that have points that have been submitted for function evaluation but have not yet been evaluated. sampler: The sampler used to draw base samples. If omitted, uses the acquisition functions's default sampler. Returns: A dict mapping kwarg names of the constructor to values. """ return { "model": model, "objective": objective, "posterior_transform": posterior_transform, "X_pending": X_pending, "sampler": sampler, }
[docs]@acqf_input_constructor(qExpectedImprovement) def construct_inputs_qEI( model: Model, training_data: TrainingData, objective: Optional[MCAcquisitionObjective] = None, posterior_transform: Optional[PosteriorTransform] = None, X_pending: Optional[Tensor] = None, sampler: Optional[MCSampler] = None, **kwargs: Any, ) -> Dict[str, Any]: r"""Construct kwargs for the `qExpectedImprovement` constructor. Args: model: The model to be used in the acquisition function. training_data: A TrainingData object contraining the model's training data. Used e.g. to extract inputs such as `best_f` for expected improvement acquisition functions. objective: The objective to be used in the acquisition function. posterior_transform: The posterior transform to be used in the acquisition function. X_pending: A `m x d`-dim Tensor of `m` design points that have been submitted for function evaluation but have not yet been evaluated. Concatenated into X upon forward call. sampler: The sampler used to draw base samples. If omitted, uses the acquisition functions's default sampler. Returns: A dict mapping kwarg names of the constructor to values. """ base_inputs = construct_inputs_mc_base( model=model, training_data=training_data, objective=objective, posterior_transform=posterior_transform, sampler=sampler, X_pending=X_pending, ) # TODO: Dedup handling of this here and in the constructor (maybe via a # shared classmethod doing this) best_f = kwargs.get( "best_f", get_best_f_mc( training_data=training_data, objective=objective, posterior_transform=posterior_transform, ), ) return {**base_inputs, "best_f": best_f}
[docs]@acqf_input_constructor(qNoisyExpectedImprovement) def construct_inputs_qNEI( model: Model, training_data: TrainingData, objective: Optional[MCAcquisitionObjective] = None, posterior_transform: Optional[PosteriorTransform] = None, X_pending: Optional[Tensor] = None, sampler: Optional[MCSampler] = None, X_baseline: Optional[Tensor] = None, prune_baseline: bool = False, **kwargs: Any, ) -> Dict[str, Any]: r"""Construct kwargs for the `qNoisyExpectedImprovement` constructor. Args: model: The model to be used in the acquisition function. training_data: A TrainingData object contraining the model's training data. Used e.g. to extract inputs such as `best_f` for expected improvement acquisition functions. Only block- design training data currently supported. objective: The objective to be used in the acquisition function. posterior_transform: The posterior transform to be used in the acquisition function. X_pending: A `m x d`-dim Tensor of `m` design points that have been submitted for function evaluation but have not yet been evaluated. Concatenated into X upon forward call. sampler: The sampler used to draw base samples. If omitted, uses the acquisition functions's default sampler. X_baseline: A `batch_shape x r x d`-dim Tensor of `r` design points that have already been observed. These points are considered as the potential best design point. If omitted, use `training_data.X`. prune_baseline: If True, remove points in `X_baseline` that are highly unlikely to be the best point. This can significantly improve performance and is generally recommended. Returns: A dict mapping kwarg names of the constructor to values. """ base_inputs = construct_inputs_mc_base( model=model, training_data=training_data, objective=objective, posterior_transform=posterior_transform, sampler=sampler, X_pending=X_pending, ) if X_baseline is None: if not training_data.is_block_design: raise NotImplementedError("Currently only block designs are supported.") X_baseline = training_data.X return { **base_inputs, "X_baseline": X_baseline, "prune_baseline": prune_baseline, }
[docs]@acqf_input_constructor(qProbabilityOfImprovement) def construct_inputs_qPI( model: Model, training_data: TrainingData, objective: Optional[MCAcquisitionObjective] = None, posterior_transform: Optional[PosteriorTransform] = None, X_pending: Optional[Tensor] = None, sampler: Optional[MCSampler] = None, tau: float = 1e-3, best_f: Optional[Union[float, Tensor]] = None, **kwargs: Any, ) -> Dict[str, Any]: r"""Construct kwargs for the `qProbabilityOfImprovement` constructor. Args: model: The model to be used in the acquisition function. training_data: A TrainingData object contraining the model's training data. Used e.g. to extract inputs such as `best_f` for expected improvement acquisition functions. objective: The objective to be used in the acquisition function. posterior_transform: The posterior transform to be used in the acquisition function. X_pending: A `m x d`-dim Tensor of `m` design points that have been submitted for function evaluation but have not yet been evaluated. Concatenated into X upon forward call. sampler: The sampler used to draw base samples. If omitted, uses the acquisition functions's default sampler. tau: The temperature parameter used in the sigmoid approximation of the step function. Smaller values yield more accurate approximations of the function, but result in gradients estimates with higher variance. best_f: The best objective value observed so far (assumed noiseless). Can be a `batch_shape`-shaped tensor, which in case of a batched model specifies potentially different values for each element of the batch. Returns: A dict mapping kwarg names of the constructor to values. """ base_inputs = construct_inputs_mc_base( model=model, training_data=training_data, objective=objective, posterior_transform=posterior_transform, sampler=sampler, X_pending=X_pending, ) # TODO: Dedup handling of this here and in the constructor (maybe via a # shared classmethod doing this) if best_f is None: best_f = get_best_f_mc(training_data=training_data, objective=objective) return { **base_inputs, "tau": tau, "best_f": best_f, }
[docs]@acqf_input_constructor(qUpperConfidenceBound) def construct_inputs_qUCB( model: Model, training_data: TrainingData, objective: Optional[MCAcquisitionObjective] = None, posterior_transform: Optional[PosteriorTransform] = None, X_pending: Optional[Tensor] = None, sampler: Optional[MCSampler] = None, beta: float = 0.2, **kwargs: Any, ) -> Dict[str, Any]: r"""Construct kwargs for the `qUpperConfidenceBound` constructor. Args: model: The model to be used in the acquisition function. training_data: A TrainingData object contraining the model's training data. Used e.g. to extract inputs such as `best_f` for expected improvement acquisition functions. objective: The objective to be used in the acquisition function. posterior_transform: The posterior transform to be used in the acquisition function. X_pending: A `m x d`-dim Tensor of `m` design points that have been submitted for function evaluation but have not yet been evaluated. Concatenated into X upon forward call. sampler: The sampler used to draw base samples. If omitted, uses the acquisition functions's default sampler. beta: Controls tradeoff between mean and standard deviation in UCB. Returns: A dict mapping kwarg names of the constructor to values. """ base_inputs = construct_inputs_mc_base( model=model, training_data=training_data, objective=objective, posterior_transform=posterior_transform, sampler=sampler, X_pending=X_pending, ) return {**base_inputs, "beta": beta}
def _get_sampler(mc_samples: int, qmc: bool) -> MCSampler: """Set up MC sampler for q(N)EHVI.""" # initialize the sampler seed = int(torch.randint(1, 10000, (1,)).item()) if qmc: return SobolQMCNormalSampler(num_samples=mc_samples, seed=seed) return IIDNormalSampler(num_samples=mc_samples, seed=seed)
[docs]@acqf_input_constructor(ExpectedHypervolumeImprovement) def construct_inputs_EHVI( model: Model, training_data: TrainingData, objective_thresholds: Tensor, objective: Optional[AnalyticMultiOutputObjective] = None, **kwargs: Any, ) -> Dict[str, Any]: r"""Construct kwargs for `ExpectedHypervolumeImprovement` constructor.""" num_objectives = objective_thresholds.shape[0] if kwargs.get("outcome_constraints") is not None: raise NotImplementedError("EHVI does not yet support outcome constraints.") X_observed = training_data.X alpha = kwargs.get( "alpha", get_default_partitioning_alpha(num_objectives=num_objectives), ) # This selects the objectives (a subset of the outcomes) and set each # objective threhsold to have the proper optimization direction. if objective is None: objective = IdentityAnalyticMultiOutputObjective() ref_point = objective(objective_thresholds) # Compute posterior mean (for ref point computation ref pareto frontier) # if one is not provided among arguments. Y_pmean = kwargs.get("Y_pmean") if Y_pmean is None: with torch.no_grad(): Y_pmean = model.posterior(X_observed).mean if alpha > 0: partitioning = NondominatedPartitioning( ref_point=ref_point, Y=objective(Y_pmean), alpha=alpha, ) else: partitioning = FastNondominatedPartitioning( ref_point=ref_point, Y=objective(Y_pmean), ) return { "model": model, "ref_point": ref_point, "partitioning": partitioning, "objective": objective, }
[docs]@acqf_input_constructor(qExpectedHypervolumeImprovement) def construct_inputs_qEHVI( model: Model, training_data: TrainingData, objective_thresholds: Tensor, objective: Optional[MCMultiOutputObjective] = None, **kwargs: Any, ) -> Dict[str, Any]: r"""Construct kwargs for `qExpectedHypervolumeImprovement` constructor.""" X_observed = training_data.X # compute posterior mean (for ref point computation ref pareto frontier) with torch.no_grad(): Y_pmean = model.posterior(X_observed).mean outcome_constraints = kwargs.pop("outcome_constraints", None) # For HV-based acquisition functions we pass the constraint transform directly if outcome_constraints is None: cons_tfs = None else: cons_tfs = get_outcome_constraint_transforms(outcome_constraints) # Adjust `Y_pmean` to contrain feasible points only. feas = torch.stack([c(Y_pmean) <= 0 for c in cons_tfs], dim=-1).all(dim=-1) Y_pmean = Y_pmean[feas] if objective is None: objective = IdentityMCMultiOutputObjective() ehvi_kwargs = construct_inputs_EHVI( model=model, training_data=training_data, objective_thresholds=objective_thresholds, objective=objective, # Pass `Y_pmean` that accounts for constraints to `construct_inputs_EHVI` # to ensure that correct non-dominated partitioning is produced. Y_pmean=Y_pmean, **kwargs, ) sampler = kwargs.get("sampler") if sampler is None: sampler = _get_sampler( mc_samples=kwargs.get("mc_samples", 128), qmc=kwargs.get("qmc", True) ) add_qehvi_kwargs = { "sampler": sampler, "X_pending": kwargs.get("X_pending"), "constraints": cons_tfs, "eta": kwargs.get("eta", 1e-3), } return {**ehvi_kwargs, **add_qehvi_kwargs}
[docs]@acqf_input_constructor(qNoisyExpectedHypervolumeImprovement) def construct_inputs_qNEHVI( model: Model, training_data: TrainingData, objective_thresholds: Tensor, objective: Optional[MCMultiOutputObjective] = None, **kwargs: Any, ) -> Dict[str, Any]: r"""Construct kwargs for `qNoisyExpectedHypervolumeImprovement` constructor.""" # This selects the objectives (a subset of the outcomes) and set each # objective threhsold to have the proper optimization direction. if objective is None: objective = IdentityMCMultiOutputObjective() outcome_constraints = kwargs.pop("outcome_constraints", None) if outcome_constraints is None: cons_tfs = None else: cons_tfs = get_outcome_constraint_transforms(outcome_constraints) sampler = kwargs.get("sampler") if sampler is None: sampler = _get_sampler( mc_samples=kwargs.get("mc_samples", 128), qmc=kwargs.get("qmc", True) ) return { "model": model, "ref_point": objective(objective_thresholds), "X_baseline": kwargs.get("X_baseline", training_data.X), "sampler": sampler, "objective": objective, "constraints": cons_tfs, "X_pending": kwargs.get("X_pending"), "eta": kwargs.get("eta", 1e-3), "prune_baseline": kwargs.get("prune_baseline", True), "alpha": kwargs.get("alpha", 0.0), "cache_pending": kwargs.get("cache_pending", True), "max_iep": kwargs.get("max_iep", 0), "incremental_nehvi": kwargs.get("incremental_nehvi", True), }
[docs]@acqf_input_constructor(qMaxValueEntropy) def construct_inputs_qMES( model: Model, training_data: TrainingData, bounds: List[Tuple[float, float]], objective: Optional[MCAcquisitionObjective] = None, posterior_transform: Optional[PosteriorTransform] = None, candidate_size: int = 1000, **kwargs: Any, ) -> Dict[str, Any]: r"""Construct kwargs for `qMaxValueEntropy` constructor.""" inputs_mc = construct_inputs_mc_base( model=model, training_data=training_data, objective=objective, **kwargs, ) _kw = {"dtype": training_data.X.dtype, "device": training_data.X.device} _rvs = torch.rand(candidate_size, len(bounds), **_kw) _bounds = torch.tensor(bounds, **_kw).transpose(0, 1) return { **inputs_mc, "candidate_set": _bounds[0] + (_bounds[1] - _bounds[0]) * _rvs, "maximize": kwargs.get("maximize", True), }
[docs]def construct_inputs_mf_base( model: Model, training_data: TrainingData, target_fidelities: Dict[int, Union[int, float]], fidelity_weights: Optional[Dict[int, float]] = None, cost_intercept: float = 1.0, num_trace_observations: int = 0, **ignore: Any, ) -> Dict[str, Any]: r"""Construct kwargs for a multifidetlity acquisition function's constructor.""" if fidelity_weights is None: fidelity_weights = {f: 1.0 for f in target_fidelities} if set(target_fidelities) != set(fidelity_weights): raise RuntimeError( "Must provide the same indices for target_fidelities " f"({set(target_fidelities)}) and fidelity_weights " f" ({set(fidelity_weights)})." ) cost_aware_utility = InverseCostWeightedUtility( cost_model=AffineFidelityCostModel( fidelity_weights=fidelity_weights, fixed_cost=cost_intercept ) ) return { "target_fidelities": target_fidelities, "cost_aware_utility": cost_aware_utility, "expand": lambda X: expand_trace_observations( X=X, fidelity_dims=sorted(target_fidelities), num_trace_obs=num_trace_observations, ), "project": lambda X: project_to_target_fidelity( X=X, target_fidelities=target_fidelities ), }
[docs]@acqf_input_constructor(qKnowledgeGradient) def construct_inputs_qKG( model: Model, training_data: TrainingData, bounds: List[Tuple[float, float]], objective: Optional[MCAcquisitionObjective] = None, posterior_transform: Optional[PosteriorTransform] = None, target_fidelities: Optional[Dict[int, float]] = None, num_fantasies: int = 64, **kwargs: Any, ) -> Dict[str, Any]: r"""Construct kwargs for `qKnowledgeGradient` constructor.""" inputs_mc = construct_inputs_mc_base( model=model, training_data=training_data, objective=objective, posterior_transform=posterior_transform, **kwargs, ) _bounds = torch.tensor( data=bounds, dtype=training_data.X.dtype, device=training_data.X.device, ) _, current_value = optimize_objective( model=model, bounds=_bounds.t(), q=1, target_fidelities=target_fidelities, objective=objective, posterior_transform=posterior_transform, **kwargs, ) return { **inputs_mc, "num_fantasies": num_fantasies, "current_value": current_value.detach().cpu().max(), }
[docs]@acqf_input_constructor(qMultiFidelityKnowledgeGradient) def construct_inputs_qMFKG( model: Model, training_data: TrainingData, bounds: List[Tuple[float, float]], target_fidelities: Dict[int, Union[int, float]], objective: Optional[MCAcquisitionObjective] = None, posterior_transform: Optional[PosteriorTransform] = None, **kwargs: Any, ) -> Dict[str, Any]: r"""Construct kwargs for `qMultiFidelityKnowledgeGradient` constructor.""" inputs_mf = construct_inputs_mf_base( model=model, training_data=training_data, target_fidelities=target_fidelities, **kwargs, ) inputs_kg = construct_inputs_qKG( model=model, training_data=training_data, bounds=bounds, objective=objective, posterior_transform=posterior_transform, **kwargs, ) return {**inputs_mf, **inputs_kg}
[docs]@acqf_input_constructor(qMultiFidelityMaxValueEntropy) def construct_inputs_qMFMES( model: Model, training_data: TrainingData, bounds: List[Tuple[float, float]], target_fidelities: Dict[int, Union[int, float]], objective: Optional[MCAcquisitionObjective] = None, posterior_transform: Optional[PosteriorTransform] = None, **kwargs: Any, ) -> Dict[str, Any]: r"""Construct kwargs for `qMultiFidelityMaxValueEntropy` constructor.""" inputs_mf = construct_inputs_mf_base( model=model, training_data=training_data, target_fidelities=target_fidelities, **kwargs, ) inputs_qmes = construct_inputs_qMES( model=model, training_data=training_data, bounds=bounds, objective=objective, posterior_transform=posterior_transform, **kwargs, ) _bounds = torch.tensor( data=bounds, dtype=training_data.X.dtype, device=training_data.X.device, ) _, current_value = optimize_objective( model=model, bounds=_bounds.t(), q=1, objective=objective, posterior_transform=posterior_transform, target_fidelities=target_fidelities, **kwargs, ) return { **inputs_mf, **inputs_qmes, "current_value": current_value.detach().cpu().max(), }
[docs]def get_best_f_analytic( training_data: TrainingData, posterior_transform: Optional[PosteriorTransform] = None, **kwargs, ) -> Tensor: if not training_data.is_block_design: raise NotImplementedError("Currently only block designs are supported.") Y = training_data.Y posterior_transform = _deprecate_objective_arg( posterior_transform=posterior_transform, objective=kwargs.get("objective", None) ) if posterior_transform is not None: return posterior_transform.evaluate(Y).max(-1).values if Y.shape[-1] > 1: raise NotImplementedError( "Analytic acquisition functions currently only work with " "multi-output models if provided with a `ScalarizedObjective`." ) return Y.max(-2).values.squeeze(-1)
[docs]def get_best_f_mc( training_data: TrainingData, objective: Optional[MCAcquisitionObjective] = None, posterior_transform: Optional[PosteriorTransform] = None, ) -> Tensor: if not training_data.is_block_design: raise NotImplementedError("Currently only block designs are supported.") Y = training_data.Y posterior_transform = _deprecate_objective_arg( posterior_transform=posterior_transform, objective=objective if not isinstance(objective, MCAcquisitionObjective) else None, ) if posterior_transform is not None: # retain the original tensor dimension since objective expects explicit # output dimension. Y_dim = Y.dim() Y = posterior_transform.evaluate(Y) if Y.dim() < Y_dim: Y = Y.unsqueeze(-1) if objective is None: if Y.shape[-1] > 1: raise UnsupportedError( "Acquisition functions require an objective when " "used with multi-output models (execpt for multi-objective" "acquisition functions)." ) objective = IdentityMCObjective() return objective(Y).max(-1).values
[docs]def optimize_objective( model: Model, bounds: Tensor, q: int, objective: Optional[MCAcquisitionObjective] = None, posterior_transform: Optional[PosteriorTransform] = None, linear_constraints: Optional[Tuple[Tensor, Tensor]] = None, fixed_features: Optional[Dict[int, float]] = None, target_fidelities: Optional[Dict[int, float]] = None, qmc: bool = True, mc_samples: int = 512, seed_inner: Optional[int] = None, optimizer_options: Dict[str, Any] = None, post_processing_func: Optional[Callable[[Tensor], Tensor]] = None, batch_initial_conditions: Optional[Tensor] = None, sequential: bool = False, **ignore, ) -> Tuple[Tensor, Tensor]: r"""Optimize an objective under the given model. Args: model: The model to be used in the objective. bounds: A `2 x d` tensor of lower and upper bounds for each column of `X`. q: The cardinality of input sets on which the objective is to be evaluated. objective: The objective to optimize. posterior_transform: The posterior transform to be used in the acquisition function. linear_constraints: A tuple of (A, b). Given `k` linear constraints on a `d`-dimensional space, `A` is `k x d` and `b` is `k x 1` such that `A x <= b`. (Not used by single task models). fixed_features: A dictionary of feature assignments `{feature_index: value}` to hold fixed during generation. target_fidelities: A dictionary mapping input feature indices to fidelity values. Defaults to `{-1: 1.0}`. qmc: Toggle for enabling (qmc=1) or disabling (qmc=0) use of Quasi Monte Carlo. mc_samples: Integer number of samples used to estimate Monte Carlo objectives. seed_inner: Integer seed used to initialize the sampler passed to MCObjective. optimizer_options: Table used to lookup keyword arguments for the optimizer. post_processing_func: A function that post-processes an optimization result appropriately (i.e. according to `round-trip` transformations). batch_initial_conditions: A Tensor of initial values for the optimizer. sequential: If False, uses joint optimization, otherwise uses sequential optimization. Returns: A tuple containing the best input locations and corresponding objective values. """ if optimizer_options is None: optimizer_options = {} if objective is not None: sampler_cls = SobolQMCNormalSampler if qmc else IIDNormalSampler acq_function = qSimpleRegret( model=model, objective=objective, posterior_transform=posterior_transform, sampler=sampler_cls(num_samples=mc_samples, seed=seed_inner), ) else: acq_function = PosteriorMean( model=model, posterior_transform=posterior_transform ) if fixed_features: acq_function = FixedFeatureAcquisitionFunction( acq_function=acq_function, d=bounds.shape[-1], columns=list(fixed_features.keys()), values=list(fixed_features.values()), ) free_feature_dims = list(range(len(bounds)) - fixed_features.keys()) free_feature_bounds = bounds[:, free_feature_dims] # (2, d' <= d) else: free_feature_bounds = bounds if linear_constraints is None: inequality_constraints = None else: A, b = linear_constraints inequality_constraints = [] k, d = A.shape for i in range(k): indicies = A[i, :].nonzero(as_tuple=False).squeeze() coefficients = -A[i, indicies] rhs = -b[i, 0] inequality_constraints.append((indicies, coefficients, rhs)) return optimize_acqf( acq_function=acq_function, bounds=free_feature_bounds, q=q, num_restarts=optimizer_options.get("num_restarts", 60), raw_samples=optimizer_options.get("raw_samples", 1024), options={ "batch_limit": optimizer_options.get("batch_limit", 8), "maxiter": optimizer_options.get("maxiter", 200), "nonnegative": optimizer_options.get("nonnegative", False), "method": optimizer_options.get("method", "L-BFGS-B"), }, inequality_constraints=inequality_constraints, fixed_features=None, # handled inside the acquisition function post_processing_func=post_processing_func, batch_initial_conditions=batch_initial_conditions, return_best_only=True, sequential=sequential, )