mlos_bench.storage.base_storage

Base interface for saving and restoring the benchmark data.

See also

mlos_bench.storage.base_storage.Storage.experiments

Retrieves a dictionary of the Experiments’ data.

mlos_bench.storage.base_experiment_data.ExperimentData.results_df

Retrieves a pandas DataFrame of the Experiment’s trials’ results data.

mlos_bench.storage.base_experiment_data.ExperimentData.trials

Retrieves a dictionary of the Experiment’s trials’ data.

mlos_bench.storage.base_experiment_data.ExperimentData.tunable_configs

Retrieves a dictionary of the Experiment’s sampled configs data.

mlos_bench.storage.base_experiment_data.ExperimentData.tunable_config_trial_groups

Retrieves a dictionary of the Experiment’s trials’ data, grouped by shared tunable config.

mlos_bench.storage.base_trial_data.TrialData

Base interface for accessing the stored benchmark trial data.

Classes

Storage

An abstract interface between the benchmarking framework and storage systems

Module Contents

class mlos_bench.storage.base_storage.Storage(config: Dict[str, Any], global_config: dict | None = None, service: mlos_bench.services.base_service.Service | None = None)

An abstract interface between the benchmarking framework and storage systems (e.g., SQLite or MLFLow).

Create a new storage object.

Parameters:

• config (dict) – Free-format key/value pairs of configuration parameters.

• global_config (Optional[dict])

• service (Optional[mlos_bench.services.base_service.Service])

class Experiment(*, tunables: mlos_bench.tunables.tunable_groups.TunableGroups, experiment_id: str, trial_id: int, root_env_config: str, description: str, opt_targets: Dict[str, Literal['min', 'max']])

Base interface for storing the results of the experiment.

This class is instantiated in the Storage.experiment() method.

Parameters:

• tunables (mlos_bench.tunables.tunable_groups.TunableGroups)

• experiment_id (str)

• trial_id (int)

• root_env_config (str)

• description (str)

• opt_targets (Dict[str, Literal['min', 'max']])

__enter__() → Storage

Enter the context of the experiment.

Override the _setup method to add custom context initialization.

Return type:

Storage

__exit__(exc_type: Type[BaseException] | None, exc_val: BaseException | None, exc_tb: types.TracebackType | None) → Literal[False]

End the context of the experiment.

Override the _teardown method to add custom context teardown logic.

Parameters:

• exc_type (Optional[Type[BaseException]])

• exc_val (Optional[BaseException])

• exc_tb (Optional[types.TracebackType])

Return type:

Literal[False]

__repr__() → str

Return type:

str

abstract load(last_trial_id: int = -1) → Tuple[List[int], List[dict], List[Dict[str, Any] | None], List[mlos_bench.environments.status.Status]]

Load (tunable values, benchmark scores, status) to warm-up the optimizer.

If last_trial_id is present, load only the data from the (completed) trials that were scheduled after the given trial ID. Otherwise, return data from ALL merged-in experiments and attempt to impute the missing tunable values.

Parameters:

last_trial_id (int) – (Optional) Trial ID to start from.

Returns:

(trial_ids, configs, scores, status) – Trial ids, Tunable values, benchmark scores, and status of the trials.

Return type:

([int], [dict], [Optional[dict]], [Status])

abstract load_telemetry(trial_id: int) → List[Tuple[datetime.datetime, str, Any]]

Retrieve the telemetry data for a given trial.

Parameters:

trial_id (int) – Trial ID.

Returns:

metrics – Telemetry data.

Return type:

List[Tuple[datetime.datetime, str, Any]]

abstract load_tunable_config(config_id: int) → Dict[str, Any]

Load tunable values for a given config ID.

Parameters:

config_id (int)

Return type:

Dict[str, Any]

abstract merge(experiment_ids: List[str]) → None

Merge in the results of other (compatible) experiments trials. Used to help warm up the optimizer for this experiment.

Parameters:

experiment_ids (List[str]) – List of IDs of the experiments to merge in.

Return type:

None

new_trial(tunables: mlos_bench.tunables.tunable_groups.TunableGroups, ts_start: datetime.datetime | None = None, config: Dict[str, Any] | None = None) → Storage

Create a new experiment run in the storage.

Parameters:

• tunables (TunableGroups) – Tunable parameters to use for the trial.

• ts_start (Optional[datetime.datetime]) – Timestamp of the trial start (can be in the future).

• config (dict) – Key/value pairs of additional non-tunable parameters of the trial.

Returns:

trial – An object that allows to update the storage with the results of the experiment trial run.

Return type:

Storage.Trial

abstract pending_trials(timestamp: datetime.datetime, *, running: bool) → Iterator[Storage]

Return an iterator over the pending trials that are scheduled to run on or before the specified timestamp.

Parameters:

• timestamp (datetime.datetime) – The time in UTC to check for scheduled trials.

• running (bool) – If True, include the trials that are already running. Otherwise, return only the scheduled trials.

Returns:

trials – An iterator over the scheduled (and maybe running) trials.

Return type:

Iterator[Storage.Trial]

property description: str

Get the Experiment’s description.

Return type:

str

property experiment_id: str

Get the Experiment’s ID.

Return type:

str

property opt_targets: Dict[str, Literal['min', 'max']]

Get the Experiment’s optimization targets and directions.

Return type:

Dict[str, Literal[‘min’, ‘max’]]

property root_env_config: str

Get the Experiment’s root Environment config file path.

Return type:

str

property trial_id: int

Get the current Trial ID.

Return type:

int

property tunables: mlos_bench.tunables.tunable_groups.TunableGroups

Get the Experiment’s tunables.

Return type:

mlos_bench.tunables.tunable_groups.TunableGroups

class Trial(*, tunables: mlos_bench.tunables.tunable_groups.TunableGroups, experiment_id: str, trial_id: int, tunable_config_id: int, opt_targets: Dict[str, Literal['min', 'max']], config: Dict[str, Any] | None = None)

Base interface for storing the results of a single run of the experiment.

This class is instantiated in the Storage.Experiment.trial() method.

Parameters:

• tunables (mlos_bench.tunables.tunable_groups.TunableGroups)

• experiment_id (str)

• trial_id (int)

• tunable_config_id (int)

• opt_targets (Dict[str, Literal['min', 'max']])

• config (Optional[Dict[str, Any]])

__repr__() → str

Return type:

str

config(global_config: Dict[str, Any] | None = None) → Dict[str, Any]

Produce a copy of the global configuration updated with the parameters of the current trial.

Note: this is not the target Environment’s “config” (i.e., tunable params), but rather the internal “config” which consists of a combination of somewhat more static variables defined in the json config files.

Parameters:

global_config (Optional[Dict[str, Any]])

Return type:

Dict[str, Any]

abstract update(status: mlos_bench.environments.status.Status, timestamp: datetime.datetime, metrics: Dict[str, Any] | None = None) → Dict[str, Any] | None

Update the storage with the results of the experiment.

Parameters:

• status (Status) – Status of the experiment run.

• timestamp (datetime.datetime) – Timestamp of the status and metrics.

• metrics (Optional[Dict[str, Any]]) – One or several metrics of the experiment run. Must contain the (float) optimization target if the status is SUCCEEDED.

Returns:

metrics – Same as metrics, but always in the dict format.

Return type:

Optional[Dict[str, Any]]

abstract update_telemetry(status: mlos_bench.environments.status.Status, timestamp: datetime.datetime, metrics: List[Tuple[datetime.datetime, str, Any]]) → None

Save the experiment’s telemetry data and intermediate status.

Parameters:

• status (Status) – Current status of the trial.

• timestamp (datetime.datetime) – Timestamp of the status (but not the metrics).

• metrics (List[Tuple[datetime.datetime, str, Any]]) – Telemetry data.

Return type:

None

property opt_targets: Dict[str, Literal['min', 'max']]

Get the Trial’s optimization targets and directions.

Return type:

Dict[str, Literal[‘min’, ‘max’]]

property status: mlos_bench.environments.status.Status

Get the status of the current trial.

Return type:

mlos_bench.environments.status.Status

property trial_id: int

ID of the current trial.

Return type:

int

property tunable_config_id: int

ID of the current trial (tunable) configuration.

Return type:

int

property tunables: mlos_bench.tunables.tunable_groups.TunableGroups

Tunable parameters of the current trial.

(e.g., application Environment’s “config”)

Return type:

mlos_bench.tunables.tunable_groups.TunableGroups

abstract experiment(*, experiment_id: str, trial_id: int, root_env_config: str, description: str, tunables: mlos_bench.tunables.tunable_groups.TunableGroups, opt_targets: Dict[str, Literal['min', 'max']]) → Storage

Create a new experiment in the storage.

We need the opt_target parameter here to know what metric to retrieve when we load the data from previous trials. Later we will replace it with full metadata about the optimization direction, multiple objectives, etc.

Parameters:

• experiment_id (str) – Unique identifier of the experiment.

• trial_id (int) – Starting number of the trial.

• root_env_config (str) – A path to the root JSON configuration file of the benchmarking environment.

• description (str) – Human-readable description of the experiment.

• tunables (TunableGroups)

• opt_targets (Dict[str, Literal["min", "max"]]) – Names of metrics we’re optimizing for and the optimization direction {min, max}.

Returns:

experiment – An object that allows to update the storage with the results of the experiment and related data.

Return type:

Storage.Experiment

property experiments: Dict[str, mlos_bench.storage.base_experiment_data.ExperimentData]

Abstractmethod:

Return type:

Dict[str, mlos_bench.storage.base_experiment_data.ExperimentData]

Retrieve the experiments’ data from the storage.

Returns:

experiments – A dictionary of the experiments’ data, keyed by experiment id.

Return type:

Dict[str, ExperimentData]