Steps — Weight Ablation

Weight-level ablation: projects out directions from model weights.

Instead of hooking activations during generation, directly modify weight matrices using an orthogonal projection P = I - dd^T. This removes the direction from ALL computations (not just the residual stream at specific hook points), yielding stronger ablation.

Read matrices (input from residual stream): W_new = W @ P Write matrices (output to residual stream): W_new = P @ W

ProjectionOperator

Orthogonal projection P = I - RR^T for ablating directions from weights.

Supports single direction (vector) or multiple directions (matrix). Directions are normalized and orthonormalized automatically so the resulting projection is valid for both single- and multi-direction ablations. All projection math runs on CPU in float32 for numerical stability.

Args:

• directions: Either [d_model] for single direction,
• or [n_dirs, d_model] for multiple directions.

__init__

def __init__(self, directions: Tensor):

project_read

def project_read(self, W_data: Tensor) -> Tensor:

Apply W @ P for read matrices (embed, q/k/v, gate_proj, up_proj).

project_write

def project_write(self, W_data: Tensor) -> Tensor:

Apply P @ W for write matrices (o_proj, down_proj).

ablate_model_weights

def ablate_model_weights(model: ModelBackend, proj_op: ProjectionOperator) -> int:

Apply directional projection to all relevant weight matrices in-place.

Rewrites the token embedding, attention (q/k/v read, o_proj write), and gated MLP (gate/up read, down write) matrices. The architecture is validated up front, so an unsupported layout raises before any weight is modified rather than leaving the model partially ablated.

Args:

• model: MuranoModel to modify.
• proj_op: ProjectionOperator built from the direction(s) to ablate.

Returns:

• Number of weight matrices modified.

Raises:

• NotImplementedError: If the model is not a supported Llama-family layout.

save_weights

def save_weights(model: ModelBackend) -> dict[str, Tensor]:

Save a copy of all model weights for later restoration.

restore_weights

def restore_weights(model: ModelBackend, saved: dict[str, Tensor]) -> None:

Restore model weights from a saved copy.

WeightAblationResult

Output of WeightAblation step.

Attributes:

• clean_generations: Model responses before weight ablation.
• modified_generations: Model responses after weight ablation.
• n_modified: Number of weight matrices modified.

__init__

def __init__(self, clean_generations: list[str], modified_generations: list[str], n_modified: int, prompts: list[str] | None = None, metadata: dict | None = None):

WeightAblation

Ablate a direction from model weights, then generate.

Unlike activation-level ablation (Intervene step), this modifies the actual weight matrices using orthogonal projection P = I - dd^T. This removes the direction from ALL computations, not just the residual stream at hook points.

Uses the best layer’s direction from SteeringResult, applied globally to all weight matrices across all layers.

Reads from results:

• results['prompts']: PromptBatch
• results['steering']: SteeringResult (uses best_layer direction)

Writes to results:

• results['weight_ablation']: WeightAblationResult
• results[intervene_key]: InterveneResult, published under the shared
• ‘intervene’ slot so downstream steps consume it exactly as they
• consume the activation-level Intervene step.

Args:

• model: MuranoModel to generate with.
• save_dir: If provided, save the ablated model in HF format to this directory.
• gen_kwargs: Keyword arguments for generation.
• intervene_key: Results key for the InterveneResult-shaped generations.
• Defaults to the shared ‘intervene’ slot; retarget it when composing
• this step with the activation-level Intervene step in one pipeline,
• so the two producers do not overwrite each other.

expected_read_types

def expected_read_types(self, results = None, available_types = None):

__init__

def __init__(self, model: ModelBackend, save_dir: str | None = None, gen_kwargs: dict | None = None, intervene_key: str = keys.INTERVENE):

expected_write_types

def expected_write_types(self, results = None, available_types = None):