| name | attention-head-output |
| description | Use this skill when you need to analyze and interpret the internal workings of large language models layer by layer, visualize hidden states and predictions across transformer layers, or understand how models like Llama-3.1-8B and Qwen-2.5-7B make predictions at each layer using the Logit Lens technique |
Demo Scripts
scripts/basic_analysis.py
"""
Basic LogitLens Analysis Script
This script demonstrates how to use LogitLens4LLMs to analyze layer-wise predictions
in large language models like Llama-3.1-8B and Qwen-2.5-7B.
Requirements:
pip install torch transformers matplotlib seaborn numpy
"""
import json
import os
from typing import List, Dict, Any
from enum import Enum
class ModelType(Enum):
"""Enumeration of supported model types"""
LLAMA_3_1_8B = "llama_3_1_8b"
QWEN_2_5_7B = "qwen_2_5_7b"
LLAMA_2_7B = "llama_2_7b"
class LogitLensAnalyzer:
"""Main analyzer class for LogitLens analysis"""
def __init__(self, model_type: ModelType, use_local: bool = False):
"""
Initialize the LogitLens analyzer.
Args:
model_type: Type of model to analyze
use_local: Whether to use locally cached model
"""
self.model_type = model_type
self.use_local = use_local
self.model_name = self._get_model_name()
def _get_model_name(self) -> str:
"""Get the Hugging Face model identifier"""
model_map = {
ModelType.LLAMA_3_1_8B: "meta-llama/Meta-Llama-3.1-8B",
ModelType.QWEN_2_5_7B: "Qwen/Qwen2.5-7B",
ModelType.LLAMA_2_7B: "meta-llama/Llama-2-7b-hf"
}
return model_map.get(self.model_type, "")
def analyze_prompt(self,
prompt: str,
max_new_tokens: int = 10,
temperature: float = 0.7,
print_details: bool = True) -> List[Dict[str, Any]]:
"""
Perform LogitLens analysis on a given prompt.
Args:
prompt: Input text to analyze
max_new_tokens: Number of tokens to generate
temperature: Sampling temperature
print_details: Whether to print detailed layer information
Returns:
List of prediction steps with layer-wise analysis
"""
prediction_steps = []
print(f"Running LogitLens analysis on {self.model_type.value}")
print(f"Prompt: {prompt}")
print("-" * 50)
generated_tokens = self._simulate_generation(prompt, max_new_tokens)
for step_idx, token in enumerate(generated_tokens):
step_data = self._analyze_step(prompt, token, step_idx, print_details)
prediction_steps.append(step_data)
if print_details:
self._print_step_analysis(step_data)
return prediction_steps
def _simulate_generation(self, prompt: str, max_tokens: int) -> List[str]:
"""Simulate token generation for demonstration"""
example_completions = {
"The cat sat on the": ["mat", "and", "looked", "at", "the", "mouse"],
"Once upon a time": ["there", "was", "a", "brave", "knight"],
"The weather today is": ["sunny", "with", "clear", "blue", "skies"]
}
for key in example_completions:
if key in prompt:
return example_completions[key][:max_tokens]
return ["token"] * min(max_tokens, 5)
def _analyze_step(self,
prompt: str,
token: str,
step_idx: int,
include_all_layers: bool = True) -> Dict[str, Any]:
"""
Analyze a single generation step.
Args:
prompt: Current prompt text
token: Generated token
step_idx: Step index
include_all_layers: Whether to include all layer predictions
Returns:
Dictionary containing step analysis data
"""
num_layers = 32 if "8b" in self.model_type.value.lower() else 28
layer_predictions = []
important_layers = []
for layer_idx in range(num_layers):
confidence = 100 - (layer_idx * 2) + (step_idx * 5)
confidence = max(10, min(100, confidence))
layer_data = {
"layer_idx": layer_idx,
"predicted_token": token,
"confidence": confidence,
"top_k_predictions": self._get_top_k_predictions(token, layer_idx)
}
layer_predictions.append(layer_data)
if confidence > 70:
important_layers.append(layer_idx)
return {
"step_idx": step_idx,
"predicted_token": token,
"current_text": f"{prompt} {token}",
"layer_predictions": layer_predictions if include_all_layers else None,
"important_layers": important_layers,
"attention_weights": self._simulate_attention_weights(num_layers),
"mlp_contributions": self._simulate_mlp_contributions(num_layers)
}
def _get_top_k_predictions(self, token: str, layer_idx: int) -> List[tuple]:
"""Get top-k token predictions for a layer"""
alternatives = {
"mat": ["floor", "carpet", "rug", "ground"],
"sunny": ["cloudy", "rainy", "warm", "bright"],
"there": ["once", "lived", "existed", "came"]
}
alt_tokens = alternatives.get(token, ["alt1", "alt2", "alt3"])
predictions = [(token, 85 - layer_idx)]
for i, alt in enumerate(alt_tokens[:3]):
score = max(5, 70 - layer_idx - (i * 20))
predictions.append((alt, score))
return predictions
def _simulate_attention_weights(self, num_layers: int) -> Dict[int, float]:
"""Simulate attention weights across layers"""
weights = {}
for i in range(num_layers):
if i < num_layers // 3:
weights[i] = 0.3 + (i * 0.02)
elif i < 2 * num_layers // 3:
weights[i] = 0.6 + (i * 0.01)
else:
weights[i] = 0.8 + (i * 0.005)
return weights
def _simulate_mlp_contributions(self, num_layers: int) -> Dict[int, float]:
"""Simulate MLP contributions across layers"""
contributions = {}
for i in range(num_layers):
contributions[i] = 0.2 + (i / num_layers) * 0.6
return contributions
def _print_step_analysis(self, step_data: Dict[str, Any]):
"""Print formatted analysis for a generation step"""
print(f"\nStep {step_data['step_idx'] + 1}: Generated '{step_data['predicted_token']}'")
print(f"Current text: {step_data['current_text']}")
if step_data['important_layers']:
print(f"Important layers: {step_data['important_layers'][:5]}")
print("Top predictions from final layers:")
if step_data['layer_predictions']:
for layer in step_data['layer_predictions'][-3:]:
preds = layer['top_k_predictions'][:3]
pred_str = ", ".join([f"{t}({s}%)" for t, s in preds])
print(f" Layer {layer['layer_idx']}: {pred_str}")
def save_analysis(self,
prediction_steps: List[Dict[str, Any]],
output_path: str = "output/analysis_results.json"):
"""
Save analysis results to JSON file.
Args:
prediction_steps: List of prediction step data
output_path: Path to save the JSON file
"""
os.makedirs(os.path.dirname(output_path), exist_ok=True)
output_data = {
"model_type": self.model_type.value,
"num_steps": len(prediction_steps),
"predictions": []
}
for step in prediction_steps:
step_summary = {
"step_idx": step["step_idx"],
"token": step["predicted_token"],
"text": step["current_text"],
"important_layers": step["important_layers"][:5] if step["important_layers"] else []
}
output_data["predictions"].append(step_summary)
with open(output_path, 'w') as f:
json.dump(output_data, f, indent=2)
print(f"\nAnalysis saved to: {output_path}")
def main():
"""Main function demonstrating LogitLens analysis workflow"""
print("=" * 60)
print("Example 1: Llama-3.1-8B Analysis")
print("=" * 60)
analyzer = LogitLensAnalyzer(ModelType.LLAMA_3_1_8B, use_local=False)
prompt = "Complete this sentence: The cat sat on the"
results = analyzer.analyze_prompt(
prompt=prompt,
max_new_tokens=5,
temperature=0.7,
print_details=True
)
analyzer.save_analysis(results, "output/llama_analysis.json")
print("\n" + "=" * 60)
print("Example 2: Qwen-2.5-7B Analysis")
print("=" * 60)
qwen_analyzer = LogitLensAnalyzer(ModelType.QWEN_2_5_7B, use_local=False)
prompt = "The weather today is"
qwen_results = qwen_analyzer.analyze_prompt(
prompt=prompt,
max_new_tokens=5,
temperature=0.8,
print_details=True
)
qwen_analyzer.save_analysis(qwen_results, "output/qwen_analysis.json")
print("\n" + "=" * 60)
print("Example 3: Batch Analysis")
print("=" * 60)
prompts = [
"Once upon a time",
"The secret to happiness is",
"In the future, AI will"
]
batch_analyzer = LogitLensAnalyzer(ModelType.LLAMA_2_7B, use_local=False)
all_results = []
for prompt in prompts:
print(f"\nAnalyzing: {prompt}")
results = batch_analyzer.analyze_prompt(
prompt=prompt,
max_new_tokens=3,
print_details=False
)
all_results.extend(results)
print(f"\nCompleted batch analysis of {len(prompts)} prompts")
print(f"Total prediction steps analyzed: {len(all_results)}")
if __name__ == "__main__":
main()
scripts/visualization_generator.py
"""
Visualization Generator for LogitLens Analysis
This script generates heatmap visualizations from LogitLens analysis results,
showing layer-wise predictions and confidence scores.
Requirements:
pip install matplotlib seaborn numpy pandas
"""
import json
import os
from typing import List, Dict, Any, Optional, Tuple
import matplotlib.pyplot as plt
import seaborn as sns
import numpy as np
import pandas as pd
from datetime import datetime
class LogitLensVisualizer:
"""Generate visualizations for LogitLens analysis results"""
def __init__(self, output_dir: str = "output/visualizations"):
"""
Initialize the visualizer.
Args:
output_dir: Directory to save visualization outputs
"""
self.output_dir = output_dir
os.makedirs(output_dir, exist_ok=True)
sns.set_style("whitegrid")
plt.rcParams['figure.figsize'] = (12, 8)
plt.rcParams['font.size'] = 10
def create_layer_heatmap(self,
layer_predictions: List[[, ]],
step_idx: ,
title: = ,
show_all_layers: = ) -> :
num_layers = (layer_predictions)
top_k =
all_tokens = ()
layer_data layer_predictions:
token, _ layer_data.get(, [])[:top_k]:
all_tokens.add(token)
tokens = ((all_tokens))
confidence_matrix = np.zeros((num_layers, (tokens)))
i, layer_data (layer_predictions):
token, score layer_data.get(, []):
token tokens:
j = tokens.index(token)
confidence_matrix[i, j] = score
fig, ax = plt.subplots(figsize=(, ))
cmap = sns.color_palette(, as_cmap=)
sns.heatmap(confidence_matrix,
xticklabels=tokens,
yticklabels=[ i (num_layers)],
cmap=cmap,
cbar_kws={: },
fmt=,
linewidths=,
linecolor=,
ax=ax)
ax.set_title(, fontsize=, fontweight=)
ax.set_xlabel(, fontsize=)
ax.set_ylabel(, fontsize=)
plt.xticks(rotation=, ha=)
plt.tight_layout()
filename_prefix = show_all_layers
filepath = os.path.join(.output_dir, )
plt.savefig(filepath, dpi=, bbox_inches=)
plt.close()
filepath
() -> :
fig, (ax1, ax2) = plt.subplots(, , figsize=(, ))
steps = []
avg_confidences = []
max_confidences = []
min_confidences = []
step prediction_steps:
step_idx = step[]
steps.append(step_idx)
confidences = []
step.get():
layer step[]:
layer[]:
confidences.append(layer[][][])
confidences:
avg_confidences.append(np.mean(confidences))
max_confidences.append(np.(confidences))
min_confidences.append(np.(confidences))
:
avg_confidences.append()
max_confidences.append()
min_confidences.append()
ax1.plot(steps, avg_confidences, , linewidth=, label=)
ax1.fill_between(steps, min_confidences, max_confidences, alpha=, color=)
ax1.plot(steps, max_confidences, , linewidth=, alpha=, label=)
ax1.plot(steps, min_confidences, , linewidth=, alpha=, label=)
ax1.set_xlabel(, fontsize=)
ax1.set_ylabel(, fontsize=)
ax1.set_title(, fontsize=, fontweight=)
ax1.legend(loc=)
ax1.grid(, alpha=)
high_conf_counts = []
threshold =
step prediction_steps:
count = ([l l step.get(, [])])
high_conf_counts.append(count)
ax2.bar(steps, high_conf_counts, color=, alpha=)
ax2.set_xlabel(, fontsize=)
ax2.set_ylabel(, fontsize=)
ax2.set_title(, fontsize=, fontweight=)
ax2.grid(, alpha=, axis=)
plt.tight_layout()
filepath = os.path.join(.output_dir, )
plt.savefig(filepath, dpi=, bbox_inches=)
plt.close()
filepath
() -> :
fig, (ax1, ax2) = plt.subplots(, , figsize=(, ))
attention_weights = step_data.get(, {})
mlp_contributions = step_data.get(, {})
attention_weights mlp_contributions:
num_layers =
attention_weights = {i: + i * i (num_layers)}
mlp_contributions = {i: + i * i (num_layers)}
layers = (((attention_weights)))
att_values = [attention_weights[i] i layers]
mlp_values = [mlp_contributions[i] i layers]
ax1.plot(layers, att_values, , linewidth=, marker=, markersize=)
ax1.fill_between(layers, , att_values, alpha=, color=)
ax1.set_xlabel(, fontsize=)
ax1.set_ylabel(, fontsize=)
ax1.set_title(, fontsize=, fontweight=)
ax1.grid(, alpha=)
ax1.set_xlim([, (layers) - ])
ax2.plot(layers, mlp_values, , linewidth=, marker=, markersize=)
ax2.fill_between(layers, , mlp_values, alpha=, color=)
ax2.set_xlabel(, fontsize=)
ax2.set_ylabel(, fontsize=)
ax2.set_title(, fontsize=, fontweight=)
ax2.grid(, alpha=)
ax2.set_xlim([, (layers) - ])
fig.suptitle(,
fontsize=, fontweight=, y=)
plt.tight_layout()
filepath = os.path.join(.output_dir, )
plt.savefig(filepath, dpi=, bbox_inches=)
plt.close()
filepath
() -> :
fig = plt.figure(figsize=(, ))
gs = fig.add_gridspec(, , hspace=, wspace=)
fig.suptitle(,
fontsize=, fontweight=)
ax1 = fig.add_subplot(gs[, :])
predictions = analysis_results.get(, [])
tokens = [p[] p predictions]
token_text = .join(tokens)
ax1.text(, , ,
ha=, va=, fontsize=,
bbox=(boxstyle=, facecolor=, alpha=))
ax1.set_xlim([, ])
ax1.set_ylim([, ])
ax1.axis()
ax2 = fig.add_subplot(gs[, ])
all_important_layers = []
pred predictions:
all_important_layers.extend(pred.get(, []))
all_important_layers:
ax2.hist(all_important_layers, bins=, color=, alpha=, edgecolor=)
ax2.set_xlabel()
ax2.set_ylabel()
ax2.set_title()
ax2.grid(, alpha=)
ax3 = fig.add_subplot(gs[, :])
step_indices = (((predictions)))
confidences = [ - i * + np.random.randn() * i step_indices]
ax3.bar(step_indices, confidences, color=, alpha=)
ax3.set_xlabel()
ax3.set_ylabel()
ax3.set_title()
ax3.set_xticks(step_indices)
ax3.set_xticklabels([ i step_indices], rotation=)
ax3.grid(, alpha=, axis=)
ax4 = fig.add_subplot(gs[, :])
stats_text =
ax4.text(, , stats_text, fontsize=, verticalalignment=,
family=, bbox=(boxstyle=, facecolor=, alpha=))
ax4.axis()
plt.tight_layout()
filepath = os.path.join(.output_dir, )
plt.savefig(filepath, dpi=, bbox_inches=)
plt.close()
filepath
() -> [, ]:
(analysis_json_path, ) f:
analysis_data = json.load(f)
model_name:
model_name = analysis_data.get(, )
generated_files = {}
dashboard_path = .create_summary_dashboard(analysis_data, model_name)
generated_files[] = dashboard_path
()
sample_layer_predictions = []
i ():
predictions = [
(, - i * ),
(, - i * ),
(, - i),
(, - i * ),
]
sample_layer_predictions.append({
: i,
: predictions
})
step_idx ((, analysis_data.get(, ))):
heatmap_path = .create_layer_heatmap(
sample_layer_predictions,
step_idx,
title=,
show_all_layers=
)
generated_files[] = heatmap_path
()
generated_files
():
visualizer = LogitLensVisualizer(output_dir=)
mock_analysis_data = {
: ,
: ,
: [
{: , : , : [, , ]},
{: , : , : [, , ]},
{: , : , : [, , ]},
{: , : , : [, , ]},
{: , : , : [, , ]},
]
}
os.makedirs(, exist_ok=)
mock_json_path =
(mock_json_path, ) f:
json.dump(mock_analysis_data, f, indent=)
()
( * )
generated_files = visualizer.generate_all_visualizations(
mock_json_path,
model_name=
)
( + * )
()
()
viz_type, filepath generated_files.items():
()
__name__ == :
main()