Skip to content
Crystallize Documentation

Plugins

module crystallize.plugins.plugins

Section titled “module crystallize.plugins.plugins”

Lifecycle Hooks

Section titled “Lifecycle Hooks”

Hooks fire in this order during Experiment.run:

  1. init_hook(experiment) – called when the plugin is attached to the experiment.
  2. before_run(experiment) – once before any replicates start.
  3. For each replicate:
    • before_replicate(experiment, ctx) – before fetching data and running steps.
    • For each step:
      • before_step(experiment, step) – just before the step executes.
      • after_step(experiment, step, data, ctx) – immediately after the step finishes and context/metrics are updated.
  4. after_run(experiment, result) – after all replicates finish and results are aggregated.

Global Variables

Section titled “Global Variables”
  • TYPE_CHECKING
  • REPLICATE_KEY
  • CONDITION_KEY
  • BASELINE_CONDITION
  • METADATA_FILENAME
  • SEED_USED_KEY

function default_seed_function

Section titled “function default_seed_function”
default_seed_function(seed: 'int') → None

Set deterministic seeds for common libraries if available.

class BasePlugin

Section titled “class BasePlugin”

Interface for extending the :class:~crystallize.experiments.experiment.Experiment lifecycle.

Subclasses can override any of the hook methods to observe or modify the behaviour of an experiment. Hooks are called in a well-defined order during :meth:Experiment.run allowing plugins to coordinate tasks such as seeding, logging, artifact storage or custom execution strategies.

method BasePlugin.after_run

Section titled “method BasePlugin.after_run”
after_run(experiment: 'Experiment', result: 'Result') → None

Execute cleanup or reporting after :meth:Experiment.run completes.

method BasePlugin.after_step

Section titled “method BasePlugin.after_step”
after_step(
    experiment: 'Experiment',
    step: 'PipelineStep',
    data: 'Any',
    ctx: 'FrozenContext'
) → None

Observe results after every :class:PipelineStep execution.

method BasePlugin.before_replicate

Section titled “method BasePlugin.before_replicate”
before_replicate(experiment: 'Experiment', ctx: 'FrozenContext') → None

Run prior to each pipeline execution for a replicate.

method BasePlugin.before_run

Section titled “method BasePlugin.before_run”
before_run(experiment: 'Experiment') → None

Execute logic before :meth:Experiment.run begins.

method BasePlugin.before_step

Section titled “method BasePlugin.before_step”
before_step(experiment: 'Experiment', step: 'PipelineStep') → None

method BasePlugin.init_hook

Section titled “method BasePlugin.init_hook”
init_hook(experiment: 'Experiment') → None

Configure the experiment instance during initialization.

method BasePlugin.run_experiment_loop

Section titled “method BasePlugin.run_experiment_loop”
run_experiment_loop(
    experiment: "'Experiment'",
    replicate_fn: 'Callable[[int], Any]'
) → List[Any]

Run all replicates and return their results.

Returning NotImplemented signals that the plugin does not provide a custom execution strategy and the default should be used instead.

class SeedPlugin

Section titled “class SeedPlugin”

Manage deterministic seeding for all random operations.

method SeedPlugin.__init__

Section titled “method SeedPlugin.__init__”
__init__(
    seed: 'Optional[int]' = None,
    auto_seed: 'bool' = True,
    seed_fn: 'Optional[Callable[[int], None]]' = None
) → None

method SeedPlugin.after_run

Section titled “method SeedPlugin.after_run”
after_run(experiment: 'Experiment', result: 'Result') → None

Execute cleanup or reporting after :meth:Experiment.run completes.

method SeedPlugin.after_step

Section titled “method SeedPlugin.after_step”
after_step(
    experiment: 'Experiment',
    step: 'PipelineStep',
    data: 'Any',
    ctx: 'FrozenContext'
) → None

Observe results after every :class:PipelineStep execution.

method SeedPlugin.before_replicate

Section titled “method SeedPlugin.before_replicate”
before_replicate(experiment: 'Experiment', ctx: 'FrozenContext') → None

method SeedPlugin.before_run

Section titled “method SeedPlugin.before_run”
before_run(experiment: 'Experiment') → None

Execute logic before :meth:Experiment.run begins.

method SeedPlugin.before_step

Section titled “method SeedPlugin.before_step”
before_step(experiment: 'Experiment', step: 'PipelineStep') → None

method SeedPlugin.init_hook

Section titled “method SeedPlugin.init_hook”
init_hook(experiment: 'Experiment') → None

method SeedPlugin.run_experiment_loop

Section titled “method SeedPlugin.run_experiment_loop”
run_experiment_loop(
    experiment: "'Experiment'",
    replicate_fn: 'Callable[[int], Any]'
) → List[Any]

Run all replicates and return their results.

Returning NotImplemented signals that the plugin does not provide a custom execution strategy and the default should be used instead.

class LoggingPlugin

Section titled “class LoggingPlugin”

Configure experiment logging using the crystallize logger.

method LoggingPlugin.__init__

Section titled “method LoggingPlugin.__init__”
__init__(verbose: 'bool' = False, log_level: 'str' = 'INFO') → None

method LoggingPlugin.after_run

Section titled “method LoggingPlugin.after_run”
after_run(experiment: 'Experiment', result: 'Result') → None

method LoggingPlugin.after_step

Section titled “method LoggingPlugin.after_step”
after_step(
    experiment: 'Experiment',
    step: 'PipelineStep',
    data: 'Any',
    ctx: 'FrozenContext'
) → None

method LoggingPlugin.before_replicate

Section titled “method LoggingPlugin.before_replicate”
before_replicate(experiment: 'Experiment', ctx: 'FrozenContext') → None

Run prior to each pipeline execution for a replicate.

method LoggingPlugin.before_run

Section titled “method LoggingPlugin.before_run”
before_run(experiment: 'Experiment') → None

method LoggingPlugin.before_step

Section titled “method LoggingPlugin.before_step”
before_step(experiment: 'Experiment', step: 'PipelineStep') → None

method LoggingPlugin.init_hook

Section titled “method LoggingPlugin.init_hook”
init_hook(experiment: 'Experiment') → None

method LoggingPlugin.run_experiment_loop

Section titled “method LoggingPlugin.run_experiment_loop”
run_experiment_loop(
    experiment: "'Experiment'",
    replicate_fn: 'Callable[[int], Any]'
) → List[Any]

Run all replicates and return their results.

Returning NotImplemented signals that the plugin does not provide a custom execution strategy and the default should be used instead.

class ArtifactPlugin

Section titled “class ArtifactPlugin”

Persist artifacts produced during pipeline execution.

method ArtifactPlugin.__init__

Section titled “method ArtifactPlugin.__init__”
__init__(
    root_dir: 'str' = './data',
    versioned: 'bool' = False,
    artifact_retention: 'int' = 3,
    big_file_threshold_mb: 'int' = 10
) → None

method ArtifactPlugin.after_run

Section titled “method ArtifactPlugin.after_run”
after_run(experiment: 'Experiment', result: 'Result') → None

method ArtifactPlugin.after_step

Section titled “method ArtifactPlugin.after_step”
after_step(
    experiment: 'Experiment',
    step: 'PipelineStep',
    data: 'Any',
    ctx: 'FrozenContext'
) → None

Write any artifacts logged in ctx.artifacts to disk.

method ArtifactPlugin.before_replicate

Section titled “method ArtifactPlugin.before_replicate”
before_replicate(experiment: 'Experiment', ctx: 'FrozenContext') → None

Run prior to each pipeline execution for a replicate.

method ArtifactPlugin.before_run

Section titled “method ArtifactPlugin.before_run”
before_run(experiment: 'Experiment') → None

method ArtifactPlugin.before_step

Section titled “method ArtifactPlugin.before_step”
before_step(experiment: 'Experiment', step: 'PipelineStep') → None

method ArtifactPlugin.init_hook

Section titled “method ArtifactPlugin.init_hook”
init_hook(experiment: 'Experiment') → None

Configure the experiment instance during initialization.

method ArtifactPlugin.run_experiment_loop

Section titled “method ArtifactPlugin.run_experiment_loop”
run_experiment_loop(
    experiment: "'Experiment'",
    replicate_fn: 'Callable[[int], Any]'
) → List[Any]

Run all replicates and return their results.

Returning NotImplemented signals that the plugin does not provide a custom execution strategy and the default should be used instead.