Orchestrator#

The Orchestrator: drives runs and owns the collected metric data.

The orchestrator is the entry point of an experiment campaign. The user creates one, then calls Orchestrator.run() once per parameter point – writing the parameter sweep as ordinary Python loops:

orch = Orchestrator("byzantine_study")
for n in [10, 20]:
    for f in [2, 3]:
        for aggregator in [Average, Krum, Bulyan]:
            for attack in [ALIEAttack, SignFlipAttack, None]:
                orch.run(
                    my_experiment,
                    n=n, f=f, aggregator=aggregator, attack=attack,
                    n_steps=100,
                )
loss = orch.get("loss")   # -> MetricDataFrame

The orchestrator registers channels and owns their data, but the data itself lives in a MetricDataFrame per channel; the orchestrator does not build pandas frames itself. Data is held in memory and never persisted; create a new orchestrator to start fresh. Execution is fail-fast: if an experiment raises, the exception propagates and the campaign stops.

This is the synchronous draft. The near-term plan (multi-process, one process per run, plus PRNG seed handling) will change how run() dispatches work, but not the Metric / get() contract.

class krum.orchestration.orchestrator.Orchestrator(name: str)[source]#

Bases: object

Runs experiments and owns the metric data they produce.

An orchestrator can drive multiple experiments and parameter sweeps. Every run is tagged with the experiment function itself (under the experiment key), so otherwise-identical parameter sets from different experiments stay distinct.

A channel is identified by its name, unique within the orchestrator. Each channel’s samples are stored in a MetricDataFrame, returned by get().

get(name: str) → MetricDataFrame[source]#

Return the storage for channel name.

Parameters:

name – The channel name.

Returns:

  • The channel’s :class:`~krum.orchestration.dataframe.MetricDataFrame`,

  • which exposes the samples as a pandas frame and can be sliced by

  • parameter values.

Raises:

KeyError – If no channel called name was ever declared.

metrics() → list[str][source]#

Return the names of all declared channels.

record(name: str, params: dict[str, Any], step: int, value: Any, skip_if_exists: bool) → None[source]#

Store one sample of channel name for the given run parameters.

Parameters:

  • name – Channel name (already registered).

  • params – The current run’s parameter values.

  • step – The step the value belongs to.

  • value – The recorded value.

  • skip_if_exists – If True and a sample for the same run and step already exists, do nothing.

register_metric(name: str, dtype: type) → None[source]#

Declare a channel, enforcing per-orchestrator name/dtype uniqueness.

Idempotent: re-declaring an existing name with the same dtype is a no-op (this happens once per run of a sweep, since the metric is created inside the experiment function).

Parameters:

  • name – Channel name.

  • dtype – Declared value type.

Raises:

ValueError – If name was already declared with a different dtype.

run(experiment: Callable[[...], Any], **params: Any) → None[source]#

Run experiment once at a single parameter point.

The parameters are first enriched with the experiment’s default arguments (see _resolve_params()), so two runs of the same experiment are tagged with the same parameter set whether or not a defaulted argument was passed explicitly. The resolved parameters are then tagged with the experiment function itself (under the experiment key). Structured parameter values (dicts, lists, sets) are frozen into hashable forms so they can be part of the run key (see _freeze()). The orchestrator publishes that enriched context, invokes experiment(**params) – which receives the original (unfrozen) values – and clears the context afterwards.

Parameters:

  • experiment – The experiment function, called as experiment(**params).

  • **params – The parameter values of this run. Names (including the experiment’s defaulted arguments) must not include the reserved column names step, value, or experiment. Values must be hashable once frozen.

Raises:

  • ValueError – If a parameter name collides with a reserved column, or if a parameter value is not hashable even after freezing.

  • RuntimeError – If experiment raises. The error names the failing run’s parameters and chains the original exception (fail-fast: the sweep stops). The run context is cleared either way.

krum.orchestration.orchestrator.active_orchestrator() → Orchestrator[source]#

Return the orchestrator of the run currently executing.

Returns:

The active :class:`Orchestrator`.

Raises:

RuntimeError – If no run is active, which usually means a Metric was created outside of Orchestrator.run().

krum.orchestration.orchestrator.begin_run(orchestrator: Orchestrator, params: dict[str, Any]) → None[source]#

Mark the start of a run, publishing its orchestrator and parameters.

Parameters:

  • orchestrator – The orchestrator driving the run.

  • params – The parameter values of the run, as a mapping of name to value.

krum.orchestration.orchestrator.current_params() → dict[str, Any][source]#

Return the parameter values of the run currently executing.

Returns:

The current run’s parameters, as a mapping of name to value.

Raises:

RuntimeError – If no run is active, which usually means a value was pushed outside of Orchestrator.run().

krum.orchestration.orchestrator.end_run() → None[source]#

Mark the end of a run, clearing the published state.

No stack is kept: the state simply returns to “no run active”.