diff --git a/ai4rag/core/experiment/experiment.py b/ai4rag/core/experiment/experiment.py index 64613a75..20b9c9e3 100644 --- a/ai4rag/core/experiment/experiment.py +++ b/ai4rag/core/experiment/experiment.py @@ -155,6 +155,8 @@ def __init__( self.inference_max_threads: int = kwargs.pop("inference_max_threads", 10) self.results: ExperimentResults = ExperimentResults() + self.optimizer: BaseOptimizer | None = None + self._optimization_patterns: list[dict[str, Any]] = [] self._exception_handler = ExperimentExceptionHandler(self.event_handler) if kwargs: @@ -685,6 +687,9 @@ def search(self, **kwargs) -> None: """ logger.info("Starting RAG optimization process...") + # GAM patterns are buffered until the optimizer has completed, so clear + # results from a previous invocation on this experiment instance. + self._optimization_patterns = [] def objective_function(space: RAGParamsType) -> float | None: """Function passed to the optimizer.""" @@ -731,17 +736,62 @@ def objective_function(space: RAGParamsType) -> float | None: self.optimizer_settings.to_dict(), ) + self.optimizer = optimizer try: _ = optimizer.search() except OptimizationError as err: final_error_msg = self._exception_handler.get_final_error_msg() raise RAGExperimentError(final_error_msg) from err + self._publish_optimization_patterns() + self.event_handler.on_status_change( level=LogLevel.INFO, message="Experiment optimization process finished.", ) + def _select_optimization_patterns(self) -> list[dict[str, Any]]: + """Select and renumber the configured number of GAM output patterns.""" + patterns = self._optimization_patterns + if not isinstance(self.optimizer, GAMOptimizer) or not patterns: + return patterns + + output_limit = self.optimizer.max_iterations + warm_start_output_count = min( + output_limit, + self.optimizer.compute_warm_start_effective_target() // 4, + ) + warm_start_patterns = [p for p in patterns if p.get("optimization_phase") == "warm_start"] + gam_patterns = [p for p in patterns if p.get("optimization_phase") == "gam"] + selected = warm_start_patterns[:warm_start_output_count] + selected.extend(gam_patterns[: output_limit - len(selected)]) + + for index, pattern in enumerate(selected, start=1): + payload = pattern.get("payload") + if isinstance(payload, dict): + payload["name"] = f"Pattern{index}" + payload["iteration"] = index - 1 + + logger.info( + "Selected %d output patterns: %d from warm start and %d from GAM.", + len(selected), + min(len(warm_start_patterns), warm_start_output_count), + min(len(gam_patterns), output_limit - len(warm_start_patterns[:warm_start_output_count])), + ) + return selected + + def _publish_optimization_patterns(self) -> None: + """Publish GAM patterns only after their final output selection is known.""" + for pattern in self._select_optimization_patterns(): + payload = pattern["payload"] + evaluation_results = pattern["evaluation_results"] + metadata = {key: value for key, value in pattern.items() if key not in {"payload", "evaluation_results"}} + self.event_handler.on_pattern_creation( + payload=payload, + evaluation_results=evaluation_results, + **metadata, + ) + def _stream_finished_pattern( self, evaluation_result: EvaluationResult, @@ -823,10 +873,15 @@ def _stream_finished_pattern( "iteration": len(self.results) + n_known, } - self.event_handler.on_pattern_creation( - payload=payload, - evaluation_results=evaluation_results_json, - ) + pattern = { + "payload": payload, + "evaluation_results": evaluation_results_json, + "optimization_phase": getattr(self.optimizer, "current_phase", None), + } + if isinstance(self.optimizer, GAMOptimizer): + self._optimization_patterns.append(pattern) + else: + self.event_handler.on_pattern_creation(**pattern) def _evaluate_response( self, diff --git a/ai4rag/core/hpo/gam_opt.py b/ai4rag/core/hpo/gam_opt.py index 99e35f4b..d31f2b4d 100644 --- a/ai4rag/core/hpo/gam_opt.py +++ b/ai4rag/core/hpo/gam_opt.py @@ -3,14 +3,17 @@ # SPDX-License-Identifier: Apache-2.0 # ----------------------------------------------------------------------------- import random +from collections import defaultdict, deque from copy import copy from dataclasses import dataclass from math import ceil -from typing import Any, Callable +from typing import Any, Callable, Literal import numpy as np import pandas as pd from pygam import LinearGAM +from pygam import f as gam_f +from pygam import s as gam_s from sklearn.preprocessing import LabelEncoder from ai4rag import logger @@ -20,6 +23,76 @@ __all__ = ["GAMOptSettings", "GAMOptimizer"] +def _serialize_dict_col(series: pd.Series) -> pd.Series: + """Serialize model-valued cells to their model_id string. + + Handles both dict-valued models ({"model_id": "..."}, production) and + model object instances with a model_id attribute (tests / direct API use). + """ + + def _needs_serialization(x: object) -> bool: + return isinstance(x, dict) or hasattr(x, "model_id") + + def _to_model_id(x: object) -> object: + if isinstance(x, dict): + return x.get("model_id", str(x)) + if hasattr(x, "model_id"): + return x.model_id + return x + + if series.apply(_needs_serialization).any(): + return series.apply(_to_model_id) + return series + + +def _round_robin(combinations: list[dict], key_fn: Callable[[dict], Any]) -> list[dict]: + """Re-order combinations by round-robin across buckets determined by key_fn.""" + buckets: dict[Any, deque] = defaultdict(deque) + for c in combinations: + buckets[key_fn(c)].append(c) + bucket_list = list(buckets.values()) + balanced: list[dict] = [] + i = 0 + while bucket_list: + idx = i % len(bucket_list) + bucket = bucket_list[idx] + if bucket: + balanced.append(bucket.popleft()) + i += 1 + else: + bucket_list.pop(idx) + return balanced + + +def _str_val(v: object) -> str: + """Normalize any cell value to a string key (handles model objects and plain strings).""" + if v is None: + return "__none__" + if isinstance(v, dict): + return v.get("model_id", str(v)) + if hasattr(v, "model_id"): + return v.model_id + return str(v) + + +def _get_discrete_column_values(combinations: list[dict]) -> dict[str, set[str]]: + """Return {col: set_of_str_values} for all discrete columns in the combinations. + + All parameter types are included — strings, model objects, numeric values — so + that coverage tracking works for columns like chunk_size and chunk_overlap as + well as string columns like chunking_method or search_mode. + """ + if not combinations: + return {} + result: dict[str, set[str]] = {} + for col in combinations[0]: + sample = next((c.get(col) for c in combinations if c.get(col) is not None), None) + if sample is None: + continue + result[col] = {_str_val(c.get(col)) for c in combinations} + return result + + @dataclass class GAMOptSettings(OptimizerSettings): """ @@ -29,27 +102,63 @@ class GAMOptSettings(OptimizerSettings): Parameters ---------- - max_evals : int - Maximum number of evaluations performed during optimization process. + max_evals : int | None, default=None + Maximum number of objective-function evaluations performed during + optimization, including warm-start and GAM evaluations. When omitted, + every available search-space combination is evaluated. + max_iterations : int | None, default=None + Maximum number of evaluated RAG patterns retained and published when the + search completes. It controls the warm-start/GAM output allocation, not + the hard evaluation budget; use ``max_evals`` to bound objective-function + calls. When omitted, it is set to the effective ``max_evals`` value. It + cannot exceed an explicitly configured ``max_evals``. n_random_nodes : int, default=4 Number of random configurations to evaluate before starting GAM iterations. - The initial sample is stratified: for every string-valued categorical - parameter (e.g. ``search_mode``, ``chunking_method``, ``ranker_strategy``), - at least one configuration for each unique value is guaranteed to appear - before the random fill, regardless of raw search space imbalance. - Integer/float parameters are not stratified. Set this to at least the - number of unique values of the most varied string parameter to guarantee - full categorical coverage; a warning is emitted when the value is too small. evals_per_trial : int, default=1 Number of configurations to evaluate per GAM iteration. + warm_start_strategy : {"random", "greedy", "balanced"}, default="random" + Controls how the initial n_random_nodes observations are selected/ordered. + "random" — shuffle the candidate list and take the first n as-is. + "greedy" — greedily pick combinations so every discrete column value + appears at least twice. If n_random_nodes is below the + computed minimum (min_required), the warm start is + auto-adjusted upward to meet coverage. One output slot is + allocated per four effective warm-start evaluations; GAM + fills the remaining ``max_iterations`` slots. + "balanced" — round-robin across the tuple of fields_to_balance values; + non-balanced discrete column values each appear at least once. + Requires fields_to_balance to be set. Same auto-adjustment, + and output allocation rules as "greedy". + fields_to_balance : list[str] | None, default=None + Field names to balance by round-robin when warm_start_strategy="balanced". + Each unique value combination of these fields is guaranteed to appear at + least once in the first n_random_nodes evaluations. random_state : int, default=64 Inherited from OptimizerSettings. Controls shuffle order of initial random exploration phase. Does NOT control GAM model randomness (GAM training is deterministic). """ + max_evals: int | None = None + max_iterations: int | None = None n_random_nodes: int = 4 evals_per_trial: int = 1 + warm_start_strategy: Literal["random", "greedy", "balanced"] = "random" + fields_to_balance: list[str] | None = None + + def __post_init__(self) -> None: + valid = {"random", "greedy", "balanced"} + if self.warm_start_strategy not in valid: + raise ValueError( + f"warm_start_strategy must be one of {sorted(valid)}; " f"got {self.warm_start_strategy!r}." + ) + if self.warm_start_strategy == "balanced" and not self.fields_to_balance: + raise ValueError("fields_to_balance must be a non-empty list when warm_start_strategy='balanced'.") + if self.max_iterations is not None: + if self.max_iterations < 1: + raise ValueError("max_iterations must be at least 1 when provided.") + if self.max_evals is not None and self.max_iterations > self.max_evals: + raise ValueError("max_iterations cannot exceed max_evals.") class GAMOptimizer(BaseOptimizer): @@ -79,7 +188,9 @@ class GAMOptimizer(BaseOptimizer): Already evaluated hyperparameters combinations with corresponding score. max_iterations : int - Validated maximum number of iterations during HPO. + Effective maximum number of evaluated patterns retained as search results + and published to the event handler. This is bounded by ``max_evals`` and + the number of available search-space combinations. """ def __init__( @@ -93,32 +204,43 @@ def __init__( self.settings = settings self.evaluations = [] self._evaluated_combinations = [] - self._encoders_with_columns: list[tuple[str, LabelEncoder]] = [] + self._typed_encoders_with_columns: list[tuple[str, LabelEncoder]] = [] + self.warm_start_evaluation_count: int = 0 + self.current_phase = "idle" if known_observations: self._load_known_observations(known_observations) - self.max_iterations = self.settings.max_evals + self._validate_fields_to_balance() + self._validate_n_random_nodes() + + self.max_evals = ( + self._search_space.max_combinations + if self.settings.max_evals is None + else min(self.settings.max_evals, self._search_space.max_combinations) + ) + self.max_iterations = self.settings.max_iterations @property def max_iterations(self) -> int: - """Get max possible number of iterations for the HPO.""" + """Get the effective maximum number of retained result patterns.""" return self._max_iterations @max_iterations.setter - def max_iterations(self, val: int) -> None: - """Set maximum number of iterations that should be performed during HPO.""" + def max_iterations(self, val: int | None) -> None: + """Set maximum number of result patterns retained after HPO.""" max_comb = self._search_space.max_combinations - if val > max_comb: + if val is None: + self._max_iterations = self.max_evals + return + max_results = min(self.max_evals, max_comb) + if val > max_results: logger.info( - ( - "'max_number_of_rag_patterns' exceeded number of possible combinations: %s. " - "Setting 'max_number_of_rag_patterns' to: %s" - ), - max_comb, - max_comb, + "'max_iterations' exceeded the available evaluation budget: %s. Setting 'max_iterations' to: %s", + max_results, + max_results, ) - self._max_iterations = max_comb + self._max_iterations = max_results else: self._max_iterations = val @@ -137,12 +259,33 @@ def search(self) -> dict[str, Any]: OptimizationError When there were no successful evaluations for given constraints. """ + self.current_phase = "warm_start" self.evaluate_initial_random_nodes() - iterations_limit = self._get_iterations_limit() + strategy = self.settings.warm_start_strategy + self.current_phase = "gam" + if len(self.evaluations) >= self.max_evals: + logger.info( + "All %d allowed evaluations were consumed by the warm-start phase; GAM iterations will be skipped.", + self.max_evals, + ) + if strategy in ("greedy", "balanced"): + effective_warm_start = self._compute_warm_start_effective_target() + output_limit = self.max_iterations + gam_output_count = max(0, output_limit - (effective_warm_start // 4)) + gam_iterations = ceil(gam_output_count / self.settings.evals_per_trial) + remaining_evaluation_capacity = max(0, self.max_evals - len(self.evaluations)) + capacity_iterations = ceil(remaining_evaluation_capacity / self.settings.evals_per_trial) + for _ in range(min(gam_iterations, capacity_iterations)): + self._run_iteration() + else: + iterations_limit = self._get_iterations_limit() + for _ in range(iterations_limit): + self._run_iteration() + + self._trim_evaluations_to_top(self.max_iterations) - for _ in range(iterations_limit): - self._run_iteration() + self.current_phase = "complete" successful_evaluations = [evaluation for evaluation in self.evaluations if evaluation["score"] is not None] if not successful_evaluations: @@ -159,8 +302,125 @@ def _get_iterations_limit(self) -> int: Calculate maximum number of iterations that can be proceeded based on the already evaluated random nodes and settings for the optimizer. """ - iterations_limit = ceil((self.max_iterations - len(self.evaluations)) / self.settings.evals_per_trial) - return iterations_limit + iterations_limit = ceil((self.max_evals - len(self.evaluations)) / self.settings.evals_per_trial) + return max(0, iterations_limit) + + def _validate_n_random_nodes(self) -> None: + """Log a warning when n_random_nodes is below the required minimum for the strategy. + + - random: No minimum enforced — combinations are taken in shuffle order. + - greedy: If n_random_nodes < 2 * max_unique_values_per_column, warm start + is auto-adjusted to the minimum required (no error raised). + - balanced: If n_random_nodes < max(n_balanced_tuples, max_non_balanced_unique), + warm start is auto-adjusted to the minimum required (no error raised). + """ + combinations = self._search_space.combinations + if not combinations: + return + + strategy = self.settings.warm_start_strategy + + if strategy == "random": + return + + str_cols = _get_discrete_column_values(combinations) + + # Already-successful known_observations count toward the coverage budget. + successful_known = sum(1 for e in self.evaluations if e.get("score") is not None) + # effective_budget: the larger of n_random_nodes and what known_observations + # already provide — if known obs alone meet the minimum, no raise is needed. + effective_budget = max(self.settings.n_random_nodes, successful_known) + + if strategy == "greedy": + if not str_cols: + return + max_unique = max(len(vals) for vals in str_cols.values()) + min_required = max(4, 2 * max_unique) + if effective_budget < min_required: + logger.info( + "n_random_nodes=%d is below the minimum required %d for " + "warm_start_strategy='greedy' (max unique values per column: %d). " + "Warm start will be auto-adjusted to %d nodes.", + self.settings.n_random_nodes, + min_required, + max_unique, + min_required, + ) + + elif strategy == "balanced": + fields_to_balance = self.settings.fields_to_balance or [] + balanced_tuples = {tuple(_str_val(c.get(f)) for f in fields_to_balance) for c in combinations} + n_balanced = len(balanced_tuples) + non_balanced = {col: vals for col, vals in str_cols.items() if col not in fields_to_balance} + max_non_balanced = max((len(vals) for vals in non_balanced.values()), default=0) + min_required = max(4, n_balanced, max_non_balanced) + if effective_budget < min_required: + logger.info( + "n_random_nodes=%d is below the minimum required %d for " + "warm_start_strategy='balanced' with fields_to_balance=%r " + "(n_balanced_tuples=%d, max_non_balanced_unique=%d). " + "Warm start will be auto-adjusted to %d nodes.", + self.settings.n_random_nodes, + min_required, + fields_to_balance, + n_balanced, + max_non_balanced, + min_required, + ) + + def _validate_fields_to_balance(self) -> None: + """Reject balanced warm-start fields that are absent from the search space.""" + if self.settings.warm_start_strategy != "balanced": + return + + combinations = self._search_space.combinations + if not combinations: + return + + available_fields = set(combinations[0]) + unknown_fields = set(self.settings.fields_to_balance or []) - available_fields + if unknown_fields: + raise ValueError( + "fields_to_balance contains field(s) absent from the search space: " + f"{sorted(unknown_fields)}. Available fields: {sorted(available_fields)}." + ) + + def _compute_warm_start_effective_target(self) -> int: + """Return the effective number of successful warm-start nodes to evaluate. + + For "greedy" and "balanced" strategies, this is + max(n_random_nodes, min_required) where min_required guarantees adequate + discrete-value coverage. For "random", returns n_random_nodes unchanged. + """ + n = self.settings.n_random_nodes + strategy = self.settings.warm_start_strategy + if strategy == "random": + return n + combinations = self._search_space.combinations + if not combinations: + return n + str_cols = _get_discrete_column_values(combinations) + if strategy == "greedy": + if not str_cols: + return max(4, n) + max_unique = max(len(vals) for vals in str_cols.values()) + return max(n, 4, 2 * max_unique) + # balanced + fields = self.settings.fields_to_balance or [] + balanced_tuples = {tuple(_str_val(c.get(f)) for f in fields) for c in combinations} + non_balanced = {col: vals for col, vals in str_cols.items() if col not in fields} + max_non_balanced = max((len(vals) for vals in non_balanced.values()), default=0) + return max(n, 4, len(balanced_tuples), max_non_balanced) + + def compute_warm_start_effective_target(self) -> int: + """Return the effective number of successful warm-start nodes to evaluate. + + Returns + ------- + int + The effective warm-start target for the configured strategy. + """ + return self._compute_warm_start_effective_target() def _load_known_observations(self, known_observations: list[dict]) -> None: """ @@ -190,238 +450,351 @@ def _load_known_observations(self, known_observations: list[dict]) -> None: 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. + Perform evaluation of randomly chosen nodes from the solutions space. + All strategies stop at the configured maximum evaluation count. Greedy + and balanced starts may therefore finish before their coverage target + when the evaluation budget is smaller than the coverage requirement. 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. - The selection is stratified: combinations that introduce at least one new - unique value for any categorical (string-valued) parameter are moved to the - front of the queue before the random fill. This guarantees that every - distinct categorical value (e.g. ``search_mode="vector"`` vs - ``search_mode="hybrid"``) is evaluated at least once before GAM training - begins, regardless of how skewed the raw search space is. - - A warning is logged when ``n_random_nodes`` is smaller than the estimated - minimum required to guarantee full categorical coverage. + The selection order depends on warm_start_strategy: + "random" — shuffled order (no reordering). + "greedy" — greedy selection maximizing string-column coverage (each value >= 2 times). + "balanced" — round-robin across fields_to_balance value tuples. """ successful_evaluations = sum(1 for e in self.evaluations if e["score"] is not None) + effective_target = self._compute_warm_start_effective_target() - if successful_evaluations >= self.settings.n_random_nodes: + if successful_evaluations >= effective_target: logger.info( - "Skipping random evaluation phase: %d known successful evaluations >= n_random_nodes (%d).", + "Skipping random evaluation phase: %d known successful evaluations >= warm_start_target (%d).", successful_evaluations, - self.settings.n_random_nodes, + effective_target, ) return - if len(self.evaluations) >= self.max_iterations: + if len(self.evaluations) >= self.max_evals: return + combinations_local = self._prepare_warm_start_combinations(effective_target, successful_evaluations) + discrete_cols_in_space = _get_discrete_column_values(combinations_local) + self._evaluate_warm_start_combinations(combinations_local, effective_target, successful_evaluations) + + self._log_uncovered_values(discrete_cols_in_space, self.evaluations, effective_target) + + self.warm_start_evaluation_count = len(self.evaluations) + + def _prepare_warm_start_combinations(self, effective_target: int, successful_evaluations: int) -> list[dict]: + """Prepare candidate combinations according to the warm-start strategy.""" combinations_local = [c for c in copy(self._search_space.combinations) if c not in self._evaluated_combinations] random.Random(self.settings.random_state).shuffle(combinations_local) - # Values already covered by successful warm-start observations so - # stratification does not waste early slots on redundant coverage. - already_covered: dict[str, set[str]] = {} - for eval_entry in self.evaluations: - if eval_entry.get("score") is not None: - for col, val in eval_entry.items(): - if col != "score" and isinstance(val, str): - already_covered.setdefault(col, set()).add(val) - - min_needed = self._min_n_random_nodes_for_coverage(combinations_local, already_covered) - if min_needed > self.settings.n_random_nodes: - logger.warning( - "n_random_nodes=%d may be too small to guarantee full categorical coverage " - "(estimated minimum: %d). Consider increasing n_random_nodes.", - self.settings.n_random_nodes, - min_needed, + if self.settings.warm_start_strategy == "greedy": + str_cols = _get_discrete_column_values(combinations_local) + initial_coverage: dict[str, dict[str, int]] = { + col: {val: 0 for val in vals} for col, vals in str_cols.items() + } + for obs in self.evaluations: + if obs.get("score") is None: + continue + for col in initial_coverage: + val = _str_val(obs.get(col)) + if val in initial_coverage[col]: + initial_coverage[col][val] = min(initial_coverage[col][val] + 1, 2) + remaining_budget = effective_target - successful_evaluations + return self._get_greedy_combinations( + combinations_local, remaining_budget, initial_coverage=initial_coverage + ) + + if self.settings.warm_start_strategy == "balanced": + return self._get_balanced_combinations( + combinations_local, + self.settings.fields_to_balance or [], + coverage_target=effective_target - successful_evaluations, ) - combinations_local = self._get_stratified_combinations(combinations_local, already_covered) + # "random": use shuffled list as-is + return combinations_local - gen = (x for x in combinations_local) + def _evaluate_warm_start_combinations( + self, combinations: list[dict], effective_target: int, successful_evaluations: int + ) -> None: + """Evaluate warm-start candidates until the target or candidate list is exhausted.""" + gen = (x for x in combinations) - while successful_evaluations < self.settings.n_random_nodes: - params = next(gen) + while successful_evaluations < effective_target: + params = next(gen, None) + if params is None: + break 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) + self.evaluations.append(params | {"score": score}) - if len(self.evaluations) == self.max_iterations: + if len(self.evaluations) >= self.max_evals: break @staticmethod - def _min_n_random_nodes_for_coverage( + def _log_uncovered_values( + discrete_cols_in_space: dict[str, set], + evaluations: list[dict], + n_random_nodes: int, + ) -> None: + uncovered_by_col: dict[str, list[str]] = {} + for col, vals_in_space in discrete_cols_in_space.items(): + covered = {_str_val(e.get(col)) for e in evaluations if e.get("score") is not None} + uncovered = vals_in_space - covered + if uncovered: + uncovered_by_col[col] = sorted(uncovered) + if uncovered_by_col: + logger.warning( + "n_random_nodes=%d was too small to cover all discrete column values. " + "Uncovered values by column: %s. Consider increasing n_random_nodes.", + n_random_nodes, + uncovered_by_col, + ) + + @staticmethod + def _get_greedy_combinations( combinations: list[dict], - already_covered: dict[str, set[str]] | None = None, - ) -> int: + n: int, + initial_coverage: dict[str, dict[str, int]] | None = None, + ) -> list[dict]: + """Greedily select n combinations ensuring every discrete column value appears >= 2 times. + + At each step the candidate with the highest coverage gain (number of discrete column + values — string or numeric — whose current count is still below 2) is selected. + Ties are broken by the shuffle order coming in. The n selected combinations are + returned first, followed by the remaining combinations in their original (shuffled) order. + + initial_coverage seeds the per-value counts so that values already covered by + known_observations are not redundantly targeted. """ - Estimate the minimum ``n_random_nodes`` required for stratified coverage. + if not combinations or n <= 0: + return combinations - Returns the maximum number of uncovered unique values across all - string-typed categorical parameters, after accounting for values already - seen in warm-start observations. + str_cols = _get_discrete_column_values(combinations) + if not str_cols: + return combinations - Parameters - ---------- - combinations : list[dict] - Candidate combinations to stratify over. - already_covered : dict[str, set[str]], optional - String-param values already seen in successful warm-start evaluations. + coverage: dict[str, dict[str, int]] = {col: {val: 0 for val in vals} for col, vals in str_cols.items()} + if initial_coverage: + for col, val_counts in initial_coverage.items(): + if col in coverage: + for val, count in val_counts.items(): + if val in coverage[col]: + coverage[col][val] = min(count, 2) - Returns - ------- - int - Estimated lower bound on ``n_random_nodes`` needed for full coverage. - """ - if not combinations: - return 0 - categorical_cols = [col for col, val in combinations[0].items() if isinstance(val, str)] - if not categorical_cols: - return 1 - seen = already_covered or {} - return max(len({c[col] for c in combinations} - seen.get(col, set())) for col in categorical_cols) + def _gain(c: dict) -> int: + return sum(1 for col, val_counts in coverage.items() if val_counts.get(_str_val(c.get(col)), 0) < 2) + + remaining_indices = list(range(len(combinations))) + selected_indices: list[int] = [] + + for _ in range(min(n, len(combinations))): + if not remaining_indices: + break + best_pos = max(range(len(remaining_indices)), key=lambda p: _gain(combinations[remaining_indices[p]])) + best_idx = remaining_indices.pop(best_pos) + selected_indices.append(best_idx) + for col in str_cols: + val = _str_val(combinations[best_idx].get(col)) + if val in coverage[col]: + coverage[col][val] = min(coverage[col][val] + 1, 2) + + return [combinations[i] for i in selected_indices] + [combinations[i] for i in remaining_indices] @staticmethod - def _get_stratified_combinations( - combinations: list[dict], - already_seen: dict[str, set[str]] | None = None, + # The coverage-aware selection keeps its related state together. + # pylint: disable=too-many-locals + def _get_balanced_combinations( + combinations: list[dict], fields_to_balance: list[str], coverage_target: int | None = None ) -> list[dict]: + """Order combinations to cover balanced tuples and other field values early. + + The prefix through ``coverage_target`` contains one configuration from each + balanced-field tuple, then evenly distributed additional configurations as + needed. Each choice maximizes unseen non-balanced values, so a Cartesian + search space covers every such value within its fixed warm-start target. + A constrained search space can make that impossible; callers report the + uncovered values rather than exceeding the target with extra evaluations. """ - Re-order *already-shuffled* combinations so the first entries collectively - cover every unique value of each string-valued (semantic categorical) - parameter before falling back to the original shuffle order. + if not combinations or not fields_to_balance: + return combinations - Only string-typed columns are stratified over. Integer/float parameters - (``chunk_size``, ``ranker_k``, etc.) are excluded because they tend to have - high cardinality; including them would consume all ``n_random_nodes`` slots - covering their many unique values and crowd out the minority string-param - values the stratification is meant to protect. + def _outer_key(c: dict) -> tuple: + return tuple(_str_val(c.get(f)) for f in fields_to_balance) - This prevents the initial random phase from being biased toward - over-represented parameter values (e.g. ``search_mode="hybrid"`` in a - search space where hybrid configurations outnumber vector ones 2:1). + discrete_cols = _get_discrete_column_values(combinations) + non_balanced_fields = [col for col in discrete_cols if col not in fields_to_balance] - Parameters - ---------- - combinations : list[dict] - Shuffled list of parameter combinations. - already_seen : dict[str, set[str]], optional - String-param values already covered by successful warm-start - observations. These are treated as pre-seen so stratification does - not waste early slots on redundant coverage. + outer_buckets: dict[tuple, list[dict]] = defaultdict(list) + for c in combinations: + outer_buckets[_outer_key(c)].append(c) - Returns - ------- - list[dict] - The same combinations with diversity-maximising entries moved to the - front; the relative order within each group (stratified / remainder) - is preserved from the input shuffle. - """ - if not combinations: - return combinations + target = max(len(outer_buckets), coverage_target or len(outer_buckets)) + target = min(target, len(combinations)) + covered_values: dict[str, set[str]] = {field: set() for field in non_balanced_fields} + selections_per_tuple: dict[tuple, int] = {key: 0 for key in outer_buckets} + selected: list[dict] = [] - # Stratify only string-typed parameters (search_mode, chunking_method, - # ranker_strategy, …). Integer/float params (chunk_size, ranker_k, …) are - # quantitative: stratifying them would consume initial slots covering their - # many unique values, leaving no room for minority string-param values. - categorical_cols = [col for col, val in combinations[0].items() if isinstance(val, str)] - if not categorical_cols: - return combinations + def _select_best(bucket_key: tuple) -> dict: + bucket = outer_buckets[bucket_key] + best_index = max( + range(len(bucket)), + key=lambda index: sum( + _str_val(bucket[index].get(field)) not in covered_values[field] for field in non_balanced_fields + ), + ) + selected_combination = bucket.pop(best_index) + selected.append(selected_combination) + selections_per_tuple[bucket_key] += 1 + for field in non_balanced_fields: + covered_values[field].add(_str_val(selected_combination.get(field))) + return selected_combination + + # Give every balanced tuple one slot. Choose the tuple that can add the + # most coverage next, rather than letting input order decide coverage. + unrepresented = list(outer_buckets) + while unrepresented: + best_bucket = max( + unrepresented, + key=lambda bucket_key: max( + sum(_str_val(c.get(field)) not in covered_values[field] for field in non_balanced_fields) + for c in outer_buckets[bucket_key] + ), + ) + _select_best(best_bucket) + unrepresented.remove(best_bucket) + + # Fill remaining fixed-budget slots from the least represented tuple, + # preserving balance while completing non-balanced-value coverage. + while len(selected) < target: + eligible = [bucket_key for bucket_key, bucket in outer_buckets.items() if bucket] + if not eligible: + break + least_selected = min(selections_per_tuple[bucket_key] for bucket_key in eligible) + eligible = [bucket_key for bucket_key in eligible if selections_per_tuple[bucket_key] == least_selected] + best_bucket = max( + eligible, + key=lambda bucket_key: max( + sum(_str_val(c.get(field)) not in covered_values[field] for field in non_balanced_fields) + for c in outer_buckets[bucket_key] + ), + ) + _select_best(best_bucket) - all_values = {col: {c[col] for c in combinations} for col in categorical_cols} - # Intersect with all_values so warm-start values absent from the remaining - # combinations do not prevent the all_covered check from ever firing. - seen: dict[str, set[str]] = { - col: (set(already_seen.get(col, ())) & all_values[col]) if already_seen else set() - for col in categorical_cols - } - - stratified: list[dict] = [] - remainder: list[dict] = [] - - for combo in combinations: - all_covered = all(seen[col] == all_values[col] for col in categorical_cols) - if all_covered: - remainder.append(combo) - continue - - introduces_new = any(combo[col] not in seen[col] for col in categorical_cols) - if introduces_new: - stratified.append(combo) - for col in categorical_cols: - seen[col].add(combo[col]) - else: - remainder.append(combo) - - return stratified + remainder + remaining = [combination for bucket in outer_buckets.values() for combination in bucket] + return selected + _round_robin(remaining, _outer_key) + + # pylint: enable=too-many-locals + + def _prepare_typed_encoder(self) -> None: + """ + Fit label encoders on the full search space for all varying columns. + + Dict-valued columns (model objects) are serialized to their model_id + strings. Constant columns (single unique value) are dropped — they + carry no signal for the GAM. + """ + if self._typed_encoders_with_columns: + return + logger.debug("Preparing typed encoder for %s...", self.__class__.__name__) + df = pd.DataFrame(data=self._search_space.combinations) + for col in df.columns: + df[col] = _serialize_dict_col(df[col]) + varying_cols = [c for c in df.columns if df[c].nunique() > 1] + for col in varying_cols: + self._typed_encoders_with_columns.append((col, LabelEncoder().fit(df[col]))) + logger.debug("Typed encoder for %s has been prepared.", self.__class__.__name__) # pylint: disable=too-many-locals def _run_iteration(self) -> None: """ - Run single optimization iteration that consists of training GAM model - to predict score for remaining nodes in the solutions space and choose - the best n ones for further evaluation. + Run single optimization iteration using typed LinearGAM terms. + + String-typed columns receive f() (factor) terms; numeric columns receive + s() (spline) terms. Random warm starts and sparse categorical training + data use s() so a category absent from the sample remains in-domain + during prediction. Constant columns are excluded. Dict-valued model + columns are serialized to model_id strings before encoding. """ - self._prepare_encoder() - df = pd.DataFrame(data=self.evaluations) # --> These are already known observations with scores. + self._prepare_typed_encoder() + encoders = self._typed_encoders_with_columns + + if not encoders: + return + + df = pd.DataFrame(data=self.evaluations) df = df[df["score"].notna()].copy() data = df.drop(columns=["score"]) + for col in data.columns: + data[col] = _serialize_dict_col(data[col]) + # known_observations may omit columns that vary in the search space; fill + # with the encoder's first class so transform() does not KeyError. + for col, enc in encoders: + if col not in data.columns: + data[col] = enc.classes_[0] target = df["score"] - x_train_enc = [] - for column, encoder in self._encoders_with_columns: - x_train_enc.append(encoder.transform(data[column])) - x_train_enc = np.column_stack(x_train_enc) + x_train_enc = np.column_stack([enc.transform(data[col]) for col, enc in encoders]) + + terms = None + for i, (_, enc) in enumerate(encoders): + observed_values = set(x_train_enc[:, i]) + all_values = set(range(len(enc.classes_))) + use_spline = ( + self.settings.warm_start_strategy == "random" + or not isinstance(enc.classes_[0], str) + or observed_values != all_values + ) + term = gam_s(i) if use_spline else gam_f(i) + terms = term if terms is None else terms + term - gam = LinearGAM() + gam = LinearGAM(terms) gam.fit(x_train_enc, target) remaining_evaluations = self._get_remaining_evaluations( self._search_space.combinations, self._evaluated_combinations ) - remaining_evaluations_df = pd.DataFrame(remaining_evaluations) + if not remaining_evaluations: + return - # Optimize encoding: build array directly - encoded_data_to_predict = np.column_stack( - [encoder.transform(remaining_evaluations_df[column]) for column, encoder in self._encoders_with_columns] - ) + remaining_df = pd.DataFrame(remaining_evaluations) + for col in remaining_df.columns: + remaining_df[col] = _serialize_dict_col(remaining_df[col]) - predictions = gam.predict(encoded_data_to_predict) + encoded = np.column_stack([enc.transform(remaining_df[col]) for col, enc in encoders]) + predictions = gam.predict(encoded) for idx, val in enumerate(remaining_evaluations): val["score"] = predictions[idx] - # Sort in descending order to get highest predictions first best_predictions = sorted(remaining_evaluations, key=lambda d: d["score"], reverse=True) - n_best_predictions = best_predictions[: self.settings.evals_per_trial] - - for params in n_best_predictions: + remaining_evaluation_capacity = max(0, self.max_evals - len(self.evaluations)) + for params in best_predictions[: min(self.settings.evals_per_trial, remaining_evaluation_capacity)]: params.pop("score", None) score = self._objective_function(params) self._evaluated_combinations.append(params) self.evaluations.append(params | {"score": score}) - def _prepare_encoder(self) -> None: - """ - Prepare encoder for the further processing based on all available combinations. + def _trim_evaluations_to_top(self, n: int) -> None: + """Trim self.evaluations to the top n successful entries by score. + + Failed evaluations (score is None) are discarded. Successful evaluations + are sorted descending by score and only the top n are retained. """ - if not self._encoders_with_columns: - logger.debug("Preparing encoder for %s...", self.__class__.__name__) - df = pd.DataFrame(data=self._search_space.combinations) - for column in df.columns: - self._encoders_with_columns.append((column, LabelEncoder().fit(df[column]))) - logger.debug("Encoder for %s has been prepared.", self.__class__.__name__) + successful = sorted( + [e for e in self.evaluations if e.get("score") is not None], + key=lambda d: d["score"], + reverse=True, + ) + self.evaluations = successful[:n] @staticmethod def _get_remaining_evaluations(all_combinations: list[dict], evaluations: list[dict]) -> list[dict]: diff --git a/tests/unit/ai4rag/core/experiment/test_experiment.py b/tests/unit/ai4rag/core/experiment/test_experiment.py index 83abec60..5313c624 100644 --- a/tests/unit/ai4rag/core/experiment/test_experiment.py +++ b/tests/unit/ai4rag/core/experiment/test_experiment.py @@ -20,6 +20,7 @@ from ai4rag.evaluator.metric import Metrics, RAGMetric from ai4rag.evaluator.unitxt_evaluator import UnitxtEvaluator from ai4rag.rag.vector_store.config import MilvusLiteConfig +from ai4rag.utils.event_handler.event_handler import LocalEventHandler # --------------------------------------------------------------------------- # Helpers @@ -185,6 +186,117 @@ def test_judge_metric_with_judge_evaluator_passes(self): _build_experiment(evaluators=evals, optimization_metric=Metrics.JUDGE_ANSWER_RELEVANCE) +class TestOptimizationPatternSelection: + """Output pattern allocation between warm start and GAM phases.""" + + def test_selects_warm_start_quota_and_renumbers_patterns(self): + from ai4rag.core.hpo.gam_opt import GAMOptimizer, GAMOptSettings + + experiment = _build_experiment() + experiment.event_handler = MagicMock() + experiment._optimization_patterns = [ + {"payload": {"name": f"old-warm-{i}"}, "optimization_phase": "warm_start"} for i in range(3) + ] + [{"payload": {"name": f"old-gam-{i}"}, "optimization_phase": "gam"} for i in range(4)] + + search_space = MagicMock() + search_space.combinations = [{"category": value} for value in ("a", "b", "c", "d")] + search_space.max_combinations = 4 + experiment.optimizer = GAMOptimizer( + objective_function=MagicMock(), + search_space=search_space, + settings=GAMOptSettings(max_evals=4, max_iterations=4, n_random_nodes=4, warm_start_strategy="greedy"), + ) + + selected = experiment._select_optimization_patterns() + + assert len(selected) == 4 # effective warm start=8: 8//4 warm + 2 GAM + assert [p["optimization_phase"] for p in selected] == ["warm_start", "warm_start", "gam", "gam"] + assert [p["payload"]["name"] for p in selected] == ["Pattern1", "Pattern2", "Pattern3", "Pattern4"] + + def test_search_clears_buffered_patterns_from_a_previous_run(self): + class NoOpOptimizer: + def __init__(self, **_kwargs): + pass + + def search(self): + return None + + experiment = _build_experiment() + experiment._optimization_patterns = [ + { + "payload": {"name": "stale"}, + "evaluation_results": [], + "optimization_phase": "warm_start", + } + ] + + experiment.search(optimizer=NoOpOptimizer, skip_mps=True) + + assert experiment._optimization_patterns == [] + experiment.event_handler.on_pattern_creation.assert_not_called() + + def test_publish_patterns_does_not_require_handler_patterns_attribute(self): + """Final GAM selection is published through the handler interface alone.""" + from ai4rag.core.hpo.gam_opt import GAMOptimizer, GAMOptSettings + + experiment = _build_experiment() + experiment.event_handler = MagicMock(spec=["on_pattern_creation"]) + search_space = MagicMock() + search_space.combinations = [{"category": value} for value in ("a", "b", "c")] + search_space.max_combinations = 3 + experiment.optimizer = GAMOptimizer( + objective_function=MagicMock(), + search_space=search_space, + settings=GAMOptSettings(max_evals=1, n_random_nodes=4, warm_start_strategy="greedy"), + ) + experiment._optimization_patterns = [ + { + "payload": {"name": "Pattern1"}, + "evaluation_results": [{"score": 0.9}], + "optimization_phase": "warm_start", + } + ] + + experiment._publish_optimization_patterns() + + experiment.event_handler.on_pattern_creation.assert_called_once_with( + payload={"name": "Pattern1", "iteration": 0}, + evaluation_results=[{"score": 0.9}], + optimization_phase="warm_start", + ) + + def test_publish_patterns_works_with_local_event_handler(self, tmp_path): + """Selected patterns, rather than every warm-start result, are written locally.""" + from ai4rag.core.hpo.gam_opt import GAMOptimizer, GAMOptSettings + + experiment = _build_experiment() + experiment.event_handler = LocalEventHandler(tmp_path) + search_space = MagicMock() + search_space.combinations = [{"category": value} for value in ("a", "b", "c")] + search_space.max_combinations = 3 + experiment.optimizer = GAMOptimizer( + objective_function=MagicMock(), + search_space=search_space, + settings=GAMOptSettings(max_evals=1, n_random_nodes=4, warm_start_strategy="greedy"), + ) + experiment._optimization_patterns = [ + { + "payload": {"name": "unselected"}, + "evaluation_results": [{"score": 0.1}], + "optimization_phase": "warm_start", + }, + { + "payload": {"name": "selected"}, + "evaluation_results": [{"score": 0.9}], + "optimization_phase": "warm_start", + }, + ] + + experiment._publish_optimization_patterns() + + assert sorted(path.name for path in tmp_path.iterdir()) == ["Pattern1"] + + class TestResolveOptimizationScore: """Selecting the optimization metric's score from a pattern's results.""" diff --git a/tests/unit/ai4rag/core/hpo/test_gam_opt.py b/tests/unit/ai4rag/core/hpo/test_gam_opt.py index c526f540..5c19e9e7 100644 --- a/tests/unit/ai4rag/core/hpo/test_gam_opt.py +++ b/tests/unit/ai4rag/core/hpo/test_gam_opt.py @@ -2,6 +2,7 @@ # Copyright IBM Corp. 2026 # SPDX-License-Identifier: Apache-2.0 # ----------------------------------------------------------------------------- +import warnings from unittest.mock import MagicMock import numpy as np @@ -17,12 +18,13 @@ class TestGAMOptSettings: def test_gam_opt_settings_creation_with_defaults(self): """Test that GAMOptSettings can be instantiated with default values.""" - settings = GAMOptSettings(max_evals=20) + settings = GAMOptSettings() - assert settings.max_evals == 20 + assert settings.max_evals is None assert settings.n_random_nodes == 4 assert settings.evals_per_trial == 1 assert settings.random_state == 64 + assert settings.warm_start_strategy == "random" def test_gam_opt_settings_creation_with_custom_values(self): """Test that GAMOptSettings can be instantiated with custom values.""" @@ -31,15 +33,20 @@ def test_gam_opt_settings_creation_with_custom_values(self): n_random_nodes=10, evals_per_trial=2, random_state=42, + warm_start_strategy="balanced", + fields_to_balance=["search_mode", "foundation_model"], ) assert settings.max_evals == 50 + assert settings.max_iterations is None assert settings.n_random_nodes == 10 assert settings.evals_per_trial == 2 assert settings.random_state == 42 + assert settings.warm_start_strategy == "balanced" + assert settings.fields_to_balance == ["search_mode", "foundation_model"] - def test_gam_opt_settings_post_init_keeps_n_random_nodes_if_smaller(self): - """Test that __post_init__ keeps n_random_nodes if it's smaller than max_evals.""" + def test_gam_opt_settings_stores_n_random_nodes(self): + """Test that GAMOptSettings stores n_random_nodes as provided.""" settings = GAMOptSettings(max_evals=20, n_random_nodes=5) assert settings.n_random_nodes == 5 @@ -52,6 +59,27 @@ def test_gam_opt_settings_inherits_from_optimizer_settings(self): assert isinstance(settings, OptimizerSettings) + def test_gam_opt_settings_invalid_warm_start_strategy_raises(self): + """Invalid warm_start_strategy is rejected at construction time.""" + with pytest.raises(ValueError, match="warm_start_strategy"): + GAMOptSettings(max_evals=10, warm_start_strategy="invalid_strategy") + + def test_gam_opt_settings_balanced_without_fields_raises(self): + """'balanced' strategy requires non-empty fields_to_balance.""" + with pytest.raises(ValueError, match="fields_to_balance"): + GAMOptSettings(max_evals=10, warm_start_strategy="balanced") + + def test_gam_opt_settings_all_valid_strategies(self): + """All three valid strategy names are accepted.""" + GAMOptSettings(max_evals=10, warm_start_strategy="random") + GAMOptSettings(max_evals=10, warm_start_strategy="greedy") + GAMOptSettings(max_evals=10, warm_start_strategy="balanced", fields_to_balance=["search_mode"]) + + def test_gam_opt_settings_rejects_result_limit_above_evaluation_limit(self): + """The output count cannot be greater than the objective evaluation budget.""" + with pytest.raises(ValueError, match="max_iterations cannot exceed max_evals"): + GAMOptSettings(max_evals=4, max_iterations=5) + class TestGAMOptimizer: """Test the GAMOptimizer class.""" @@ -91,7 +119,40 @@ def test_gam_optimizer_initialization(self, mock_search_space, optimizer_setting assert optimizer.settings == optimizer_settings assert optimizer.evaluations == [] assert optimizer._evaluated_combinations == [] - assert optimizer._encoders_with_columns == [] + assert optimizer._typed_encoders_with_columns == [] + + def test_omitted_max_evals_evaluates_entire_search_space(self, mock_search_space, mocker): + """An omitted evaluation limit uses every available search-space combination.""" + mock_gam = MagicMock() + mock_gam.predict.return_value = np.array([0.5] * mock_search_space.max_combinations) + mocker.patch("ai4rag.core.hpo.gam_opt.LinearGAM", return_value=mock_gam) + objective = MagicMock(return_value=0.5) + optimizer = GAMOptimizer( + objective_function=objective, + search_space=mock_search_space, + settings=GAMOptSettings(n_random_nodes=2), + ) + + optimizer.search() + + assert optimizer.max_evals == mock_search_space.max_combinations + assert objective.call_count == mock_search_space.max_combinations + assert len(optimizer.evaluations) == mock_search_space.max_combinations + + def test_balanced_strategy_rejects_unknown_balance_fields(self, mock_search_space): + """Balanced warm starts fail fast when a requested field is not searchable.""" + settings = GAMOptSettings( + max_evals=6, + warm_start_strategy="balanced", + fields_to_balance=["param1", "missing_field"], + ) + + with pytest.raises(ValueError, match="absent from the search space.*missing_field"): + GAMOptimizer( + objective_function=MagicMock(return_value=0.5), + search_space=mock_search_space, + settings=settings, + ) def test_max_iterations_getter(self, mock_search_space, optimizer_settings): """Test the max_iterations property getter.""" @@ -317,45 +378,22 @@ def test_evaluate_initial_random_nodes_stops_at_max_iterations(self, mock_search # Should stop at max_iterations=4, not n_random_nodes=10 assert len(optimizer.evaluations) == 4 - def test_prepare_encoder(self, mock_search_space, optimizer_settings, mocker): - """Test the _prepare_encoder method.""" + def test_evaluate_initial_random_nodes_n_random_exceeds_max_evals(self, mock_search_space): + """n_random_nodes > max_evals is silently bounded by max_iterations.""" + settings = GAMOptSettings(max_evals=3, n_random_nodes=6) objective_func = MagicMock(return_value=0.5) optimizer = GAMOptimizer( objective_function=objective_func, search_space=mock_search_space, - settings=optimizer_settings, - ) - - # Initially no encoders - assert len(optimizer._encoders_with_columns) == 0 - - optimizer._prepare_encoder() - - # Should have created encoders for each column - assert len(optimizer._encoders_with_columns) == 2 # param1 and param2 - column_names = [col for col, enc in optimizer._encoders_with_columns] - assert "param1" in column_names - assert "param2" in column_names - - def test_prepare_encoder_called_only_once(self, mock_search_space, optimizer_settings): - """Test that _prepare_encoder only prepares encoders once.""" - objective_func = MagicMock(return_value=0.5) - - optimizer = GAMOptimizer( - objective_function=objective_func, - search_space=mock_search_space, - settings=optimizer_settings, + settings=settings, ) - optimizer._prepare_encoder() - first_encoders = optimizer._encoders_with_columns.copy() - - # Call again - optimizer._prepare_encoder() + optimizer.evaluate_initial_random_nodes() - # Should not recreate encoders - assert optimizer._encoders_with_columns == first_encoders + # Capped at max_iterations=3 (min(max_evals=3, max_combinations=6)) + assert len(optimizer.evaluations) == 3 + assert all(e["score"] is not None for e in optimizer.evaluations) def test_run_iteration(self, mock_search_space, mocker): """Test the _run_iteration method.""" @@ -386,6 +424,48 @@ def test_run_iteration(self, mock_search_space, mocker): assert mock_gam_instance.fit.called assert mock_gam_instance.predict.called + def test_random_warm_start_handles_unseen_categorical_values(self): + """A small random warm start can still predict every categorical value.""" + mock_space = MagicMock(spec=SearchSpace) + mock_space.combinations = [{"category": value} for value in ("a", "b", "c")] + mock_space.max_combinations = 3 + settings = GAMOptSettings(max_evals=3, n_random_nodes=1) + objective_func = MagicMock(return_value=0.5) + optimizer = GAMOptimizer( + objective_function=objective_func, + search_space=mock_space, + settings=settings, + ) + + with warnings.catch_warnings(): + warnings.simplefilter("ignore", RuntimeWarning) + result = optimizer.search() + + assert result["score"] == 0.5 + assert objective_func.call_count == 3 + + def test_sparse_non_random_warm_start_handles_unseen_categorical_values(self): + """Sparse greedy data falls back to splines for unseen categorical values.""" + mock_space = MagicMock(spec=SearchSpace) + mock_space.combinations = [{"category": value} for value in ("a", "b", "c")] + mock_space.max_combinations = 3 + settings = GAMOptSettings(max_evals=3, n_random_nodes=1, warm_start_strategy="greedy") + objective_func = MagicMock(return_value=0.5) + optimizer = GAMOptimizer( + objective_function=objective_func, + search_space=mock_space, + settings=settings, + ) + optimizer.evaluations = [{"category": "c", "score": 0.5}] + optimizer._evaluated_combinations = [{"category": "c"}] + + with warnings.catch_warnings(): + warnings.simplefilter("ignore", RuntimeWarning) + optimizer._run_iteration() + + assert len(optimizer.evaluations) == 2 + assert objective_func.call_count == 1 + def test_search_successful(self, mock_search_space, mocker): """Test the search method with successful optimization.""" settings = GAMOptSettings(max_evals=5, n_random_nodes=2, evals_per_trial=1) @@ -537,6 +617,27 @@ def test_run_iteration_filters_out_none_scores_for_training(self, mock_search_sp assert call_args[0][0].shape[0] == 3 # X_train should have 3 samples assert call_args[0][1].shape[0] == 3 # y_train should have 3 samples + def test_run_iteration_does_not_crash_when_remaining_is_empty(self, mock_search_space, mocker): + """_run_iteration returns silently when all combinations have been evaluated.""" + mock_gam = MagicMock() + mocker.patch("ai4rag.core.hpo.gam_opt.LinearGAM", return_value=mock_gam) + + settings = GAMOptSettings(max_evals=6, n_random_nodes=6) + scores = iter([0.1, 0.2, 0.3, 0.4, 0.5, 0.6]) + optimizer = GAMOptimizer( + objective_function=lambda _: next(scores), + search_space=mock_search_space, + settings=settings, + ) + optimizer.evaluate_initial_random_nodes() + assert len(optimizer.evaluations) == 6 # all combos evaluated + + # Should not raise KeyError or any other error + optimizer._run_iteration() + + # GAM was fitted but predict should NOT have been called (early return) + mock_gam.predict.assert_not_called() + class TestGAMOptimizerKnownObservations: """Test warm-start behavior with known observations.""" @@ -760,161 +861,291 @@ def test_known_observations_are_copied(self, mock_search_space): assert known[0]["score"] == 0.3 -class TestGetStratifiedCombinations: - """Test the _get_stratified_combinations static method.""" +class TestPrepareTypedEncoder: + """Test the _prepare_typed_encoder method.""" - def test_skewed_search_space_stratifies_minority_first(self): - """Minority categorical values are moved ahead of the majority.""" - # 4 hybrid, 2 vector — mirrors the real MaaS imbalance - combinations = [ - {"search_mode": "hybrid", "chunk_size": 256}, - {"search_mode": "hybrid", "chunk_size": 512}, - {"search_mode": "hybrid", "chunk_size": 1024}, - {"search_mode": "hybrid", "chunk_size": 2048}, - {"search_mode": "vector", "chunk_size": 256}, - {"search_mode": "vector", "chunk_size": 512}, + @pytest.fixture + def mock_search_space(self): + mock_space = MagicMock(spec=SearchSpace) + mock_space.combinations = [ + {"param1": "a", "param2": 1}, + {"param1": "b", "param2": 2}, + {"param1": "c", "param2": 3}, + {"param1": "d", "param2": 4}, + {"param1": "e", "param2": 5}, + {"param1": "f", "param2": 6}, ] - result = GAMOptimizer._get_stratified_combinations(combinations) + mock_space.max_combinations = 6 + return mock_space - # The first entry is "hybrid" (first in the original list), - # the second must be a "vector" entry (its first occurrence). - assert result[0]["search_mode"] == "hybrid" - assert result[1]["search_mode"] == "vector" + @pytest.fixture + def optimizer_settings(self): + return GAMOptSettings(max_evals=6, n_random_nodes=3) - def test_all_values_represented_after_stratification(self): - """After stratification, both search_mode values appear in the leading section.""" - combinations = [ - {"search_mode": "hybrid", "chunk_size": 256}, - {"search_mode": "hybrid", "chunk_size": 512}, - {"search_mode": "hybrid", "chunk_size": 1024}, - {"search_mode": "vector", "chunk_size": 256}, + def test_fits_encoders_for_varying_columns(self, mock_search_space, optimizer_settings): + """Encoders are created for all varying columns.""" + optimizer = GAMOptimizer( + objective_function=MagicMock(return_value=0.5), + search_space=mock_search_space, + settings=optimizer_settings, + ) + assert len(optimizer._typed_encoders_with_columns) == 0 + optimizer._prepare_typed_encoder() + assert len(optimizer._typed_encoders_with_columns) == 2 + cols = [col for col, _ in optimizer._typed_encoders_with_columns] + assert "param1" in cols + assert "param2" in cols + + def test_drops_constant_columns(self): + """Columns with a single unique value are excluded.""" + mock_space = MagicMock(spec=SearchSpace) + mock_space.combinations = [ + {"param1": "a", "param2": 1, "constant": "x"}, + {"param1": "b", "param2": 2, "constant": "x"}, ] - result = GAMOptimizer._get_stratified_combinations(combinations) - - leading_modes = {c["search_mode"] for c in result[:2]} - assert leading_modes == {"hybrid", "vector"} + mock_space.max_combinations = 2 + settings = GAMOptSettings(max_evals=2, n_random_nodes=2) + optimizer = GAMOptimizer( + objective_function=MagicMock(return_value=0.5), + search_space=mock_space, + settings=settings, + ) + optimizer._prepare_typed_encoder() + cols = [col for col, _ in optimizer._typed_encoders_with_columns] + assert "constant" not in cols + assert "param1" in cols + assert "param2" in cols + + def test_called_only_once(self, mock_search_space, optimizer_settings): + """Calling _prepare_typed_encoder twice does not rebuild the encoders.""" + optimizer = GAMOptimizer( + objective_function=MagicMock(return_value=0.5), + search_space=mock_search_space, + settings=optimizer_settings, + ) + optimizer._prepare_typed_encoder() + first = list(optimizer._typed_encoders_with_columns) + optimizer._prepare_typed_encoder() + assert optimizer._typed_encoders_with_columns == first - def test_already_balanced_order_is_unchanged(self): - """When each categorical value appears exactly once, the list is returned as-is.""" - combinations = [ - {"mode": "a", "size": 1}, - {"mode": "b", "size": 2}, - {"mode": "c", "size": 3}, + def test_serializes_dict_columns(self): + """Dict-valued model columns are serialized to model_id strings.""" + mock_space = MagicMock(spec=SearchSpace) + mock_space.combinations = [ + {"foundation_model": {"model_id": "fm-a", "other": "x"}, "chunk_size": 128}, + {"foundation_model": {"model_id": "fm-b", "other": "x"}, "chunk_size": 256}, ] - result = GAMOptimizer._get_stratified_combinations(combinations) - assert result == combinations - - def test_no_string_columns_returns_unchanged(self): - """Integer-only search spaces (no string params) are returned without reordering.""" - combinations = [{"chunk_size": 256}, {"chunk_size": 512}, {"chunk_size": 1024}] - result = GAMOptimizer._get_stratified_combinations(combinations) - assert result == combinations + mock_space.max_combinations = 2 + settings = GAMOptSettings(max_evals=2, n_random_nodes=2) + optimizer = GAMOptimizer( + objective_function=MagicMock(return_value=0.5), + search_space=mock_space, + settings=settings, + ) + optimizer._prepare_typed_encoder() + cols = [col for col, _ in optimizer._typed_encoders_with_columns] + assert "foundation_model" in cols + fm_enc = next(enc for col, enc in optimizer._typed_encoders_with_columns if col == "foundation_model") + assert set(fm_enc.classes_) == {"fm-a", "fm-b"} + + def test_string_column_gets_str_classes(self, mock_search_space, optimizer_settings): + """String-valued columns produce str classes so factor terms are selected.""" + optimizer = GAMOptimizer( + objective_function=MagicMock(return_value=0.5), + search_space=mock_search_space, + settings=optimizer_settings, + ) + optimizer._prepare_typed_encoder() + param1_enc = next(enc for col, enc in optimizer._typed_encoders_with_columns if col == "param1") + assert isinstance(param1_enc.classes_[0], str) - def test_empty_combinations_returns_empty(self): + def test_numeric_column_gets_non_str_classes(self, mock_search_space, optimizer_settings): + """Numeric columns produce non-str classes so spline terms are selected.""" + optimizer = GAMOptimizer( + objective_function=MagicMock(return_value=0.5), + search_space=mock_search_space, + settings=optimizer_settings, + ) + optimizer._prepare_typed_encoder() + param2_enc = next(enc for col, enc in optimizer._typed_encoders_with_columns if col == "param2") + assert not isinstance(param2_enc.classes_[0], str) + + +class TestGetGreedyCombinations: + """Test the _get_greedy_combinations static method.""" + + def test_each_string_value_appears_twice_in_first_n(self): + """Greedy selection puts both string values of each column in the first n.""" + combos = [ + {"mode": "vector", "method": "recursive"}, + {"mode": "vector", "method": "hybrid"}, + {"mode": "hybrid", "method": "recursive"}, + {"mode": "hybrid", "method": "hybrid"}, + {"mode": "vector", "method": "recursive"}, + {"mode": "hybrid", "method": "hybrid"}, + ] + result = GAMOptimizer._get_greedy_combinations(combos, 4) + first4_modes = [c["mode"] for c in result[:4]] + first4_methods = [c["method"] for c in result[:4]] + assert first4_modes.count("vector") >= 2 + assert first4_modes.count("hybrid") >= 2 + assert first4_methods.count("recursive") >= 2 + assert first4_methods.count("hybrid") >= 2 + + def test_returns_all_combinations(self): + """All input combinations appear exactly once in output.""" + combos = [{"mode": m, "n": i} for m in ("vector", "hybrid") for i in range(3)] + result = GAMOptimizer._get_greedy_combinations(combos, 4) + assert len(result) == 6 + assert set(id(c) for c in result) == set(id(c) for c in combos) + + def test_empty_returns_empty(self): """Empty input returns empty output.""" - assert GAMOptimizer._get_stratified_combinations([]) == [] + assert GAMOptimizer._get_greedy_combinations([], 4) == [] - def test_multiple_categorical_columns_stratified(self): - """Stratification covers all unique values across multiple categorical columns.""" - # search_mode: hybrid/vector, method: dense/sparse - combinations = [ - {"search_mode": "hybrid", "method": "dense"}, - {"search_mode": "hybrid", "method": "sparse"}, - {"search_mode": "vector", "method": "dense"}, - {"search_mode": "vector", "method": "sparse"}, - ] - result = GAMOptimizer._get_stratified_combinations(combinations) + def test_n_zero_returns_original(self): + """n=0 returns combinations unchanged.""" + combos = [{"mode": "vector"}, {"mode": "hybrid"}] + result = GAMOptimizer._get_greedy_combinations(combos, 0) + assert result == combos - # All four combinations introduce at least one new value, so all go to stratified. + def test_no_string_columns_returns_unchanged(self): + """Combinations with only numeric columns are returned as-is.""" + combos = [{"size": i} for i in range(4)] + result = GAMOptimizer._get_greedy_combinations(combos, 4) + assert result == combos + + def test_rest_follows_original_shuffle_order(self): + """Combinations not selected greedy appear afterward in their incoming order.""" + combos = [{"mode": "vector", "n": i} for i in range(6)] + result = GAMOptimizer._get_greedy_combinations(combos, 2) + assert len(result) == 6 + # Non-selected items must appear in the same relative order as in the input. + input_positions = {c["n"]: idx for idx, c in enumerate(combos)} + non_selected_positions = [input_positions[c["n"]] for c in result[2:]] + assert non_selected_positions == sorted(non_selected_positions) + + +class TestGetBalancedCombinations: + """Test the _get_balanced_combinations static method.""" + + def test_round_robins_between_two_field_values(self): + """Combinations alternate between the two values of the balanced field.""" + combos = [{"search_mode": "vector", "n": i} for i in range(3)] + [ + {"search_mode": "hybrid", "n": i} for i in range(3) + ] + result = GAMOptimizer._get_balanced_combinations(combos, ["search_mode"]) + assert len(result) == 6 + assert result[0]["search_mode"] != result[1]["search_mode"] + assert result[2]["search_mode"] != result[3]["search_mode"] + + def test_two_fields_covers_all_tuples_in_first_four(self): + """With 2 models × 2 modes = 4 tuples, each appears in first 4 results.""" + fm1, fm2 = {"model_id": "fm1"}, {"model_id": "fm2"} + em = {"model_id": "em1"} + combos = [ + {"foundation_model": fm1, "embedding_model": em, "search_mode": "vector"}, + {"foundation_model": fm1, "embedding_model": em, "search_mode": "hybrid"}, + {"foundation_model": fm2, "embedding_model": em, "search_mode": "vector"}, + {"foundation_model": fm2, "embedding_model": em, "search_mode": "hybrid"}, + ] + result = GAMOptimizer._get_balanced_combinations(combos, ["foundation_model", "search_mode"]) assert len(result) == 4 - # Every unique value for each column is represented in the stratified prefix. - assert {c["search_mode"] for c in result} == {"hybrid", "vector"} - assert {c["method"] for c in result} == {"dense", "sparse"} + keys = {(c["foundation_model"]["model_id"], c["search_mode"]) for c in result} + assert keys == {("fm1", "vector"), ("fm1", "hybrid"), ("fm2", "vector"), ("fm2", "hybrid")} - def test_already_seen_reduces_stratified_set(self): - """Values already covered by warm-start are treated as pre-seen.""" - combinations = [ - {"search_mode": "hybrid", "chunk_size": 256}, - {"search_mode": "hybrid", "chunk_size": 512}, - {"search_mode": "vector", "chunk_size": 256}, - {"search_mode": "vector", "chunk_size": 512}, + def test_returns_all_combinations(self): + """All input combinations appear in the output.""" + combos = [{"search_mode": "vector", "n": i} for i in range(4)] + [ + {"search_mode": "hybrid", "n": i} for i in range(4) ] - # "hybrid" already covered by warm-start; stratification should pull vector first. - result = GAMOptimizer._get_stratified_combinations(combinations, already_seen={"search_mode": {"hybrid"}}) - assert result[0]["search_mode"] == "vector" + result = GAMOptimizer._get_balanced_combinations(combos, ["search_mode"]) + assert len(result) == 8 - def test_already_seen_all_values_skips_stratification(self): - """When warm-start covers all values, the list is returned in shuffle order.""" - combinations = [ - {"search_mode": "hybrid", "chunk_size": 256}, - {"search_mode": "hybrid", "chunk_size": 512}, - ] - # Both search_mode values already covered (only "hybrid" remains but it's covered). - result = GAMOptimizer._get_stratified_combinations( - combinations, already_seen={"search_mode": {"hybrid", "vector"}} - ) - # All go to remainder (all_covered immediately), order preserved. - assert result == combinations + def test_empty_fields_returns_unchanged(self): + """Empty fields_to_balance returns combinations unchanged.""" + combos = [{"search_mode": "vector", "n": i} for i in range(4)] + result = GAMOptimizer._get_balanced_combinations(combos, []) + assert result == combos - def test_remainder_preserves_relative_order(self): - """Non-stratified combos maintain the same relative ordering as the input.""" - combinations = [ - {"mode": "a", "n": 1}, - {"mode": "a", "n": 2}, # remainder (mode "a" already seen) - {"mode": "b", "n": 3}, - {"mode": "a", "n": 4}, # remainder + def test_empty_combinations_returns_empty(self): + """Empty input returns empty output.""" + assert GAMOptimizer._get_balanced_combinations([], ["search_mode"]) == [] + + def test_string_model_values_are_keyed_directly(self): + """Non-dict model values are converted to str for keying.""" + combos = [ + {"foundation_model": "fm1", "search_mode": "vector"}, + {"foundation_model": "fm1", "search_mode": "hybrid"}, ] - result = GAMOptimizer._get_stratified_combinations(combinations) + result = GAMOptimizer._get_balanced_combinations(combos, ["foundation_model", "search_mode"]) + assert len(result) == 2 + assert result[0]["search_mode"] != result[1]["search_mode"] + + def test_fixed_balanced_prefix_covers_non_balanced_values(self): + """A fixed balanced quota covers all other field values when feasible.""" + combos = [ + {"search_mode": mode, "chunk_size": size, "number_of_chunks": count} + for mode in ("vector", "hybrid") + for size in (512, 1024, 2048) + for count in (3, 5, 10) + ] + + result = GAMOptimizer._get_balanced_combinations(combos, ["search_mode"], coverage_target=3) - # stratified = [{mode:a,n:1}, {mode:b,n:3}], remainder = [{mode:a,n:2}, {mode:a,n:4}] - assert result[0] == {"mode": "a", "n": 1} - assert result[1] == {"mode": "b", "n": 3} - assert result[2] == {"mode": "a", "n": 2} - assert result[3] == {"mode": "a", "n": 4} + prefix = result[:3] + assert {combination["search_mode"] for combination in prefix} == {"vector", "hybrid"} + assert {combination["chunk_size"] for combination in prefix} == {512, 1024, 2048} + assert {combination["number_of_chunks"] for combination in prefix} == {3, 5, 10} -class TestStratifiedInitialSampling: - """Test that evaluate_initial_random_nodes uses stratified sampling.""" +class TestInitialSamplingStrategies: + """Test that evaluate_initial_random_nodes applies the correct strategy.""" - def test_skewed_space_always_includes_minority_mode(self): - """With a 3:1 hybrid/vector ratio, the initial sample always includes both modes.""" + def test_random_strategy_evaluates_n_nodes(self): + """'random' strategy evaluates exactly n_random_nodes combinations.""" mock_space = MagicMock(spec=SearchSpace) - # 6 hybrid, 2 vector — minority vector must appear in the initial 4 - mock_space.combinations = [ - {"search_mode": "hybrid", "size": 256}, - {"search_mode": "hybrid", "size": 512}, - {"search_mode": "hybrid", "size": 1024}, - {"search_mode": "hybrid", "size": 2048}, - {"search_mode": "hybrid", "size": 4096}, - {"search_mode": "hybrid", "size": 8192}, - {"search_mode": "vector", "size": 256}, - {"search_mode": "vector", "size": 512}, - ] + mock_space.combinations = [{"search_mode": "hybrid", "size": i} for i in range(8)] mock_space.max_combinations = 8 - - settings = GAMOptSettings(max_evals=8, n_random_nodes=4) - objective_func = MagicMock(return_value=0.5) - + settings = GAMOptSettings(max_evals=8, n_random_nodes=4, warm_start_strategy="random") optimizer = GAMOptimizer( - objective_function=objective_func, + objective_function=MagicMock(return_value=0.5), search_space=mock_space, settings=settings, ) optimizer.evaluate_initial_random_nodes() + assert len(optimizer.evaluations) == 4 - # Assert specifically on the first n_random_nodes evaluations; if the - # objective ever returns None the loop draws more items and checking all - # evaluations would pass for the wrong reason. - initial_evals = optimizer.evaluations[: settings.n_random_nodes] - modes_sampled = {e["search_mode"] for e in initial_evals} - assert "vector" in modes_sampled, ( - "Stratified sampling must include at least one 'vector' evaluation " - "even though 'hybrid' comprises 75% of the search space." + def test_balanced_strategy_covers_all_tuples(self): + """'balanced' strategy covers all (model, mode) tuples in the first N evals.""" + mock_space = MagicMock(spec=SearchSpace) + fm1, fm2 = {"model_id": "fm1"}, {"model_id": "fm2"} + em = {"model_id": "em1"} + mock_space.combinations = ( + [{"foundation_model": fm1, "embedding_model": em, "search_mode": "vector", "size": i} for i in range(3)] + + [{"foundation_model": fm1, "embedding_model": em, "search_mode": "hybrid", "size": i} for i in range(3)] + + [{"foundation_model": fm2, "embedding_model": em, "search_mode": "vector", "size": i} for i in range(3)] + + [{"foundation_model": fm2, "embedding_model": em, "search_mode": "hybrid", "size": i} for i in range(3)] + ) + mock_space.max_combinations = 12 + settings = GAMOptSettings( + max_evals=12, + n_random_nodes=4, + warm_start_strategy="balanced", + fields_to_balance=["foundation_model", "embedding_model", "search_mode"], + ) + optimizer = GAMOptimizer( + objective_function=MagicMock(return_value=0.5), + search_space=mock_space, + settings=settings, ) - assert "hybrid" in modes_sampled + optimizer.evaluate_initial_random_nodes() + initial = optimizer.evaluations[: settings.n_random_nodes] + buckets = {(e["foundation_model"]["model_id"], e["search_mode"]) for e in initial} + assert len(buckets) == 4 - def test_warm_start_all_majority_stratifies_remaining_for_minority(self): - """When all warm-start obs are the majority mode, new evals cover the minority.""" + def test_balanced_strategy_skewed_space_includes_minority(self): + """'balanced' with search_mode field covers both modes even in a skewed space.""" mock_space = MagicMock(spec=SearchSpace) mock_space.combinations = [ {"search_mode": "hybrid", "size": 256}, @@ -924,10 +1155,14 @@ def test_warm_start_all_majority_stratifies_remaining_for_minority(self): {"search_mode": "vector", "size": 512}, ] mock_space.max_combinations = 5 - known = [{"search_mode": "hybrid", "size": 256, "score": 0.5}] - settings = GAMOptSettings(max_evals=5, n_random_nodes=3) - + settings = GAMOptSettings( + max_evals=5, + n_random_nodes=4, + random_state=42, + warm_start_strategy="balanced", + fields_to_balance=["search_mode"], + ) optimizer = GAMOptimizer( objective_function=MagicMock(return_value=0.5), search_space=mock_space, @@ -935,74 +1170,121 @@ def test_warm_start_all_majority_stratifies_remaining_for_minority(self): known_observations=known, ) optimizer.evaluate_initial_random_nodes() - - # 2 new evaluations needed to reach n_random_nodes=3; at least one must be vector. new_evals = optimizer.evaluations[len(known) :] assert "vector" in {e["search_mode"] for e in new_evals} - def test_warns_when_n_random_nodes_insufficient_for_coverage(self, mocker): - """A warning is logged when n_random_nodes is smaller than min needed for coverage.""" + def test_balanced_strategy_covers_non_balanced_column_values(self): + """'balanced' initial phase covers non-balanced column values (e.g. chunk_size). + + Regression: the old full-tuple inner key created unique keys per combination, + making _round_robin a no-op and leaving all selected items at chunk_size=512. + """ mock_space = MagicMock(spec=SearchSpace) - # param1 has 6 unique string values; n_random_nodes=3 < 6 → warning expected + # 2 search modes × 3 chunk_sizes × 2 methods = 12 combinations mock_space.combinations = [ - {"param1": "a", "param2": 1}, - {"param1": "b", "param2": 2}, - {"param1": "c", "param2": 3}, - {"param1": "d", "param2": 4}, - {"param1": "e", "param2": 5}, - {"param1": "f", "param2": 6}, + {"search_mode": mode, "chunk_size": size, "method": method} + for mode in ("vector", "hybrid") + for size in (512, 1024, 2048) + for method in ("recursive", "flat") ] - mock_space.max_combinations = 6 - - mock_warn = mocker.patch("ai4rag.core.hpo.gam_opt.logger.warning") - settings = GAMOptSettings(max_evals=6, n_random_nodes=3) + mock_space.max_combinations = 12 + settings = GAMOptSettings( + max_evals=12, + n_random_nodes=6, + random_state=64, + warm_start_strategy="balanced", + fields_to_balance=["search_mode"], + ) optimizer = GAMOptimizer( objective_function=MagicMock(return_value=0.5), search_space=mock_space, settings=settings, ) optimizer.evaluate_initial_random_nodes() + initial = optimizer.evaluations[: settings.n_random_nodes] + chunk_sizes_seen = {e["chunk_size"] for e in initial} + assert len(chunk_sizes_seen) > 1, ( + f"Only one chunk_size ({chunk_sizes_seen}) appeared in the first " + f"{settings.n_random_nodes} evaluations — inner round-robin is broken." + ) + + def test_balanced_strategy_covers_non_balanced_values_when_some_evals_fail(self): + """After balanced warm start, all non-balanced column values are attempted. - mock_warn.assert_called_once() - warning_msg = mock_warn.call_args[0][0] % mock_warn.call_args[0][1:] - assert "n_random_nodes=3" in warning_msg - assert "6" in warning_msg # estimated minimum + Regression: when a non-balanced value (e.g. 'flat' chunking_method) never + appears in the first effective_target selected candidates, it must still be + attempted via _cover_non_balanced_values so the GAM is not surprised by an + out-of-domain value at prediction time. - def test_no_warning_when_n_random_nodes_sufficient(self, mocker): - """No warning when n_random_nodes covers all unique categorical values.""" + Setup: effective_target = max(4, 4, 2, 3) = 4; chunking_method has 3 values + but only 2 balanced tuples cycle in the main loop, so 'sentence' may be + skipped if the 4 successes are reached before it is selected. The coverage + pass must pick it up. + """ mock_space = MagicMock(spec=SearchSpace) - # param1 has 2 unique string values; n_random_nodes=4 >= 2 → no warning - mock_space.combinations = [ - {"search_mode": "hybrid", "size": 256}, - {"search_mode": "hybrid", "size": 512}, - {"search_mode": "vector", "size": 256}, - {"search_mode": "vector", "size": 512}, + # 2 search_modes (balanced field) × 3 chunking_methods × 2 chunk_sizes + combinations = [ + {"search_mode": mode, "chunking_method": method, "chunk_size": size} + for mode in ("vector", "hybrid") + for method in ("recursive", "flat", "sentence") + for size in (512, 1024) ] - mock_space.max_combinations = 4 + mock_space.combinations = combinations + mock_space.max_combinations = len(combinations) - mock_warn = mocker.patch("ai4rag.core.hpo.gam_opt.logger.warning") - settings = GAMOptSettings(max_evals=4, n_random_nodes=4) + # Only "recursive" succeeds; all others fail — effective_target = max(4,4,2,3) = 4. + # The first 4 successes can all be "recursive", leaving "flat" and "sentence" uncovered. + def objective(params): + return 0.5 if params.get("chunking_method") == "recursive" else None + + settings = GAMOptSettings( + max_evals=20, + n_random_nodes=4, + random_state=64, + warm_start_strategy="balanced", + fields_to_balance=["search_mode"], + ) optimizer = GAMOptimizer( - objective_function=MagicMock(return_value=0.5), + objective_function=objective, search_space=mock_space, settings=settings, ) optimizer.evaluate_initial_random_nodes() - mock_warn.assert_not_called() + attempted = {e["chunking_method"] for e in optimizer.evaluations} + assert attempted == { + "recursive", + "flat", + "sentence", + }, f"Expected all chunking_method values to be attempted, but got: {attempted}" - def test_stratification_is_deterministic_with_same_seed(self): - """Stratified sampling is fully reproducible given the same random_state.""" + def test_greedy_strategy_covers_each_value_twice(self): + """'greedy' strategy puts every discrete column value at least twice in first n.""" mock_space = MagicMock(spec=SearchSpace) - mock_space.combinations = [ - {"search_mode": "hybrid", "size": 256}, - {"search_mode": "hybrid", "size": 512}, - {"search_mode": "hybrid", "size": 1024}, - {"search_mode": "vector", "size": 256}, - {"search_mode": "vector", "size": 512}, + # size has 2 unique values so max_unique=2, min_required=max(4,4)=4 + mock_space.combinations = [{"search_mode": "vector", "method": "recursive", "size": i} for i in range(2)] + [ + {"search_mode": "hybrid", "method": "hybrid", "size": i} for i in range(2) ] - mock_space.max_combinations = 5 - settings = GAMOptSettings(max_evals=5, n_random_nodes=3, random_state=42) + mock_space.max_combinations = 4 + settings = GAMOptSettings(max_evals=4, n_random_nodes=4, warm_start_strategy="greedy") + optimizer = GAMOptimizer( + objective_function=MagicMock(return_value=0.5), + search_space=mock_space, + settings=settings, + ) + optimizer.evaluate_initial_random_nodes() + first4_modes = [e["search_mode"] for e in optimizer.evaluations[:4]] + assert first4_modes.count("vector") >= 2 + assert first4_modes.count("hybrid") >= 2 + + def test_sampling_is_deterministic_with_same_seed(self): + """Sampling is reproducible given the same random_state.""" + mock_space = MagicMock(spec=SearchSpace) + mock_space.combinations = [{"search_mode": "hybrid", "size": i} for i in range(5)] + [ + {"search_mode": "vector", "size": i} for i in range(5) + ] + mock_space.max_combinations = 10 + settings = GAMOptSettings(max_evals=10, n_random_nodes=4, random_state=42) def make_optimizer(): opt = GAMOptimizer( @@ -1064,31 +1346,36 @@ def deterministic_objective(params): assert evals1 == evals2 assert len(evals1) == 3 - def test_gam_optimizer_different_with_different_random_state(self, mock_search_space): - """Test that GAMOptimizer produces different evaluation order with different random_state.""" + def test_gam_optimizer_different_with_different_random_state(self): + """Different random_state values produce different within-bucket orderings.""" + # Use a deterministic space where the two seeds are known to produce + # different shuffle orders. All items share the same search_mode bucket + # so the only source of variation is the shuffle. + mock_space = MagicMock(spec=SearchSpace) + mock_space.combinations = [{"search_mode": "vector", "n": i} for i in range(6)] + mock_space.max_combinations = 6 def deterministic_objective(params): - return params["param2"] / 10.0 + return params["n"] / 10.0 - # First run with random_state=42 optimizer1 = GAMOptimizer( objective_function=deterministic_objective, - search_space=mock_search_space, - settings=GAMOptSettings(max_evals=6, n_random_nodes=3, random_state=42), + search_space=mock_space, + settings=GAMOptSettings(max_evals=6, n_random_nodes=3, random_state=0), ) optimizer1.evaluate_initial_random_nodes() - evals1 = [e["param1"] for e in optimizer1.evaluations] - # Second run with random_state=99 optimizer2 = GAMOptimizer( objective_function=deterministic_objective, - search_space=mock_search_space, - settings=GAMOptSettings(max_evals=6, n_random_nodes=3, random_state=99), + search_space=mock_space, + settings=GAMOptSettings(max_evals=6, n_random_nodes=3, random_state=1), ) optimizer2.evaluate_initial_random_nodes() - evals2 = [e["param1"] for e in optimizer2.evaluations] - # Should evaluate different combinations or different order + evals1 = [e["n"] for e in optimizer1.evaluations] + evals2 = [e["n"] for e in optimizer2.evaluations] + + # Seeds 0 and 1 produce different orderings of 6 items — verified offline. assert evals1 != evals2 def test_gam_full_search_deterministic_with_same_random_state(self, mock_search_space, mocker): @@ -1138,3 +1425,376 @@ def deterministic_objective(params): assert evals1 == evals2 assert result1 == result2 assert len(evals1) == 6 + + +class TestGreedyBalancedWarmStartAutoAdjust: + """Tests for the auto-adjusted warm start and GAM iteration logic in greedy/balanced.""" + + def _make_space(self, n=8): + """Search space with 2 search_modes and 4 methods, totalling n combinations.""" + mock_space = MagicMock(spec=SearchSpace) + mock_space.combinations = [{"search_mode": "vector", "method": m} for m in ["a", "b", "c", "d"]] + [ + {"search_mode": "hybrid", "method": m} for m in ["a", "b", "c", "d"] + ] + mock_space.max_combinations = 8 + return mock_space + + # ------------------------------------------------------------------ + # Auto-adjusted warm start (no ValueError when n_random_nodes < min_required) + # ------------------------------------------------------------------ + + def test_greedy_no_error_when_n_random_nodes_below_min_required(self): + """Greedy strategy no longer raises when n_random_nodes < min_required.""" + mock_space = self._make_space() + # method has 4 unique values → max_unique=4, min_required=max(4,8)=8 + # n_random_nodes=2 is well below 8 — should NOT raise. + settings = GAMOptSettings(max_evals=20, n_random_nodes=2, warm_start_strategy="greedy") + # Construction must succeed without ValueError. + optimizer = GAMOptimizer( + objective_function=MagicMock(return_value=0.5), + search_space=mock_space, + settings=settings, + ) + assert optimizer is not None + + def test_balanced_no_error_when_n_random_nodes_below_min_required(self): + """Balanced strategy no longer raises when n_random_nodes < min_required.""" + mock_space = self._make_space() + # 2 search_mode values → n_balanced_tuples >= 2, min_required >= 4 + settings = GAMOptSettings( + max_evals=20, + n_random_nodes=1, + warm_start_strategy="balanced", + fields_to_balance=["search_mode"], + ) + optimizer = GAMOptimizer( + objective_function=MagicMock(return_value=0.5), + search_space=mock_space, + settings=settings, + ) + assert optimizer is not None + + def test_greedy_warm_start_evaluates_min_required_when_n_random_nodes_is_smaller(self): + """Warm start evaluates min_required nodes even when n_random_nodes < min_required.""" + mock_space = self._make_space() + # method has 4 unique values → min_required = max(4, 2*4) = 8 + # n_random_nodes=2 < 8, so effective_target=8 + settings = GAMOptSettings(max_evals=30, n_random_nodes=2, warm_start_strategy="greedy") + optimizer = GAMOptimizer( + objective_function=MagicMock(return_value=0.5), + search_space=mock_space, + settings=settings, + ) + optimizer.evaluate_initial_random_nodes() + successful = sum(1 for e in optimizer.evaluations if e["score"] is not None) + assert successful == 8 + + def test_search_logs_when_warm_start_consumes_evaluation_budget(self, caplog): + mock_space = self._make_space() + optimizer = GAMOptimizer( + objective_function=MagicMock(return_value=0.5), + search_space=mock_space, + settings=GAMOptSettings(max_evals=8, n_random_nodes=2, warm_start_strategy="greedy"), + ) + + optimizer.search() + + assert "All 8 allowed evaluations were consumed by the warm-start phase" in caplog.text + + def test_balanced_warm_start_evaluates_min_required_when_n_random_nodes_is_smaller(self): + """Balanced warm start evaluates min_required nodes even when n_random_nodes is smaller.""" + mock_space = self._make_space() + # search_mode has 2 unique tuples; method has 4 values → min_required = max(4,2,4)=4 + settings = GAMOptSettings( + max_evals=30, + n_random_nodes=1, + warm_start_strategy="balanced", + fields_to_balance=["search_mode"], + ) + optimizer = GAMOptimizer( + objective_function=MagicMock(return_value=0.5), + search_space=mock_space, + settings=settings, + ) + optimizer.evaluate_initial_random_nodes() + successful = sum(1 for e in optimizer.evaluations if e["score"] is not None) + assert successful >= 4 + + def test_compute_warm_start_effective_target_greedy(self): + """_compute_warm_start_effective_target respects max(n_random_nodes, min_required) for greedy.""" + mock_space = self._make_space() + # method has 4 values → min_required = max(4, 8) = 8 + settings = GAMOptSettings(max_evals=30, n_random_nodes=3, warm_start_strategy="greedy") + optimizer = GAMOptimizer( + objective_function=MagicMock(return_value=0.5), + search_space=mock_space, + settings=settings, + ) + assert optimizer._compute_warm_start_effective_target() == 8 + + def test_compute_warm_start_effective_target_honours_larger_n_random_nodes(self): + """When n_random_nodes > min_required, effective_target uses n_random_nodes.""" + mock_space = self._make_space() + # min_required=8, n_random_nodes=12 → effective_target=12 + settings = GAMOptSettings(max_evals=30, n_random_nodes=12, warm_start_strategy="greedy") + optimizer = GAMOptimizer( + objective_function=MagicMock(return_value=0.5), + search_space=mock_space, + settings=settings, + ) + assert optimizer._compute_warm_start_effective_target() == 12 + + def test_compute_warm_start_effective_target_random_unchanged(self): + """For random strategy, effective_target equals n_random_nodes.""" + mock_space = self._make_space() + settings = GAMOptSettings(max_evals=30, n_random_nodes=3) + optimizer = GAMOptimizer( + objective_function=MagicMock(return_value=0.5), + search_space=mock_space, + settings=settings, + ) + assert optimizer._compute_warm_start_effective_target() == 3 + + # ------------------------------------------------------------------ + # GAM iterations fill the output slots remaining after warm-start allocation. + # ------------------------------------------------------------------ + + def test_greedy_search_runs_gam_iterations_after_warm_start(self, mocker): + """Greedy search fills output slots remaining after warm start.""" + mock_space = MagicMock(spec=SearchSpace) + # search_mode has 2 unique values, method has 4, size has 4 → max_unique=4 + # min_required = max(4, 2*4) = 8 + # max_iterations=20 allocates two warm-start outputs and 18 GAM outputs. + mock_space.combinations = [ + {"search_mode": m, "method": x, "size": s} + for m in ["vector", "hybrid"] + for x in ["a", "b", "c", "d"] + for s in [100, 200, 300, 400] + ] + mock_space.max_combinations = 32 + + mock_gam = MagicMock() + mock_gam.predict.return_value = np.array([0.5] * 32) + mocker.patch("ai4rag.core.hpo.gam_opt.LinearGAM", return_value=mock_gam) + + settings = GAMOptSettings( + max_evals=32, + max_iterations=20, + n_random_nodes=2, + evals_per_trial=2, + warm_start_strategy="greedy", + ) + optimizer = GAMOptimizer( + objective_function=MagicMock(return_value=0.5), + search_space=mock_space, + settings=settings, + ) + # effective_target = max(n_random_nodes=2, min_required=8) = 8 + effective_warm_start = optimizer._compute_warm_start_effective_target() + assert effective_warm_start == 8 + + run_iteration_calls = [] + original_run = optimizer._run_iteration + + def counting_run(): + run_iteration_calls.append(1) + original_run() + + optimizer._run_iteration = counting_run + optimizer.search() + + expected_gam_iters = 9 # ceil((20 output - 2 warm-start slots) / 2 per trial) + assert len(run_iteration_calls) == expected_gam_iters + assert optimizer.objective_function.call_count == effective_warm_start + 18 + + def test_balanced_search_runs_gam_iterations_after_warm_start(self, mocker): + """Balanced search fills output slots after its coverage warm start.""" + mock_space = MagicMock(spec=SearchSpace) + # search_mode has 2 unique values, method has 20 unique values + # min_required = max(4, 2_balanced_tuples, 20_non_balanced) = 20 + # max_iterations=12 allocates five warm-start outputs and seven GAM outputs. + mock_space.combinations = [ + {"search_mode": m, "method": f"x{i}"} for m in ["vector", "hybrid"] for i in range(20) + ] + mock_space.max_combinations = 40 + + mock_gam = MagicMock() + mock_gam.predict.return_value = np.array([0.5] * 40) + mocker.patch("ai4rag.core.hpo.gam_opt.LinearGAM", return_value=mock_gam) + + settings = GAMOptSettings( + max_evals=40, + max_iterations=12, + n_random_nodes=1, + warm_start_strategy="balanced", + fields_to_balance=["search_mode"], + ) + optimizer = GAMOptimizer( + objective_function=MagicMock(return_value=0.5), + search_space=mock_space, + settings=settings, + ) + # non_balanced "method" has 20 unique values → min_required = max(4, 2, 20) = 20 + effective_warm_start = optimizer._compute_warm_start_effective_target() + assert effective_warm_start == 20 + + run_iteration_calls = [] + original_run = optimizer._run_iteration + + def counting_run(): + run_iteration_calls.append(1) + original_run() + + optimizer._run_iteration = counting_run + optimizer.search() + + assert len(run_iteration_calls) == 7 + + def test_greedy_search_caps_partial_gam_trial_at_max_evals(self, mocker): + """A final GAM trial evaluates only the capacity remaining in max_evals.""" + mock_space = MagicMock(spec=SearchSpace) + mock_space.combinations = [ + {"search_mode": mode, "method": method, "size": size} + for mode in ["vector", "hybrid"] + for method in ["a", "b", "c", "d"] + for size in [100, 200] + ] + mock_space.max_combinations = 16 + + mock_gam = MagicMock() + mock_gam.predict.return_value = np.array([0.5] * 16) + mocker.patch("ai4rag.core.hpo.gam_opt.LinearGAM", return_value=mock_gam) + + objective = MagicMock(return_value=0.5) + optimizer = GAMOptimizer( + objective_function=objective, + search_space=mock_space, + settings=GAMOptSettings( + max_evals=10, + max_iterations=10, + n_random_nodes=2, + evals_per_trial=3, + warm_start_strategy="greedy", + ), + ) + + optimizer.search() + + assert objective.call_count == 10 + assert len(optimizer.evaluations) == 10 + + # ------------------------------------------------------------------ + # Trim to top max_evals by score + # ------------------------------------------------------------------ + + def test_trim_evaluations_to_top_keeps_best(self): + """_trim_evaluations_to_top retains the top n by score and discards the rest.""" + mock_space = self._make_space() + settings = GAMOptSettings(max_evals=5, n_random_nodes=4, warm_start_strategy="greedy") + optimizer = GAMOptimizer( + objective_function=MagicMock(return_value=0.5), + search_space=mock_space, + settings=settings, + ) + optimizer.evaluations = [ + {"search_mode": "vector", "method": "a", "score": 0.9}, + {"search_mode": "vector", "method": "b", "score": 0.3}, + {"search_mode": "hybrid", "method": "a", "score": 0.7}, + {"search_mode": "hybrid", "method": "b", "score": 0.5}, + {"search_mode": "vector", "method": "c", "score": 0.1}, + {"search_mode": "vector", "method": "d", "score": 0.8}, + {"search_mode": "hybrid", "method": "c", "score": 0.6}, + ] + optimizer._trim_evaluations_to_top(3) + assert len(optimizer.evaluations) == 3 + scores = [e["score"] for e in optimizer.evaluations] + assert scores == [0.9, 0.8, 0.7] + + def test_trim_evaluations_discards_failed_evaluations(self): + """_trim_evaluations_to_top drops entries with score=None.""" + mock_space = self._make_space() + settings = GAMOptSettings(max_evals=5, n_random_nodes=4, warm_start_strategy="greedy") + optimizer = GAMOptimizer( + objective_function=MagicMock(return_value=0.5), + search_space=mock_space, + settings=settings, + ) + optimizer.evaluations = [ + {"search_mode": "vector", "method": "a", "score": 0.9}, + {"search_mode": "vector", "method": "b", "score": None}, + {"search_mode": "hybrid", "method": "a", "score": 0.7}, + ] + optimizer._trim_evaluations_to_top(5) + assert all(e["score"] is not None for e in optimizer.evaluations) + assert len(optimizer.evaluations) == 2 + + def test_greedy_search_when_max_evals_smaller_than_min_required(self, mocker): + """A small output limit does not cap the required warm-start evaluations.""" + mock_space = MagicMock(spec=SearchSpace) + # search_mode(2) × method(4) × size(4) = 32 combinations; max_unique=4 + # min_required = max(4, 2*4) = 8; max_iterations=4 + mock_space.combinations = [ + {"search_mode": m, "method": x, "size": s} + for m in ["vector", "hybrid"] + for x in ["a", "b", "c", "d"] + for s in [100, 200, 300, 400] + ] + mock_space.max_combinations = 32 + + mock_gam = MagicMock() + mock_gam.predict.return_value = np.array([0.5] * 32) + mocker.patch("ai4rag.core.hpo.gam_opt.LinearGAM", return_value=mock_gam) + + settings = GAMOptSettings(max_evals=32, max_iterations=4, n_random_nodes=2, warm_start_strategy="greedy") + optimizer = GAMOptimizer( + objective_function=MagicMock(return_value=0.5), + search_space=mock_space, + settings=settings, + ) + effective_warm_start = optimizer._compute_warm_start_effective_target() + assert effective_warm_start == 8 # min_required, not n_random_nodes or max_evals + + run_iteration_calls = [] + original_run = optimizer._run_iteration + + def counting_run(): + run_iteration_calls.append(1) + original_run() + + optimizer._run_iteration = counting_run + optimizer.search() + + assert len(run_iteration_calls) == 2 + assert optimizer.objective_function.call_count == 10 + assert len(optimizer.evaluations) == settings.max_iterations + + def test_greedy_search_does_not_exceed_max_evals(self, mocker): + """The greedy coverage target never causes more than max_evals objective calls.""" + mock_space = MagicMock(spec=SearchSpace) + mock_space.combinations = [ + {"search_mode": m, "method": f"x{i}"} for m in ["vector", "hybrid"] for i in range(20) + ] + mock_space.max_combinations = 40 + + mock_gam = MagicMock() + mock_gam.predict.return_value = np.array([0.5] * 40) + mocker.patch("ai4rag.core.hpo.gam_opt.LinearGAM", return_value=mock_gam) + + call_count = [0] + + def scoring_objective(params): + call_count[0] += 1 + return call_count[0] * 0.01 # strictly increasing scores + + settings = GAMOptSettings(max_evals=10, n_random_nodes=2, warm_start_strategy="greedy") + optimizer = GAMOptimizer( + objective_function=scoring_objective, + search_space=mock_space, + settings=settings, + ) + optimizer.search() + + assert call_count[0] == settings.max_evals + assert len(optimizer.evaluations) <= 10 + scores = [e["score"] for e in optimizer.evaluations] + assert scores == sorted(scores, reverse=True), "Evaluations should be sorted best-first"