Approach: Use Optuna's study API with a pruner (e.g., MedianPruner). Inside the objective, sample hyperparameters, build model, train in epochs; after each epoch evaluate on validation set, report intermediate value to Optuna (trial.report) and call trial.should_prune(); log metrics and hyperparams to an experiment store (could be MLflow, a DB, or simple JSON file); checkpoint model after each improved val metric to a trial-specific path. Catch TrialPruned to persist final state and re-raise so Optuna records pruning.python
import optuna
import json
import os
import shutil
import random
import numpy as np
EXPERIMENT_DIR = "/tmp/experiment_store"
def save_metadata(trial_id, metadata):
os.makedirs(EXPERIMENT_DIR, exist_ok=True)
with open(os.path.join(EXPERIMENT_DIR, f"trial_{trial_id}.json"), "w") as f:
json.dump(metadata, f)
def save_checkpoint(trial_id, epoch, model_state, best=False):
path = os.path.join(EXPERIMENT_DIR, f"trial_{trial_id}_epoch_{epoch}.ckpt")
# pseudo-save: replace with torch.save or joblib.dump
with open(path, "wb") as f:
f.write(b"MODEL_STATE")
if best:
best_path = os.path.join(EXPERIMENT_DIR, f"trial_{trial_id}_best.ckpt")
shutil.copy(path, best_path)
def objective(trial):
# reproducibility per trial
seed = 42 + trial.number
random.seed(seed); np.random.seed(seed)
# sample hyperparams
lr = trial.suggest_loguniform("lr", 1e-5, 1e-1)
hidden = trial.suggest_int("hidden", 32, 512)
# build model/trainer (pseudo)
model = build_model(hidden)
optimizer = build_optimizer(model, lr)
best_val = float("inf")
metadata = {"trial_id": trial.number, "params": trial.params, "metrics": []}
try:
for epoch in range(1, 51):
train_one_epoch(model, optimizer, epoch)
val_metric = evaluate(model, epoch) # lower is better (e.g., val loss)
# report to Optuna for pruning decisions
trial.report(val_metric, epoch)
# log to experiment store
metadata["metrics"].append({"epoch": epoch, "val": float(val_metric)})
save_metadata(trial.number, metadata)
# checkpoint policy: save on improvement and every N epochs
improved = val_metric < best_val
if improved:
best_val = val_metric
save_checkpoint(trial.number, epoch, model.state_dict(), best=True)
elif epoch % 10 == 0:
save_checkpoint(trial.number, epoch, model.state_dict(), best=False)
# pruning check
if trial.should_prune():
# optional: save final metadata/checkpoint before pruning
save_metadata(trial.number, metadata)
raise optuna.exceptions.TrialPruned()
# finalize
save_metadata(trial.number, metadata)
return best_val
except Exception:
# ensure metadata saved on unexpected errors
save_metadata(trial.number, metadata)
raise
# set up study with Bayesian sampler and pruner
sampler = optuna.samplers.TPESampler(seed=123)
pruner = optuna.pruners.MedianPruner(n_startup_trials=5, n_warmup_steps=3)
study = optuna.create_study(direction="minimize", sampler=sampler, pruner=pruner)
study.optimize(objective, n_trials=100, n_jobs=1)
Key points:- report intermediate values with trial.report and call trial.should_prune to enable pruning.- checkpoint consistently (best and periodic), store trial metadata atomically.- Catch TrialPruned to persist state so experiments are reproducible.Complexity: overhead proportional to epochs × eval cost; pruning reduces overall compute by stopping poor trials early.Edge cases: ensure checkpoint atomicity, handle failed saves, choose appropriate n_warmup_steps to avoid premature pruning.Alternative: scikit-optimize can be used but lacks built-in pruning — you'd implement pruning logic manually.