| name | static-parameters |
| description | Analyze and manipulate massive values in LLM attention mechanisms, particularly for understanding contextual knowledge processing in transformer models with RoPE |
Rope with LLM - Massive Values in Self-Attention Analysis
When to Use
This skill should be activated when you need to:
- Analyze massive values appearing in transformer attention mechanisms (Q, K, V matrices)
- Understand how LLMs process contextual vs parametric knowledge
- Investigate the impact of RoPE (Rotary Positional Encoding) on attention patterns
- Perform experiments on attention value disruption and its effects
- Evaluate quantization methods' impact on contextual knowledge understanding
- Generate synthetic datasets for passkey retrieval and knowledge QA tasks
- Extract and visualize attention maps from various LLMs (Llama, Mistral, Qwen, Gemma, etc.)
Trigger keywords: massive values, attention mechanism, RoPE, contextual knowledge, attention maps, Q/K/V matrices, quantization impact, passkey retrieval, knowledge QA
Quick Reference
Installation/Setup
Environment Setup
conda create -n myenv python=3.9
conda activate myenv
pip install -r requirements.txt
Environment Variables Configuration
Create a .env file with:
Demo Scripts
scripts/attention_analysis.py
"""
Attention Analysis for Massive Values in LLMs
This script demonstrates how to extract and analyze attention matrices (Q, K, V)
from transformer models to identify massive values in low-frequency dimensions.
"""
import torch
import torch.nn as nn
import numpy as np
import matplotlib.pyplot as plt
from transformers import AutoModelForCausalLM, AutoTokenizer
import os
from typing import Dict, List, Tuple, Optional
import argparse
class AttentionAnalyzer:
"""
Analyzer for extracting and visualizing attention patterns in LLMs.
"""
def __init__(self, model_name: str, device: str = 'cuda'):
"""
Initialize the attention analyzer with a specific model.
Args:
model_name: HuggingFace model identifier
device: Device to run the model on ('cuda' or 'cpu')
"""
self.model_name = model_name
self.device = device
self.tokenizer = AutoTokenizer.from_pretrained(model_name)
self.model = AutoModelForCausalLM.from_pretrained(
model_name,
torch_dtype=torch.float16,
device_map='auto'
)
self.attention_weights = {}
def extract_attention_states(self,
text: ,
layers: [] = ) -> [, torch.Tensor]:
inputs = .tokenizer(text, return_tensors=).to(.device)
attention_states = {
: {},
: {},
: {}
}
hooks = []
():
():
(module, ):
hidden_states = []
query_states = module.q_proj(hidden_states)
key_states = module.k_proj(hidden_states)
value_states = module.v_proj(hidden_states)
attention_states[][layer_idx] = query_states.detach().cpu()
attention_states[][layer_idx] = key_states.detach().cpu()
attention_states[][layer_idx] = value_states.detach().cpu()
hook_fn
idx, layer (.model.model.layers):
layers idx layers:
hook = layer.self_attn.register_forward_hook(create_hook(idx))
hooks.append(hook)
torch.no_grad():
outputs = .model(**inputs)
hook hooks:
hook.remove()
attention_states
() -> [, []]:
massive_values = {
: [],
: [],
: []
}
matrix_type [, , ]:
layer_idx, tensor attention_states[matrix_type].items():
norms = torch.norm(tensor, dim=-)
threshold = torch.quantile(norms.flatten(), threshold_percentile / )
positions = torch.where(norms > threshold)
i ((positions[])):
massive_values[matrix_type].append({
: layer_idx,
: positions[][i].item(),
: positions[][i].item() (positions) > ,
: positions[][i].item() (positions) > ,
: norms[(p[i] p positions)].item()
})
massive_values
() -> :
results = {}
matrix_type [, , ]:
results[matrix_type] = {}
layer_idx, tensor attention_states[matrix_type].items():
reshaped = tensor.view(-, tensor.size(-))
fft_result = torch.fft.rfft(reshaped, dim=-)
magnitude = torch.(fft_result)
freq_bins = magnitude.size(-)
low_freq_cutoff = freq_bins //
low_freq_energy = torch.(magnitude[:, :low_freq_cutoff], dim=-)
high_freq_energy = torch.(magnitude[:, low_freq_cutoff:], dim=-)
results[matrix_type][layer_idx] = {
: low_freq_energy.mean().item(),
: high_freq_energy.mean().item(),
: low_freq_energy.std().item(),
: high_freq_energy.std().item(),
: (low_freq_energy.mean() / high_freq_energy.mean()).item()
}
results
():
fig, axes = plt.subplots(, , figsize=(, ))
idx, (matrix_type, ax) (([, , ], axes)):
layer_idx attention_states[matrix_type]:
tensor = attention_states[matrix_type][layer_idx]
norms = torch.norm(tensor, dim=-).squeeze()
im = ax.imshow(norms.cpu().numpy(), aspect=, cmap=)
ax.set_title()
ax.set_xlabel()
ax.set_ylabel()
plt.colorbar(im, ax=ax)
plt.suptitle()
plt.tight_layout()
save_path:
plt.savefig(save_path, dpi=, bbox_inches=)
()
:
plt.show()
plt.close()
():
parser = argparse.ArgumentParser(description=)
parser.add_argument(, =,
default=,
=)
parser.add_argument(, =,
default=,
=)
parser.add_argument(, =, nargs=,
default=[, , ],
=)
parser.add_argument(, =,
default=,
=)
args = parser.parse_args()
os.makedirs(args.save_dir, exist_ok=)
()
analyzer = AttentionAnalyzer(args.model_name)
()
attention_states = analyzer.extract_attention_states(args.text, args.layers)
()
massive_values = analyzer.identify_massive_values(attention_states)
matrix_type [, , ]:
count = (massive_values[matrix_type])
()
count > :
sorted_values = (massive_values[matrix_type],
key= x: x[],
reverse=)[:]
item sorted_values:
()
()
freq_analysis = analyzer.analyze_frequency_distribution(attention_states)
matrix_type [, , ]:
()
layer_idx, stats freq_analysis[matrix_type].items():
()
()
()
()
layer_idx args.layers:
save_path = os.path.join(args.save_dir, )
analyzer.visualize_attention_maps(attention_states, layer_idx, save_path)
matrix_type [, , ]:
layer_idx, tensor attention_states[matrix_type].items():
save_path = os.path.join(args.save_dir,
)
torch.save(tensor, save_path)
()
()
__name__ == :
main()
scripts/disruption_experiment.py
"""
Massive Value Disruption Experiments
This script performs experiments to test the impact of disrupting massive values
in attention mechanisms on model performance, particularly for contextual knowledge
understanding.
"""
import torch
import torch.nn as nn
import numpy as np
from transformers import AutoModelForCausalLM, AutoTokenizer
from typing import Dict, List, Optional, Tuple
import json
import argparse
from dataclasses import dataclass
import os
from tqdm import tqdm
@dataclass
class DisruptionConfig:
"""Configuration for disruption experiments."""
disruption_type: str
target_matrix: str
num_outliers: int
layers_to_disrupt: List[int]
class MassiveValueDisruptor:
"""
Handles disruption of massive values in attention mechanisms.
"""
def __init__(self, model_name: str, device: str = 'cuda'):
"""
Initialize the disruptor with a model.
Args:
model_name: HuggingFace model identifier
device: Device to run experiments on
"""
.model_name = model_name
.device = device
.tokenizer = AutoTokenizer.from_pretrained(model_name)
.model = AutoModelForCausalLM.from_pretrained(
model_name,
torch_dtype=torch.float16,
device_map=
)
.original_forward_methods = {}
():
layer_idx config.layers_to_disrupt:
layer = .model.model.layers[layer_idx]
layer_idx .original_forward_methods:
.original_forward_methods[layer_idx] = layer.self_attn.forward
():
():
bsz, q_len, _ = hidden_states.size()
query_states = layer.self_attn.q_proj(hidden_states)
key_states = layer.self_attn.k_proj(hidden_states)
value_states = layer.self_attn.v_proj(hidden_states)
query_states = query_states.view(bsz, q_len,
layer.self_attn.num_heads,
layer.self_attn.head_dim).transpose(, )
key_states = key_states.view(bsz, q_len,
layer.self_attn.num_key_value_heads,
layer.self_attn.head_dim).transpose(, )
value_states = value_states.view(bsz, q_len,
layer.self_attn.num_key_value_heads,
layer.self_attn.head_dim).transpose(, )
config.target_matrix [, ]:
target_states = query_states config.target_matrix == key_states
num_heads = target_states.size()
head_idx (num_heads):
head_states = target_states[:, head_idx, :, :]
norms = torch.norm(head_states, dim=-)
values, indices = torch.topk(norms.flatten(),
(config.num_outliers, norms.numel()))
config.disruption_type == :
mean_val = head_states.mean()
idx indices:
pos = (idx // head_states.size(-), idx % head_states.size(-))
target_states[:, head_idx, pos[], pos[]] = mean_val
config.disruption_type == :
idx indices:
pos = (idx // head_states.size(-), idx % head_states.size(-))
target_states[:, head_idx, pos[], pos[]] =
config.disruption_type == :
idx indices:
pos = (idx // head_states.size(-), idx % head_states.size(-))
target_states[:, head_idx, pos[], pos[]] = torch.randn_like(
target_states[:, head_idx, pos[], pos[]]
)
config.target_matrix == :
query_states = target_states
:
key_states = target_states
attn_weights = torch.matmul(query_states, key_states.transpose(, ))
attn_weights = attn_weights / np.sqrt(layer.self_attn.head_dim)
attention_mask :
attn_weights = attn_weights + attention_mask
attn_weights = nn.functional.softmax(attn_weights, dim=-)
attn_output = torch.matmul(attn_weights, value_states)
attn_output = attn_output.transpose(, ).contiguous()
attn_output = attn_output.reshape(bsz, q_len, -)
attn_output = layer.self_attn.o_proj(attn_output)
attn_output, , past_key_value
disrupted_forward
layer.self_attn.forward = create_disrupted_forward(
layer.self_attn.forward, layer_idx
)
():
layer_idx, original_forward .original_forward_methods.items():
.model.model.layers[layer_idx].self_attn.forward = original_forward
.original_forward_methods.clear()
() -> [, ]:
correct =
total =
results = []
samples = test_data[:max_samples] max_samples test_data
sample tqdm(samples, desc=):
context = sample.get(, )
question = sample.get(, )
expected_answer = sample.get(, )
prompt =
inputs = .tokenizer(prompt, return_tensors=,
max_length=, truncation=).to(.device)
torch.no_grad():
outputs = .model.generate(
**inputs,
max_new_tokens=,
temperature=,
do_sample=
)
generated_text = .tokenizer.decode(outputs[], skip_special_tokens=)
generated_answer = generated_text.split()[-].strip()
is_correct = expected_answer.lower() generated_answer.lower()
correct += (is_correct)
total +=
results.append({
: question,
: expected_answer,
: generated_answer,
: is_correct
})
accuracy = correct / total total >
{
: accuracy,
: correct,
: total,
: results
}
() -> []:
file_path = os.path.join(data_path, )
os.path.exists(file_path):
(file_path, ) f:
json.load(f)
:
()
[
{
: ,
: ,
:
},
{
: ,
: ,
:
}
]
():
parser = argparse.ArgumentParser(description=)
parser.add_argument(, =,
default=,
=)
parser.add_argument(, =, default=,
choices=[, , , , , , ],
=)
parser.add_argument(, =, default=,
choices=[, , , ],
=)
parser.add_argument(, =, default=,
choices=[, , ],
=)
parser.add_argument(, =, default=,
=)
parser.add_argument(, =, nargs=, default=[, , ],
=)
parser.add_argument(, =, default=,
=)
parser.add_argument(, =, default=,
=)
args = parser.parse_args()
os.makedirs(args.output_dir, exist_ok=)
()
disruptor = MassiveValueDisruptor(args.model_name)
()
test_data = load_test_data(args.pattern)
()
baseline_results = disruptor.evaluate_contextual_knowledge(
test_data, args.max_samples
)
()
args.disruption_type != :
()
config = DisruptionConfig(
disruption_type=args.disruption_type,
target_matrix=args.target_matrix,
num_outliers=args.num_outliers,
layers_to_disrupt=args.layers
)
disruptor.apply_disruption(config)
disrupted_results = disruptor.evaluate_contextual_knowledge(
test_data, args.max_samples
)
()
performance_drop = baseline_results[] - disrupted_results[]
()
disruptor.restore_original()
:
disrupted_results = baseline_results
performance_drop =
results = {
: args.model_name,
: args.pattern,
: {
: args.disruption_type,
: args.target_matrix,
: args.num_outliers,
: args.layers
},
: baseline_results[],
: disrupted_results[],
: performance_drop,
: baseline_results[][:],
: disrupted_results[][:]
}
output_file = os.path.join(
args.output_dir,
)
(output_file, ) f:
json.dump(results, f, indent=)
()
( + *)
()
(*)
()
()
()
()
()
()
__name__ == :
main()
