API Reference¶

OptimizerClient¶

class kubeflow.optimizer.OptimizerClient(backend_config: KubernetesBackendConfig | None = None)[source]¶

Bases: object

__init__(backend_config: KubernetesBackendConfig | None = None)[source]¶

Initialize a Kubeflow Optimizer client.

Parameters:: backend_config (KubernetesBackendConfig | None) – Backend configuration. Either KubernetesBackendConfig or None to use the default config class. Defaults to None (uses KubernetesBackendConfig).
Raises:: ValueError – If the backend configuration is invalid.

Examples

>>> from kubeflow.optimizer import OptimizerClient
>>> client = OptimizerClient()

optimize(trial_template: TrainJobTemplate, *, trial_config: TrialConfig | None = None, search_space: dict[str, Any], objectives: list[Objective] | None = None, algorithm: BaseAlgorithm | None = None) → str[source]¶

Create and submit an OptimizationJob for hyperparameter tuning.

Parameters:

trial_template (TrainJobTemplate) – The TrainJob template defining the training script.
trial_config (TrialConfig | None) – Optional configuration to run Trials.
search_space (dict[str, Any]) – Dictionary mapping parameter names to Search specifications using Search.uniform(), Search.loguniform(), Search.choice(), etc.
objectives (list[Objective] | None) – Optional list of objectives to optimize. Defaults to minimizing the “loss” metric.
algorithm (BaseAlgorithm | None) – The optimization algorithm to use. Defaults to RandomSearch.

Returns:

The unique name generated for the OptimizationJob (Experiment).

Raises:

ValueError – If input arguments are invalid.
TimeoutError – Timeout occurred while creating the Experiment.
RuntimeError – Failed to create the Experiment.

Examples

>>> from kubeflow.trainer import TrainJobTemplate, CustomTrainer
>>> from kubeflow.optimizer import OptimizerClient, Search, TrialConfig
>>> def train_fn(learning_rate, num_epochs):
...     pass
>>> template = TrainJobTemplate(runtime="torch-distributed", trainer=CustomTrainer(func=train_fn))
>>> client = OptimizerClient()
>>> opt_id = client.optimize(
...     trial_template=template,
...     trial_config=TrialConfig(num_trials=5, parallel_trials=2),
...     search_space={
...         "learning_rate": Search.loguniform(0.001, 0.1),
...         "num_epochs": Search.choice([5, 10]),
...     },
... )
>>> print(opt_id)

list_jobs() → list[OptimizationJob][source]¶

List created OptimizationJobs.

Returns:

List of created OptimizationJobs. If no OptimizationJobs exist, an empty list is returned.

Raises:

TimeoutError – Timeout occurred while listing OptimizationJobs.
RuntimeError – Failed to list OptimizationJobs.

Examples

>>> from kubeflow.optimizer import OptimizerClient
>>> client = OptimizerClient()
>>> jobs = client.list_jobs()
>>> for job in jobs:
...     print(job.name)

get_job(name: str) → OptimizationJob[source]¶

Get the OptimizationJob object by name.

Parameters:

name (str) – Name of the OptimizationJob.

Returns:

An OptimizationJob object.

Raises:

TimeoutError – Timeout occurred while getting the OptimizationJob.
RuntimeError – Failed to get the OptimizationJob.

Examples

>>> from kubeflow.optimizer import OptimizerClient
>>> client = OptimizerClient()
>>> job = client.get_job("opt-12345")
>>> print(job.status)

get_job_logs(name: str, trial_name: str | None = None, follow: bool = False) → Iterator[str][source]¶

Get logs from a specific trial of an OptimizationJob.

Parameters:

name (str) – Name of the OptimizationJob.
trial_name (str | None) – Optional name of a specific Trial. If not provided, logs from the current best trial are returned. If no best trial is available yet, logs from the first trial are returned.
follow (bool) – Whether to stream logs in realtime as they are produced. Defaults to False.

Returns:

Iterator of log lines.

Raises:

TimeoutError – Timeout occurred while getting the OptimizationJob logs.
RuntimeError – Failed to get the OptimizationJob logs.

Examples

>>> from kubeflow.optimizer import OptimizerClient
>>> client = OptimizerClient()
>>> for line in client.get_job_logs(name="opt-12345"):
...     print(line)

get_best_results(name: str) → Result | None[source]¶

Get the best hyperparameters and metrics from an OptimizationJob.

This method retrieves the optimal hyperparameters and their corresponding metrics from the best trial found during the optimization process.

Parameters:

name (str) – Name of the OptimizationJob.

Returns:

A Result object containing the best hyperparameters and metrics, or None if no best trial is available yet.

Raises:

TimeoutError – Timeout occurred while getting the best results.
RuntimeError – Failed to get the best results for the OptimizationJob.

Examples

>>> from kubeflow.optimizer import OptimizerClient
>>> client = OptimizerClient()
>>> best_res = client.get_best_results("opt-12345")
>>> if best_res:
...     print(best_res.parameters)

wait_for_job_status(name: str, status: set[str] = {'Complete'}, timeout: int = 3600, polling_interval: int = 2, callbacks: list[Callable[[OptimizationJob], None]] | None = None) → OptimizationJob[source]¶

Wait for an OptimizationJob to reach a desired status.

Parameters:

name (str) – Name of the OptimizationJob.
status (set[str]) – Expected statuses. Must be a subset of Created, Running, Complete, and Failed statuses. Defaults to Complete.
timeout (int) – Maximum number of seconds to wait for the OptimizationJob to reach one of the expected statuses. Defaults to 3600.
polling_interval (int) – The polling interval in seconds to check OptimizationJob status. Defaults to 2.
callbacks (list[Callable[[OptimizationJob], None]] | None) – Optional list of callback functions to be invoked after each polling interval. Each callback should accept a single argument: the OptimizationJob object.

Returns:

An OptimizationJob object that reaches the desired status.

Raises:

ValueError – The input values are incorrect.
RuntimeError – Failed to get OptimizationJob or OptimizationJob reaches unexpected Failed status.
TimeoutError – Timeout occurred while waiting for the OptimizationJob status.

Examples

>>> from kubeflow.optimizer import OptimizerClient
>>> client = OptimizerClient()
>>> job = client.wait_for_job_status(name="opt-12345")
>>> print(job.status)

delete_job(name: str)[source]¶

Delete the OptimizationJob.

Parameters:

name (str) – Name of the OptimizationJob.

Raises:

TimeoutError – Timeout occurred while deleting the OptimizationJob.
RuntimeError – Failed to delete the OptimizationJob.

Examples

>>> from kubeflow.optimizer import OptimizerClient
>>> client = OptimizerClient()
>>> client.delete_job("opt-12345")

get_job_events(name: str) → list[Event][source]¶

Get events for an OptimizationJob.

This provides additional clarity about the state of the OptimizationJob when logs alone are not sufficient. Events include information about trial state changes, errors, and other significant occurrences.

Parameters:

name (str) – Name of the OptimizationJob.

Returns:

A list of Event objects associated with the OptimizationJob.

Raises:

TimeoutError – Timeout occurred while getting the OptimizationJob events.
RuntimeError – Failed to get the OptimizationJob events.

Examples

>>> from kubeflow.optimizer import OptimizerClient
>>> client = OptimizerClient()
>>> events = client.get_job_events("opt-12345")
>>> for event in events:
...     print(f"[{event.event_time}] {event.message}")

Search Space¶

class kubeflow.optimizer.Search[source]¶

Bases: object

Helper class for defining search space parameters.

static uniform(min: float, max: float) → V1beta1ParameterSpec[source]¶

Sample a float value uniformly between min and max.

Parameters:

min (float) – Lower boundary for the float value.
max (float) – Upper boundary for the float value.

Returns:

Katib ParameterSpec object.

static loguniform(min: float, max: float) → V1beta1ParameterSpec[source]¶

Sample a float value with log-uniform distribution between min and max.

Parameters:

min (float) – Lower boundary for the float value.
max (float) – Upper boundary for the float value.

Returns:

Katib ParameterSpec object.

static choice(values: list) → V1beta1ParameterSpec[source]¶

Sample a categorical value from the list.

Parameters:: values (list) – List of categorical values.
Returns:: Katib ParameterSpec object.

Configuration¶

class kubeflow.optimizer.TrialConfig(num_trials: int = 10, parallel_trials: int = 1, max_failed_trials: int | None = None) → None[source]¶

Bases: object

Trial configuration for hyperparameter optimization.

Parameters:

num_trials (int) – Number of trials to run. Defaults to 10.
parallel_trials (int) – Number of trials to run in parallel. Defaults to 1.
max_failed_trials (Optional[int]) – Maximum number of failed trials before stopping.

num_trials: int = 10¶

parallel_trials: int = 1¶

max_failed_trials: int | None = None¶

class kubeflow.optimizer.Objective(metric: str = 'loss', direction: Direction = Direction.MINIMIZE) → None[source]¶

Bases: object

Objective configuration for hyperparameter optimization.

Parameters:

metric (str) – The name of the metric to optimize. Defaults to “loss”.
direction (Direction) – Whether to maximize or minimize the metric. Defaults to “minimize”.

metric: str = 'loss'¶

direction: Direction = 'minimize'¶