Experiment

module crystallize.experiments.experiment

Global Variables

• TYPE_CHECKING
• VALID_EXECUTOR_TYPES
• METADATA_FILENAME
• BASELINE_CONDITION
• REPLICATE_KEY
• CONDITION_KEY
• SEED_USED_KEY

---

class Experiment

method Experiment.__init__

__init__(
    datasource: 'DataSource',
    pipeline: 'Pipeline',
    plugins: 'Optional[List[BasePlugin]]' = None,
    description: 'str | None' = None,
    name: 'str | None' = None,
    initial_ctx: 'Dict[str, Any] | None' = None,
    outputs: 'List[Artifact] | None' = None,
    treatments: 'List[Treatment] | None' = None,
    hypotheses: 'List[Hypothesis] | None' = None,
    replicates: 'int' = 1
) → None

Instantiate an experiment configuration.

Args:

• datasource: Object that provides the initial data for each run.
• pipeline: Pipeline executed for every replicate.
• plugins: Optional list of plugins controlling experiment behaviour.
• description: Optional text describing this experiment.
• name: Optional experiment name used for artifact storage.

---

property Experiment.hypotheses

---

property Experiment.replicates

---

property Experiment.treatments

---

method Experiment.aoptimize

aoptimize(
    optimizer: "'BaseOptimizer'",
    num_trials: 'int',
    replicates_per_trial: 'int' = 1
) → Treatment

---

method Experiment.apply

apply(
    treatment: 'Treatment | None' = None,
    data: 'Any | None' = None,
    seed: 'Optional[int]' = None
) → Any

Run the pipeline once and return the output.

This method mirrors :meth:run for a single replicate. Plugin hooks are executed and all pipeline steps receive setup and teardown calls.

---

method Experiment.artifact_datasource

artifact_datasource(
    step: 'str',
    name: 'str' = 'data.json',
    condition: 'str' = 'baseline',
    require_metadata: 'bool' = False
) → DataSource

Return a datasource providing :class:pathlib.Path objects to artifacts.

Parameters ---------- step: Pipeline step name that produced the artifact. name: Artifact file name. condition: Condition directory to load from. Defaults to "baseline". require_metadata: If True and metadata.json does not exist, raise a FileNotFoundError. When False (default), missing metadata means replicates are inferred from the experiment instance.

---

method Experiment.arun

arun(
    treatments: 'List[Treatment] | None' = None,
    hypotheses: 'List[Hypothesis] | None' = None,
    replicates: 'int | None' = None,
    strategy: 'str | None' = None
) → Result

Execute the experiment and return a :class:Result instance.

The lifecycle proceeds as follows:

1. before_run hooks for all plugins are invoked. 2. Each replicate is executed via run_experiment_loop. The default implementation runs serially, but plugins may provide parallel or distributed strategies. 3. After all replicates complete, metrics are aggregated and hypotheses are verified. 4. after_run hooks for all plugins are executed.

The returned :class:~crystallize.experiments.result.Result contains aggregated metrics, any captured errors and a provenance record of context mutations for every pipeline step.

---

classmethod Experiment.builder

builder(name: 'str | None' = None) → 'ExperimentBuilder'

Return a fluent builder for constructing an Experiment.

---

classmethod Experiment.from_yaml

from_yaml(config_path: 'str | Path') → 'Experiment'

Instantiate an experiment from a folder-based YAML config.

---

method Experiment.get_plugin

get_plugin(plugin_class: 'type') → Optional[BasePlugin]

Return the first plugin instance matching plugin_class.

---

method Experiment.optimize

optimize(
    optimizer: "'BaseOptimizer'",
    num_trials: 'int',
    replicates_per_trial: 'int' = 1
) → Treatment

Synchronous wrapper for :meth:aoptimize.

---

method Experiment.run

run(
    treatments: 'List[Treatment] | None' = None,
    hypotheses: 'List[Hypothesis] | None' = None,
    replicates: 'int | None' = None,
    strategy: 'str | None' = None
) → Result

Synchronous wrapper for the async run method. Convenient for tests and scripts. Checks for an active event loop before dispatching work and raises a clear error in notebook contexts so you can switch to the async variant. Replicates flow through the pipeline, and ResultAggregator collates metrics once execution completes.

Use this pattern in Jupyter and other environments that keep an event loop alive:

import asyncio
if asyncio.get_running_loop():
    await exp.arun()
else:
    exp.run()

---

method Experiment.set_default_plugins

set_default_plugins() → None

---

method Experiment.validate

validate() → None