Source code for ax.service.utils.early_stopping

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

# pyre-strict

from typing import Dict, List, Optional, Set

from ax.core.experiment import Experiment
from ax.early_stopping.strategies import BaseEarlyStoppingStrategy
from ax.utils.common.typeutils import not_none

[docs]def should_stop_trials_early( early_stopping_strategy: Optional[BaseEarlyStoppingStrategy], trial_indices: Set[int], experiment: Experiment, ) -> Dict[int, Optional[str]]: """Evaluate whether to early-stop running trials. Args: early_stopping_strategy: A ``BaseEarlyStoppingStrategy`` that determines whether a trial should be stopped given the state of an experiment. trial_indices: Indices of trials to consider for early stopping. experiment: The experiment containing the trials. Returns: A dictionary mapping trial indices that should be early stopped to (optional) messages with the associated reason. """ if early_stopping_strategy is None: return {} early_stopping_strategy = not_none(early_stopping_strategy) return early_stopping_strategy.should_stop_trials_early( trial_indices=trial_indices, experiment=experiment )
[docs]def get_early_stopping_metrics( experiment: Experiment, early_stopping_strategy: Optional[BaseEarlyStoppingStrategy] ) -> List[str]: """A helper function that returns a list of metric names on which a given `early_stopping_strategy` is operating.""" if early_stopping_strategy is None: return [] if early_stopping_strategy.metric_names is not None: return list(early_stopping_strategy.metric_names) # TODO: generalize this to multi-objective ess default_objective, _ = early_stopping_strategy._default_objective_and_direction( experiment=experiment ) return [default_objective]