| name | neuron-value-weight |
| description | Analyze transformer feed-forward layers as key-value memories, extract activations, identify trigger examples, and compute key-value agreement in transformer language models |
Demo Scripts
scripts/compute_key_value_agreement.py
"""
Compute Key-Value Agreement in Transformer Feed-Forward Layers
This script demonstrates the key-value agreement analysis for transformer
feed-forward layers, showing how values correspond to their associated keys.
Requirements:
- ff-layers installed
- Pre-extracted trigger examples (textual format)
- ~150GB RAM for full analysis
"""
import argparse
import json
import os
import sys
from pathlib import Path
from typing import Dict, List, Tuple, Optional
import pandas as pd
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
def check_memory_requirements():
"""
Check if system has sufficient memory for analysis.
"""
import psutil
mem = psutil.virtual_memory()
total_gb = mem.total / (1024 ** 3)
available_gb = mem.available / (1024 ** 3)
print(f"System Memory Status:")
print(f" Total: {total_gb:.1f} GB")
print(f" Available: {available_gb:.1f} GB")
if total_gb < 150:
print("WARNING: Full key-value agreement analysis requires ~150GB RAM")
print("Consider using a subset of data or a machine with more memory")
return False
return True
def compute_agreement(
model_dir: str,
data_dir: str,
output_base: str
) -> None:
"""
Compute agreement between keys and values.
Args:
model_dir: Path to model checkpoint directory
data_dir: Directory with trigger examples (textual format)
output_base: Base name for output files (will create .tsv and .json)
"""
cmd_parts = [
'python', 'analysis/key_value_agreement.py',
'--model_dir', model_dir,
'--data_dir', data_dir,
'--output_base', output_base
]
import subprocess
print(f"Computing key-value agreement...")
print(f"This may take significant time and memory...")
try:
result = subprocess.run(
cmd_parts,
check=True,
capture_output=True,
text=True
)
print(f"Agreement computation completed!")
print(f"Output files: {output_base}.tsv and {output_base}.json")
except subprocess.CalledProcessError as e:
print(f"Error during computation: {e}")
if e.stdout:
print(f"stdout: {e.stdout}")
if e.stderr:
print(f"stderr: {e.stderr}")
raise
def analyze_agreement_results(tsv_file: str) -> Dict:
"""
Analyze key-value agreement results from TSV file.
Args:
tsv_file: Path to TSV file with agreement results
Returns:
Dictionary with analysis statistics
"""
try:
df = pd.read_csv(tsv_file, sep='\t')
stats = {
'total_keys': len(df),
'mean_agreement': df['agreement'].mean() if 'agreement' in df.columns else 0,
'std_agreement': df['agreement'].std() if 'agreement' in df.columns else 0,
'layers': df['layer'].unique().tolist() if 'layer' in df.columns else [],
}
if 'agreement' in df.columns:
top_keys = df.nlargest(10, 'agreement')[['layer', 'dimension', 'agreement']]
stats['top_agreement_keys'] = top_keys.to_dict('records')
return stats
except FileNotFoundError:
print(f"File not found: {tsv_file}")
return {}
except Exception as e:
print(f"Error analyzing results: {e}")
return {}
def create_subset_data(
input_dir: str,
output_dir: str,
max_keys: int = 100
) -> str:
"""
Create a subset of trigger example data for testing.
Args:
input_dir: Directory with full trigger examples
output_dir: Directory for subset output
max_keys: Maximum number of keys to include
Returns:
Path to subset directory
"""
import shutil
Path(output_dir).mkdir(parents=True, exist_ok=True)
input_path = Path(input_dir)
files = list(input_path.glob('*.txt'))[:max_keys]
print(f"Creating subset with {len(files)} keys...")
for file in files:
dest = Path(output_dir) / file.name
shutil.copy2(file, dest)
print(f"Subset created in: {output_dir}")
return output_dir
def visualize_agreement(json_file: str, output_plot: str = None):
"""
Create visualization of key-value agreement patterns.
Args:
json_file: Path to JSON file with agreement data
output_plot: Path to save plot (optional)
"""
try:
import matplotlib.pyplot as plt
import numpy as np
with open(json_file, 'r') as f:
data = json.load(f)
layer_agreements = {}
for key, value in data.items():
if isinstance(value, dict) and 'layer' in value:
layer = value['layer']
agreement = value.get('agreement', 0)
if layer not in layer_agreements:
layer_agreements[layer] = []
layer_agreements[layer].append(agreement)
fig, ax = plt.subplots(figsize=(12, 6))
layers = sorted(layer_agreements.keys())
agreements = [layer_agreements[l] for l in layers]
bp = ax.boxplot(agreements, labels=layers)
ax.set_xlabel('Layer')
ax.set_ylabel('Agreement Score')
ax.set_title('Key-Value Agreement Across Layers')
ax.grid(True, alpha=0.3)
if output_plot:
plt.savefig(output_plot, dpi=150, bbox_inches='tight')
print(f"Plot saved to: {output_plot}")
else:
plt.show()
except ImportError:
print("Matplotlib not installed. Skipping visualization.")
except Exception as e:
print(f"Error creating visualization: {e}")
def main():
"""
Main function for key-value agreement analysis.
"""
parser = argparse.ArgumentParser(
description='Compute key-value agreement in transformer FF layers'
)
parser.add_argument(
'--model-dir',
type=str,
default='checkpoints/adaptive_lm_wiki103.v2/',
help='Path to model checkpoint directory'
)
parser.add_argument(
'--data-dir',
type=str,
help='Directory with trigger examples (textual format)'
)
parser.add_argument(
'--output-base',
type=str,
default='key_value_agreement',
help='Base name for output files'
)
parser.add_argument(
'--analyze-only',
type=str,
help='Only analyze existing TSV file'
)
parser.add_argument(
'--subset',
type=int,
help='Create and use subset with N keys (for testing)'
)
parser.add_argument(
'--visualize',
action='store_true',
help='Create visualization of results'
)
args = parser.parse_args()
if args.analyze_only:
stats = analyze_agreement_results(args.analyze_only)
print("\n=== Key-Value Agreement Analysis ===")
print(f"Total keys: {stats.get('total_keys', 0)}")
print(f"Mean agreement: {stats.get('mean_agreement', 0):.4f}")
print(f"Std agreement: {stats.get('std_agreement', 0):.4f}")
print(f"Number of layers: {len(stats.get('layers', []))}")
if 'top_agreement_keys' in stats:
print("\nTop 10 Keys by Agreement:")
for key in stats['top_agreement_keys']:
print(f" Layer {key['layer']}, Dim {key['dimension']}: {key['agreement']:.4f}")
if args.visualize:
json_file = args.analyze_only.replace('.tsv', '.json')
if os.path.exists(json_file):
visualize_agreement(json_file, 'agreement_plot.png')
else:
if not check_memory_requirements():
response = input("\nContinue anyway? (y/n): ")
if response.lower() != 'y':
print("Exiting...")
return
data_dir = args.data_dir
if args.subset and data_dir:
subset_dir = f"{data_dir}_subset_{args.subset}"
data_dir = create_subset_data(data_dir, subset_dir, args.subset)
if not data_dir:
print("Error: --data-dir is required")
return
compute_agreement(
model_dir=args.model_dir,
data_dir=data_dir,
output_base=args.output_base
)
tsv_file = f"{args.output_base}.tsv"
if os.path.exists(tsv_file):
stats = analyze_agreement_results(tsv_file)
print("\n=== Results Summary ===")
print(f"Mean agreement: {stats.get('mean_agreement', 0):.4f}")
if args.visualize:
json_file = f"{args.output_base}.json"
if os.path.exists(json_file):
visualize_agreement(json_file, f"{args.output_base}_plot.png")
if __name__ == "__main__":
main()
scripts/extract_predictions.py
"""
Extract Layer and Value Predictions from Transformer Feed-Forward Layers
This script demonstrates extraction of predictions at both dimension and
layer levels from transformer models, useful for analyzing the memory-like
behavior of feed-forward layers.
Requirements:
- ff-layers installed
- Model checkpoint and preprocessed data
"""
import argparse
import pickle
import os
import sys
from pathlib import Path
from typing import Dict, List, Optional, Tuple
import pandas as pd
import numpy as np
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
def extract_ffn_predictions(
data_file: str,
model_dir: str,
output_file: str,
extract_mode: str = 'layer',
max_sentences: int = 1000
) -> None:
"""
Extract FFN predictions from transformer model.
Args:
data_file: Path to tokenized data file
model_dir: Path to model checkpoint directory
output_file: Path for output pickle file
extract_mode: 'dim' for dimension-level, 'layer' for layer-level
max_sentences: Number of sentences to process (-1 for all)
"""
cmd_parts = [
'python', 'analysis/generate_outputs.py',
'--data_file', data_file,
'--model_dir', model_dir,
'--extract_ffn_info',
'--max_sentences', (max_sentences),
, extract_mode,
, output_file
]
subprocess
()
()
extract_mode == max_sentences > :
estimated_hours = (max_sentences / ) *
()
extract_mode == max_sentences == -:
()
:
result = subprocess.run(
cmd_parts,
check=,
capture_output=,
text=
)
()
()
subprocess.CalledProcessError e:
()
e.stdout:
()
e.stderr:
()
() -> :
:
(pickle_file, ) f:
df = pickle.load(f)
()
()
stats = {
: (df),
: df.columns.tolist(),
: df.memory_usage(deep=).() / ( * )
}
df.columns:
all_preds = []
pred_list df[]:
(pred_list, ):
all_preds.extend(pred_list)
unique_preds = ((all_preds))
stats[] = unique_preds
df.columns:
layer_counts = df[].value_counts().to_dict()
stats[] = layer_counts
stats[] = (layer_counts)
df.columns:
activation_shapes = []
act df[].dropna()[:]:
(act, np.ndarray):
activation_shapes.append(act.shape)
stats[] = activation_shapes
stats
FileNotFoundError:
()
{}
Exception e:
()
{}
() -> :
:
(pickle_file, ) f:
df = pickle.load(f)
max_rows:
df = df.head(max_rows)
col df.columns:
df[col].dtype == :
first_val = df[col].dropna().iloc[] df[col].dropna().empty
(first_val, (, np.ndarray)):
df[col] = df[col].apply( x: (x) x )
df.to_csv(output_csv, index=)
()
Exception e:
()
() -> :
:
(dim_pickle, ) f:
df_dim = pickle.load(f)
(layer_pickle, ) f:
df_layer = pickle.load(f)
()
()
()
()
()
()
()
df_dim.columns df_layer.columns:
dim_layers = df_dim[].nunique()
layer_layers = df_layer[].nunique()
()
()
()
Exception e:
()
() -> [, ]:
Path(output_dir).mkdir(parents=, exist_ok=)
dim_output = os.path.join(output_dir, )
extract_ffn_predictions(
data_file=data_file,
model_dir=model_dir,
output_file=dim_output,
extract_mode=,
max_sentences=num_sentences
)
layer_output = os.path.join(output_dir, )
extract_ffn_predictions(
data_file=data_file,
model_dir=model_dir,
output_file=layer_output,
extract_mode=,
max_sentences=num_sentences
)
dim_output, layer_output
():
parser = argparse.ArgumentParser(
description=
)
parser.add_argument(
,
=,
default=,
=
)
parser.add_argument(
,
=,
default=,
=
)
parser.add_argument(
,
=,
choices=[, ],
default=,
=
)
parser.add_argument(
,
=,
default=,
=
)
parser.add_argument(
,
=,
=
)
parser.add_argument(
,
=,
=
)
parser.add_argument(
,
=,
=
)
parser.add_argument(
,
nargs=,
metavar=(, ),
=
)
parser.add_argument(
,
action=,
=
)
args = parser.parse_args()
args.analyze:
stats = analyze_predictions(args.analyze)
()
()
()
()
()
stats:
()
args.export_csv:
export_predictions_to_csv(args.analyze, args.export_csv, max_rows=)
args.compare:
compare_extraction_modes(args.compare[], args.compare[])
args.sample_analysis:
()
output_dir =
dim_pkl, layer_pkl = create_sample_analysis(
data_file=args.data_file,
model_dir=args.model_dir,
output_dir=output_dir,
num_sentences=
)
()
dim_stats = analyze_predictions(dim_pkl)
()
()
layer_stats = analyze_predictions(layer_pkl)
()
compare_extraction_modes(dim_pkl, layer_pkl)
:
args.output_file:
args.output_file =
extract_ffn_predictions(
data_file=args.data_file,
model_dir=args.model_dir,
output_file=args.output_file,
extract_mode=args.extract_mode,
max_sentences=args.max_sentences
)
os.path.exists(args.output_file):
stats = analyze_predictions(args.output_file)
()
__name__ == :
main()
scripts/extract_trigger_examples.py
"""
Extract Trigger Examples from Transformer Feed-Forward Layers
This script demonstrates how to use the ff-layers library to identify
trigger examples for keys in transformer neural networks.
Requirements:
- ff-layers installed (pip install --editable .)
- Downloaded model: transformer_lm.wiki103.adaptive
- Preprocessed WikiText-103 data
"""
import argparse
import json
import os
import sys
from pathlib import Path
from typing import List, Dict, Any, Optional
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
def setup_paths():
"""
Set up default paths for model and data.
Modify these paths based on your installation.
"""
paths = {
'model_dir': 'checkpoints/adaptive_lm_wiki103.v2/',
'data_file': 'examples/language_model/wikitext-103/wiki.train.tokens',
'output_dir': 'analysis_output/'
}
Path(paths['output_dir']).mkdir(parents=True, exist_ok=True)
return paths
def extract_trigger_examples(
data_file: str,
model_dir: str,
output_file: str,
max_sentences: int = 1000,
top_k: int = 50,
dims: Optional[[]] = ,
extract_mode: =
) -> :
cmd_parts = [
, ,
, data_file,
, model_dir,
,
, (max_sentences),
, (top_k),
, extract_mode,
, output_file
]
dims:
cmd_parts.extend([] + [(d) d dims])
subprocess
()
:
result = subprocess.run(
cmd_parts,
check=,
capture_output=,
text=
)
()
()
subprocess.CalledProcessError e:
()
e.stdout:
()
e.stderr:
()
() -> :
cmd_parts = [
, ,
, input_jsonl,
, model_dir,
, output_dir
]
subprocess
()
:
result = subprocess.run(
cmd_parts,
check=,
capture_output=,
text=
)
()
subprocess.CalledProcessError e:
()
() -> [, ]:
stats = {
: ,
: (),
: (),
: []
}
:
(jsonl_file, ) f:
line f:
data = json.loads(line)
stats[] +=
data:
stats[].add(data[])
data:
stats[].add(data[])
data:
stats[].append((data[]))
stats[] = ((stats[]))
stats[] = ((stats[]))
stats[]:
stats[] = (stats[]) / (stats[])
stats
FileNotFoundError:
()
{}
json.JSONDecodeError e:
()
{}
():
parser = argparse.ArgumentParser(
description=
)
parser.add_argument(
,
action=,
=
)
parser.add_argument(
,
=,
=
)
args = parser.parse_args()
paths = setup_paths()
args.analyze_only:
stats = analyze_trigger_examples(args.analyze_only)
()
()
()
()
()
args.quick_demo:
()
()
demo_output = os.path.join(paths[], )
demo_text_dir = os.path.join(paths[], )
extract_trigger_examples(
data_file=paths[],
model_dir=paths[],
output_file=demo_output,
max_sentences=,
top_k=,
dims=[, ],
extract_mode=
)
convert_to_textual(
input_jsonl=demo_output,
model_dir=paths[],
output_dir=demo_text_dir
)
stats = analyze_trigger_examples(demo_output)
()
()
:
()
()
()
()
__name__ == :
main()
