Skip to content

HPO Optimizers API

GAM Optimizer

GAMOptimizer

GAMOptimizer(
    objective_function: Callable[[dict], float],
    search_space: SearchSpace,
    settings: GAMOptSettings,
    known_observations: list[dict] | None = None,
)

Bases: BaseOptimizer

Optimizer based on Generalized Additive Models. Trained model is used to suggest next node in the search space for evaluation.

Parameters:

  • objective_function (Callable[[dict], float]) –

    Target function that will be used in every evaluation. Output of this function should be 'float', as this is the value for which algorithms try to optimize solution. Function should take dict filled with 'key: value' pairs that are 'argument: corresponding value'.

  • search_space (SearchSpace) –

    Instance containing information about nodes in the solutions space that will be evaluated during the optimization.

  • settings (GAMOptSettings) –

    Instance with settings required for configuring the optimization process.

Attributes:

  • evaluations (list[dict]) –

    Already evaluated hyperparameters combinations with corresponding score.

  • max_iterations (int) –

    Validated maximum number of iterations during HPO.

Source code in ai4rag/core/hpo/gam_opt.py
def __init__(
    self,
    objective_function: Callable[[dict], float],
    search_space: SearchSpace,
    settings: GAMOptSettings,
    known_observations: list[dict] | None = None,
):
    super().__init__(objective_function, search_space, settings)
    self.settings = settings
    self.evaluations = []
    self._evaluated_combinations = []
    self._encoders_with_columns: list[tuple[str, LabelEncoder]] = []

    if known_observations:
        self._load_known_observations(known_observations)

    self.max_iterations = self.settings.max_evals

Attributes

max_iterations property writable

max_iterations: int

Get max possible number of iterations for the HPO.

Functions

search

search() -> dict[str, Any]

Actual function performing hyperparameter optimization for the selected objective function.

Returns:

  • dict[str, Any]

    The best set of parameters with achieved score.

Raises:

  • OptimizationError

    When there were no successful evaluations for given constraints.

Source code in ai4rag/core/hpo/gam_opt.py
def search(self) -> dict[str, Any]:
    """
    Actual function performing hyperparameter optimization for the selected
    objective function.

    Returns
    -------
    dict[str, Any]
        The best set of parameters with achieved score.

    Raises
    ------
    OptimizationError
        When there were no successful evaluations for given constraints.
    """
    self.evaluate_initial_random_nodes()

    iterations_limit = self._get_iterations_limit()

    for _ in range(iterations_limit):
        self._run_iteration()

    successful_evaluations = [evaluation for evaluation in self.evaluations if evaluation["score"] is not None]
    if not successful_evaluations:
        raise OptimizationError("Number of evaluations has reached limit. All iterations have failed.")

    # Sort in ascending order and take the last element (highest score).
    # This assumes we're maximizing the score.
    best_config_with_score = sorted(successful_evaluations, key=lambda d: d["score"])[-1]

    return best_config_with_score

evaluate_initial_random_nodes

evaluate_initial_random_nodes() -> None

Perform evaluation of randomly chosen n nodes from the solutions space. Evaluations are performed until desired number of successful evaluations is reached or maximum number of evaluations is reached.

When the optimizer has been warm-started with known observations, already-successful evaluations count toward the n_random_nodes target and already-evaluated combinations are excluded from candidates.

Source code in ai4rag/core/hpo/gam_opt.py
def evaluate_initial_random_nodes(self) -> None:
    """
    Perform evaluation of randomly chosen n nodes from the solutions space.
    Evaluations are performed until desired number of successful evaluations
    is reached or maximum number of evaluations is reached.

    When the optimizer has been warm-started with known observations,
    already-successful evaluations count toward the n_random_nodes target
    and already-evaluated combinations are excluded from candidates.
    """
    successful_evaluations = sum(1 for e in self.evaluations if e["score"] is not None)

    if successful_evaluations >= self.settings.n_random_nodes:
        logger.info(
            "Skipping random evaluation phase: %d known successful evaluations >= n_random_nodes (%d).",
            successful_evaluations,
            self.settings.n_random_nodes,
        )
        return

    if len(self.evaluations) >= self.max_iterations:
        return

    combinations_local = [c for c in copy(self._search_space.combinations) if c not in self._evaluated_combinations]
    random.shuffle(combinations_local)

    gen = (x for x in combinations_local)

    while successful_evaluations < self.settings.n_random_nodes:
        params = next(gen)
        score = self._objective_function(params=params)
        if score is not None:
            successful_evaluations += 1
        self._evaluated_combinations.append(params)
        params_with_score = params | {"score": score}
        self.evaluations.append(params_with_score)

        if len(self.evaluations) == self.max_iterations:
            break

GAMOptSettings dataclass

GAMOptSettings(max_evals: int, n_random_nodes: int = 4, evals_per_trial: int = 1, random_state: int = 64)

Bases: OptimizerSettings

Settings for the GAMOptimizer. For the detailed description of parameters for Generalized Additive Models, please see pygam documentation.

Random Optimizer

RandomOptimizer

RandomOptimizer(objective_function: Callable[[dict], float], search_space: SearchSpace, settings: RandomOptSettings)

Bases: BaseOptimizer

Optimizer running random search on the given search space.

Source code in ai4rag/core/hpo/random_opt.py
def __init__(
    self, objective_function: Callable[[dict], float], search_space: SearchSpace, settings: RandomOptSettings
):
    super().__init__(objective_function, search_space, settings)
    self._evaluated_combinations = []

Functions

search

search() -> dict[str, Any]

Actual function performing hyperparameter optimization for the selected objective function.

Returns:

  • dict[str, Any]

    The best set of parameters with achieved score.

Raises:

  • OptimizationError

    When there were no successful evaluations for given constraints.

Source code in ai4rag/core/hpo/random_opt.py
def search(self) -> dict[str, Any]:
    """
    Actual function performing hyperparameter optimization for the selected
    objective function.

    Returns
    -------
    dict[str, Any]
        The best set of parameters with achieved score.

    Raises
    ------
    OptimizationError
        When there were no successful evaluations for given constraints.
    """
    combinations = self._search_space.combinations
    random.shuffle(combinations)

    for idx in range(self.settings.max_evals):
        score = self._objective_function(combinations[idx])
        self._evaluated_combinations.append(combinations[idx] | {"score": score})

    successful_evaluations = [ev for ev in self._evaluated_combinations if ev["score"] is not None]

    if not successful_evaluations:
        raise OptimizationError("Number of evaluations has reached limit. All iterations have failed.")

    best_config_with_score = sorted(successful_evaluations, key=lambda d: d["score"])[-1]

    return best_config_with_score

RandomOptSettings dataclass

RandomOptSettings(max_evals: int)

Bases: OptimizerSettings

Settings for random optimizer.