interpretune.analysis#

Analysis submodule.

class interpretune.analysis.ActivationSumm(*, custom_repr=<factory>, run_name=None, mean_activation, num_samples_active)[source]#

Container for activation summary metrics.

Parameters:
class interpretune.analysis.AnalysisBackend(*args, **kwargs)[source]#

Protocol defining analysis-adapter functionality layered above model execution backends.

supports(capability)[source]#

Check whether this backend supports a given analysis capability.

Return type:

bool

Parameters:

capability (AnalysisBackendCapability)

property capabilities: frozenset[AnalysisBackendCapability]#

Return the set of analysis capabilities this backend supports.

class interpretune.analysis.AnalysisBackendCapability(*values)[source]#

Capabilities exposed by analysis adapters/backends rather than model execution backends.

ATTRIBUTION_GRAPH = 'attribution_graph'#

Module exposes attribution graph analysis support via an attached analysis backend.

FEATURE_INTERVENTION = 'feature_intervention'#

Module exposes feature intervention support via an attached analysis backend.

class interpretune.analysis.AnalysisBatch(*args, **kwargs)[source]#
bind_resolution_context(module, *, analysis_inputs=None, batch_idx=None, input_schema=None)[source]#

Bind execution-time lookup context used by scoped batch accessors.

Return type:

AnalysisBatch

Parameters:

  • module (Any | None)

  • analysis_inputs (Any)

  • batch_idx (int | None)

  • input_schema (Any)

clear_resolution_context()[source]#

Clear any previously bound execution-time lookup context.

Return type:

AnalysisBatch

get(key, default=None, *, scopes=None, resolve=True)[source]#

Return a direct batch value or, when bound, resolve via scoped inputs.

Parameters:
require(key, *, scopes=None, message=None)[source]#

Resolve a required value from the bound analysis execution context.

Return type:

Any

Parameters:
resolution_context(module, *, analysis_inputs=None, batch_idx=None, input_schema=None)[source]#

Temporarily bind execution-time scoped lookup state for the current op call.

Parameters:

  • module (Any | None)

  • analysis_inputs (Any)

  • batch_idx (int | None)

  • input_schema (Any)

resolve(key, default=None, *, scopes=None)[source]#

Resolve a value using the bound analysis execution context.

Return type:

Any

Parameters:
to_cpu()[source]#

Detach and move all field tensors to CPU.

update([E, ]**F) → None.  Update D from mapping/iterable E and F.[source]#

If E is present and has a .keys() method, then does: for k in E.keys(): D[k] = E[k] If E is present and lacks a .keys() method, then does: for k, v in E: D[k] = v In either case, this is followed by: for k in F: D[k] = F[k]

class interpretune.analysis.AnalysisCfgProtocol(*args, **kwargs)[source]#

Protocol verifying core analysis configuration functionality.

class interpretune.analysis.AnalysisInputs(*, row=None, batch=None, run=None, store=None)[source]#

Explicit scoped analysis inputs used during op execution.

Parameters:
class interpretune.analysis.AnalysisOp(name, description, output_schema, input_schema=None, aliases=None, impl_params=None, required_capabilities=None)[source]#

Base class for analysis operations.

Parameters:

  • name (str)

  • description (str)

  • output_schema (OpSchema)

  • input_schema (OpSchema | None)

  • aliases (Sequence[str] | None)

  • impl_params (dict[str, Any] | None)

  • required_capabilities (Sequence[str | Any] | None)

static process_batch(analysis_batch, batch, output_schema, tokenizer=None, save_prompts=False, save_tokens=False, decode_kwargs=None)[source]#

Process analysis batch using provided output schema.

This static method handles the common processing logic for analysis batches, including token handling and schema-based transformations.

Parameters:

  • analysis_batch (BaseAnalysisBatchProtocol) – The analysis batch to process

  • batch (BatchEncoding) – The raw batch data

  • output_schema (OpSchema) – Schema defining the structure of the output

  • tokenizer (PreTrainedTokenizerBase | None) – Optional tokenizer for decoding prompts

  • save_prompts (bool) – Whether to save prompts

  • save_tokens (bool) – Whether to save tokens

  • decode_kwargs (dict[str, Any] | None) – Additional keyword arguments for decoding

Return type:

BaseAnalysisBatchProtocol

Returns:

Processed analysis batch

active_ctx_key(ctx_key)[source]#

Context manager for temporarily setting the active context key.

Parameters:

ctx_key – The context key to set during the context execution

save_batch(analysis_batch, batch, tokenizer=None, save_prompts=False, save_tokens=False, decode_kwargs=None)[source]#

Save analysis batch using process_batch static method.

Return type:

BaseAnalysisBatchProtocol

Parameters:
property ctx_key: str#

Return the context key if set, otherwise return the name.

property impl: Callable | None#

Get the implementation function.

class interpretune.analysis.AnalysisOpProtocol(*args, **kwargs)[source]#

Protocol defining required interface for analysis operations.

class interpretune.analysis.AnalysisStoreProtocol(*args, **kwargs)[source]#

Protocol verifying core analysis store functionality.

class interpretune.analysis.BackendCapability(*values)[source]#

Capabilities that a model backend may support.

Ops and the dispatcher can query backend.capabilities to check support before calling optional methods. Backends that do not support a capability should fall back to a simpler code path (e.g., looping instead of batching).

BATCHED_HOOKS = 'batched_hooks'#

Backend can run multiple forward passes with different hook configs in a single batched execution (e.g., NNsight multi-invoke within one trace).

GRADIENTS = 'gradients'#

Backend supports forward + backward with gradient caching.

class interpretune.analysis.BaseAnalysisBatchProtocol(*args, **kwargs)[source]#

Base protocol defining methods all analysis batches should implement.

Subclasses should define which dataset columns will have attribute-based access enabled for associated AnalysisStore objects.

class interpretune.analysis.ColCfg(datasets_dtype, required=True, dyn_dim=None, dyn_dim_ceil=None, non_tensor=False, per_latent=False, per_latent_model_hook=False, intermediate_only=False, connected_obj='analysis_store', array_shape=None, sequence_type=True, array_dtype=None, default=None)[source]#

Configuration for a dataset column.

Parameters:

  • datasets_dtype (str)

  • required (bool)

  • dyn_dim (int | None)

  • dyn_dim_ceil (Literal['batch_size', 'max_answer_tokens', 'num_classes', 'vocab_size', 'max_seq_len'] | None)

  • non_tensor (bool)

  • per_latent (bool)

  • per_latent_model_hook (bool)

  • intermediate_only (bool)

  • connected_obj (Literal['analysis_store', 'datamodule'])

  • array_shape (tuple[int | Literal['batch_size', 'max_answer_tokens', 'num_classes', 'vocab_size', 'max_seq_len'] | None, ...] | None)

  • sequence_type (bool)

  • array_dtype (str | None)

  • default (Any)

classmethod from_dict(data)[source]#

Create from dict representation.

Return type:

ColCfg

Parameters:

data (dict)

to_dict()[source]#

Convert to JSON serializable dict.

Return type:

dict

class interpretune.analysis.DefaultAnalysisBatchProtocol(*args, **kwargs)[source]#

Default analysis batch protocol defining which dataset columns should have attribute-based access enabled for AnalysisStore objects. Subclasses can extend this protocol (or the base one) to add additional attributes or change existing attributes as needed.

logit_diffs#

Per batch logit differences with shape [batch_size]

Type:

torch.Tensor | dict[str, dict[int, torch.Tensor]] | None

answer_logits#

Model output logits with shape [batch_size, 1, num_classes]

Type:

torch.Tensor | dict[str, dict[int, torch.Tensor]] | None

loss#

Loss values with shape [batch_size]

Type:

torch.Tensor | dict[str, dict[int, torch.Tensor]] | None

label_ids#

Input labels translated to token ids with shape [batch_size] (if labels provided & translation is needed)

Type:

torch.Tensor | None

orig_labels#

Ground truth unmodified labels with shape [batch_size]

Type:

torch.Tensor | None

preds#

Model predictions with shape [batch_size]

Type:

torch.Tensor | dict[str, dict[int, torch.Tensor]] | None

cache#

Forward pass activation cache

Type:

ActivationCacheProtocol | None

grad_cache#

Backward pass gradient cache

Type:

ActivationCacheProtocol | None

answer_indices#

Indices of answers with shape [batch_size]

Type:

torch.Tensor | None

alive_latents#

Active latent indices per latent model hook

Type:

dict[str, list[int]] | None

correct_activations#

Latent model activations after corrections with shape [batch_size, d_sae] for each latent model

Type:

dict[str, torch.Tensor] | None

attribution_values#

Attribution values per latent model hook

Type:

dict[str, torch.Tensor] | None

tokens#

Input token IDs

Type:

torch.Tensor | None

prompts#

Text prompts

Type:

list[str] | None

class interpretune.analysis.HubAnalysisOpManager(cache_dir=None, token=None)[source]#

Manages downloading and uploading analysis operation definitions from/to Hugging Face Hub.

Initialize the hub manager.

Parameters:

  • cache_dir (Path | None) – Directory for caching hub downloads. Defaults to IT_ANALYSIS_HUB_CACHE.

  • token (str | None) – HuggingFace token for authentication. If None, uses HF_TOKEN env var.

discover_hub_ops(search_patterns=None)[source]#

Discover and cache analysis operations from the Hub.

Parameters:

search_patterns (list[str] | None) – List of repo_id patterns to search for (default: auto-discover)

Return type:

list[HubOpCollection]

Returns:

List of HubOpCollection objects for discovered collections

download_ops(repo_id, revision='main', force_download=False)[source]#

Download analysis operations from HF Hub.

Parameters:

  • repo_id (str) – Repository ID in format ‘username/repo-name’

  • revision (str) – Git revision to download (default: “main”)

  • force_download (bool) – Whether to force re-download even if cached

Return type:

HubOpCollection

Returns:

HubOpCollection with information about the downloaded collection

Raises:

  • RepositoryNotFoundError – If the repository doesn’t exist

  • ValueError – If repo_id format is invalid

get_cached_collections()[source]#

Get analysis operation collections that are already cached locally.

Return type:

list[HubOpCollection]

Returns:

List of HubOpCollection objects for cached collections

list_available_collections(username=None)[source]#

List available analysis operation collections on the Hub.

Parameters:

username (str | None) – Filter by username (optional)

Return type:

list[str]

Returns:

List of repository IDs for analysis operation collections

upload_ops(local_dir, repo_id, commit_message='Upload analysis operations', revision='main', create_pr=False, private=False, clean_existing=False, delete_patterns=None)[source]#

Upload analysis operations to HuggingFace Hub.

Parameters:

  • local_dir (Path) – Local directory containing operations to upload

  • repo_id (str) – Repository ID on HuggingFace Hub

  • commit_message (str) – Commit message for the upload

  • revision (str) – Git revision/branch to upload to

  • create_pr (bool) – Whether to create a pull request

  • private (bool) – Whether the repository should be private

  • clean_existing (bool) – Whether to remove existing operation files before upload

  • delete_patterns (list[str] | None) – Custom patterns for files to delete (overrides default when clean_existing=True)

Return type:

str

Returns:

Commit URL or PR URL if create_pr=True

class interpretune.analysis.ITAnalysisFormatter(features=None, **format_kwargs)[source]#

Formatter for Interpretune analysis operations that extends TorchFormatter with operation schema extensions.

format_batch(pa_table)[source]#

Format a batch with enhanced tensorization that respects field contexts.

Return type:

Mapping

Parameters:

pa_table (Table)

format_column(pa_table)[source]#

Format a column with enhanced tensorization.

Return type:

Tensor

Parameters:

pa_table (Table)

format_row(pa_table)[source]#

Format a row with enhanced tensorization that respects field contexts.

Return type:

Mapping

Parameters:

pa_table (Table)

class interpretune.analysis.LatentAnalysisDict[source]#

Dictionary for latent model analysis data where values must be torch.Tensor or list[torch.Tensor].

apply_op_by_latent_model(operation, *args, **kwargs)[source]#

Apply an operation to each tensor value while preserving latent model keys.

Parameters:

  • operation (Union[Callable, str]) – Either callable or string name of torch.Tensor method

  • *args – Additional positional arguments passed to the operation

  • **kwargs – Additional keyword arguments passed to the operation

Returns:

New dictionary mapping latent model names to operated tensor values

Return type:

LatentAnalysisDict

Examples:

# Apply mean
my_dict.batch_join().apply_op_by_latent_model('mean', dim=0)

# Apply custom function
my_dict.batch_join().apply_op_by_latent_model(torch.mean, dim=0)
batch_join(across_saes=False, join_fn=<built-in method cat of type object>)[source]#

Join field values either by SAE or across SAEs.

Parameters:

  • join_across_saes – If True, joins values across SAEs for each batch. If False, joins batches for each SAE separately.

  • join_fn (Callable) – Function to use for joining (default: torch.cat)

  • across_saes (bool)

Returns:

List of tensors, one per batch, with values joined across SAEs If join_across_saes=False: LatentAnalysisDict with batches joined for each SAE

Return type:

If join_across_saes=True

property shapes: dict[str, Size | list[Size]]#

Return shapes for each tensor or list of tensors in the dictionary.

Returns:

Dictionary mapping latent model names to either single tensor shapes or lists of tensor shapes

class interpretune.analysis.LatentAnalysisTargets(latent_model_fqns=<factory>, sae_release='gpt2-small-hook-z-kk', target_sae_ids=<factory>, sae_id_factory_fn=<function default_sae_id_factory_fn>, target_layers=<factory>, sae_hook_match_fn=<function default_sae_hook_match_fn>)[source]#

Encapsulation of latent model FQNs and specific hooks involved in a latent-model-mediated analysis, along with helper functions for explicit or pattern based matching by name and/or layer.

Parameters:
class interpretune.analysis.LatentMetrics(*, custom_repr=<factory>, run_name=None, mean_activation, num_samples_active, total_effect, mean_effect, proportion_samples_active)[source]#

Container for latent analysis metrics.

Each metric maps sae names to tensors of latent-level statistics.

Parameters:
create_attribution_tables(sort_by='total_effect', top_k=10, filter_type='both', per_latent_model=False)[source]#

Creates formatted tables of attribution metrics.

Parameters:

  • sort_by (str) – Attribute name from metrics instance to sort by

  • top_k (int) – Number of top entries to include

  • filter_type (Literal['positive', 'negative', 'both']) – Which values to include (‘positive’, ‘negative’, or ‘both’)

  • per_latent_model (bool | None) – Whether to create separate tables per latent model

Return type:

dict[str, str]

class interpretune.analysis.LatentModelFqn(release, sae_id)[source]#

Create new instance of LatentModelFqn(release, sae_id)

Parameters:
release: str#

Alias for field number 0

sae_id: str#

Alias for field number 1

class interpretune.analysis.ModuleCapabilities(model, analysis)[source]#

Execution and analysis capabilities exposed by a module.

Parameters:
class interpretune.analysis.OpSchema(*args, **kwargs)[source]#

Schema defining column specifications for analysis operations.

class interpretune.analysis.OpWrapper(op_name)[source]#

A special wrapper for operations that ensures the op is instantiated when accessed directly or when attributes are accessed.

classmethod initialize(target_module)[source]#

Set the target module where operations will be registered.

classmethod register_operations(module, dispatcher)[source]#

Register all operations from the dispatcher to the module as lazy OpWrapper instances.

Parameters:

  • module – The module where operations will be registered

  • dispatcher – The operations dispatcher instance

property dispatcher#

Lazily load the dispatcher only when needed.

class interpretune.analysis.PredSumm(total_correct, percentage_correct, batch_predictions)[source]#

Create new instance of PredSumm(total_correct, percentage_correct, batch_predictions)

Parameters:

  • total_correct (int)

  • percentage_correct (float)

  • batch_predictions (list | None)

batch_predictions: list | None#

Alias for field number 2

percentage_correct: float#

Alias for field number 1

total_correct: int#

Alias for field number 0

interpretune.analysis.base_vs_sae_logit_diffs(sae, base_ref, tokenizer, top_k=10, max_prompt_width=80)[source]#

Display a table comparing reference vs SAE logit differences.

Parameters:

  • sae (AnalysisStoreProtocol) – Analysis cache from clean with SAE run

  • no_sae_ref – Analysis cache from clean without SAE reference run

  • tokenizer (PreTrainedTokenizerBase) – Tokenizer for decoding labels

  • top_k (int) – Number of top samples to show

  • max_prompt_width (int) – Maximum width for prompt column

  • base_ref (AnalysisStoreProtocol)

Return type:

None

interpretune.analysis.compute_correct(analysis_obj, op=None)[source]#

Compute correct prediction statistics for a given analysis mode.

Return type:

PredSumm

Parameters:
interpretune.analysis.execute_analysis_op(module, batch=None, batch_idx=0, *, analysis_batch=None, analysis_cfg=None, analysis_inputs=None, **kwargs)[source]#

Execute the configured analysis op through a shared helper.

Return type:

AnalysisBatch

Parameters:
interpretune.analysis.execute_analysis_step(module, batch=None, batch_idx=0, *, dataloader_idx=0, analysis_batch=None, analysis_cfg=None, analysis_inputs=None, **kwargs)[source]#

Execute and serialize one analysis step through the shared helper path.

Return type:

Generator[Union[Tensor, Mapping[str, Any], None], None, None]

Parameters:
interpretune.analysis.latent_metrics_scatter(metrics1, metrics2, metric_field='total_effect', label1='Metrics 1', label2='Metrics 2', width=800, height=600)[source]#

Create scatter plots comparing two sets of LatentMetrics.

Parameters:

  • metrics1 (LatentMetrics) – First LatentMetrics to compare

  • metrics2 (LatentMetrics) – Second LatentMetrics to compare

  • metric_field (str) – Name of metric field to compare (default: ‘total_effect’)

  • label1 (str) – Label for first metrics set

  • label2 (str) – Label for second metrics set

  • width (int) – Plot width in pixels

  • height (int) – Plot height in pixels

Return type:

None

interpretune.analysis.schema_to_features(module, op=None, schema=None, default_dtype='float32', int_dtype='int64')[source]#

Convert an operation schema or direct schema to features for Dataset.from_generator.

Parameters:

  • module (Any) – The module being analyzed

  • op (str | AnalysisOp | None) – An optional AnalysisOp to get the schema from

  • schema (OpSchema | None) – An optional direct schema to use instead of op.output_schema

  • default_dtype (str)

  • int_dtype (str)

Return type:

Features

Returns:

A features dict compatible with Dataset.from_generator

On this page
Show Source