Spaces:
Running on Zero
Running on Zero
File size: 3,645 Bytes
b784950 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 | from __future__ import annotations
import sys
import types
from pathlib import Path
import numpy as np
import pandas as pd
import torch
def _stub_transformers() -> None:
if 'transformers' in sys.modules:
return
stub = types.ModuleType('transformers')
stub.AutoModelForCausalLM = type('AutoModelForCausalLM', (), {})
stub.AutoTokenizer = type('AutoTokenizer', (), {})
sys.modules['transformers'] = stub
def test_feature_trace_and_position_policy() -> None:
_stub_transformers()
from experiments.run_causal import choose_intervention_position, feature_activation_trace
from featurelens.sae import SparseEncoding
enc = SparseEncoding(
indices=torch.tensor([[1, 8], [7, 2], [7, 3]]),
values=torch.tensor([[2.0, 1.0], [4.0, 3.0], [9.0, 1.0]]),
)
trace = feature_activation_trace(enc, 7)
assert trace.tolist() == [0.0, 4.0, 9.0]
idx, activation, active = choose_intervention_position(
trace, prompt_len=3, position_policy='max_feature_activation'
)
assert (idx, activation, active) == (2, 9.0, True)
idx, activation, active = choose_intervention_position(
trace, prompt_len=3, position_policy='final_token'
)
assert (idx, activation, active) == (2, 9.0, True)
def test_max_active_inactive_falls_back_to_final_token() -> None:
_stub_transformers()
from experiments.run_causal import choose_intervention_position
idx, activation, active = choose_intervention_position(
torch.zeros(4), prompt_len=4, position_policy='max_feature_activation'
)
assert idx == 3
assert activation == 0.0
assert not active
def test_legacy_causal_baseline_migration(tmp_path: Path) -> None:
from experiments.run_causal_addendum import migrate_final_token_baseline
source = tmp_path / 'causal_results.csv'
destination = tmp_path / 'causal_results_final_token.csv'
pd.DataFrame(
[
{'task_id': 'a', 'feature_activation': 3.0, 'condition': 'sae_feature'},
{'task_id': 'a', 'feature_activation': 3.0, 'condition': 'random_norm_matched'},
{'task_id': 'b', 'feature_activation': 0.0, 'condition': 'sae_feature'},
]
).to_csv(source, index=False)
migrate_final_token_baseline(source, destination)
frame = pd.read_csv(destination)
assert set(frame['position_policy']) == {'final_token'}
assert frame['feature_active_at_intervention'].tolist() == [1, 1, 0]
assert frame['feature_active_at_final_token'].tolist() == [1, 1, 0]
assert frame['feature_active_anywhere'].isna().all()
def test_task_level_specificity_averages_interventions_within_task() -> None:
from experiments.analyze_study import _task_level_specificity
rows = []
for intervention, sae_effect, random_effects in [
('ablate', 0.4, [0.1, -0.1]),
('amplify_2x', -0.2, [0.05, -0.15]),
]:
rows.append({'task_id': 'a', 'intervention': intervention, 'condition': 'sae_feature', 'target_mean_logprob_delta': sae_effect})
for value in random_effects:
rows.append({'task_id': 'a', 'intervention': intervention, 'condition': 'random_norm_matched', 'target_mean_logprob_delta': value})
result = _task_level_specificity(pd.DataFrame(rows), effect_column='target_mean_logprob_delta', seed=1)
# One task: SAE mean absolute intervention effect=(.4+.2)/2=.3;
# random means=(.1,.1), then averaged within task=.1.
assert result['n_tasks'] == 1
assert np.isclose(result['sae_abs_mean'], 0.3)
assert np.isclose(result['random_abs_mean'], 0.1)
assert np.isclose(result['specificity_ratio'], 3.0)
|