Artifacts

Shared artifact types used across Murano pipelines.

PromptBatch

Prompt inputs for generation-style experiments.

Attributes:

• prompts: Prompts actually fed to the model.
• raw_prompts: Original prompts before templating, if available.
• source: Human-readable description of where the prompts came from.
• metadata: Arbitrary prompt-level metadata.

__init__

def __init__(self, prompts: list[str], raw_prompts: list[str] | None = None, source: str = 'manual', metadata: dict[str, Any] = dict()) -> None:

GenerationComparison

Paired baseline vs modified generations for the same prompts.

Attributes:

• baseline_generations: Generations from the unmodified pipeline.
• modified_generations: Generations from the post-intervention pipeline,
• paired by index with baseline_generations.
• prompts: Prompts used for generation, paired by index. May be None
• when the upstream step did not record them.
• baseline_label: Display label for the baseline column.
• modified_label: Display label for the modified column.
• metadata: Arbitrary comparison-level metadata.

__init__

def __init__(self, baseline_generations: list[str], modified_generations: list[str], prompts: list[str] | None = None, baseline_label: str = 'clean', modified_label: str = 'modified', metadata: dict[str, Any] = dict()) -> None:

ComponentSelection

An ordered, scored set of component addresses chosen by a discovery step.

A discovery step (for example ranking heads by direct logit attribution) writes this so a downstream :class:~murano.steps.patch.Patch or :class:~murano.steps.path_patch.PathPatch can read its targets at run time, letting attribute-then-patch compose in a single pipeline. It iterates over its :attr:nodes (best first), so it drops straight into the same target coercion the patch steps use for a hand-written address list.

Attributes:

• nodes: Selected component addresses, best first.
• scores: {Node: float} score behind each selected node (the ranking
• value that put it in the selection).
• metadata: Provenance (source key, ranking criterion, cutoff).

__init__

def __init__(self, nodes: list[Node], scores: dict[Node, float] = dict(), metadata: dict[str, Any] = dict()) -> None:

SweepResult

Per-item scores from running one step chain once per swept item.

A :class:~murano.steps.sweep.Sweep forks the pipeline’s Results once per item, applies the steps built for that item, and harvests one or more metric keys. Each harvested key becomes a column: an ordered {item: float} map, in the order the items were swept.

When every swept item is a :class:~murano.nodes.Node, the sweep is a component sweep and :attr:contributions exposes the primary column as a :class:~murano.nodes.NodeDict. That is the same shape :class:~murano.steps.logit_attribution.LogitAttributionResult publishes, so a component sweep drops straight into :class:~murano.steps.select.SelectComponents without an adapter.

Attributes:

• columns: {read_key: {item: float}}, one entry per harvested key.
• primary: Read key whose column :attr:scores returns; the first key
• passed to Sweep(read=...).
• contributions: The primary column as a {Node: float} NodeDict when
• every swept item is a Node, otherwise None.
• metadata: Provenance (harvested keys, item labels, swept step names).

column

def column(self, key: str | None = None) -> dict[Any, float]:

Return one harvested column, defaulting to the primary one.

Args:

• key: Harvested read key. None selects :attr:primary.

Returns:

• The ``{item: value}map forkey“.

Raises:

• KeyError: If key was not harvested by the sweep.

head_matrix

def head_matrix(self, n_layers: int | None = None, n_heads: int | None = None, column: str | None = None, fill: float = float('nan')) -> list[list[float]]:

Return the scores as a dense [n_layers, n_heads] grid.

The grid is what :func:~murano.plotting.plot_head_matrix consumes. Positions the sweep never visited hold fill; the default NaN renders as a blank cell, so an unswept head never reads as a head with no effect.

Args:

• n_layers: Grid height. Defaults to one past the deepest swept layer.
• n_heads: Grid width. Defaults to one past the highest swept head.
• column: Harvested key to render. None selects :attr:primary.
• fill: Value for positions the sweep did not visit.

Returns:

• A nested list of floats, indexed [layer][head].

Raises:

• TypeError: If the sweep’s items are not all attention-head Nodes.

__init__

def __init__(self, columns: dict[str, dict[Any, float]], primary: str, metadata: dict[str, Any] = dict()) -> None:

MetricComparison

Aggregate metric comparing baseline vs modified generations.

Attributes:

• metric_name: Identifier of the metric (e.g., "logit_diff").
• baseline_score: Aggregate score on the baseline generations.
• modified_score: Aggregate score on the modified generations.
• baseline_scores: Per-item scores on the baseline generations, when
• the metric exposes them.
• modified_scores: Per-item scores on the modified generations, when
• the metric exposes them.
• baseline_label: Display label for the baseline column.
• modified_label: Display label for the modified column.
• metadata: Arbitrary metric-level metadata (method name, parameters).

__init__

def __init__(self, metric_name: str, baseline_score: float, modified_score: float, baseline_scores: list[float] | None = None, modified_scores: list[float] | None = None, baseline_label: str = 'clean', modified_label: str = 'modified', metadata: dict[str, Any] = dict()) -> None:

MetricScore

Scalar result of a forward-pass evaluation metric.

Holds one comparable number, plus an optional per-example breakdown, so a causal experiment ends in a value that can be compared across runs. Unlike :class:MetricComparison, which is shaped for baseline-vs-modified generation comparisons, this carries a single forward-pass score (logit difference, KL divergence, answer log-probability, recovered effect).

Attributes:

• metric_name: Identifier of the metric (e.g. "logit_diff").
• value: Aggregate scalar score, typically the mean over examples.
• per_example: Per-example scores, when the metric exposes them.
• metadata: Arbitrary metric-level metadata (input keys, answer position,
• direction, recovered-metric endpoints).

__init__

def __init__(self, metric_name: str, value: float, per_example: list[float] | None = None, metadata: dict[str, Any] = dict()) -> None: