Cost cooling

neps.optimizers.bayesian_optimization.cost_cooling

CostCooling

CostCooling(
    pipeline_space: SearchSpace,
    initial_design_size: int = 10,
    surrogate_model: str | Any = "gp",
    cost_model: str | Any = "gp",
    surrogate_model_args: dict = None,
    cost_model_args: dict = None,
    optimal_assignment: bool = False,
    domain_se_kernel: str = None,
    graph_kernels: list = None,
    hp_kernels: list = None,
    acquisition: str | BaseAcquisition = "EI",
    log_prior_weighted: bool = False,
    acquisition_sampler: (
        str | AcquisitionSampler
    ) = "mutation",
    random_interleave_prob: float = 0.0,
    patience: int = 100,
    budget: None | int | float = None,
    ignore_errors: bool = False,
    loss_value_on_error: None | float = None,
    cost_value_on_error: None | float = None,
    logger=None,
)

Bases: BayesianOptimization

Implements a basic cost-cooling as described in "Cost-aware Bayesian Optimization" (arxiv.org/abs/2003.10870) by Lee et al.

PARAMETER	DESCRIPTION
pipeline_space	Space in which to search TYPE: SearchSpace
initial_design_size	Number of 'x' samples that need to be evaluated before selecting a sample using a strategy instead of randomly. TYPE: int DEFAULT: 10
surrogate_model	Surrogate model TYPE: str | Any DEFAULT: 'gp'
cost_model	Cost model TYPE: str | Any DEFAULT: 'gp'
surrogate_model_args	Arguments that will be given to the surrogate model (the Gaussian processes model). TYPE: dict DEFAULT: None
cost_model_args	Arguments that will be given to the cost model (the Gaussian processes model). TYPE: dict DEFAULT: None
optimal_assignment	whether the optimal assignment kernel should be used. TYPE: bool DEFAULT: False
domain_se_kernel	Stationary kernel name TYPE: str DEFAULT: None
graph_kernels	Kernels for NAS TYPE: list DEFAULT: None
hp_kernels	Kernels for HPO TYPE: list DEFAULT: None
acquisition	Acquisition strategy TYPE: str | BaseAcquisition DEFAULT: 'EI'
log_prior_weighted	if to use log for prior TYPE: bool DEFAULT: False
acquisition_sampler	Acquisition function fetching strategy TYPE: str | AcquisitionSampler DEFAULT: 'mutation'
random_interleave_prob	Frequency at which random configurations are sampled instead of configurations from the acquisition strategy. TYPE: float DEFAULT: 0.0
patience	How many times we try something that fails before giving up. TYPE: int DEFAULT: 100
budget	Maximum budget TYPE: None | int | float DEFAULT: None
ignore_errors	Ignore hyperparameter settings that threw an error and do not raise an error. Error configs still count towards max_evaluations_total. TYPE: bool DEFAULT: False
loss_value_on_error	Setting this and cost_value_on_error to any float will supress any error during bayesian optimization and will use given loss value instead. default: None TYPE: None | float DEFAULT: None
cost_value_on_error	Setting this and loss_value_on_error to any float will supress any error during bayesian optimization and will use given cost value instead. default: None TYPE: None | float DEFAULT: None
logger	logger object, or None to use the neps logger DEFAULT: None

RAISES	DESCRIPTION
ValueError	if patience < 1
ValueError	if initial_design_size < 1
ValueError	if random_interleave_prob is not between 0.0 and 1.0
ValueError	if no kernel is provided

Source code in neps/optimizers/bayesian_optimization/cost_cooling.py

def __init__(
    self,
    pipeline_space: SearchSpace,
    initial_design_size: int = 10,
    surrogate_model: str | Any = "gp",
    cost_model: str | Any = "gp",
    surrogate_model_args: dict = None,
    cost_model_args: dict = None,
    optimal_assignment: bool = False,
    domain_se_kernel: str = None,
    graph_kernels: list = None,
    hp_kernels: list = None,
    acquisition: str | BaseAcquisition = "EI",
    log_prior_weighted: bool = False,
    acquisition_sampler: str | AcquisitionSampler = "mutation",
    random_interleave_prob: float = 0.0,
    patience: int = 100,
    budget: None | int | float = None,
    ignore_errors: bool = False,
    loss_value_on_error: None | float = None,
    cost_value_on_error: None | float = None,
    logger=None,
):
    """Initialise the BO loop.

    Args:
        pipeline_space: Space in which to search
        initial_design_size: Number of 'x' samples that need to be evaluated before
            selecting a sample using a strategy instead of randomly.
        surrogate_model: Surrogate model
        cost_model: Cost model
        surrogate_model_args: Arguments that will be given to the surrogate model
            (the Gaussian processes model).
        cost_model_args: Arguments that will be given to the cost model
            (the Gaussian processes model).
        optimal_assignment: whether the optimal assignment kernel should be used.
        domain_se_kernel: Stationary kernel name
        graph_kernels: Kernels for NAS
        hp_kernels: Kernels for HPO
        acquisition: Acquisition strategy
        log_prior_weighted: if to use log for prior
        acquisition_sampler: Acquisition function fetching strategy
        random_interleave_prob: Frequency at which random configurations are sampled
            instead of configurations from the acquisition strategy.
        patience: How many times we try something that fails before giving up.
        budget: Maximum budget
        ignore_errors: Ignore hyperparameter settings that threw an error and do not
            raise an error. Error configs still count towards max_evaluations_total.
        loss_value_on_error: Setting this and cost_value_on_error to any float will
            supress any error during bayesian optimization and will use given loss
            value instead. default: None
        cost_value_on_error: Setting this and loss_value_on_error to any float will
            supress any error during bayesian optimization and will use given cost
            value instead. default: None
        logger: logger object, or None to use the neps logger

    Raises:
        ValueError: if patience < 1
        ValueError: if initial_design_size < 1
        ValueError: if random_interleave_prob is not between 0.0 and 1.0
        ValueError: if no kernel is provided
    """
    super().__init__(
        pipeline_space=pipeline_space,
        patience=patience,
        logger=logger,
        budget=budget,
        ignore_errors=ignore_errors,
        loss_value_on_error=loss_value_on_error,
        cost_value_on_error=cost_value_on_error,
    )

    if initial_design_size < 1:
        raise ValueError(
            "BayesianOptimization needs initial_design_size to be at least 1"
        )
    if not 0 <= random_interleave_prob <= 1:
        raise ValueError("random_interleave_prob should be between 0.0 and 1.0")

    self._initial_design_size = initial_design_size
    self._random_interleave_prob = random_interleave_prob
    self._num_train_x: int = 0
    self._pending_evaluations: list = []
    self._model_update_failed: bool = False

    if ignore_errors:
        self.logger.warning(
            "ignore_errors was set, but this optimizer does not support it"
        )

    surrogate_model_args = surrogate_model_args or {}
    cost_model_args = cost_model_args or {}
    graph_kernels, hp_kernels = get_kernels(
        self.pipeline_space,
        domain_se_kernel,
        graph_kernels,
        hp_kernels,
        optimal_assignment,
    )
    if "graph_kernels" not in surrogate_model_args:
        surrogate_model_args["graph_kernels"] = graph_kernels
    if "hp_kernels" not in surrogate_model_args:
        surrogate_model_args["hp_kernels"] = hp_kernels

    if (
        not surrogate_model_args["graph_kernels"]
        and not surrogate_model_args["hp_kernels"]
    ):
        raise ValueError("No kernels are provided!")

    if "vectorial_features" not in surrogate_model_args:
        surrogate_model_args[
            "vectorial_features"
        ] = self.pipeline_space.get_vectorial_dim()

    self.surrogate_model = instance_from_map(
        SurrogateModelMapping,
        surrogate_model,
        name="surrogate model",
        kwargs=surrogate_model_args,
    )

    if "graph_kernels" not in cost_model_args:
        cost_model_args["graph_kernels"] = graph_kernels
    if "hp_kernels" not in cost_model_args:
        cost_model_args["hp_kernels"] = hp_kernels

    if not cost_model_args["graph_kernels"] and not cost_model_args["hp_kernels"]:
        raise ValueError("No kernels are provided!")

    if "vectorial_features" not in cost_model_args:
        cost_model_args[
            "vectorial_features"
        ] = self.pipeline_space.get_vectorial_dim()

    self.cost_model = instance_from_map(
        SurrogateModelMapping,
        cost_model,
        name="cost model",  # does changing this string work?
        kwargs=cost_model_args,
    )

    orig_acquisition = instance_from_map(
        AcquisitionMapping,
        acquisition,
        name="acquisition function",
    )

    self.acquisition = CostCooler(orig_acquisition)

    if self.pipeline_space.has_prior:
        self.acquisition = DecayingPriorWeightedAcquisition(
            self.acquisition, log=log_prior_weighted
        )

    self.acquisition_sampler = instance_from_map(
        AcquisitionSamplerMapping,
        acquisition_sampler,
        name="acquisition sampler function",
        kwargs={"patience": self.patience, "pipeline_space": self.pipeline_space},
    )

get_cost

get_cost(result: ERROR | ResultDict | float) -> float | Any

Calls result.utils.get_cost() and passes the error handling through. Please use self.get_cost() instead of get_cost() in all optimizer classes.

Source code in neps/optimizers/base_optimizer.py

def get_cost(self, result: ERROR | ResultDict | float) -> float | Any:
    """Calls result.utils.get_cost() and passes the error handling through.
    Please use self.get_cost() instead of get_cost() in all optimizer classes."""
    return _get_cost(
        result,
        cost_value_on_error=self.cost_value_on_error,
        ignore_errors=self.ignore_errors,
    )

get_learning_curve

get_learning_curve(
    result: str | dict | float,
) -> float | Any

Calls result.utils.get_loss() and passes the error handling through. Please use self.get_loss() instead of get_loss() in all optimizer classes.

Source code in neps/optimizers/base_optimizer.py

def get_learning_curve(self, result: str | dict | float) -> float | Any:
    """Calls result.utils.get_loss() and passes the error handling through.
    Please use self.get_loss() instead of get_loss() in all optimizer classes."""
    return _get_learning_curve(
        result,
        learning_curve_on_error=self.learning_curve_on_error,
        ignore_errors=self.ignore_errors,
    )

get_loss

get_loss(result: ERROR | ResultDict | float) -> float | Any

Calls result.utils.get_loss() and passes the error handling through. Please use self.get_loss() instead of get_loss() in all optimizer classes.

Source code in neps/optimizers/base_optimizer.py

def get_loss(self, result: ERROR | ResultDict | float) -> float | Any:
    """Calls result.utils.get_loss() and passes the error handling through.
    Please use self.get_loss() instead of get_loss() in all optimizer classes."""
    return _get_loss(
        result,
        loss_value_on_error=self.loss_value_on_error,
        ignore_errors=self.ignore_errors,
    )

is_init_phase

is_init_phase() -> bool

Decides if optimization is still under the warmstart phase/model-based search.

Source code in neps/optimizers/bayesian_optimization/optimizer.py

def is_init_phase(self) -> bool:
    """Decides if optimization is still under the warmstart phase/model-based search."""
    if self._num_train_x >= self._initial_design_size:
        return False
    return True

is_out_of_budget

is_out_of_budget() -> bool

Check if the optimizer has used all of its budget, if any.

Source code in neps/optimizers/base_optimizer.py

def is_out_of_budget(self) -> bool:
    """Check if the optimizer has used all of its budget, if any."""
    return self.budget is not None and self.used_budget >= self.budget