Pipelines

Murano’s Pipeline chains steps together sequentially, passing a shared Results object between them. Each step reads from and writes to named keys in Results, and the pipeline validates the chain before running.

Basic usage

from murano import MuranoDataset, MuranoModel, Pipeline
from murano.steps import Load, Record, SteeringVector

model = MuranoModel("meta-llama/Llama-3.2-1B-Instruct")

dataset = MuranoDataset.contrastive(
    positive=["What a wonderful, delightful day"],
    negative=["What a miserable, dreadful day"],
    template_fn=model.chat_template,
)

results = Pipeline([
    Load(dataset),
    Record(model, layers="all", position="last"),
    SteeringVector(normalize=True),
]).run()

How it works

1. Each step declares reads and writes — the keys it needs and produces
2. Pipeline.run() calls each step in order: step(results) -> results
3. Steps that inherit from Step get automatic validation before execution

The data flow for a steering-direction experiment:

Load(dataset)
  writes: dataset, prompts

Record(model)
  reads: dataset    writes: record

SteeringVector()
  reads: record     writes: steering

Intervene(model, fn)
  reads: prompts    writes: intervene

Validation

Call .validate() to dry-run the pipeline without executing any steps. It checks that every step’s reads are satisfied by a prior step’s writes:

pipeline = Pipeline([
    Load(dataset),
    Record(model),
    SteeringVector(),
])

available_keys = pipeline.validate()
print(available_keys)  # ['dataset', 'prompts', 'record', 'steering']

If a step reads a key that no prior step writes, validate() raises a KeyError. Type compatibility between steps is also checked.

Multi-phase pipelines

Many experiments need a train phase and an eval phase with different data. Run two pipelines and pass results between them:

from murano.steps import Intervene
from murano.steps.intervene import ablate_direction

# Phase 1: Train
train_results = Pipeline([
    Load(train_dataset),
    Record(model, layers="all", position="last"),
    SteeringVector(normalize=True),
]).run()

# Phase 2: Evaluate
eval_results = Pipeline([
    Load(eval_dataset),
    Intervene(
        model,
        ablate_direction(train_results["steering"].direction_per_layer),
    ),
]).run()

comparison = eval_results["intervene"]
print("Clean:", comparison.clean_generations)
print("Ablated:", comparison.modified_generations)

Saving results

Add a Save step at the end, or call results.save() directly:

from murano.steps import Save

# Option 1: as a step
pipeline = Pipeline([
    Load(dataset),
    Record(model),
    SteeringVector(),
    Save(output_dir="my_experiment", model_id=model.model_id),
])

# Option 2: after the pipeline
results = pipeline.run()
results.save(output_dir="my_experiment", model_id=model.model_id)

Both default output_dir to murano_outputs/ in the current directory, and the Plot step writes there too, so a run’s artifacts and plots share one tree. Results go directly into output_dir, so re-running overwrites the previous run; pass run_name="..." to keep runs side by side under murano_outputs/<run_name>/.

The output structure:

my_experiment/
├── direction/
│   └── steering.pt
├── evaluation/
│   └── generations.json
├── probe/
│   └── probe.json
└── metadata.json

Available steps

Step	Reads	Writes	Description
Load	—	dataset, prompts	Loads a dataset into the pipeline
Record	dataset	record	Captures activations via nnsight
SteeringVector	record	steering	Computes contrastive mean diff
Intervene	prompts	intervene	Generates with activation patching
WeightAblation	prompts, steering	intervene, weight_ablation	Ablates via weight projection
Probe	record	probe	Linear probe per layer
GenerationMetric	intervene	metric	Custom evaluation metric
Save	—	output_dir	Saves all results to disk

Writing custom steps

Subclass Step, declare reads and writes, and implement __call__:

from murano.steps.base import Step
from murano.results import Results

class MyStep(Step):
    reads = ["record"]
    writes = ["my_output"]

    def __call__(self, results: Results) -> Results:
        store = results["record"]
        # ... your logic ...
        results["my_output"] = my_result
        return results