Imported from experimaestro/experimaestro-python (.agents/skills/experimaestro/SKILL.md). Install upstream with npx skills add experimaestro/experimaestro-python --skill experimaestro. Copyright stays with the author.

Experimaestro

Official experimaestro skill for writing experiment code with best practices and correct patterns.

Experimaestro is a Python experiment manager and workflow orchestrator. It manages task automation, dependencies, resource allocation, and reproducibility through a type system based on Config and Task classes.

Documentation: https://experimaestro-python.readthedocs.io

Python 3.10+ Syntax

Always use modern Python syntax:

• dict[str, int], list[Path], tuple[int, ...] instead of Dict, List, Tuple
• X | Y instead of Union[X, Y]; X | None instead of Optional[X]

Defining Configurations

Configurations inherit from Config and use Param[T] for parameters:

from experimaestro import Config, Param, field

class MyModel(Config):
    """A model configuration.

    Attributes:
        hidden_size: Size of hidden layers
        num_layers: Number of transformer layers
    """
    hidden_size: Param[int]
    num_layers: Param[int] = field(default=6, ignore_default=True)

Supported types

• Primitives: str, int, float, bool, Path
• Collections: list[T], set[T], dict[U, V]
• Enumerations: enum.Enum subclasses
• Other Config subclasses (nested configurations)

Parameter defaults with field()

Always use field() for defaults:

from experimaestro import Config, Param, field

class MyConfig(Config):
    # Default always included in identifier computation
    x: Param[int] = field(default=1)

    # Default ignored in identifier when value == default
    # Use this when adding new parameters to existing configs
    y: Param[int] = field(default=1, ignore_default=True)

    # Factory default
    z: Param[SomeConfig] = field(default_factory=SomeConfig.C)

When to use which:

Metadata (not part of identifier)

Use Meta[T] for parameters ignored during identifier computation:

from experimaestro import Config, Meta

class MyConfig(Config):
    name: Meta[str]  # Not included in identifier

Path generators

For output paths relative to the task directory:

from experimaestro import Task, Meta, PathGenerator, field
from pathlib import Path

class MyTask(Task):
    output: Meta[Path] = field(default_factory=PathGenerator("output.txt"))

Constants

Use Constant[T] for values that signal behavior changes:

from experimaestro import Config, Constant

class MyConfig(Config):
    version: Constant[str] = "2.1"

Sets

Sets are supported. Order does not matter for identifiers.

For Config objects in sets, elements must be sealed first:

from experimaestro import Config, Param, sealed_set

class Model(Config):
    lr: Param[float]

class Ensemble(Config):
    models: Param[set[Model]]

m1 = Model.C(lr=0.01)
m2 = Model.C(lr=0.02)
ensemble = Ensemble.C(models=sealed_set(m1, m2))

Composition operator (@)

The @ operator composes configurations when the target parameter type is unambiguous:

class Inner(Config):
    x: Param[int]

class Outer(Config):
    inner: Param[Inner]

# These are equivalent:
outer1 = Outer.C(inner=Inner.C(x=42))
outer2 = Outer.C() @ Inner.C(x=42)

Validation

Use __validate__ for parameter validation before submission:

class ModelLearn(Config):
    batch_size: Param[int] = field(default=100, ignore_default=True)
    micro_batch_size: Param[int] = field(default=100, ignore_default=True)

    def __validate__(self):
        assert self.batch_size % self.micro_batch_size == 0

Object lifecycle

During task execution:

1. __init__() is called
2. Attributes are set
3. __post_init__() is called (if defined)
4. Pre-tasks run (if any)

Use @initializer for one-time deferred initialization.

Defining Tasks

Tasks inherit from Task and implement execute():

from experimaestro import Task, Param, Meta, PathGenerator, field
from pathlib import Path

class ModelLearn(Task):
    """Train a model.

    Attributes:
        epochs: Number of training epochs
        model: The model to train
    """
    epochs: Param[int] = field(default=100, ignore_default=True)
    model: Param[Model]
    parameters: Meta[Path] = field(default_factory=PathGenerator("parameters.pth"))

    def execute(self):
        # Task implementation
        # self.__taskdir__ is the task directory
        # self.__tags__ contains tags as a dictionary
        for epoch in range(self.epochs):
            train_one_epoch(self.model)
        save(self.parameters)

Task outputs with __submit__

Override __submit__ to customize task outputs and register actions. The add_action callable registers actions with the experiment. Use **kwargs to ensure forward compatibility:

class RandomFold(Task):
    dataset: Param[Dataset]
    topics: Param[Path] = field(default_factory=PathGenerator("topics.tsv"))

    def __submit__(self, dep, add_action, **kwargs) -> Dataset:
        return dep(Dataset.C(
            topics=dep(Topics.C(path=self.topics)),
            documents=self.dataset.documents,  # No dep() = no dependency
        ))

Legacy task_outputs(self, dep) still works for backward compatibility.

Task dependencies

Tasks automatically resolve dependencies:

task1 = Task1.C(...).submit()
task2 = Task2.C(...).submit()
task3 = Task3.C(inputs=[task1, task2]).submit()  # Waits for task1 and task2

Getting a task's identifier

Every Config/Task has a unique identifier computed from its parameters. Use __identifier__() to retrieve it:

task = MyTask.C(x=1, y="a")

ident = task.__identifier__()    # Identifier object
hexid = ident.all.hex()          # hex hash that names the job directory
str(ident)                       # same hex string

# Task class id (the __xpmid__ / type name)
type_id = str(task.__xpmtype__.identifier)

The job directory is workdir/jobs/<type_id>/<hexid>/. The identifier reflects the current parameter values; it is frozen once the config is sealed (at submit time), so changing a parameter afterwards is not allowed.

Resumable Tasks

For long-running tasks that may timeout (e.g., SLURM walltime):

from experimaestro import ResumableTask, GracefulTimeout, Param, Meta, PathGenerator, field
from pathlib import Path

class LongTraining(ResumableTask):
    epochs: Param[int] = field(default=1000, ignore_default=True)
    checkpoint: Meta[Path] = field(default_factory=PathGenerator("checkpoint.pth"))

    def execute(self):
        start_epoch = 0
        if self.checkpoint.exists():
            start_epoch = load_checkpoint(self.checkpoint)

        for epoch in range(start_epoch, self.epochs):
            remaining = self.remaining_time()
            if remaining is not None and remaining < 300:
                save_checkpoint(self.checkpoint, epoch)
                raise GracefulTimeout("Not enough time for another epoch")

            train_one_epoch()
            save_checkpoint(self.checkpoint, epoch)

Key points:

• remaining_time() returns seconds remaining (SLURM) or None (local)
• Raise GracefulTimeout to stop cleanly and trigger retry
• Only timeout failures trigger retries (not other errors)
• Configure retries: task.submit(max_retries=10) or globally in settings

Graceful termination (SIGTERM/SIGINT)

Catch TaskCancelled for custom cleanup:

from experimaestro import Task, TaskCancelled

class MyTask(Task):
    def execute(self):
        try:
            do_work()
        except TaskCancelled:
            save_partial_state()
            raise  # Always re-raise

Running Experiments

Python API

from experimaestro import experiment

with experiment(workdir, "my-experiment", port=12345) as xp:
    task = MyTask.C(param1="value", param2=42).submit()

CLI with YAML configuration

# experiment.py
from experimaestro.experiments import ExperimentHelper, configuration, ConfigurationBase

@configuration
class Configuration(ConfigurationBase):
    learning_rate: float = 1e-3

def run(helper: ExperimentHelper, cfg: Configuration):
    with helper.experiment() as xp:
        MyTask.C(lr=cfg.learning_rate).submit()

# experiment.yaml
id: my-experiment
file: experiment
learning_rate: 1e-4

experimaestro run-experiment --run-mode NORMAL experiment.yaml

Override from CLI:

experimaestro run-experiment -c learning_rate=1e-5 experiment.yaml

Launchers

Direct (local)

Default launcher, runs tasks as local processes.

SLURM

from experimaestro.launchers.slurm import SlurmLauncher, SlurmOptions

launcher = SlurmLauncher(options=SlurmOptions(nodes=1))
# .config(**kwargs) returns a new launcher with the options merged in
gpulauncher = launcher.config(gpus_per_node=1)

with experiment(workdir, "my-experiment", launcher=launcher) as xp:
    # Default launcher
    task1.submit()

    # GPU launcher for specific tasks
    task2.submit(launcher=gpulauncher)

Launcher finder (launchers.py)

Define how requirements map to launchers in ~/.config/experimaestro/launchers.py:

from experimaestro.launcherfinder import HostRequirement, HostSpecification, CudaSpecification, CPUSpecification
from experimaestro.launchers.slurm import SlurmLauncher, SlurmOptions
from experimaestro.launchers.direct import DirectLauncher
from experimaestro.connectors.local import LocalConnector

def find_launcher(requirements: HostRequirement, tags: set[str] = set()):
    if match := requirements.match(HostSpecification(accelerators=[])):
        return DirectLauncher(connector=LocalConnector.instance())

    if match := requirements.match(
        HostSpecification(
            max_duration=100 * 3600,
            cpu=CPUSpecification(cores=32, memory=129 * (1024**3)),
            accelerators=[CudaSpecification(memory=24 * (1024**3)) for _ in range(8)],
        )
    ):
        return SlurmLauncher(
            connector=LocalConnector.instance(),
            options=SlurmOptions(gpus_per_node=len(match.requirement.accelerators)),
        )

    return None

Use in code:

from experimaestro.launcherfinder import find_launcher

launcher = find_launcher("duration=4h & cuda(mem=8GiB) * 2 & cpu(mem=16GiB)")

Dynamic launcher

Select from multiple launchers based on priority:

from experimaestro.launchers import DynamicLauncher
from experimaestro.launchers.slurm import SlurmLauncher, SlurmOptions

fast = SlurmLauncher(options=SlurmOptions(partition="fast"), priority=10)
slow = SlurmLauncher(options=SlurmOptions(partition="slow"), priority=1)
dynamic = DynamicLauncher([fast, slow])

Dynamic Task Outputs

For tasks producing outputs during execution (e.g., checkpoints):

from experimaestro import ResumableTask, Config, Param, DependentMarker

class Checkpoint(Config):
    step: Param[int]
    model: Param[Model]

class Validation(Config):
    model: Param[Model]

    def checkpoint(self, dep: DependentMarker, *, step: int) -> Checkpoint:
        return dep(Checkpoint.C(model=self.model, step=step))

    def compute(self, step: int):
        self.register_task_output(self.checkpoint, step=step)

class Learn(ResumableTask):
    model: Param[Model]
    validation: Param[Validation]

    def execute(self):
        for step in range(100):
            train_step()
            if step % 10 == 0:
                self.validation.compute(step)

# Watch outputs
def on_checkpoint(checkpoint: Checkpoint):
    Evaluate.C(checkpoint=checkpoint).submit()

learn = Learn.C(model=model, validation=validation)
learn.watch_output(validation.checkpoint, on_checkpoint)
learn.submit()

Prepare (data preparation)

Prepare is a Config declaring an in-process step that runs before any task depending on it (downloading a dataset, fetching credentials, warming a cache). Subclass it and override prepare():

from experimaestro import Prepare, Task, Param

class DatasetPrep(Prepare):
    name: Param[str]

    def prepare(self) -> None:
        # Must be idempotent: the underlying tool should no-op when cache is warm
        actually_fetch_dataset(self.name)

class Train(Task):
    dataset: Param[DatasetPrep]

    def execute(self):
        ...

# Referencing a Prepare in a task's params auto-attaches an in-memory dependency
Train.C(dataset=DatasetPrep.C(name="foo")).submit()

Key points:

• No on-disk footprint: a Prepare is a resource, not a job — no workdir under jobs/, no .done marker. Idempotence is prepare()'s responsibility.
• Dedup by identifier: instances with the same params share one prepare() call per process; distinct ones run concurrently.
• prepare() runs in the driver process (via asyncio.to_thread).
• RunMode.PREPARE: run only the discovered Prepares, skipping tasks — useful to pre-warm caches before submitting to an offline cluster:

experimaestro run-experiment --run-mode PREPARE my_experiment.yaml

Deprecation

Deprecating a configuration

When renaming or restructuring a config, use @deprecate to preserve identifier compatibility:

from experimaestro import Config, Param, deprecate

class NewConfig(Config):
    values: Param[list[int]]

@deprecate(NewConfig)
class OldConfig(Config):
    value: Param[int]

    def __convert__(self):
        return NewConfig(values=[self.value])

Deprecating a parameter

class Learning(Config):
    losses: Param[list[Loss]]

    @deprecate
    def loss(self, value):
        assert len(self.losses) == 0
        self.losses.append(value)

Fix deprecated identifiers:

experimaestro deprecated list WORKDIR

Value Classes

Separate configuration from implementation (e.g., avoid importing PyTorch at config time):

from experimaestro import Config, Param

class Model(Config):
    hidden_size: Param[int]

@Model.value_class()
class TorchModel(Model):
    def __post_init__(self):
        import torch.nn as nn
        self.layers = nn.Linear(self.hidden_size, self.hidden_size)

    def forward(self, x):
        return self.layers(x)

Caching with @cache

For lightweight computations worth caching without a full task:

from experimaestro import Config, cache
from pathlib import Path
import numpy as np

class Terms(Config):
    @cache("terms.npy")
    def load(self, path: Path):
        if path.is_file():
            return np.load(path)
        weights = self.compute_weights()
        np.save(path, weights)
        return weights

Workspace Settings

Configure in ~/.config/experimaestro/settings.yaml:

workspaces:
  - id: my-workspace
    path: ~/experiments
    env:
      CUDA_VISIBLE_DEVICES: "0"
    triggers:
      - "my-experiment-*"
    max_retries: 3

history:
  max_done: 5
  max_failed: 1

Monitoring

# TUI (terminal)
experimaestro experiments --workdir /path/to/workdir monitor --console

# Web UI
experimaestro experiments --workdir /path/to/workdir monitor --port 12345

Experiment Actions (alpha)

Actions are user-defined operations (Config subclasses) registered during submission and executable after completion. They support interactive user input.

Defining an Action

from experimaestro import Action, Interaction, Param

class ExportToHub(Action):
    model: Param[Model]

    def describe(self) -> str:
        return "Export model to HF Hub"

    def execute(self, interaction: Interaction) -> None:
        name = interaction.text("name", "Hub model name:", default="my-model")
        self.model.push_to_hub(name)

Registering actions via __submit__

class TrainModel(Task):
    model: Param[Model]

    def __submit__(self, dep, add_action, **kwargs):
        add_action(ExportToHub.C(model=self.model))
        return self

Running actions

# List actions
experimaestro experiments actions list my-experiment

# Run interactively
experimaestro experiments actions run my-experiment <action_id>

# Pre-fill answers
experimaestro experiments actions run my-experiment <action_id> --set name=my-model

In the TUI, select an experiment, go to the Actions tab, and press Enter.

Analyzing Experiment Results

Loading experiment data

Job configs and actions are streamed to objects.jsonl during submission. Use load_xp_info() to load them:

from experimaestro import load_xp_info

info = load_xp_info("/path/to/workspace/experiments/my-exp/20260319_120000")
# info.jobs: dict[str, Config] — job configs with shared references preserved
# info.actions: dict[str, Action] — registered actions

Via WorkspaceStateProvider (lookup by experiment name):

from experimaestro.scheduler.workspace_state_provider import WorkspaceStateProvider
provider = WorkspaceStateProvider(workspace_path)
info = provider.load_xp_info("my-experiment")

Building DataFrames with tags

Use provider.get_tags_map() to combine tags with configs:

from experimaestro import load_xp_info
import pandas as pd

info = load_xp_info("/path/to/run_dir")
# tags_map from provider.get_tags_map()
rows = []
for job_id, config in info.jobs.items():
    row = tags_map.get(job_id, {}).copy()
    row["job_id"] = job_id
    row["param"] = config.some_param
    rows.append(row)
df = pd.DataFrame(rows)

Common Pitfalls

1. Identifier stability: Changing parameter names, types, or __xpmid__ breaks caching. Use deprecate for backward compatibility.
2. Unsealed configs in sets: Always use sealed_set() or .seal() before putting configs in set[T] parameters.
3. RPyC serialization: Keep task parameters simple (primitives, Paths, Config objects).
4. Workspace locking: Don't run multiple experiments in the same workspace simultaneously.