sae-gpu-backend / app.py
justforkicks's picture
Update app.py
eda1ec9 verified
Raw
History Blame Contribute Delete
360 kB
"""
Local SAE Feature Analysis Service
This Flask service runs SAE inference locally using SAELens and TransformerLens,
bypassing Neuronpedia's rate limits for corpus analysis.
Supports:
- GPT-2 Small with res-jb SAE (Layer 6, residual stream)
- Pythia-70M with EleutherAI SAEs (MLP outputs, all layers)
- Gemma 2 2B with Gemma Scope SAEs (Layer 12):
- Residual stream (gemma-scope-2b-pt-res)
- Attention output (gemma-scope-2b-pt-att)
- MLP output (gemma-scope-2b-pt-mlp)
- Batched feature analysis for corpus workflows
Model Selection Rationale:
- GPT-2 Small: The "fruit fly" of interpretability - small, fast, well-documented
- Pythia-70M: EleutherAI's interpretability-focused model with full training checkpoints
- Gemma 2 2B: Google's official Gemma Scope SAEs with residual/attention/MLP variants
Larger models (Llama, Mistral, Pythia 1B+) require GPU and are available via
Neuronpedia API, not local inference.
"""
import os
import json
import math
import threading
from flask import Flask, request, jsonify
from flask_cors import CORS
import torch
import numpy as np
class NumpyEncoder(json.JSONEncoder):
"""JSON encoder that handles numpy types transparently."""
def default(self, obj):
if isinstance(obj, (np.integer,)):
return int(obj)
if isinstance(obj, (np.floating,)):
return float(obj)
if isinstance(obj, np.ndarray):
return obj.tolist()
return super().default(obj)
# Use cached models without network checks when possible
# This prevents hangs when HuggingFace has connectivity issues
os.environ.setdefault("HF_HUB_OFFLINE", "0") # Set to "1" after first successful load
os.environ.setdefault("TRANSFORMERS_OFFLINE", "0")
os.environ.setdefault("HF_HUB_DISABLE_SYMLINKS_WARNING", "1")
# Lazy loading to reduce startup time
_model = None
_saes = {}
_current_model_id = None
import threading
import functools
_model_swap_lock = threading.RLock()
def gpu_exclusive(f):
"""Decorator: hold the model lock for the entire request lifecycle.
Prevents concurrent requests from swapping the model mid-inference."""
@functools.wraps(f)
def wrapper(*args, **kwargs):
with _model_swap_lock:
return f(*args, **kwargs)
return wrapper
def _cleanup_model_vram():
"""Free VRAM from current model, ALL cached SAEs, AND OpenReviewer."""
global _model, _saes, _current_model_id, _openreviewer_model, _openreviewer_tokenizer
import gc
if _model is not None:
try:
_model.cpu()
except Exception:
pass
del _model
_model = None
all_sae_keys = list(_saes.keys())
for k in all_sae_keys:
sae = _saes.pop(k, None)
if sae is not None:
try:
if hasattr(sae, 'cpu'):
sae.cpu()
del sae
except Exception:
pass
_current_model_id = None
if _openreviewer_model is not None:
print("[SAE Service] Unloading OpenReviewer during VRAM cleanup")
try:
_openreviewer_model.cpu()
except Exception:
pass
del _openreviewer_model
_openreviewer_model = None
if _openreviewer_tokenizer is not None:
del _openreviewer_tokenizer
_openreviewer_tokenizer = None
gc.collect()
if torch.cuda.is_available():
torch.cuda.synchronize()
torch.cuda.empty_cache()
gc.collect()
torch.cuda.empty_cache()
free_mb = torch.cuda.mem_get_info()[0] // (1024 * 1024)
print(f"[SAE Service] VRAM cleanup done. Free: {free_mb} MB")
GEMMA_SCOPE_L0_MAP = {
"gemma-2-2b": {
0: 46, 1: 40, 2: 53, 3: 59, 4: 60, 5: 68,
6: 70, 7: 69, 8: 71, 9: 73, 10: 77, 11: 80,
12: 82, 13: 83, 14: 83, 15: 78, 16: 78, 17: 77,
18: 74, 19: 73, 20: 71, 21: 70, 22: 72, 23: 74,
24: 73, 25: 55,
},
"gemma-2-9b": {
0: 35, 1: 31, 2: 29, 3: 37, 4: 37, 5: 37,
6: 47, 7: 46, 8: 51, 9: 51, 10: 57, 11: 60,
12: 64, 13: 65, 14: 67, 15: 65, 16: 75, 17: 73,
18: 71, 19: 67, 20: 68, 21: 66, 22: 65, 23: 63,
24: 61, 25: 61, 26: 63, 27: 65, 28: 65, 29: 66,
30: 66, 31: 63, 32: 61, 33: 63, 34: 60, 35: 61,
36: 61, 37: 63, 38: 64, 39: 64, 40: 61, 41: 52,
},
}
DEFAULT_LAYERS = {
"gpt2-small": 6,
"gemma-2-2b": 12,
"gemma-2-2b-res": 12,
"gemma-2-2b-att": 12,
"gemma-2-2b-mlp": 12,
"gemma-2-9b": 20,
"pythia-70m": 3,
"salamandra-2b": 12,
}
MODEL_LAYER_COUNTS = {
"gpt2-small": 12,
"gemma-2-2b": 26, "gemma-2-2b-res": 26, "gemma-2-2b-att": 26, "gemma-2-2b-mlp": 26,
"gemma-2-9b": 42,
"pythia-70m": 6,
"salamandra-2b": 24,
}
def clamp_layer(model_id: str, layer: int) -> int:
max_layers = MODEL_LAYER_COUNTS.get(model_id, 26)
return max(0, min(layer, max_layers - 1))
def resolve_gemma_sae_id(model_id: str, layer: int) -> str:
base = "gemma-2-2b" if model_id.startswith("gemma-2-2b") else model_id
l0_map = GEMMA_SCOPE_L0_MAP.get(base, {})
default_l0 = 82 if "2b" in model_id else 68
l0 = l0_map.get(layer, default_l0)
return f"layer_{layer}/width_16k/average_l0_{l0}"
# Serialize model access — TransformerLens hooks use shared mutable state on the
# global model object. When Flask handles concurrent requests (threaded=True by
# default), multiple threads calling model.run_with_cache() simultaneously corrupt
# each other's hook activations and return identical results. This lock ensures
# only one request uses the model at a time.
_model_lock = threading.Lock()
def sanitize_nan(obj):
"""Recursively replace NaN/Inf floats with 0.0 to produce valid JSON."""
if isinstance(obj, float):
if math.isnan(obj) or math.isinf(obj):
return 0.0
return obj
elif isinstance(obj, dict):
return {k: sanitize_nan(v) for k, v in obj.items()}
elif isinstance(obj, (list, tuple)):
return [sanitize_nan(v) for v in obj]
return obj
app = Flask(__name__)
CORS(app)
def sanitize_for_json(obj):
"""Recursively convert numpy types to native Python types for JSON serialization."""
if isinstance(obj, dict):
return {k: sanitize_for_json(v) for k, v in obj.items()}
elif isinstance(obj, (list, tuple)):
return [sanitize_for_json(v) for v in obj]
elif isinstance(obj, (np.integer,)):
return int(obj)
elif isinstance(obj, (np.floating,)):
v = float(obj)
if math.isnan(v) or math.isinf(v):
return 0
return v
elif isinstance(obj, np.ndarray):
return sanitize_for_json(obj.tolist())
elif isinstance(obj, float):
if math.isnan(obj) or math.isinf(obj):
return 0
return obj
return obj
@app.errorhandler(TypeError)
def handle_type_error(error):
"""Catch TypeError from jsonify when numpy types slip through — return a clean error."""
error_msg = str(error)
if 'JSON serializable' in error_msg:
print(f"[JSON] TypeError caught: {error_msg} — this usually means numpy types weren't converted")
return jsonify({"error": "Internal serialization error", "details": error_msg}), 500
raise error
_GPU_EXEMPT_ROUTES = {'/health', '/cleanup-vram', '/cleanup-tensors'}
@app.before_request
def acquire_gpu_lock():
"""Serialize all GPU-using requests so model swaps can't corrupt running inference."""
from flask import g, request as req
if req.path not in _GPU_EXEMPT_ROUTES:
_model_swap_lock.acquire()
g._gpu_lock_held = True
@app.teardown_request
def release_gpu_lock(exc=None):
from flask import g
if getattr(g, '_gpu_lock_held', False):
_model_swap_lock.release()
g._gpu_lock_held = False
@app.after_request
def cleanup_gpu_tensors(response):
"""Free intermediate GPU tensors after every request.
Route handlers create cache objects from run_with_cache() that hold ALL
intermediate activations on GPU. Once the handler returns, those caches are
out of scope but not yet collected by Python's GC. Force collection here
so CUDA memory is available for the next request."""
from flask import request as req
if req.path not in _GPU_EXEMPT_ROUTES:
import gc
gc.collect()
if torch.cuda.is_available():
torch.cuda.empty_cache()
return response
@app.after_request
def fix_nan_json(response):
"""Intercept JSON responses and sanitize NaN/Inf values that break JSON parsing."""
if response.content_type and 'application/json' in response.content_type:
try:
data = response.get_json(silent=True)
if data is not None:
cleaned = sanitize_for_json(data)
response.set_data(json.dumps(cleaned, allow_nan=False))
except Exception:
pass
return response
def get_hook_name(model_id: str, layer: int = None) -> str:
"""Get the correct hook name for a given model/SAE type"""
default_layer = DEFAULT_LAYERS.get(model_id, 12)
l = layer if layer is not None else default_layer
if model_id == "gpt2-small":
return f"blocks.{l}.hook_resid_pre"
elif model_id == "pythia-70m":
return f"blocks.{l}.hook_mlp_out"
elif model_id == "gemma-2-2b-att":
return f"blocks.{l}.attn.hook_result"
elif model_id == "gemma-2-2b-mlp":
return f"blocks.{l}.hook_mlp_out"
elif model_id in ("gemma-2-2b", "gemma-2-2b-res", "gemma-2-9b"):
return f"blocks.{l}.hook_resid_pre"
else:
return f"blocks.{l}.hook_resid_pre"
def safe_get_cache(cache, hook_name):
"""Safely get activations from cache, trying alternative hook names if needed."""
if hook_name in cache:
return cache[hook_name]
alt = hook_name.replace("hook_resid_pre", "hook_resid_post") if "hook_resid_pre" in hook_name else hook_name.replace("hook_resid_post", "hook_resid_pre")
if alt in cache:
print(f"[SAE Service] Hook '{hook_name}' not found, using '{alt}'")
return cache[alt]
attn_fallbacks = [
("attn.hook_result", "attn.hook_z"),
("attn.hook_z", "attn.hook_result"),
("attn.hook_result", "attn.hook_attn_out"),
("attn.hook_z", "attn.hook_attn_out"),
]
for src, dst in attn_fallbacks:
if src in hook_name:
attn_alt = hook_name.replace(src, dst)
if attn_alt in cache:
print(f"[SAE Service] Hook '{hook_name}' not found, using attention fallback '{attn_alt}'")
return cache[attn_alt]
parts = hook_name.split(".")
block_prefix = ".".join(parts[:2]) if len(parts) >= 2 else hook_name
available = [k for k in cache.keys() if block_prefix in k]
if available:
print(f"[SAE Service] Hook '{hook_name}' not found, using '{available[0]}' from {len(available)} options")
return cache[available[0]]
raise KeyError(f"'{hook_name}' not found in cache. Available: {list(cache.keys())[:10]}")
GATED_MODELS = {"meta-llama/Meta-Llama-3.1-8B"}
ACTIVATION_ONLY_MODELS = {"qwen2.5-1.5b", "bloom-3b", "bloom-560m", "salamandra-2b"}
LLAMA_SAE_LAYERS = 32
def _load_model_only(model_id: str):
"""Load just the HookedTransformer model (no SAE) for activation-only analysis.
Uses global _model cache, handles VRAM cleanup when switching models."""
global _model, _saes, _current_model_id
with _model_swap_lock:
if _openreviewer_model is not None:
print("[SAE Service] Unloading OpenReviewer to free VRAM for SAE model")
_unload_openreviewer()
base_model = get_base_model_id(model_id)
if _model is not None and get_base_model_id(_current_model_id) == base_model:
_model.reset_hooks()
return _model
if _model is not None:
_cleanup_model_vram()
_model = _load_hooked_transformer(base_model)
_saes = {}
_current_model_id = model_id
return _model
def get_base_model_id(model_id: str) -> str:
"""Get the base transformer model for loading"""
if model_id is None:
return None
if model_id.startswith("gemma-2-2b"):
return "gemma-2-2b"
elif model_id == "gemma-2-9b":
return "gemma-2-9b"
elif model_id == "pythia-70m":
return "EleutherAI/pythia-70m"
elif model_id in ("llama-3.1-8b", "llama-3.1-8B", "Llama-3.1-8B"):
return "meta-llama/Meta-Llama-3.1-8B"
elif model_id in ("salamandra-2b", "salamandra-2B"):
return "BSC-LT/salamandra-2b"
elif model_id.startswith("gpt2"):
return "gpt2-small"
return model_id
def _ensure_qwen_rope_theta():
"""Monkey-patch transformers Qwen2Config to provide rope_theta default if missing."""
try:
from transformers import Qwen2Config
_orig_getattr = Qwen2Config.__getattribute__
def _patched_getattr(self, key):
if key == "rope_theta":
try:
return _orig_getattr(self, key)
except AttributeError:
return 1000000.0
return _orig_getattr(self, key)
Qwen2Config.__getattribute__ = _patched_getattr
print("[Model Loader] Patched Qwen2Config to provide rope_theta default")
except ImportError:
pass
except Exception as e:
print(f"[Model Loader] Qwen patch warning: {e}")
_ensure_qwen_rope_theta()
def _load_hooked_transformer(base_model: str, device: str = None):
"""Load a HookedTransformer model, handling gated models and HF token.
For models not in the TransformerLens registry (e.g. Llama 3.1),
we use from_pretrained_no_processing which loads directly from HuggingFace
without requiring a built-in model config.
"""
from transformer_lens import HookedTransformer
if device is None:
device = _get_device()
kwargs = dict(device=device, dtype=torch.float16)
if base_model in GATED_MODELS:
hf_token = os.environ.get("HF_TOKEN")
if not hf_token:
raise RuntimeError(f"Model {base_model} is gated and requires HF_TOKEN environment variable")
kwargs["token"] = hf_token
print(f"[Model Loader] Loading {base_model} on {device}")
try:
return HookedTransformer.from_pretrained(base_model, **kwargs)
except Exception as e:
error_msg = str(e)
if "not found" in error_msg.lower() or "not in" in error_msg.lower() or "valid" in error_msg.lower():
print(f"[Model Loader] {base_model} not in TransformerLens registry, trying from_pretrained_no_processing...")
try:
return HookedTransformer.from_pretrained_no_processing(base_model, **kwargs)
except Exception as e2:
print(f"[Model Loader] from_pretrained_no_processing also failed: {e2}")
raise e2
raise e
GEMMA_MLP_L0 = 60
def _get_device():
"""Auto-detect best available device: CUDA GPU if available, else CPU"""
if torch.cuda.is_available():
return "cuda"
return "cpu"
def get_model_and_sae(model_id: str = "gpt2-small", layer: int = None):
"""Lazy load model and SAE on first request, using GPU if available.
SAEs are cached by model_id:layer so switching layers doesn't re-download.
If layer is None, uses the default layer for the model.
"""
global _model, _saes, _current_model_id
with _model_swap_lock:
if _openreviewer_model is not None:
print("[SAE Service] Unloading OpenReviewer to free VRAM for SAE model")
_unload_openreviewer()
if _model is not None:
_model.reset_hooks()
default_layer = DEFAULT_LAYERS.get(model_id, 12)
actual_layer = layer if layer is not None else default_layer
sae_cache_key = f"{model_id}:{actual_layer}"
if _model is not None and _current_model_id == model_id and sae_cache_key in _saes:
return _model, _saes[sae_cache_key]
device = _get_device()
print(f"[SAE Service] Loading model: {model_id} layer {actual_layer} on {device}")
from transformer_lens import HookedTransformer
from sae_lens import SAE
base_model = get_base_model_id(model_id)
current_base = get_base_model_id(_current_model_id) if _current_model_id else None
if _model is None or current_base != base_model:
if _model is not None:
print(f"[SAE Service] Swapping model: {_current_model_id} -> {model_id}, cleaning up VRAM")
_cleanup_model_vram()
print(f"[SAE Service] Loading transformer: {base_model} on {device}")
try:
_model = _load_hooked_transformer(base_model, device=device)
except torch.cuda.OutOfMemoryError:
if torch.cuda.is_available():
torch.cuda.empty_cache()
raise RuntimeError(f"CUDA out of memory loading {model_id}. GPU does not have enough VRAM for this model. Try restarting the Space to clear stale allocations.")
if sae_cache_key not in _saes:
if model_id == "gpt2-small":
sae, _, _ = SAE.from_pretrained(
release="gpt2-small-res-jb",
sae_id=f"blocks.{actual_layer}.hook_resid_pre",
device=device
)
elif model_id in ("gemma-2-2b", "gemma-2-2b-res"):
sae_id = resolve_gemma_sae_id(model_id, actual_layer)
print(f"[SAE Service] Gemma SAE ID: {sae_id}")
sae, _, _ = SAE.from_pretrained(
release="gemma-scope-2b-pt-res",
sae_id=sae_id,
device=device
)
elif model_id == "gemma-2-2b-att":
sae, _, _ = SAE.from_pretrained(
release="gemma-scope-2b-pt-att",
sae_id=f"layer_{actual_layer}/width_16k/average_l0_77",
device=device
)
elif model_id == "gemma-2-2b-mlp":
sae, _, _ = SAE.from_pretrained(
release="gemma-scope-2b-pt-mlp",
sae_id=f"layer_{actual_layer}/width_16k/average_l0_{GEMMA_MLP_L0}",
device=device
)
elif model_id == "gemma-2-9b":
sae_id = resolve_gemma_sae_id(model_id, actual_layer)
print(f"[SAE Service] Gemma 9B SAE ID: {sae_id}")
sae, _, _ = SAE.from_pretrained(
release="gemma-scope-9b-pt-res",
sae_id=sae_id,
device=device
)
elif model_id == "pythia-70m":
sae, _, _ = SAE.from_pretrained(
release="EleutherAI/sae-pythia-70m-32k",
sae_id=f"blocks.{actual_layer}.hook_mlp_out",
device=device
)
elif model_id == "llama-3.1-8b":
llama_layer = min(actual_layer, LLAMA_SAE_LAYERS - 1)
llama_sae_id = f"blocks.{llama_layer}.hook_resid_post"
print(f"[SAE Service] Llama Scope SAE: layer {llama_layer}, id {llama_sae_id}")
try:
sae, _, _ = SAE.from_pretrained(
release="llama_scope_lxr_32x",
sae_id=llama_sae_id,
device=device
)
except Exception as llama_sae_err:
print(f"[SAE Service] Llama Scope 32x failed ({llama_sae_err}), trying 8x release...")
try:
sae, _, _ = SAE.from_pretrained(
release="llama_scope_lxr_8x",
sae_id=llama_sae_id,
device=device
)
except Exception as llama_sae_err2:
print(f"[SAE Service] Llama Scope 8x also failed ({llama_sae_err2}), falling back to activation-only")
sae = None
elif model_id in ACTIVATION_ONLY_MODELS:
sae = None
print(f"[SAE Service] {model_id} is activation-only (no SAE)")
else:
raise ValueError(f"Unsupported model: {model_id}")
_saes[sae_cache_key] = sae
print(f"[SAE Service] SAE cached as {sae_cache_key}")
_current_model_id = model_id
print(f"[SAE Service] Model and SAE loaded successfully on {device}")
return _model, _saes[sae_cache_key]
@app.route('/health', methods=['GET'])
def health():
"""Health check endpoint"""
free_mb = 0
model_loaded = _model is not None
if torch.cuda.is_available():
free_mb = torch.cuda.mem_get_info()[0] // (1024 * 1024)
return jsonify({"status": "healthy", "service": "sae-local", "free_mb": free_mb, "model_loaded": model_loaded})
@app.route('/cleanup-vram', methods=['POST'])
def cleanup_vram():
"""Force full VRAM cleanup — unloads model + SAEs. Used before new protocol rounds."""
global _model, _saes, _current_model_id
try:
with _model_swap_lock:
_cleanup_model_vram()
free_mb = 0
if torch.cuda.is_available():
torch.cuda.synchronize()
import gc
gc.collect()
torch.cuda.empty_cache()
gc.collect()
torch.cuda.empty_cache()
free_mb = torch.cuda.mem_get_info()[0] // (1024 * 1024)
return jsonify({"status": "ok", "free_mb": free_mb})
except Exception as e:
return jsonify({"status": "error", "error": str(e)}), 500
@app.route('/cleanup-tensors', methods=['POST'])
def cleanup_tensors():
"""Lightweight VRAM cleanup — frees intermediate tensors but keeps model loaded.
Used between consecutive GPU tool calls within a protocol run."""
try:
import gc
gc.collect()
free_mb = 0
if torch.cuda.is_available():
torch.cuda.synchronize()
torch.cuda.empty_cache()
gc.collect()
torch.cuda.empty_cache()
free_mb = torch.cuda.mem_get_info()[0] // (1024 * 1024)
return jsonify({"status": "ok", "free_mb": free_mb, "model_loaded": _model is not None})
except Exception as e:
return jsonify({"status": "error", "error": str(e)}), 500
@app.route('/gpu-analyze', methods=['POST'])
def gpu_analyze():
"""
GPU-compatible analyze: get top SAE features at a specific layer.
Request body:
{
"prompt": "The Eiffel Tower is located in",
"model_id": "gpt2-small",
"layer": 6,
"top_k": 32
}
"""
try:
data = request.json
prompt = data.get('prompt', '')
model_id = data.get('model_id', data.get('model', 'gpt2-small'))
layer = data.get('layer', 0)
top_k = data.get('top_k', 32)
if not prompt:
return jsonify({"error": "No prompt provided"}), 400
with _model_lock:
actual_layer = clamp_layer(model_id, layer)
model, sae = get_model_and_sae(model_id, layer=actual_layer)
tokens = model.to_tokens(prompt)
str_tokens = [t.replace('\u0120', ' ').replace('\u010a', '\n') for t in model.to_str_tokens(prompt)]
hook_name = f"blocks.{actual_layer}.hook_resid_post"
if sae is None:
return jsonify({"error": f"SAE not available for {model_id} (activation-only mode). SAE weights could not be loaded."}), 400
with torch.no_grad():
_, cache = model.run_with_cache(tokens, names_filter=[hook_name])
resid = safe_get_cache(cache, hook_name)
last_pos_resid = resid[0, -1, :]
sae_input = last_pos_resid.unsqueeze(0)
feature_acts = sae.encode(sae_input)
acts = feature_acts[0]
top_values, top_indices = torch.topk(acts, min(top_k, acts.shape[0]))
features = []
for i in range(top_values.shape[0]):
val = top_values[i].item()
if val > 0:
features.append({
"feature": top_indices[i].item(),
"index": top_indices[i].item(),
"activation": round(val, 6)
})
del cache
target_token = str_tokens[-1] if str_tokens else ""
device_name = "cuda" if torch.cuda.is_available() else "cpu"
return jsonify({
"success": True,
"model": model_id,
"layer": actual_layer,
"tokens": str_tokens,
"targetToken": target_token,
"features": features,
"device": device_name
})
except Exception as e:
print(f"[SAE Service] GPU-analyze error: {e}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/feature-activation-map', methods=['POST'])
def feature_activation_map():
"""
Get per-token activations for a specific SAE feature across all token positions.
Returns a diverging activation map suitable for Anthropic-style token coloring.
Request body:
{
"prompt": "The model is reasoning about this problem...",
"model": "gemma-2-2b",
"layer": 20,
"feature_index": 1234,
"normalize": true
}
"""
try:
data = request.json
prompt = data.get('prompt', '')
model_id = data.get('model', data.get('model_id', 'gemma-2-2b'))
layer = data.get('layer', 20)
feature_index = data.get('feature_index', 0)
normalize = data.get('normalize', True)
if not prompt:
return jsonify({"error": "No prompt provided"}), 400
with _model_lock:
actual_layer = clamp_layer(model_id, layer)
model, sae = get_model_and_sae(model_id, layer=actual_layer)
if sae is None:
return jsonify({"error": f"SAE not available for {model_id}. Feature activation map requires SAE support."}), 400
tokens = model.to_tokens(prompt)
str_tokens = [t.replace('\u0120', ' ').replace('\u010a', '\n') for t in model.to_str_tokens(prompt)]
hook_name = f"blocks.{actual_layer}.hook_resid_post"
with torch.no_grad():
_, cache = model.run_with_cache(tokens, names_filter=[hook_name])
resid = safe_get_cache(cache, hook_name)
seq_len = resid.shape[1]
all_acts = sae.encode(resid[0])
feature_acts = all_acts[:, feature_index].float().cpu().tolist()
top_features_last = []
last_acts = all_acts[-1]
top_vals, top_idxs = torch.topk(last_acts, min(20, last_acts.shape[0]))
for i in range(top_vals.shape[0]):
v = top_vals[i].item()
if v > 0:
top_features_last.append({
"feature": top_idxs[i].item(),
"activation": round(v, 6),
})
del cache
max_abs = max(abs(v) for v in feature_acts) if feature_acts else 1.0
if normalize and max_abs > 0:
normalized = [round(v / max_abs, 6) for v in feature_acts]
else:
normalized = [round(v, 6) for v in feature_acts]
return jsonify({
"success": True,
"model": model_id,
"layer": actual_layer,
"featureIndex": feature_index,
"tokens": str_tokens,
"activations": [round(v, 6) for v in feature_acts],
"normalizedActivations": normalized,
"maxActivation": round(max_abs, 6),
"topFeaturesAtLastToken": top_features_last,
})
except Exception as e:
print(f"[SAE Service] Feature activation map error: {e}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/gpu-analyze-batch', methods=['POST'])
def gpu_analyze_batch():
"""
Batch GPU analyze: get top SAE features across multiple layers in one request.
Acquires the model lock once, runs the model once caching all layers, then
extracts features at each layer. Much faster than N individual /gpu-analyze calls.
Request body:
{
"prompts": ["The Eiffel Tower is in", "A calculator is a"],
"model_id": "gpt2-small",
"layers": [0, 1, 2, ..., 11],
"top_k": 20
}
Returns:
{
"success": true,
"results": [
{"prompt_index": 0, "layer": 0, "features": [...]},
{"prompt_index": 0, "layer": 1, "features": [...]},
{"prompt_index": 1, "layer": 0, "features": [...]},
...
]
}
"""
try:
data = request.json
prompts = data.get('prompts', [])
model_id = data.get('model_id', data.get('model', 'gpt2-small'))
layers = data.get('layers', list(range(12)))
top_k = data.get('top_k', 20)
if not prompts:
return jsonify({"error": "No prompts provided"}), 400
import time
start_time = time.time()
with _model_lock:
model, sae = get_model_and_sae(model_id)
if sae is None:
return jsonify({"error": f"SAE not available for {model_id} (activation-only mode). SAE weights could not be loaded."}), 400
actual_layers = [min(l, model.cfg.n_layers - 1) for l in layers]
unique_layers = sorted(set(actual_layers))
hook_names = [f"blocks.{l}.hook_resid_post" for l in unique_layers]
all_results = []
with torch.no_grad():
for pi, prompt in enumerate(prompts):
tokens = model.to_tokens(prompt)
_, cache = model.run_with_cache(tokens, names_filter=hook_names)
for layer in unique_layers:
hook_name = f"blocks.{layer}.hook_resid_post"
resid = safe_get_cache(cache, hook_name)
last_pos_resid = resid[0, -1, :]
sae_input = last_pos_resid.unsqueeze(0)
feature_acts = sae.encode(sae_input)
acts = feature_acts[0]
top_values, top_indices = torch.topk(acts, min(top_k, acts.shape[0]))
features = []
for i in range(top_values.shape[0]):
val = top_values[i].item()
if val > 0:
features.append({
"feature": top_indices[i].item(),
"index": top_indices[i].item(),
"activation": round(val, 6)
})
all_results.append({
"prompt_index": pi,
"layer": layer,
"features": features
})
del cache
elapsed = round(time.time() - start_time, 2)
device_name = "cuda" if torch.cuda.is_available() else "cpu"
print(f"[SAE Service] Batch analyze: {len(prompts)} prompts x {len(unique_layers)} layers in {elapsed}s on {device_name}")
return jsonify({
"success": True,
"results": all_results,
"model": model_id,
"device": device_name,
"elapsed": elapsed
})
except Exception as e:
print(f"[SAE Service] Batch analyze error: {e}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/analyze', methods=['POST'])
def analyze():
"""
Analyze text with SAE features locally.
Request body:
{
"text": "Hello world",
"model": "gpt2-small",
"features": [100, 200, 300, ...] // Feature indices to check
}
Response:
{
"tokens": ["Hello", " world"],
"activations": {
"100": [0.0, 0.5, ...], // Activation per token
"200": [0.1, 0.0, ...],
...
},
"model": "gpt2-small"
}
"""
try:
data = request.json
text = data.get('text', '')
model_id = data.get('model', 'gpt2-small')
requested_features = data.get('features', [])
if not text:
return jsonify({"error": "No text provided"}), 400
with _model_lock:
model, sae = get_model_and_sae(model_id)
tokens = model.to_tokens(text)
str_tokens = model.to_str_tokens(text)
with torch.no_grad():
_, cache = model.run_with_cache(tokens)
hook_name = get_hook_name(model_id)
activations = safe_get_cache(cache, hook_name)
feature_acts = sae.encode(activations)
result_activations = {}
for feat_idx in requested_features:
if feat_idx < feature_acts.shape[-1]:
feat_values = feature_acts[0, :, feat_idx].tolist()
result_activations[str(feat_idx)] = feat_values
clean_tokens = [t.replace('Ġ', ' ').replace('Ċ', '\n') for t in str_tokens]
return jsonify({
"tokens": clean_tokens,
"activations": result_activations,
"model": model_id,
"tokenCount": len(clean_tokens),
"featuresAnalyzed": len(requested_features)
})
except Exception as e:
print(f"[SAE Service] Error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/batch-analyze', methods=['POST'])
def batch_analyze():
"""
Analyze multiple texts in a batch for corpus analysis.
Request body:
{
"texts": ["text1", "text2", ...],
"model": "gpt2-small",
"features": [100, 200, 300, ...]
}
Response:
{
"results": [
{"text": "text1", "activatingFeatures": [100, 200]},
{"text": "text2", "activatingFeatures": [100]},
...
],
"featureStats": {
"100": {"activationCount": 2, "texts": [0, 1]},
"200": {"activationCount": 1, "texts": [0]},
...
}
}
"""
try:
data = request.json
texts = data.get('texts', [])
model_id = data.get('model', 'gpt2-small')
requested_features = data.get('features', [])
threshold = data.get('threshold', 0.1) # Activation threshold
if not texts:
return jsonify({"error": "No texts provided"}), 400
model, sae = get_model_and_sae(model_id)
results = []
feature_stats = {str(f): {"activationCount": 0, "texts": [], "maxActivation": 0.0} for f in requested_features}
for text_idx, text in enumerate(texts):
tokens = model.to_tokens(text)
with torch.no_grad():
_, cache = model.run_with_cache(tokens)
hook_name = get_hook_name(model_id)
activations = safe_get_cache(cache, hook_name)
feature_acts = sae.encode(activations)
# Check which features activated above threshold
activating_features = []
for feat_idx in requested_features:
if feat_idx < feature_acts.shape[-1]:
max_act = feature_acts[0, :, feat_idx].max().item()
if max_act > threshold:
activating_features.append(feat_idx)
feat_key = str(feat_idx)
feature_stats[feat_key]["activationCount"] += 1
feature_stats[feat_key]["texts"].append(text_idx)
feature_stats[feat_key]["maxActivation"] = max(
feature_stats[feat_key]["maxActivation"], max_act
)
results.append({
"textIndex": text_idx,
"activatingFeatures": activating_features
})
print(f"[SAE Service] Processed text {text_idx + 1}/{len(texts)}: {len(activating_features)} features activated")
return jsonify({
"results": results,
"featureStats": feature_stats,
"model": model_id,
"textsAnalyzed": len(texts),
"featuresAnalyzed": len(requested_features)
})
except Exception as e:
print(f"[SAE Service] Batch error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/steer', methods=['POST'])
def steer():
"""
Generate text with a feature ablated (set to 0) or amplified.
Shows how the feature affects model output.
Request body:
{
"text": "Hello world",
"model": "gpt2-small",
"feature": 16758,
"mode": "ablate" | "amplify",
"strength": 5.0 // For amplify mode, multiplier for feature activation
}
Response:
{
"original": {
"text": "Hello world",
"continuation": " is a common greeting...",
"featureActivation": 0.5
},
"modified": {
"text": "Hello world",
"continuation": " is used in programming...",
"featureActivation": 0.0 // or amplified value
}
}
"""
try:
data = request.json
text = data.get('text', '')
model_id = data.get('model', 'gpt2-small')
feature_idx = data.get('feature', 0)
mode = data.get('mode', 'ablate') # "ablate" or "amplify"
strength = data.get('strength', 5.0) # Multiplier for amplify
max_new_tokens = data.get('maxNewTokens', 20)
if not text:
return jsonify({"error": "No text provided"}), 400
model, sae = get_model_and_sae(model_id)
# Get hook name based on model
hook_name = get_hook_name(model_id)
# Tokenize
tokens = model.to_tokens(text)
# Generate original continuation
with torch.no_grad():
original_output = model.generate(
tokens,
max_new_tokens=max_new_tokens,
temperature=0.7,
do_sample=True
)
original_text = model.to_string(original_output[0])
# Get original feature activation
_, cache = model.run_with_cache(tokens)
activations = safe_get_cache(cache, hook_name)
feature_acts = sae.encode(activations)
original_activation = feature_acts[0, :, feature_idx].max().item()
# Validate feature index
sae_dim = sae.W_dec.shape[0] if hasattr(sae, 'W_dec') else 24576
if feature_idx < 0 or feature_idx >= sae_dim:
return jsonify({"error": f"Feature index {feature_idx} out of range (0-{sae_dim-1})"}), 400
# Create steering hook that preserves reconstruction error
def steering_hook(activations, hook):
# Encode to get feature activations
original_feature_acts = sae.encode(activations)
# Compute reconstruction error (what SAE doesn't capture)
original_reconstruction = sae.decode(original_feature_acts)
reconstruction_error = activations - original_reconstruction
# Modify the target feature
modified_feature_acts = original_feature_acts.clone()
if mode == "ablate":
# Set target feature to zero
modified_feature_acts[:, :, feature_idx] = 0.0
else: # amplify
# Multiply target feature activation
modified_feature_acts[:, :, feature_idx] = modified_feature_acts[:, :, feature_idx] * strength
# Decode modified features and ADD BACK the reconstruction error
modified_reconstruction = sae.decode(modified_feature_acts)
return modified_reconstruction + reconstruction_error
# Generate with steering
with torch.no_grad():
model.reset_hooks()
model.add_hook(hook_name, steering_hook)
steered_output = model.generate(
tokens,
max_new_tokens=max_new_tokens,
temperature=0.7,
do_sample=True
)
steered_text = model.to_string(steered_output[0])
# Get modified feature activation (should be near 0 for ablate)
_, cache = model.run_with_cache(tokens)
activations = safe_get_cache(cache, hook_name)
modified_acts = steering_hook(activations, None)
modified_feature_acts = sae.encode(modified_acts)
modified_activation = modified_feature_acts[0, :, feature_idx].max().item()
model.reset_hooks()
# Extract just the continuation
original_continuation = original_text[len(model.to_string(tokens[0])):]
steered_continuation = steered_text[len(model.to_string(tokens[0])):]
return jsonify({
"original": {
"text": text,
"continuation": original_continuation,
"featureActivation": original_activation
},
"modified": {
"text": text,
"continuation": steered_continuation,
"featureActivation": modified_activation,
"mode": mode,
"strength": strength if mode == "amplify" else 0
},
"feature": feature_idx,
"model": model_id
})
except Exception as e:
print(f"[SAE Service] Steer error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/feature-ablation', methods=['POST'])
def feature_ablation():
"""
Measure causal impact of a specific SAE feature by ablating it.
Zeros out one feature's contribution and measures output change.
Request body:
{
"model": "gpt2-small",
"prompt": "The capital of France is",
"feature": 16758,
"layer": 6 (optional, uses model default)
}
"""
try:
data = request.json
model_id = data.get('model', 'gpt2-small')
prompt = data.get('prompt', '')
feature_idx = data.get('feature', 0)
if not prompt:
return jsonify({"error": "No prompt provided"}), 400
model, sae = get_model_and_sae(model_id)
hook_name = get_hook_name(model_id)
tokens = model.to_tokens(prompt)
with torch.no_grad():
baseline_logits = model(tokens)
baseline_probs = torch.softmax(baseline_logits[0, -1, :], dim=-1)
_, cache = model.run_with_cache(tokens)
activations = safe_get_cache(cache, hook_name)
feature_acts = sae.encode(activations)
feature_activation = feature_acts[0, :, feature_idx].max().item()
def ablation_hook(acts, hook):
original_feature_acts = sae.encode(acts)
original_reconstruction = sae.decode(original_feature_acts)
reconstruction_error = acts - original_reconstruction
modified_feature_acts = original_feature_acts.clone()
modified_feature_acts[:, :, feature_idx] = 0.0
modified_reconstruction = sae.decode(modified_feature_acts)
return modified_reconstruction + reconstruction_error
model.reset_hooks()
ablated_logits = model.run_with_hooks(
tokens,
fwd_hooks=[(hook_name, ablation_hook)],
reset_hooks_end=True
)
ablated_probs = torch.softmax(ablated_logits[0, -1, :], dim=-1)
top_k = 10
top_baseline_idx = torch.topk(baseline_probs, k=top_k).indices
top_ablated_idx = torch.topk(ablated_probs, k=top_k).indices
combined_idx = torch.unique(torch.cat([top_baseline_idx, top_ablated_idx]))
logit_diff = (baseline_probs[combined_idx] - ablated_probs[combined_idx]).abs().sum().item()
baseline_token = model.to_single_str_token(baseline_probs.argmax().item())
ablated_token = model.to_single_str_token(ablated_probs.argmax().item())
return jsonify({
"feature": feature_idx,
"featureActivation": feature_activation,
"logitDifference": logit_diff,
"baselineTopToken": baseline_token.replace('\u0120', ' '),
"ablatedTopToken": ablated_token.replace('\u0120', ' '),
"tokenChanged": baseline_token != ablated_token,
"model": model_id
})
except Exception as e:
print(f"[SAE Service] Feature ablation error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/feature-ablation-batch', methods=['POST'])
def feature_ablation_batch():
"""
Measure causal impact of a specific SAE feature across multiple prompts.
Returns per-prompt and averaged results.
Request body:
{
"model": "gpt2-small",
"prompts": ["prompt1", "prompt2", ...],
"feature": 16758
}
"""
try:
data = request.json
model_id = data.get('model', 'gpt2-small')
prompts = data.get('prompts', [])
feature_idx = data.get('feature', 0)
if not prompts:
return jsonify({"error": "No prompts provided"}), 400
model, sae = get_model_and_sae(model_id)
hook_name = get_hook_name(model_id)
results = []
for prompt in prompts:
tokens = model.to_tokens(prompt)
with torch.no_grad():
baseline_logits = model(tokens)
baseline_probs = torch.softmax(baseline_logits[0, -1, :], dim=-1)
_, cache = model.run_with_cache(tokens)
activations = safe_get_cache(cache, hook_name)
feature_acts = sae.encode(activations)
feature_activation = feature_acts[0, :, feature_idx].max().item()
def ablation_hook(acts, hook):
original_feature_acts = sae.encode(acts)
original_reconstruction = sae.decode(original_feature_acts)
reconstruction_error = acts - original_reconstruction
modified = original_feature_acts.clone()
modified[:, :, feature_idx] = 0.0
return sae.decode(modified) + reconstruction_error
model.reset_hooks()
ablated_logits = model.run_with_hooks(
tokens,
fwd_hooks=[(hook_name, ablation_hook)],
reset_hooks_end=True
)
ablated_probs = torch.softmax(ablated_logits[0, -1, :], dim=-1)
del cache
if torch.cuda.is_available():
torch.cuda.empty_cache()
top_k = 10
top_b = torch.topk(baseline_probs, k=top_k).indices
top_a = torch.topk(ablated_probs, k=top_k).indices
combined = torch.unique(torch.cat([top_b, top_a]))
logit_diff = (baseline_probs[combined] - ablated_probs[combined]).abs().sum().item()
baseline_token = model.to_single_str_token(baseline_probs.argmax().item())
ablated_token = model.to_single_str_token(ablated_probs.argmax().item())
results.append({
"prompt": prompt,
"featureActivation": feature_activation,
"logitDifference": logit_diff,
"baselineTopToken": baseline_token.replace('\u0120', ' '),
"ablatedTopToken": ablated_token.replace('\u0120', ' '),
"tokenChanged": baseline_token != ablated_token
})
avg_logit_diff = sum(r["logitDifference"] for r in results) / len(results)
avg_activation = sum(r["featureActivation"] for r in results) / len(results)
tokens_changed = sum(1 for r in results if r["tokenChanged"])
return jsonify({
"feature": feature_idx,
"results": results,
"averageLogitDiff": avg_logit_diff,
"averageActivation": avg_activation,
"tokensChanged": tokens_changed,
"totalPrompts": len(prompts),
"model": model_id
})
except Exception as e:
print(f"[SAE Service] Feature ablation batch error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/feature-ablation-sweep', methods=['POST'])
def feature_ablation_sweep():
"""
Ablate MULTIPLE features across ALL prompts in one call.
Returns per-feature causal impact with KL divergence.
Request body:
{
"model": "gpt2-small",
"prompts": ["prompt1", "prompt2", ...],
"features": [6471, 7393, 10543, ...]
}
"""
try:
data = request.json
model_id = data.get('model', 'gpt2-small')
prompts = data.get('prompts', [])
feature_indices = data.get('features', [])
if not prompts:
return jsonify({"error": "No prompts provided"}), 400
if not feature_indices:
return jsonify({"error": "No features provided"}), 400
model, sae = get_model_and_sae(model_id)
hook_name = get_hook_name(model_id)
sweep_results = {}
for fi_idx, feature_idx in enumerate(feature_indices):
print(f"[SAE Service] Ablation sweep: feature {feature_idx} ({fi_idx+1}/{len(feature_indices)})")
per_prompt = []
for prompt in prompts:
tokens = model.to_tokens(prompt)
with torch.no_grad():
baseline_logits = model(tokens)
baseline_probs = torch.softmax(baseline_logits[0, -1, :], dim=-1)
_, cache = model.run_with_cache(tokens)
activations = safe_get_cache(cache, hook_name)
feature_acts = sae.encode(activations)
feature_activation = feature_acts[0, :, feature_idx].max().item()
current_feature = feature_idx
def ablation_hook(acts, hook, feat_idx=current_feature):
orig_acts = sae.encode(acts)
orig_recon = sae.decode(orig_acts)
recon_error = acts - orig_recon
modified = orig_acts.clone()
modified[:, :, feat_idx] = 0.0
return sae.decode(modified) + recon_error
model.reset_hooks()
ablated_logits = model.run_with_hooks(
tokens,
fwd_hooks=[(hook_name, ablation_hook)],
reset_hooks_end=True
)
ablated_probs = torch.softmax(ablated_logits[0, -1, :], dim=-1)
del cache
if torch.cuda.is_available():
torch.cuda.empty_cache()
eps = 1e-10
kl_div = (baseline_probs * torch.log((baseline_probs + eps) / (ablated_probs + eps))).sum().item()
top_k = 10
top_b = torch.topk(baseline_probs, k=top_k).indices
top_a = torch.topk(ablated_probs, k=top_k).indices
combined = torch.unique(torch.cat([top_b, top_a]))
logit_diff = (baseline_probs[combined] - ablated_probs[combined]).abs().sum().item()
baseline_token = model.to_single_str_token(baseline_probs.argmax().item())
ablated_token = model.to_single_str_token(ablated_probs.argmax().item())
per_prompt.append({
"prompt": prompt,
"featureActivation": round(feature_activation, 6),
"logitDifference": round(logit_diff, 6),
"klDivergence": round(kl_div, 6),
"baselineTopToken": baseline_token.replace('\u0120', ' '),
"ablatedTopToken": ablated_token.replace('\u0120', ' '),
"tokenChanged": baseline_token != ablated_token
})
avg_logit_diff = sum(r["logitDifference"] for r in per_prompt) / len(per_prompt)
avg_kl = sum(r["klDivergence"] for r in per_prompt) / len(per_prompt)
avg_activation = sum(r["featureActivation"] for r in per_prompt) / len(per_prompt)
tokens_changed = sum(1 for r in per_prompt if r["tokenChanged"])
sweep_results[str(feature_idx)] = {
"feature": feature_idx,
"averageLogitDiff": round(avg_logit_diff, 6),
"averageKLDivergence": round(avg_kl, 6),
"averageActivation": round(avg_activation, 6),
"tokensChanged": tokens_changed,
"totalPrompts": len(prompts),
"perPrompt": per_prompt
}
return jsonify({
"success": True,
"results": sweep_results,
"featuresTotal": len(feature_indices),
"promptsTotal": len(prompts),
"model": model_id
})
except Exception as e:
print(f"[SAE Service] Feature ablation sweep error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/patch', methods=['POST'])
def activation_patch():
"""Single-layer activation patching between clean and corrupted prompts."""
try:
data = request.json
model_id = data.get('model', 'gpt2-small')
clean_prompt = data.get('cleanPrompt', '')
corrupted_prompt = data.get('corruptedPrompt', '')
patch_layer = data.get('patchLayer', 6)
patch_component = data.get('patchComponent', 'residual')
patch_direction = data.get('patchDirection', 'noising')
if not clean_prompt or not corrupted_prompt:
return jsonify({"error": "Both clean and corrupted prompts are required"}), 400
with _model_lock:
if model_id in ACTIVATION_ONLY_MODELS:
model = _load_model_only(model_id)
else:
model, sae = get_model_and_sae(model_id)
clean_tokens = model.to_tokens(clean_prompt)
corrupted_tokens = model.to_tokens(corrupted_prompt)
n_layers = model.cfg.n_layers
if patch_layer < 0 or patch_layer >= n_layers:
return jsonify({"error": f"Layer {patch_layer} out of range (0-{n_layers-1})"}), 400
if patch_component == "attention":
hook_name = f"blocks.{patch_layer}.attn.hook_result"
elif patch_component == "mlp":
hook_name = f"blocks.{patch_layer}.hook_mlp_out"
else:
hook_name = f"blocks.{patch_layer}.hook_resid_post"
is_attn = patch_component == "attention"
with torch.no_grad():
_, clean_cache = model.run_with_cache(clean_tokens)
_, corrupted_cache = model.run_with_cache(corrupted_tokens)
clean_activations = safe_get_cache(clean_cache, hook_name)
corrupted_activations = safe_get_cache(corrupted_cache, hook_name)
if patch_direction == "noising":
run_tokens = clean_tokens
patch_in_activations = corrupted_activations
else:
run_tokens = corrupted_tokens
patch_in_activations = clean_activations
baseline_logits = model(run_tokens)
baseline_probs = torch.softmax(baseline_logits[0, -1, :], dim=-1)
baseline_top_tokens = torch.topk(baseline_probs, k=10)
def patch_hook(activations, hook):
new_activations = activations.clone()
last_pos = min(activations.shape[1], patch_in_activations.shape[1]) - 1
if is_attn and activations.dim() == 4 and patch_in_activations.dim() == 4:
new_activations[:, last_pos, :, :] = patch_in_activations[:, last_pos, :, :].clone()
else:
new_activations[:, last_pos, :] = patch_in_activations[:, last_pos, :].clone()
return new_activations
model.reset_hooks()
patched_logits = model.run_with_hooks(
run_tokens,
fwd_hooks=[(hook_name, patch_hook)],
reset_hooks_end=True
)
patched_probs = torch.softmax(patched_logits[0, -1, :], dim=-1)
patched_top_tokens = torch.topk(patched_probs, k=10)
del clean_cache, corrupted_cache
if torch.cuda.is_available():
torch.cuda.empty_cache()
with torch.no_grad():
baseline_gen = model.generate(run_tokens, max_new_tokens=20, temperature=0.7)
baseline_output = model.to_string(baseline_gen[0])
def gen_patch_hook(activations, hook):
min_len = min(activations.shape[1], patch_in_activations.shape[1])
if is_attn and activations.dim() == 4 and patch_in_activations.dim() == 4:
activations[:, :min_len, :, :] = patch_in_activations[:, :min_len, :, :]
else:
activations[:, :min_len, :] = patch_in_activations[:, :min_len, :]
return activations
model.add_hook(hook_name, gen_patch_hook)
try:
patched_gen = model.generate(run_tokens, max_new_tokens=20, temperature=0.7)
patched_output = model.to_string(patched_gen[0])
except Exception as gen_err:
print(f"[SAE Service] Patched generation failed (non-fatal): {gen_err}")
patched_output = "(generation failed)"
finally:
model.reset_hooks()
top_k = 10
top_baseline_idx = torch.topk(baseline_probs, k=top_k).indices
top_patch_idx = torch.topk(patched_probs, k=top_k).indices
combined_idx = torch.unique(torch.cat([top_baseline_idx, top_patch_idx]))
logit_diff = (baseline_probs[combined_idx] - patched_probs[combined_idx]).abs().sum().item()
token_changes = []
for i in range(min(10, len(baseline_top_tokens.indices))):
token_idx = baseline_top_tokens.indices[i].item()
token_str = model.to_single_str_token(token_idx)
baseline_prob = baseline_probs[token_idx].item()
patch_prob = patched_probs[token_idx].item()
token_changes.append({
"token": token_str.replace('\u0120', ' ').replace('\u010a', '\n'),
"baselineProb": round(baseline_prob, 4),
"patchedProb": round(patch_prob, 4),
"change": round(patch_prob - baseline_prob, 4)
})
return jsonify({
"success": True,
"baselineOutput": baseline_output,
"patchedOutput": patched_output,
"logitDifference": round(logit_diff, 4),
"tokenChanges": token_changes,
"patchLayer": patch_layer,
"patchComponent": patch_component,
"patchDirection": patch_direction
})
except Exception as e:
if _model is not None:
_model.reset_hooks()
print(f"[SAE Service] Patch error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/sweep', methods=['POST'])
def layer_sweep():
"""
Sweep across all layers to find which layer has strongest patching effect.
Request body:
{
"model": "gpt2-small",
"cleanPrompt": "I am an AI assistant",
"corruptedPrompt": "A calculator is a device",
"patchComponent": "residual",
"patchDirection": "noising",
"numLayers": 12
}
"""
try:
data = request.json
model_id = data.get('model', 'gpt2-small')
clean_prompt = data.get('cleanPrompt', '')
corrupted_prompt = data.get('corruptedPrompt', '')
patch_component = data.get('patchComponent', 'residual')
patch_direction = data.get('patchDirection', 'noising')
num_layers = data.get('numLayers', 12)
skip_generate = data.get('skipGenerate', False)
if not clean_prompt or not corrupted_prompt:
return jsonify({"error": "Both clean and corrupted prompts are required"}), 400
with _model_lock:
if model_id in ACTIVATION_ONLY_MODELS:
model = _load_model_only(model_id)
else:
model, sae = get_model_and_sae(model_id)
actual_layers = min(num_layers, model.cfg.n_layers)
clean_tokens = model.to_tokens(clean_prompt)
corrupted_tokens = model.to_tokens(corrupted_prompt)
primary_hook_names = []
for layer in range(actual_layers):
if patch_component == "attention":
primary_hook_names.append(f"blocks.{layer}.attn.hook_result")
elif patch_component == "mlp":
primary_hook_names.append(f"blocks.{layer}.hook_mlp_out")
else:
primary_hook_names.append(f"blocks.{layer}.hook_resid_post")
results = []
with torch.no_grad():
clean_logits, clean_cache = model.run_with_cache(clean_tokens)
corrupted_logits, corrupted_cache = model.run_with_cache(corrupted_tokens)
if patch_direction == "noising":
run_tokens = clean_tokens
patch_cache = corrupted_cache
baseline_probs = torch.softmax(clean_logits[0, -1, :], dim=-1)
else:
run_tokens = corrupted_tokens
patch_cache = clean_cache
baseline_probs = torch.softmax(corrupted_logits[0, -1, :], dim=-1)
baseline_output = None
if not skip_generate:
baseline_gen = model.generate(run_tokens, max_new_tokens=15, temperature=0.7)
baseline_output = model.to_string(baseline_gen[0])
for layer in range(actual_layers):
hook_name = primary_hook_names[layer]
try:
patch_in_acts = safe_get_cache(patch_cache, hook_name)
except KeyError as ke:
print(f"[SAE Service] Skipping layer {layer}: {ke}")
continue
resolved_hook = hook_name if hook_name in patch_cache else None
if resolved_hook is None:
for candidate in [hook_name, hook_name.replace("hook_result", "hook_z"), hook_name.replace("hook_result", "hook_attn_out")]:
if candidate in patch_cache:
resolved_hook = candidate
break
if resolved_hook is None:
resolved_hook = hook_name
def make_patch_hook(patch_activations, layer_num, is_attn=False):
def patch_hook(activations, hook):
new_activations = activations.clone()
last_pos = min(activations.shape[1], patch_activations.shape[1]) - 1
if is_attn and activations.dim() == 4 and patch_activations.dim() == 4:
new_activations[:, last_pos, :, :] = patch_activations[:, last_pos, :, :].clone()
else:
new_activations[:, last_pos, :] = patch_activations[:, last_pos, :].clone()
return new_activations
return patch_hook
is_attn = patch_component == "attention"
model.reset_hooks()
try:
patched_logits = model.run_with_hooks(
run_tokens,
fwd_hooks=[(resolved_hook, make_patch_hook(patch_in_acts.clone(), layer, is_attn))],
reset_hooks_end=True
)
except Exception as hook_err:
print(f"[SAE Service] Layer {layer} patching failed: {hook_err}")
model.reset_hooks()
continue
patched_probs = torch.softmax(patched_logits[0, -1, :], dim=-1)
top_k = 10
top_baseline_indices = torch.topk(baseline_probs, k=top_k).indices
top_patch_indices = torch.topk(patched_probs, k=top_k).indices
combined_indices = torch.unique(torch.cat([top_baseline_indices, top_patch_indices]))
logit_diff = (baseline_probs[combined_indices] - patched_probs[combined_indices]).abs().sum().item()
token_details = []
top_indices = torch.topk(baseline_probs, k=5).indices
for idx in top_indices:
token_str = model.to_single_str_token(idx.item())
baseline_p = baseline_probs[idx].item()
patch_p = patched_probs[idx].item()
token_details.append({
"token": token_str.replace('\u0120', ' ').replace('\u010a', '\n'),
"position": 0,
"baselineProb": round(baseline_p, 4),
"patchedProb": round(patch_p, 4),
"change": round(patch_p - baseline_p, 4)
})
patched_output = None
if not skip_generate:
try:
model.add_hook(resolved_hook, make_patch_hook(patch_in_acts, layer, is_attn))
patched_gen = model.generate(run_tokens, max_new_tokens=15, temperature=0.7)
patched_output = model.to_string(patched_gen[0])
except Exception as gen_err:
print(f"[SAE Service] Layer {layer} generate failed (non-fatal): {gen_err}")
patched_output = "(generation failed)"
finally:
model.reset_hooks()
else:
model.reset_hooks()
results.append({
"layer": layer,
"logitDifference": round(logit_diff, 4),
"component": patch_component,
"tokenDetails": token_details,
"baselineOutput": baseline_output,
"patchedOutput": patched_output
})
del clean_cache, corrupted_cache, patch_cache
return jsonify({
"success": True,
"results": results,
"patchDirection": patch_direction,
"model": model_id
})
except Exception as e:
if _model is not None:
_model.reset_hooks()
print(f"[SAE Service] Sweep error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/feature-layer-activations', methods=['POST'])
def feature_layer_activations():
"""
Get per-feature activation projections across all layers.
Projects residual stream activations at each layer onto each feature's
SAE decoder direction. This shows where in the network each feature's
representation emerges and strengthens.
Request body:
{
"model": "gpt2-small",
"prompt": "What are your main limitations as an AI model?",
"featureIndices": [6471, 7393, 10543],
"promptGroup": "A" // optional, for labeling
}
"""
try:
data = request.json
model_id = data.get('model', 'gpt2-small')
prompt = data.get('prompt', '')
feature_indices = data.get('featureIndices', [])
if not prompt:
return jsonify({"error": "Prompt is required"}), 400
if not feature_indices:
return jsonify({"error": "Feature indices are required"}), 400
model, sae = get_model_and_sae(model_id)
num_layers = model.cfg.n_layers
tokens = model.to_tokens(prompt)
with torch.no_grad():
_, cache = model.run_with_cache(tokens)
decoder_weights = sae.W_dec.detach()
results = {}
for feat_idx in feature_indices:
if feat_idx >= decoder_weights.shape[0]:
continue
feat_direction = decoder_weights[feat_idx]
feat_direction_norm = feat_direction / (feat_direction.norm() + 1e-8)
layer_activations = {}
for l in range(num_layers):
hook_name = f"blocks.{l}.hook_resid_post"
if hook_name not in cache:
hook_name = f"blocks.{l}.hook_resid_pre"
if hook_name not in cache:
layer_activations[l] = 0.0
continue
resid = safe_get_cache(cache, hook_name)
last_token_resid = resid[0, -1, :]
projection = torch.dot(last_token_resid, feat_direction_norm).item()
layer_activations[l] = round(max(0, projection), 4)
results[str(feat_idx)] = layer_activations
return jsonify({
"success": True,
"results": results,
"numLayers": num_layers,
"model": model_id
})
except Exception as e:
print(f"[SAE Service] Feature layer activations error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/feature-layer-activations-batch', methods=['POST'])
def feature_layer_activations_batch():
"""
Batch version: average feature projections across multiple prompts.
This gives more robust per-feature, per-layer activation profiles.
Request body:
{
"model": "gpt2-small",
"prompts": ["prompt1", "prompt2", ...],
"featureIndices": [6471, 7393, 10543]
}
"""
try:
data = request.json
model_id = data.get('model', 'gpt2-small')
prompts = data.get('prompts', [])
feature_indices = data.get('featureIndices', [])
if not prompts:
return jsonify({"error": "Prompts are required"}), 400
if not feature_indices:
return jsonify({"error": "Feature indices are required"}), 400
model, sae = get_model_and_sae(model_id)
num_layers = model.cfg.n_layers
decoder_weights = sae.W_dec.detach()
accumulated = {str(fi): {l: 0.0 for l in range(num_layers)} for fi in feature_indices if fi < decoder_weights.shape[0]}
valid_features = [fi for fi in feature_indices if fi < decoder_weights.shape[0]]
feat_directions = {}
for fi in valid_features:
d = decoder_weights[fi]
feat_directions[fi] = d / (d.norm() + 1e-8)
with torch.no_grad():
for prompt in prompts:
tokens = model.to_tokens(prompt)
_, cache = model.run_with_cache(tokens)
for fi in valid_features:
for l in range(num_layers):
hook_name = f"blocks.{l}.hook_resid_post"
if hook_name not in cache:
hook_name = f"blocks.{l}.hook_resid_pre"
if hook_name not in cache:
continue
resid = safe_get_cache(cache, hook_name)
last_token_resid = resid[0, -1, :]
projection = torch.dot(last_token_resid, feat_directions[fi]).item()
accumulated[str(fi)][l] += max(0, projection)
num_prompts = len(prompts)
results = {}
for fi_str, layers in accumulated.items():
results[fi_str] = {l: round(v / num_prompts, 4) for l, v in layers.items()}
return jsonify({
"success": True,
"results": results,
"numLayers": num_layers,
"numPrompts": num_prompts,
"model": model_id
})
except Exception as e:
print(f"[SAE Service] Batch feature layer activations error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
# ============================================
# Chain of Thought Monitoring Endpoints (Fidelity)
# ============================================
def parse_cot_steps(text: str) -> list:
"""Parse numbered reasoning steps from text"""
import re
steps = []
# Match patterns like "1.", "Step 1:", "1)", etc.
pattern = r'(?:^|\n)\s*(?:Step\s*)?(\d+)[.:)\]]\s*(.+?)(?=(?:\n\s*(?:Step\s*)?\d+[.:)\]]|\Z))'
matches = re.findall(pattern, text, re.DOTALL | re.IGNORECASE)
for num, content in matches:
steps.append({
"stepNumber": int(num),
"content": content.strip()
})
# If no numbered steps found, treat each sentence as a step
if not steps:
sentences = [s.strip() for s in text.split('.') if s.strip()]
for i, s in enumerate(sentences[:10]): # Limit to 10
steps.append({"stepNumber": i + 1, "content": s})
return steps
@app.route('/cot/analyze', methods=['POST'])
def cot_analyze():
"""
Analyze chain of thought reasoning for faithfulness.
Request body:
{
"prompt": "Solve this step by step: What is 15 + 27?",
"model": "gpt2-small",
"layers": [4, 6, 8, 10] // Which layers to analyze
}
Response:
{
"generatedText": "Step 1: Add 15 and 27...",
"steps": [
{
"stepNumber": 1,
"content": "Add 15 and 27",
"tokenSpan": [5, 10],
"faithfulness": {
"activationAlignment": 0.85,
"causalImpact": 0.72,
"score": 0.78,
"verdict": "faithful"
},
"layerActivations": {...}
}
],
"overallFaithfulness": 0.75
}
"""
try:
data = request.json
prompt = data.get('prompt', '')
model_id = data.get('model', 'gpt2-small')
layers = data.get('layers', [4, 6, 8, 10])
max_tokens = min(data.get('maxTokens', 40), 50)
if not prompt:
return jsonify({"error": "No prompt provided"}), 400
model, sae = get_model_and_sae(model_id)
n_layers = model.cfg.n_layers
layers = [l for l in layers if l < n_layers]
if not layers:
layers = [n_layers // 3, n_layers // 2, n_layers * 2 // 3, n_layers - 1]
layers = [l for l in layers if l < n_layers]
hook_names = [f"blocks.{l}.hook_resid_post" for l in layers]
names_filter = lambda name: name in hook_names
tokens = model.to_tokens(prompt)
prompt_len = tokens.shape[1]
with torch.no_grad():
output = model.generate(tokens, max_new_tokens=max_tokens, temperature=0.7)
generated_text = model.to_string(output[0])
_, cache = model.run_with_cache(output, names_filter=names_filter)
layer_activations = {}
for layer in layers:
hook_name = f"blocks.{layer}.hook_resid_post"
acts = safe_get_cache(cache, hook_name)[0, prompt_len:, :]
layer_activations[layer] = {
"mean": acts.mean().item(),
"norm": acts.norm(dim=-1).mean().item(),
"shape": list(acts.shape)
}
final_emb = cache[f"blocks.{layers[-1]}.hook_resid_post"][0, -1, :]
response_text = generated_text[len(model.to_string(tokens[0])):]
steps = parse_cot_steps(response_text)
analyzed_steps = []
analysis_layer = layers[-1]
analysis_hook = f"blocks.{analysis_layer}.hook_resid_post"
analysis_filter = lambda name: name == analysis_hook
with torch.no_grad():
baseline_logits = model(output)
baseline_probs = torch.softmax(baseline_logits[0, -1, :], dim=-1)
full_text = model.to_string(output[0])
for step in steps:
step_text = step["content"]
with torch.no_grad():
step_tokens = model.to_tokens(step_text)
_, step_cache = model.run_with_cache(step_tokens, names_filter=analysis_filter)
step_emb = step_cache[analysis_hook][0, -1, :]
cos_sim = torch.nn.functional.cosine_similarity(
step_emb.unsqueeze(0), final_emb.unsqueeze(0)
).item()
causal_impact = 0.0
try:
step_start = full_text.find(step_text)
if step_start >= 0:
ablated_text = full_text[:step_start] + full_text[step_start + len(step_text):]
ablated_tokens = model.to_tokens(ablated_text)
with torch.no_grad():
ablated_logits = model(ablated_tokens)
ablated_probs = torch.softmax(ablated_logits[0, -1, :], dim=-1)
top_k = 10
top_baseline = torch.topk(baseline_probs, k=top_k).indices
top_ablated = torch.topk(ablated_probs, k=top_k).indices
combined = torch.unique(torch.cat([top_baseline, top_ablated]))
causal_impact = (baseline_probs[combined] - ablated_probs[combined]).abs().sum().item()
except Exception as ablation_err:
print(f"[CoT Service] Ablation failed for step {step['stepNumber']}: {ablation_err}")
causal_impact = 0.0
faithfulness_score = (abs(cos_sim) * 0.4 + min(causal_impact, 1.0) * 0.6)
verdict = "faithful" if faithfulness_score > 0.5 else "suspicious" if faithfulness_score > 0.25 else "unfaithful"
analyzed_steps.append({
"stepNumber": step["stepNumber"],
"content": step["content"],
"faithfulness": {
"activationAlignment": round(abs(cos_sim), 3),
"causalImpact": round(causal_impact, 4),
"score": round(faithfulness_score, 3),
"verdict": verdict
}
})
del cache
if analyzed_steps:
overall = sum(s["faithfulness"]["score"] for s in analyzed_steps) / len(analyzed_steps)
else:
overall = 0.0
return jsonify({
"success": True,
"prompt": prompt,
"generatedText": response_text,
"steps": analyzed_steps,
"layerActivations": layer_activations,
"overallFaithfulness": round(overall, 3),
"model": model_id
})
except Exception as e:
print(f"[CoT Service] Analyze error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/cot/ablate', methods=['POST'])
def cot_ablate():
"""
Ablate a specific reasoning step to measure its causal impact.
Request body:
{
"prompt": "Solve step by step: 15 + 27",
"generatedText": "Step 1: ... Step 2: ...",
"stepToAblate": 1,
"model": "gpt2-small"
}
Response:
{
"baselineOutput": "42",
"ablatedOutput": "38",
"logitDifference": 0.35,
"stepCausalImpact": 0.72
}
"""
try:
data = request.json
prompt = data.get('prompt', '')
step_num = data.get('stepToAblate', 1)
model_id = data.get('model', 'gpt2-small')
layer = data.get('layer', 6)
if not prompt:
return jsonify({"error": "No prompt provided"}), 400
model, sae = get_model_and_sae(model_id, layer=layer)
# Generate baseline
tokens = model.to_tokens(prompt)
with torch.no_grad():
# Baseline generation
baseline_output = model.generate(tokens, max_new_tokens=50, temperature=0.7)
baseline_text = model.to_string(baseline_output[0])
# Get baseline activations
_, cache = model.run_with_cache(baseline_output)
baseline_acts = cache[f"blocks.{layer}.hook_resid_post"].clone()
# Create ablation hook that zeros out middle portion of activations
# (representing the "step" being ablated)
prompt_len = tokens.shape[1]
total_len = baseline_output.shape[1]
step_start = prompt_len + (step_num - 1) * 5 # Approximate
step_end = min(step_start + 10, total_len)
def ablation_hook(activations, hook):
new_acts = activations.clone()
if step_start < new_acts.shape[1]:
end = min(step_end, new_acts.shape[1])
new_acts[:, step_start:end, :] = 0.0
return new_acts
# Generate with ablation
model.reset_hooks()
model.add_hook(f"blocks.{layer}.hook_resid_post", ablation_hook)
ablated_output = model.generate(tokens, max_new_tokens=50, temperature=0.7)
ablated_text = model.to_string(ablated_output[0])
model.reset_hooks()
# Compute logit difference
baseline_logits = model(baseline_output)
model.add_hook(f"blocks.{layer}.hook_resid_post", ablation_hook)
ablated_logits = model(tokens)
model.reset_hooks()
baseline_probs = torch.softmax(baseline_logits[0, -1, :], dim=-1)
ablated_probs = torch.softmax(ablated_logits[0, -1, :], dim=-1)
top_k = 10
top_idx = torch.topk(baseline_probs, k=top_k).indices
logit_diff = (baseline_probs[top_idx] - ablated_probs[top_idx]).abs().sum().item()
# Causal impact: how much did ablating change the output?
causal_impact = min(1.0, logit_diff / 2.0) # Normalize
return jsonify({
"success": True,
"baselineOutput": baseline_text[len(model.to_string(tokens[0])):],
"ablatedOutput": ablated_text[len(model.to_string(tokens[0])):],
"logitDifference": round(logit_diff, 4),
"stepCausalImpact": round(causal_impact, 3),
"stepAblated": step_num,
"layer": layer
})
except Exception as e:
print(f"[CoT Service] Ablate error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
# ============================================
# Attention Analysis Endpoints
# ============================================
@app.route('/attention/analyze', methods=['POST'])
def attention_analyze():
"""
Analyze attention patterns across all heads.
Request body:
{
"prompt": "The capital of France is Paris",
"model": "gpt2-small",
"topK": 10
}
"""
try:
data = request.json
prompt = data.get('prompt', '')
model_id = data.get('model', 'gpt2-small')
top_k = data.get('topK', 10)
if not prompt:
return jsonify({"error": "Prompt required"}), 400
from transformer_lens import HookedTransformer
global _model, _saes, _current_model_id
base_model = get_base_model_id(model_id)
with _model_swap_lock:
if _model is None or get_base_model_id(_current_model_id) != base_model:
if _model is not None:
_cleanup_model_vram()
print(f"[Attention] Loading model: {base_model}")
_model = _load_hooked_transformer(base_model)
_saes = {}
_current_model_id = model_id
tokens = _model.to_tokens(prompt)
token_strs = _model.to_str_tokens(prompt)
with torch.no_grad():
_, cache = _model.run_with_cache(tokens)
n_layers = _model.cfg.n_layers
n_heads = _model.cfg.n_heads
heads_data = []
top_heads = []
for layer in range(n_layers):
attn_pattern = cache[f"blocks.{layer}.attn.hook_pattern"][0] # [n_heads, seq, seq]
for head in range(n_heads):
pattern = attn_pattern[head].detach().float().cpu().numpy().tolist()
# Calculate entropy of attention distribution
attn_probs = attn_pattern[head].detach().float().cpu().numpy()
entropy = -np.sum(attn_probs * np.log(attn_probs + 1e-10), axis=-1).mean()
# Max attention value
max_attn = float(attn_pattern[head].max())
# Classify head type based on pattern
pat_len = len(pattern)
head_type = "unknown"
if pat_len > 1:
# Check for previous token attention (diagonal -1)
prev_score = sum(pattern[i][i-1] if i > 0 else 0 for i in range(pat_len)) / pat_len
# Check for first token attention
first_score = sum(pattern[i][0] for i in range(pat_len)) / pat_len
if prev_score > 0.3:
head_type = "previous"
elif first_score > 0.4:
head_type = "global"
elif entropy < 1.0:
head_type = "local"
elif max_attn > 0.5 and pat_len > 5:
head_type = "induction"
heads_data.append({
"layer": layer,
"head": head,
"pattern": pattern,
"entropy": round(float(entropy), 3),
"maxAttention": round(max_attn, 3),
"headType": head_type
})
# Score for ranking - high max attention + low entropy = interesting
score = max_attn - entropy * 0.1
top_heads.append({
"layer": layer,
"head": head,
"score": round(score, 3),
"type": head_type
})
# Sort and take top K
top_heads.sort(key=lambda x: x["score"], reverse=True)
top_heads = top_heads[:top_k]
return jsonify(sanitize_for_json({
"success": True,
"tokens": token_strs,
"heads": heads_data,
"topHeads": top_heads
}))
except Exception as e:
print(f"[Attention] Analyze error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
# ============================================
# Logit Lens Endpoints
# ============================================
@app.route('/logit-lens/analyze', methods=['POST'])
def logit_lens_analyze():
"""
Project intermediate layer activations to vocabulary space.
Request body:
{
"prompt": "The Eiffel Tower is located in",
"model": "gpt2-small",
"topK": 5
}
"""
try:
data = request.json
prompt = data.get('prompt', '')
model_id = data.get('model', 'gpt2-small')
top_k = data.get('topK', 5)
if not prompt:
return jsonify({"error": "Prompt required"}), 400
if model_id in ACTIVATION_ONLY_MODELS:
model = _load_model_only(model_id)
else:
model, sae = get_model_and_sae(model_id)
tokens = model.to_tokens(prompt)
token_strs = model.to_str_tokens(prompt)
n_layers = model.cfg.n_layers
resid_filter = lambda name: name.endswith("hook_resid_post")
with torch.no_grad():
_, cache = model.run_with_cache(tokens, names_filter=resid_filter)
layers_data = []
W_U = model.W_U
b_U = model.b_U if hasattr(model, 'b_U') and model.b_U is not None else None
for layer in range(n_layers):
resid = cache[f"blocks.{layer}.hook_resid_post"][0, -1, :]
if hasattr(model, 'ln_final'):
resid_normed = model.ln_final(resid.unsqueeze(0)).squeeze(0)
else:
resid_normed = resid
logits = (resid_normed.float() @ W_U.float())
if b_U is not None:
logits = logits + b_U.float()
probs = torch.softmax(logits, dim=-1)
top_probs, top_indices = probs.topk(top_k)
top_tokens = []
for i in range(top_k):
tok_id = top_indices[i].item()
tok_str = model.tokenizer.decode([tok_id])
prob_val = top_probs[i].item()
logit_val = logits[tok_id].item()
top_tokens.append({
"token": tok_str,
"prob": round(prob_val, 4) if not np.isnan(prob_val) else 0.0,
"logit": round(logit_val, 3) if not np.isnan(logit_val) else 0.0
})
entropy = -torch.sum(probs * torch.log(probs + 1e-10)).item()
if np.isnan(entropy) or np.isinf(entropy):
entropy = 0.0
layers_data.append({
"layer": layer,
"topTokens": top_tokens,
"entropy": round(entropy, 3)
})
final_resid = cache[f"blocks.{n_layers - 1}.hook_resid_post"][0, -1, :]
if hasattr(model, 'ln_final'):
final_normed = model.ln_final(final_resid.unsqueeze(0)).squeeze(0)
else:
final_normed = final_resid
final_logits = final_normed.float() @ W_U.float()
if b_U is not None:
final_logits = final_logits + b_U.float()
final_probs = torch.softmax(final_logits, dim=-1)
final_token_idx = final_probs.argmax().item()
final_token = model.tokenizer.decode([final_token_idx])
top_probs, top_indices = final_probs.topk(top_k)
unembed_tokens = []
for i in range(top_k):
tok_id = top_indices[i].item()
tok_str = model.tokenizer.decode([tok_id])
prob_val = top_probs[i].item()
logit_val = final_logits[tok_id].item()
unembed_tokens.append({
"token": tok_str,
"prob": round(prob_val, 4) if not np.isnan(prob_val) else 0.0,
"logit": round(logit_val, 3) if not np.isnan(logit_val) else 0.0
})
del cache
return jsonify({
"success": True,
"prompt": prompt,
"tokens": token_strs,
"targetPosition": len(token_strs) - 1,
"layers": layers_data,
"unembedding": unembed_tokens,
"finalPrediction": final_token
})
except Exception as e:
print(f"[LogitLens] Analyze error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
_tuned_lens_probes = {}
TUNED_LENS_CALIBRATION_TEXTS = [
"The capital of France is Paris, which is known for the Eiffel Tower.",
"Machine learning models can be trained on large datasets to make predictions.",
"The quick brown fox jumps over the lazy dog near the river bank.",
"In 1969, Neil Armstrong became the first person to walk on the moon.",
"Water boils at 100 degrees Celsius under standard atmospheric pressure.",
"Shakespeare wrote many plays including Hamlet, Macbeth, and Romeo and Juliet.",
"The mitochondria is often called the powerhouse of the cell in biology.",
"Python is a popular programming language used for data science and web development.",
"The stock market experienced significant volatility during the financial crisis.",
"Photosynthesis converts sunlight into chemical energy that plants use to grow.",
"Albert Einstein developed the theory of relativity in the early twentieth century.",
"The Pacific Ocean is the largest and deepest ocean on Earth.",
"Democracy is a system of government where citizens vote to elect their leaders.",
"Antibiotics are medications used to treat bacterial infections in humans.",
"The Great Wall of China stretches thousands of miles across northern China.",
"Neural networks consist of layers of interconnected nodes that process information.",
]
def _train_tuned_lens_probes(model, base_model_id):
"""Train affine probes per layer: map raw resid_post to final-layer resid_post (pre-ln_final)."""
global _tuned_lens_probes
if base_model_id in _tuned_lens_probes:
return _tuned_lens_probes[base_model_id]
print(f"[TunedLens] Training probes for {base_model_id}...")
n_layers = model.cfg.n_layers
d_model = model.cfg.d_model
all_resids = {layer: [] for layer in range(n_layers)}
all_targets = []
with torch.no_grad():
for text in TUNED_LENS_CALIBRATION_TEXTS:
tokens = model.to_tokens(text)
_, cache = model.run_with_cache(tokens)
seq_len = tokens.shape[1]
for pos in range(max(0, seq_len - 8), seq_len):
final_resid = cache[f"blocks.{n_layers - 1}.hook_resid_post"][0, pos, :]
all_targets.append(final_resid)
for layer in range(n_layers):
resid = cache[f"blocks.{layer}.hook_resid_post"][0, pos, :]
all_resids[layer].append(resid)
targets = torch.stack(all_targets)
probes = {}
for layer in range(n_layers):
X = torch.stack(all_resids[layer])
Y = targets
X_bias = torch.cat([X, torch.ones(X.shape[0], 1)], dim=1)
solution = torch.linalg.lstsq(X_bias, Y).solution
W = solution[:d_model, :]
b = solution[d_model, :]
probes[layer] = {"W": W, "b": b}
with torch.no_grad():
pred = X @ W + b
mse = ((pred - Y) ** 2).mean().item()
print(f" Layer {layer}: MSE = {mse:.6f}")
_tuned_lens_probes[base_model_id] = probes
print(f"[TunedLens] Probes trained for {base_model_id}")
return probes
@app.route('/logit-lens/tuned-analyze', methods=['POST'])
def tuned_lens_analyze():
"""
Tuned Lens: uses learned per-layer affine probes for cleaner intermediate projections.
Request body:
{
"prompt": "The Eiffel Tower is located in",
"model": "gpt2-small",
"topK": 5
}
"""
try:
data = request.json
prompt = data.get('prompt', '')
model_id = data.get('model', 'gpt2-small')
top_k = data.get('topK', 5)
if not prompt:
return jsonify({"error": "Prompt required"}), 400
from transformer_lens import HookedTransformer
global _model, _saes, _current_model_id
base_model = get_base_model_id(model_id)
with _model_swap_lock:
if _model is None or _current_model_id is None or get_base_model_id(_current_model_id) != base_model:
if _model is not None:
_cleanup_model_vram()
print(f"[TunedLens] Loading model: {base_model}")
_model = _load_hooked_transformer(base_model)
_saes = {}
_current_model_id = model_id
probes = _train_tuned_lens_probes(_model, base_model)
tokens = _model.to_tokens(prompt)
token_strs = _model.to_str_tokens(prompt)
with torch.no_grad():
_, cache = _model.run_with_cache(tokens)
n_layers = _model.cfg.n_layers
W_U = _model.W_U
b_U = _model.b_U if hasattr(_model, 'b_U') and _model.b_U is not None else None
layers_data = []
for layer in range(n_layers):
resid = cache[f"blocks.{layer}.hook_resid_post"][0, -1, :]
probe = probes[layer]
tuned_resid = resid @ probe["W"] + probe["b"]
if hasattr(_model, 'ln_final'):
tuned_resid = _model.ln_final(tuned_resid.unsqueeze(0)).squeeze(0)
logits = tuned_resid @ W_U
if b_U is not None:
logits = logits + b_U
probs = torch.softmax(logits, dim=-1)
top_probs, top_indices = probs.topk(top_k)
top_tokens = []
for i in range(top_k):
tok_id = top_indices[i].item()
tok_str = _model.tokenizer.decode([tok_id])
top_tokens.append({
"token": tok_str,
"prob": round(top_probs[i].item(), 4),
"logit": round(logits[tok_id].item(), 3)
})
entropy = -torch.sum(probs * torch.log(probs + 1e-10)).item()
layers_data.append({
"layer": layer,
"topTokens": top_tokens,
"entropy": round(entropy, 3)
})
final_resid = cache[f"blocks.{n_layers - 1}.hook_resid_post"][0, -1, :]
if hasattr(_model, 'ln_final'):
final_normed = _model.ln_final(final_resid.unsqueeze(0)).squeeze(0)
else:
final_normed = final_resid
final_logits = final_normed @ W_U
if b_U is not None:
final_logits = final_logits + b_U
final_probs = torch.softmax(final_logits, dim=-1)
final_token_idx = final_probs.argmax().item()
final_token = _model.tokenizer.decode([final_token_idx])
top_probs, top_indices = final_probs.topk(top_k)
unembed_tokens = []
for i in range(top_k):
tok_id = top_indices[i].item()
tok_str = _model.tokenizer.decode([tok_id])
unembed_tokens.append({
"token": tok_str,
"prob": round(top_probs[i].item(), 4),
"logit": round(final_logits[tok_id].item(), 3)
})
return jsonify({
"success": True,
"prompt": prompt,
"tokens": token_strs,
"targetPosition": len(token_strs) - 1,
"layers": layers_data,
"unembedding": unembed_tokens,
"finalPrediction": final_token,
"mode": "tuned"
})
except Exception as e:
print(f"[TunedLens] Analyze error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
# ============================================
# Probing Classifier Endpoints
# ============================================
@app.route('/probe/train', methods=['POST'])
def probe_train():
"""
Train linear probes on activations at each layer.
Request body:
{
"task": "sentiment",
"examples": [{"text": "I love this", "label": "positive"}, ...],
"model": "gpt2-small",
"testSplit": 0.3
}
"""
try:
data = request.json
task = data.get('task', 'custom')
examples = data.get('examples', [])
model_id = data.get('model', 'gpt2-small')
test_split = data.get('testSplit', 0.3)
if not isinstance(test_split, (int, float)) or test_split <= 0 or test_split >= 1:
test_split = 0.3
if not examples or len(examples) < 20:
unique_lbls = set(ex.get("label", "") for ex in (examples or []))
test_split_val = request.json.get("test_split", 0.3)
est_test = max(1, round(len(examples or []) * test_split_val))
return jsonify({
"error": f"Insufficient sample size: {len(examples or [])} examples provided, minimum 20 required.",
"message": f"With {len(examples or [])} examples and {int(test_split_val*100)}% test split, only ~{est_test} test samples would be used — too few for reliable results. Provide at least 20 examples (10+ per label).",
"minimum": 20,
"provided": len(examples or []),
"labels": list(unique_lbls),
"estimatedTestSamples": est_test
}), 400
unique_labels = set(ex.get("label", "") for ex in examples)
if len(unique_labels) < 2:
return jsonify({"error": f"Need at least 2 different labels for classification, but all examples have the same label: '{next(iter(unique_labels))}'. Provide examples with contrasting labels (e.g. 'true'/'false', 'positive'/'negative')."}), 400
label_counts = {}
for lbl in unique_labels:
label_counts[lbl] = sum(1 for ex in examples if ex.get("label") == lbl)
if label_counts[lbl] < 2:
return jsonify({"error": f"Label '{lbl}' has only {label_counts[lbl]} example(s). Each label needs at least 2 examples for train/test splitting."}), 400
min_class_count = min(label_counts.values())
n_test_needed = max(len(unique_labels), 2)
total = len(examples)
test_count = int(total * test_split)
if test_count < n_test_needed:
test_split = min(0.5, n_test_needed / total)
test_count = int(total * test_split)
if min_class_count < 2 or test_count < 1 or (total - test_count) < 1:
return jsonify({"error": f"Not enough examples ({total}) for {len(unique_labels)} classes with test_split={test_split}. Provide more examples (at least {n_test_needed * 3} total recommended)."}), 400
k_folds = data.get('kFolds', None)
if k_folds is not None:
try:
k_folds = int(k_folds)
if k_folds < 2 or k_folds > 10:
k_folds = None
elif min_class_count < k_folds:
k_folds = min(min_class_count, 3)
if k_folds < 2:
k_folds = None
except (ValueError, TypeError):
k_folds = None
from transformer_lens import HookedTransformer
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split, StratifiedKFold
global _model, _saes, _current_model_id
base_model = get_base_model_id(model_id)
if _model is None or get_base_model_id(_current_model_id) != base_model:
if _model is not None:
_cleanup_model_vram()
print(f"[Probe] Loading model: {base_model}")
_model = _load_hooked_transformer(base_model)
_saes = {}
_current_model_id = model_id
# Get unique labels
labels = list(set(ex["label"] for ex in examples))
label_to_idx = {l: i for i, l in enumerate(labels)}
# Collect activations for each example at each layer
n_layers = _model.cfg.n_layers
layer_activations = {layer: [] for layer in range(n_layers)}
y_labels = []
texts = []
for ex in examples:
text = ex["text"]
label = ex["label"]
tokens = _model.to_tokens(text)
with torch.no_grad():
_, cache = _model.run_with_cache(tokens)
for layer in range(n_layers):
act = cache[f"blocks.{layer}.hook_resid_post"][0, -1, :].detach().float().cpu().numpy()
layer_activations[layer].append(act)
del cache
if torch.cuda.is_available():
torch.cuda.empty_cache()
y_labels.append(label_to_idx[label])
texts.append(text)
y_labels = np.array(y_labels)
indices = np.arange(len(examples))
layers_data = []
best_layer = 0
best_accuracy = 0
if k_folds and k_folds >= 2:
skf = StratifiedKFold(n_splits=k_folds, shuffle=True, random_state=42)
for layer in range(n_layers):
X = np.array(layer_activations[layer])
fold_accuracies = []
fold_details = []
all_predictions = []
for fold_idx, (train_idx, test_idx) in enumerate(skf.split(X, y_labels)):
X_train, X_test = X[train_idx], X[test_idx]
y_train, y_test = y_labels[train_idx], y_labels[test_idx]
clf = LogisticRegression(max_iter=1000, random_state=42)
clf.fit(X_train, y_train)
y_pred = clf.predict(X_test)
acc = float((y_pred == y_test).mean())
fold_accuracies.append(acc)
fold_details.append({"fold": fold_idx + 1, "accuracy": round(acc, 3), "nTrain": len(train_idx), "nTest": len(test_idx)})
for i, idx in enumerate(test_idx):
all_predictions.append({
"input": texts[idx],
"predicted": labels[y_pred[i]],
"actual": labels[y_test[i]],
"correct": bool(y_pred[i] == y_test[i]),
"fold": fold_idx + 1,
})
mean_acc = float(np.mean(fold_accuracies))
std_acc = float(np.std(fold_accuracies))
layers_data.append({
"layer": layer,
"accuracy": round(mean_acc, 3),
"accuracyStd": round(std_acc, 3),
"folds": fold_details,
"kFolds": k_folds,
"predictions": all_predictions,
})
if mean_acc > best_accuracy:
best_accuracy = mean_acc
best_layer = layer
return jsonify({
"success": True,
"task": task,
"layers": layers_data,
"bestLayer": best_layer,
"bestAccuracy": round(float(best_accuracy), 3),
"crossValidation": {"kFolds": k_folds, "method": "stratified_k_fold"},
})
else:
train_idx, test_idx = train_test_split(indices, test_size=test_split, random_state=42, stratify=y_labels)
for layer in range(n_layers):
X = np.array(layer_activations[layer])
X_train, X_test = X[train_idx], X[test_idx]
y_train, y_test = y_labels[train_idx], y_labels[test_idx]
clf = LogisticRegression(max_iter=1000, random_state=42)
clf.fit(X_train, y_train)
y_pred = clf.predict(X_test)
accuracy = (y_pred == y_test).mean()
predictions = []
for i, idx in enumerate(test_idx):
predictions.append({
"input": texts[idx],
"predicted": labels[y_pred[i]],
"actual": labels[y_test[i]],
"correct": bool(y_pred[i] == y_test[i])
})
layers_data.append({
"layer": layer,
"accuracy": round(float(accuracy), 3),
"predictions": predictions
})
if accuracy > best_accuracy:
best_accuracy = accuracy
best_layer = layer
return jsonify({
"success": True,
"task": task,
"layers": layers_data,
"bestLayer": best_layer,
"bestAccuracy": round(float(best_accuracy), 3)
})
except Exception as e:
print(f"[Probe] Train error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
# ============================================
# SAE Feature Comparison
# ============================================
@app.route('/sae/compare', methods=['POST'])
def sae_compare():
"""
Compare feature activations across two prompt groups.
Request body:
{
"groupA": ["prompt 1", "prompt 2", ...],
"groupB": ["prompt 3", "prompt 4", ...],
"model": "gpt2-small",
"layer": 6,
"topK": 20
}
Response:
{
"success": true,
"totalFeatures": 24576,
"topFeatures": [
{
"featureIndex": 1234,
"groupARate": 0.8,
"groupBRate": 0.2,
"differential": 0.6,
"groupAPrompts": [...],
"groupBPrompts": [...]
}
],
"dominantGroup": "A"
}
"""
try:
data = request.json
group_a = data.get('groupA', [])
group_b = data.get('groupB', [])
model_id = data.get('model', 'gpt2-small')
layer = data.get('layer', 6)
top_k = data.get('topK', 20)
robust_config = data.get('robustExtraction', None)
feature_ids = data.get('featureIds', None)
if not group_a or not group_b:
return jsonify({"error": "Both prompt groups required"}), 400
model, sae = get_model_and_sae(model_id, layer=layer)
n_features = sae.cfg.d_sae
hook_name = get_hook_name(model_id, layer=layer)
# Collect activations for each group
def get_group_activations(prompts):
all_activations = []
for prompt in prompts:
tokens = model.to_tokens(prompt)
with torch.no_grad():
_, cache = model.run_with_cache(tokens)
acts = safe_get_cache(cache, hook_name)[0, -1, :]
sae_acts = sae.encode(acts)
all_activations.append(sae_acts)
del cache
if torch.cuda.is_available():
torch.cuda.empty_cache()
return torch.stack(all_activations)
group_a_acts = get_group_activations(group_a) # [n_prompts_a, n_features]
group_b_acts = get_group_activations(group_b) # [n_prompts_b, n_features]
# Compute activation rates (fraction of prompts where feature activates above threshold)
threshold = 0.1
group_a_rates = (group_a_acts > threshold).float().mean(dim=0) # [n_features]
group_b_rates = (group_b_acts > threshold).float().mean(dim=0) # [n_features]
# Count features that activated in any prompt (either group)
combined_acts = torch.cat([group_a_acts, group_b_acts], dim=0) # [n_all_prompts, n_features]
activated_features = int((combined_acts.max(dim=0).values > threshold).sum().item())
# Differential score: A rate minus B rate
differential = group_a_rates - group_b_rates
# Get top features - either from specified feature set or all features
if feature_ids and len(feature_ids) > 0:
# Filter to only the specified features, then get top_k from those
valid_ids = [fid for fid in feature_ids if 0 <= fid < n_features]
if not valid_ids:
return jsonify({"error": "No valid feature IDs in specified range"}), 400
# Create tensor of differential values for specified features
subset_diffs = torch.tensor([differential[fid].item() for fid in valid_ids])
# Get top_k indices within the subset
k = min(top_k, len(valid_ids))
top_subset_indices = subset_diffs.abs().topk(k).indices.tolist()
# Map back to actual feature indices
top_indices = [valid_ids[i] for i in top_subset_indices]
scanned_features = len(valid_ids)
else:
# Scan all features
top_indices = differential.abs().topk(top_k).indices.tolist()
scanned_features = n_features
# Statistical significance helper functions
import numpy as np
def bootstrap_ci(a_binary, b_binary, n_bootstrap=500, alpha=0.05):
"""Compute bootstrap 95% CI for difference in activation rates"""
diffs = []
n_a, n_b = len(a_binary), len(b_binary)
for _ in range(n_bootstrap):
a_sample = np.random.choice(a_binary, size=n_a, replace=True)
b_sample = np.random.choice(b_binary, size=n_b, replace=True)
diffs.append(a_sample.mean() - b_sample.mean())
diffs = np.sort(diffs)
low_idx = int(alpha / 2 * n_bootstrap)
high_idx = int((1 - alpha / 2) * n_bootstrap)
return float(diffs[low_idx]), float(diffs[high_idx])
def permutation_pvalue(a_binary, b_binary, observed_diff, n_perm=500):
"""Compute permutation test p-value"""
pooled = np.concatenate([a_binary, b_binary])
n_a = len(a_binary)
extreme_count = 0
for _ in range(n_perm):
np.random.shuffle(pooled)
perm_a = pooled[:n_a]
perm_b = pooled[n_a:]
perm_diff = abs(perm_a.mean() - perm_b.mean())
if perm_diff >= abs(observed_diff):
extreme_count += 1
return extreme_count / n_perm
def compute_robustness(feature_idx, group_a_acts_tensor, group_b_acts_tensor,
threshold, n_bootstrap=100, holdout_ratio=0.2, min_stability=0.7):
"""Compute robustness metrics for a feature via bootstrap and cross-validation"""
n_a = group_a_acts_tensor.shape[0]
n_b = group_b_acts_tensor.shape[0]
a_acts = group_a_acts_tensor[:, feature_idx].numpy()
b_acts = group_b_acts_tensor[:, feature_idx].numpy()
# Bootstrap stability: how often does this feature remain in top differential set?
stability_count = 0
original_diff = abs((a_acts > threshold).mean() - (b_acts > threshold).mean())
for _ in range(n_bootstrap):
a_sample = np.random.choice(a_acts, size=n_a, replace=True)
b_sample = np.random.choice(b_acts, size=n_b, replace=True)
boot_diff = abs((a_sample > threshold).mean() - (b_sample > threshold).mean())
# Feature is "stable" if bootstrap diff is at least 50% of original
if boot_diff >= original_diff * 0.5:
stability_count += 1
stability = stability_count / n_bootstrap
# Cross-validation: holdout some prompts and test if feature still differentiates
holdout_a = max(1, int(n_a * holdout_ratio))
holdout_b = max(1, int(n_b * holdout_ratio))
cv_scores = []
for _ in range(min(50, n_bootstrap)):
# Random split
a_indices = np.random.permutation(n_a)
b_indices = np.random.permutation(n_b)
train_a = a_acts[a_indices[holdout_a:]]
test_a = a_acts[a_indices[:holdout_a]]
train_b = b_acts[b_indices[holdout_b:]]
test_b = b_acts[b_indices[:holdout_b]]
# Train differential
train_diff = (train_a > threshold).mean() - (train_b > threshold).mean()
# Test if direction holds
test_diff = (test_a > threshold).mean() - (test_b > threshold).mean()
# Score 1 if same sign, 0 otherwise
if train_diff * test_diff > 0 or (abs(train_diff) < 0.05 and abs(test_diff) < 0.05):
cv_scores.append(1)
else:
cv_scores.append(0)
cv_score = np.mean(cv_scores) if cv_scores else 0.5
bootstrap_consistent = stability >= min_stability and cv_score >= 0.6
return {
"stability": round(stability, 3),
"crossValidationScore": round(cv_score, 3),
"bootstrapConsistent": bootstrap_consistent
}
# Build response with statistical significance
top_features = []
for idx in top_indices:
# Get binary activation arrays for this feature
a_binary = (group_a_acts[:, idx] > threshold).float().numpy()
b_binary = (group_b_acts[:, idx] > threshold).float().numpy()
obs_diff = differential[idx].item()
# Compute stats only for meaningful differences with enough data
p_value = None
ci_low, ci_high = None, None
total_prompts = len(a_binary) + len(b_binary)
if abs(obs_diff) > 0.1 and total_prompts >= 6:
p_value = round(permutation_pvalue(a_binary, b_binary, obs_diff), 3)
ci_low, ci_high = bootstrap_ci(a_binary, b_binary)
ci_low, ci_high = round(ci_low, 3), round(ci_high, 3)
# Compute robustness metrics if enabled
robustness = None
if robust_config and robust_config.get('enabled'):
robustness = compute_robustness(
idx, group_a_acts, group_b_acts, threshold,
n_bootstrap=robust_config.get('bootstrapSamples', 100),
holdout_ratio=robust_config.get('holdoutRatio', 0.2),
min_stability=robust_config.get('minStability', 0.7)
)
feature_data = {
"featureIndex": idx,
"groupARate": round(group_a_rates[idx].item(), 3),
"groupBRate": round(group_b_rates[idx].item(), 3),
"differential": round(obs_diff, 3),
"pValue": p_value,
"confidenceInterval": {"low": ci_low, "high": ci_high} if ci_low is not None else None,
"robustness": robustness,
"groupAPrompts": [
{"prompt": p, "activation": round(group_a_acts[i, idx].item(), 3)}
for i, p in enumerate(group_a)
],
"groupBPrompts": [
{"prompt": p, "activation": round(group_b_acts[i, idx].item(), 3)}
for i, p in enumerate(group_b)
]
}
top_features.append(feature_data)
# Sort by absolute differential (highest first)
top_features.sort(key=lambda x: abs(x["differential"]), reverse=True)
# Determine dominant group
mean_diff = differential.mean().item()
if mean_diff > 0.05:
dominant = "A"
elif mean_diff < -0.05:
dominant = "B"
else:
dominant = "neutral"
return jsonify({
"success": True,
"totalFeatures": n_features,
"scannedFeatures": scanned_features,
"activatedFeatures": activated_features,
"topFeatures": top_features,
"dominantGroup": dominant,
"model": model_id,
"layer": layer
})
except Exception as e:
print(f"[Backlight] Compare error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
# ============================================
# Transcoder Circuit Discovery Endpoints (Circuits)
# ============================================
# Lazy loading for transcoders
_transcoder = None
_transcoder_model_id = None
def get_transcoder(model_id: str = "gpt2-small", layer: int = 6):
"""Load transcoder (MLP SAE) for circuit discovery"""
global _transcoder, _transcoder_model_id, _model, _current_model_id
cache_key = f"{model_id}_layer{layer}"
if _transcoder is not None and _transcoder_model_id == cache_key:
return _transcoder, _model
print(f"[Circuits] Loading transcoder for: {model_id} layer {layer}")
from transformer_lens import HookedTransformer
base_model = get_base_model_id(model_id)
with _model_swap_lock:
if _model is None or get_base_model_id(_current_model_id) != base_model:
if _model is not None:
_cleanup_model_vram()
_model = _load_hooked_transformer(base_model)
_current_model_id = model_id
try:
from sae_lens import SAE
if "gpt2" in model_id:
_transcoder, _, _ = SAE.from_pretrained(
release="gpt2-small-mlp-out-v5-32k",
sae_id=f"blocks.{layer}.hook_mlp_out",
device=_get_device()
)
else:
mlp_sae_id = f"layer_{layer}/width_16k/average_l0_{GEMMA_MLP_L0}"
print(f"[Circuits] Loading MLP SAE: gemma-scope-2b-pt-mlp / {mlp_sae_id}")
_transcoder, _, _ = SAE.from_pretrained(
release="gemma-scope-2b-pt-mlp",
sae_id=mlp_sae_id,
device=_get_device()
)
_transcoder_model_id = cache_key
print(f"[Circuits] Loaded MLP SAE as transcoder proxy for layer {layer}")
except Exception as e:
print(f"[Circuits] Could not load transcoder: {e}")
_transcoder = None
return _transcoder, _model
@app.route('/transcoder/discover', methods=['POST'])
def transcoder_discover():
"""
Discover feature circuits from MLP input to output.
Request body:
{
"prompt": "The capital of France is",
"model": "gemma-2-2b",
"layer": 12,
"topK": 32,
"threshold": 0.1
}
Response:
{
"nodes": [
{"id": "in_123", "type": "input", "feature": 123, "activation": 0.5},
{"id": "out_456", "type": "output", "feature": 456, "activation": 0.8}
],
"edges": [
{"source": "in_123", "target": "out_456", "weight": 0.3}
],
"tokens": ["The", " capital", ...],
"targetToken": "Paris"
}
"""
try:
data = request.json
prompt = data.get('prompt', '')
model_id = data.get('model', 'gemma-2-2b')
layer = data.get('layer', 12)
top_k = data.get('topK', 32)
threshold = data.get('threshold', 0.1)
if not prompt:
return jsonify({"error": "No prompt provided"}), 400
transcoder, model = get_transcoder(model_id, layer)
if transcoder is None:
return jsonify({"error": "Transcoder not available for this model"}), 400
tokens = model.to_tokens(prompt)
str_tokens = [t.replace('Ġ', ' ').replace('Ċ', '\n') for t in model.to_str_tokens(prompt)]
with torch.no_grad():
# Get MLP input and output
_, cache = model.run_with_cache(tokens)
mlp_in = cache[f"blocks.{layer}.hook_resid_pre"][0, -1, :] # Last token
mlp_out = cache[f"blocks.{layer}.hook_mlp_out"][0, -1, :]
# Encode through transcoder
input_features = transcoder.encode(mlp_in.unsqueeze(0).unsqueeze(0))
output_features = transcoder.encode(mlp_out.unsqueeze(0).unsqueeze(0))
input_acts = input_features[0, 0, :]
output_acts = output_features[0, 0, :]
# Get top-k input features
top_in_vals, top_in_idx = torch.topk(input_acts, k=min(top_k, len(input_acts)))
# Get top-k output features
top_out_vals, top_out_idx = torch.topk(output_acts, k=min(top_k, len(output_acts)))
# Build circuit graph
nodes = []
edges = []
# Input feature nodes
for i, (idx, val) in enumerate(zip(top_in_idx.tolist(), top_in_vals.tolist())):
if val > threshold:
nodes.append({
"id": f"in_{idx}",
"type": "input",
"feature": idx,
"activation": round(val, 4),
"layer": layer
})
# Output feature nodes
for i, (idx, val) in enumerate(zip(top_out_idx.tolist(), top_out_vals.tolist())):
if val > threshold:
nodes.append({
"id": f"out_{idx}",
"type": "output",
"feature": idx,
"activation": round(val, 4),
"layer": layer
})
# Compute edges based on decoder weight contributions
# Edge weight = input_activation * W_dec contribution to output
with torch.no_grad():
W_dec = transcoder.W_dec # [d_sae, d_model]
for in_node in [n for n in nodes if n["type"] == "input"]:
in_idx = in_node["feature"]
in_act = in_node["activation"]
for out_node in [n for n in nodes if n["type"] == "output"]:
out_idx = out_node["feature"]
# Approximate edge weight via decoder correlation
# This is simplified - real implementation would use transcoder decoder
weight = abs(in_act * output_acts[out_idx].item()) / 10.0
if weight > 0.01: # Only significant edges
edges.append({
"source": in_node["id"],
"target": out_node["id"],
"weight": round(weight, 4)
})
# Get model's next token prediction
with torch.no_grad():
logits = model(tokens)
probs = torch.softmax(logits[0, -1, :], dim=-1)
top_token_idx = probs.argmax().item()
target_token = model.to_single_str_token(top_token_idx).replace('Ġ', ' ')
return jsonify({
"success": True,
"nodes": nodes,
"edges": edges,
"tokens": str_tokens,
"targetToken": target_token,
"layer": layer,
"model": model_id
})
except Exception as e:
print(f"[Circuits] Discover error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
# Cache for multi-layer SAEs
_multilayer_saes = {}
@app.route('/transcoder/discover-multilayer', methods=['POST'])
def transcoder_discover_multilayer():
"""
Discover feature activations across multiple layers to trace circuit evolution.
Supports both GPT-2 Small and Gemma 2 2B.
Request body:
{
"prompt": "The capital of France is",
"model": "gemma-2-2b",
"layers": [6, 12, 18, 24],
"topK": 10,
"threshold": 0.1
}
"""
global _multilayer_saes, _model, _current_model_id
try:
data = request.json
prompt = data.get('prompt', '')
model_id = data.get('model', 'gemma-2-2b')
layers = data.get('layers', [6, 12, 18, 24])
top_k = data.get('topK', 10)
threshold = data.get('threshold', 0.1)
if not prompt:
return jsonify({"error": "No prompt provided"}), 400
is_gemma = "gemma" in model_id
base_model = get_base_model_id(model_id)
from transformer_lens import HookedTransformer
from sae_lens import SAE
with _model_swap_lock:
if _model is None or get_base_model_id(_current_model_id) != base_model:
if _model is not None:
_cleanup_model_vram()
print(f"[Circuits] Loading {base_model} for multi-layer analysis...")
_model = _load_hooked_transformer(base_model)
_current_model_id = model_id
for layer in layers:
cache_key = f"{base_model}_mlp_{layer}"
if cache_key not in _multilayer_saes:
try:
print(f"[Circuits] Loading MLP SAE for {base_model} layer {layer}...")
if is_gemma:
sae, _, _ = SAE.from_pretrained(
release="gemma-scope-2b-pt-mlp",
sae_id=f"layer_{layer}/width_16k/average_l0_{GEMMA_MLP_L0}",
device=_get_device()
)
else:
sae, _, _ = SAE.from_pretrained(
release="gpt2-small-mlp-out-v5-32k",
sae_id=f"blocks.{layer}.hook_mlp_out",
device=_get_device()
)
_multilayer_saes[cache_key] = sae
print(f"[Circuits] Loaded SAE for {base_model} layer {layer}")
except Exception as e:
print(f"[Circuits] Could not load SAE for layer {layer}: {e}")
_multilayer_saes[cache_key] = None
tokens = _model.to_tokens(prompt)
str_tokens = _model.to_str_tokens(prompt)
if not is_gemma:
str_tokens = [t.replace('Ġ', ' ').replace('Ċ', '\n') for t in str_tokens]
else:
str_tokens = list(str_tokens)
with torch.no_grad():
_, cache = _model.run_with_cache(tokens)
logits = _model(tokens)
probs = torch.softmax(logits[0, -1, :], dim=-1)
top_token_idx = probs.argmax().item()
target_token = _model.to_single_str_token(top_token_idx)
if not is_gemma:
target_token = target_token.replace('Ġ', ' ')
layer_results = []
all_feature_activations = {}
with torch.no_grad():
for layer in layers:
cache_key = f"{base_model}_mlp_{layer}"
sae = _multilayer_saes.get(cache_key)
if sae is None:
layer_results.append({
"layer": layer,
"features": [],
"error": "SAE not available for this layer"
})
continue
hook_name = f"blocks.{layer}.hook_mlp_out"
try:
mlp_out = cache[hook_name][0, -1, :]
except KeyError:
resid_hook = f"blocks.{layer}.hook_resid_post"
mlp_out = cache[resid_hook][0, -1, :]
features = sae.encode(mlp_out.unsqueeze(0).unsqueeze(0))
acts = features[0, 0, :]
top_vals, top_idx = torch.topk(acts, k=min(top_k, len(acts)))
layer_features = []
for idx, val in zip(top_idx.tolist(), top_vals.tolist()):
if val > threshold:
layer_features.append({
"feature": idx,
"activation": round(val, 4)
})
if idx not in all_feature_activations:
all_feature_activations[idx] = {}
all_feature_activations[idx][layer] = val
layer_results.append({
"layer": layer,
"features": layer_features
})
persistence = []
for feature_id, layer_acts in all_feature_activations.items():
if len(layer_acts) >= 2:
sorted_layers = sorted(layer_acts.keys())
peak_layer = max(layer_acts, key=layer_acts.get)
persistence.append({
"feature": feature_id,
"firstLayer": sorted_layers[0],
"lastLayer": sorted_layers[-1],
"peakLayer": peak_layer,
"peakActivation": round(layer_acts[peak_layer], 4),
"layerCount": len(layer_acts),
"trajectory": {str(l): round(a, 4) for l, a in sorted(layer_acts.items())}
})
persistence.sort(key=lambda x: (-x["layerCount"], -x["peakActivation"]))
return jsonify({
"success": True,
"layers": layer_results,
"persistence": persistence[:20],
"tokens": str_tokens,
"targetToken": target_token,
"model": model_id,
"requestedLayers": layers
})
except Exception as e:
print(f"[Circuits] Multi-layer discover error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/transcoder/ablate', methods=['POST'])
def transcoder_ablate():
"""
Ablate specific transcoder features to verify circuit causality.
Request body:
{
"prompt": "The capital of France is",
"model": "gemma-2-2b",
"layer": 12,
"featuresToAblate": [123, 456]
}
Response:
{
"baselineOutput": "Paris",
"ablatedOutput": "London",
"logitDifference": 0.45,
"featuresCausalImpact": [
{"feature": 123, "impact": 0.3},
{"feature": 456, "impact": 0.15}
]
}
"""
try:
data = request.json
prompt = data.get('prompt', '')
model_id = data.get('model', 'gemma-2-2b')
layer = data.get('layer', 12)
features_to_ablate = data.get('featuresToAblate', [])
if not prompt:
return jsonify({"error": "No prompt provided"}), 400
transcoder, model = get_transcoder(model_id)
if transcoder is None:
return jsonify({"error": "Transcoder not available"}), 400
tokens = model.to_tokens(prompt)
with torch.no_grad():
# Baseline
baseline_logits = model(tokens)
baseline_probs = torch.softmax(baseline_logits[0, -1, :], dim=-1)
baseline_token = model.to_single_str_token(baseline_probs.argmax().item())
baseline_gen = model.generate(tokens, max_new_tokens=5, temperature=0.1)
baseline_output = model.to_string(baseline_gen[0])[len(model.to_string(tokens[0])):]
# Create ablation hook
def ablation_hook(activations, hook):
# Encode, ablate features, decode
features = transcoder.encode(activations)
for feat_idx in features_to_ablate:
if feat_idx < features.shape[-1]:
features[:, :, feat_idx] = 0.0
return transcoder.decode(features)
# Ablated
model.reset_hooks()
model.add_hook(f"blocks.{layer}.hook_mlp_out", ablation_hook)
ablated_logits = model(tokens)
ablated_probs = torch.softmax(ablated_logits[0, -1, :], dim=-1)
ablated_token = model.to_single_str_token(ablated_probs.argmax().item())
ablated_gen = model.generate(tokens, max_new_tokens=5, temperature=0.1)
ablated_output = model.to_string(ablated_gen[0])[len(model.to_string(tokens[0])):]
model.reset_hooks()
# Compute logit difference
top_k = 10
top_idx = torch.topk(baseline_probs, k=top_k).indices
logit_diff = (baseline_probs[top_idx] - ablated_probs[top_idx]).abs().sum().item()
# Compute per-feature causal impact
feature_impacts = []
for feat in features_to_ablate:
# Simplified: distribute total impact equally
impact = logit_diff / len(features_to_ablate) if features_to_ablate else 0
feature_impacts.append({
"feature": feat,
"impact": round(impact, 4)
})
return jsonify({
"success": True,
"baselineOutput": baseline_output,
"ablatedOutput": ablated_output,
"baselineToken": baseline_token.replace('Ġ', ' '),
"ablatedToken": ablated_token.replace('Ġ', ' '),
"logitDifference": round(logit_diff, 4),
"featuresCausalImpact": feature_impacts,
"layer": layer
})
except Exception as e:
print(f"[Circuits] Ablate error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/domain-profile', methods=['POST'])
def domain_profile():
"""
Profile which SAE features are domain-specific vs shared vs irrelevant.
Runs domain and baseline texts through the model, encodes with SAE,
and classifies features by activation frequency.
Request body:
{
"model": "gpt2-small",
"domain_texts": ["text1", "text2", ...],
"baseline_texts": ["text1", "text2", ...],
"activation_threshold": 0.5
}
"""
try:
data = request.json
model_id = data.get('model', 'gpt2-small')
domain_texts = data.get('domain_texts', [])
baseline_texts = data.get('baseline_texts', [])
threshold = data.get('activation_threshold', 0.5)
if not domain_texts:
return jsonify({"error": "No domain texts provided"}), 400
if not baseline_texts:
return jsonify({"error": "No baseline texts provided"}), 400
model, sae = get_model_and_sae(model_id)
hook_name = get_hook_name(model_id)
sae_dim = sae.W_dec.shape[0] if hasattr(sae, 'W_dec') else 24576
domain_activation_counts = np.zeros(sae_dim)
baseline_activation_counts = np.zeros(sae_dim)
domain_activation_strengths = np.zeros(sae_dim)
baseline_activation_strengths = np.zeros(sae_dim)
print(f"[SAE Service] Domain profiling: {len(domain_texts)} domain texts, {len(baseline_texts)} baseline texts")
with torch.no_grad():
for i, text in enumerate(domain_texts):
print(f"[SAE Service] Profiling domain text {i+1}/{len(domain_texts)}")
tokens = model.to_tokens(text)
_, cache = model.run_with_cache(tokens)
activations = safe_get_cache(cache, hook_name)
feature_acts = sae.encode(activations)
max_acts = feature_acts[0].max(dim=0).values.detach().float().cpu().numpy()
domain_activation_counts += (max_acts > threshold).astype(float)
domain_activation_strengths += max_acts
for i, text in enumerate(baseline_texts):
print(f"[SAE Service] Profiling baseline text {i+1}/{len(baseline_texts)}")
tokens = model.to_tokens(text)
_, cache = model.run_with_cache(tokens)
activations = safe_get_cache(cache, hook_name)
feature_acts = sae.encode(activations)
max_acts = feature_acts[0].max(dim=0).values.detach().float().cpu().numpy()
baseline_activation_counts += (max_acts > threshold).astype(float)
baseline_activation_strengths += max_acts
domain_rates = domain_activation_counts / len(domain_texts)
baseline_rates = baseline_activation_counts / len(baseline_texts)
domain_avg_strength = domain_activation_strengths / len(domain_texts)
baseline_avg_strength = baseline_activation_strengths / len(baseline_texts)
domain_critical = []
shared = []
irrelevant = []
feature_details = {}
for idx in range(sae_dim):
dr = float(domain_rates[idx])
br = float(baseline_rates[idx])
ds = float(domain_avg_strength[idx])
bs = float(baseline_avg_strength[idx])
if dr >= 0.3 and (dr - br) >= 0.2:
classification = "domain_critical"
domain_critical.append(idx)
elif dr >= 0.2 or br >= 0.2:
classification = "shared"
shared.append(idx)
else:
classification = "irrelevant"
irrelevant.append(idx)
if dr > 0.01 or br > 0.01:
feature_details[str(idx)] = {
"domainRate": round(dr, 4),
"baselineRate": round(br, 4),
"domainStrength": round(ds, 4),
"baselineStrength": round(bs, 4),
"classification": classification
}
top_domain = sorted(domain_critical, key=lambda x: float(domain_rates[x]), reverse=True)[:50]
top_shared = sorted(shared, key=lambda x: max(float(domain_rates[x]), float(baseline_rates[x])), reverse=True)[:30]
return jsonify({
"success": True,
"totalFeatures": sae_dim,
"domainCritical": len(domain_critical),
"shared": len(shared),
"irrelevant": len(irrelevant),
"topDomainFeatures": top_domain,
"topSharedFeatures": top_shared,
"featureDetails": feature_details,
"model": model_id,
"domainTextsCount": len(domain_texts),
"baselineTextsCount": len(baseline_texts)
})
except Exception as e:
print(f"[SAE Service] Domain profile error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/pruned-inference', methods=['POST'])
def pruned_inference():
"""
Run inference with only retained features active, zeroing all others.
Compares full model output with pruned model output.
Request body:
{
"model": "gpt2-small",
"prompt": "The patient presented with",
"retained_features": [100, 200, 300, ...],
"max_new_tokens": 30
}
"""
try:
data = request.json
model_id = data.get('model', 'gpt2-small')
prompt = data.get('prompt', '')
retained_features = set(data.get('retained_features', []))
max_new_tokens = data.get('max_new_tokens', 30)
if not prompt:
return jsonify({"error": "No prompt provided"}), 400
if not retained_features:
return jsonify({"error": "No retained features provided"}), 400
model, sae = get_model_and_sae(model_id)
hook_name = get_hook_name(model_id)
sae_dim = sae.W_dec.shape[0] if hasattr(sae, 'W_dec') else 24576
tokens = model.to_tokens(prompt)
with torch.no_grad():
baseline_logits = model(tokens)
baseline_probs = torch.softmax(baseline_logits[0, -1, :], dim=-1)
baseline_top5 = torch.topk(baseline_probs, k=5)
baseline_tokens = [model.to_single_str_token(t.item()).replace('\u0120', ' ') for t in baseline_top5.indices]
baseline_probs_list = [round(p.item(), 4) for p in baseline_top5.values]
original_output = model.generate(
tokens, max_new_tokens=max_new_tokens, temperature=0.1
)
original_text = model.to_string(original_output[0])
original_continuation = original_text[len(model.to_string(tokens[0])):]
retained_set = retained_features
mask_tensor = torch.zeros(sae_dim, device=_get_device())
for f in retained_set:
if f < sae_dim:
mask_tensor[f] = 1.0
def pruning_hook(activations, hook):
orig_acts = sae.encode(activations)
orig_recon = sae.decode(orig_acts)
recon_error = activations - orig_recon
pruned_acts = orig_acts * mask_tensor
return sae.decode(pruned_acts) + recon_error
try:
with torch.no_grad():
model.reset_hooks()
pruned_logits = model.run_with_hooks(
tokens,
fwd_hooks=[(hook_name, pruning_hook)],
reset_hooks_end=True
)
pruned_probs = torch.softmax(pruned_logits[0, -1, :], dim=-1)
pruned_top5 = torch.topk(pruned_probs, k=5)
pruned_tokens = [model.to_single_str_token(t.item()).replace('\u0120', ' ') for t in pruned_top5.indices]
pruned_probs_list = [round(p.item(), 4) for p in pruned_top5.values]
eps = 1e-10
kl_div = (baseline_probs * torch.log((baseline_probs + eps) / (pruned_probs + eps))).sum().item()
model.reset_hooks()
model.add_hook(hook_name, pruning_hook)
pruned_output = model.generate(
tokens, max_new_tokens=max_new_tokens, temperature=0.1
)
pruned_text = model.to_string(pruned_output[0])
pruned_continuation = pruned_text[len(model.to_string(tokens[0])):]
finally:
model.reset_hooks()
fidelity = max(0, 1.0 - kl_div) * 100
return jsonify({
"success": True,
"original": {
"continuation": original_continuation,
"topTokens": baseline_tokens,
"topProbs": baseline_probs_list
},
"pruned": {
"continuation": pruned_continuation,
"topTokens": pruned_tokens,
"topProbs": pruned_probs_list
},
"klDivergence": round(kl_div, 6),
"fidelity": round(fidelity, 2),
"totalFeatures": sae_dim,
"retainedFeatures": len(retained_features),
"prunedFeatures": sae_dim - len(retained_features),
"reductionPercent": round((1 - len(retained_features) / sae_dim) * 100, 1),
"model": model_id
})
except Exception as e:
print(f"[SAE Service] Pruned inference error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/tokenize-compare', methods=['POST'])
def tokenize_compare():
"""
Compare how different tokenizers break down the same text.
Loads only tokenizers (not full models) for speed.
Request body:
{
"text": "Hello world こんにちは世界",
"tokenizers": ["gpt2", "cl100k_base", "llama3", "gemma"]
}
"""
try:
data = request.json
text = data.get('text', '')
requested = data.get('tokenizers', ['gpt2', 'cl100k_base'])
if not text:
return jsonify({"error": "No text provided"}), 400
def tiktoken_decode_tokens(enc, token_ids):
"""Decode tiktoken tokens, showing hex for incomplete UTF-8 byte fragments.
Keeps 1:1 correspondence with token_ids (no grouping)."""
str_tokens = []
byte_lengths = []
for t in token_ids:
raw = enc.decode_single_token_bytes(t)
byte_lengths.append(len(raw))
try:
str_tokens.append(raw.decode('utf-8'))
except UnicodeDecodeError:
str_tokens.append("<0x" + raw.hex().upper() + ">")
return str_tokens, byte_lengths
results = {}
for tok_id in requested:
try:
if tok_id == 'gpt2':
import tiktoken
enc = tiktoken.get_encoding("gpt2")
token_ids = enc.encode(text, allowed_special="all")
str_tokens, byte_lengths = tiktoken_decode_tokens(enc, token_ids)
results[tok_id] = {
"name": "GPT-2",
"vocabSize": enc.n_vocab,
"tokenCount": len(token_ids),
"tokens": str_tokens,
"tokenIds": token_ids,
"byteLengths": byte_lengths,
"algorithm": "BPE (Byte-Pair Encoding)",
"charsPerToken": round(len(text) / max(len(token_ids), 1), 2)
}
elif tok_id == 'cl100k_base':
import tiktoken
enc = tiktoken.get_encoding("cl100k_base")
token_ids = enc.encode(text, allowed_special="all")
str_tokens, byte_lengths = tiktoken_decode_tokens(enc, token_ids)
results[tok_id] = {
"name": "GPT-4 (cl100k)",
"vocabSize": enc.n_vocab,
"tokenCount": len(token_ids),
"tokens": str_tokens,
"tokenIds": token_ids,
"byteLengths": byte_lengths,
"algorithm": "BPE (Byte-Pair Encoding)",
"charsPerToken": round(len(text) / max(len(token_ids), 1), 2)
}
elif tok_id == 'o200k_base':
import tiktoken
enc = tiktoken.get_encoding("o200k_base")
token_ids = enc.encode(text, allowed_special="all")
str_tokens, byte_lengths = tiktoken_decode_tokens(enc, token_ids)
results[tok_id] = {
"name": "GPT-4o (o200k)",
"vocabSize": enc.n_vocab,
"tokenCount": len(token_ids),
"tokens": str_tokens,
"tokenIds": token_ids,
"byteLengths": byte_lengths,
"algorithm": "BPE (Byte-Pair Encoding)",
"charsPerToken": round(len(text) / max(len(token_ids), 1), 2)
}
elif tok_id == 'llama3':
from transformers import AutoTokenizer
tokenizer = AutoTokenizer.from_pretrained(
"philschmid/meta-llama-3-tokenizer",
use_fast=True,
token=os.environ.get("HF_TOKEN", None)
)
token_ids = tokenizer.encode(text, add_special_tokens=False)
str_tokens = [tokenizer.decode([t]) for t in token_ids]
byte_lengths = [len(s.encode('utf-8')) for s in str_tokens]
results[tok_id] = {
"name": "Llama 3",
"vocabSize": tokenizer.vocab_size,
"tokenCount": len(token_ids),
"tokens": str_tokens,
"tokenIds": token_ids,
"byteLengths": byte_lengths,
"algorithm": "BPE (SentencePiece)",
"charsPerToken": round(len(text) / max(len(token_ids), 1), 2)
}
elif tok_id == 'gemma':
from transformers import AutoTokenizer
tokenizer = AutoTokenizer.from_pretrained(
"pcuenq/gemma-tokenizer",
use_fast=True,
token=os.environ.get("HF_TOKEN", None)
)
token_ids = tokenizer.encode(text, add_special_tokens=False)
str_tokens = [tokenizer.decode([t]) for t in token_ids]
byte_lengths = [len(s.encode('utf-8')) for s in str_tokens]
results[tok_id] = {
"name": "Gemma",
"vocabSize": tokenizer.vocab_size,
"tokenCount": len(token_ids),
"tokens": str_tokens,
"tokenIds": token_ids,
"byteLengths": byte_lengths,
"algorithm": "SentencePiece (Unigram)",
"charsPerToken": round(len(text) / max(len(token_ids), 1), 2)
}
elif tok_id == 'rinna':
from transformers import AutoTokenizer
tokenizer = AutoTokenizer.from_pretrained(
"rinna/japanese-gpt2-medium",
use_fast=False,
token=os.environ.get("HF_TOKEN", None)
)
token_ids = tokenizer.encode(text, add_special_tokens=False)
str_tokens = [tokenizer.decode([t]) for t in token_ids]
byte_lengths = [len(s.encode('utf-8')) for s in str_tokens]
results[tok_id] = {
"name": "Rinna",
"vocabSize": tokenizer.vocab_size,
"tokenCount": len(token_ids),
"tokens": str_tokens,
"tokenIds": token_ids,
"byteLengths": byte_lengths,
"algorithm": "SentencePiece (Unigram)",
"charsPerToken": round(len(text) / max(len(token_ids), 1), 2)
}
elif tok_id == 'calm2':
from transformers import AutoTokenizer
tokenizer = AutoTokenizer.from_pretrained(
"cyberagent/calm2-7b",
use_fast=True,
token=os.environ.get("HF_TOKEN", None)
)
token_ids = tokenizer.encode(text, add_special_tokens=False)
str_tokens = [tokenizer.decode([t]) for t in token_ids]
byte_lengths = [len(s.encode('utf-8')) for s in str_tokens]
results[tok_id] = {
"name": "CyberAgent CALM2",
"vocabSize": tokenizer.vocab_size,
"tokenCount": len(token_ids),
"tokens": str_tokens,
"tokenIds": token_ids,
"byteLengths": byte_lengths,
"algorithm": "BPE (SentencePiece)",
"charsPerToken": round(len(text) / max(len(token_ids), 1), 2)
}
elif tok_id == 'stablelm_jp':
from transformers import AutoTokenizer
tokenizer = AutoTokenizer.from_pretrained(
"stabilityai/japanese-stablelm-base-gamma-7b",
use_fast=True,
token=os.environ.get("HF_TOKEN", None)
)
token_ids = tokenizer.encode(text, add_special_tokens=False)
str_tokens = [tokenizer.decode([t]) for t in token_ids]
byte_lengths = [len(s.encode('utf-8')) for s in str_tokens]
results[tok_id] = {
"name": "StableLM JP",
"vocabSize": tokenizer.vocab_size,
"tokenCount": len(token_ids),
"tokens": str_tokens,
"tokenIds": token_ids,
"byteLengths": byte_lengths,
"algorithm": "SentencePiece (BPE+Unigram)",
"charsPerToken": round(len(text) / max(len(token_ids), 1), 2)
}
else:
results[tok_id] = {"error": f"Unknown tokenizer: {tok_id}"}
except Exception as tok_err:
print(f"[SAE Service] Tokenizer {tok_id} error: {str(tok_err)}")
results[tok_id] = {"error": str(tok_err)}
return jsonify({
"success": True,
"text": text,
"textLength": len(text),
"textBytes": len(text.encode('utf-8')),
"results": results
})
except Exception as e:
print(f"[SAE Service] Tokenize compare error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
def extract_all_layer_activations(model, texts, position="last"):
"""
Shared utility: extract residual stream activations at every layer for a list of texts.
Returns dict with per-text, per-layer activations as numpy arrays.
position: "last" = last token, "mean" = mean across all tokens
"""
n_layers = model.cfg.n_layers
resid_hooks = [f"blocks.{l}.hook_resid_post" for l in range(n_layers)]
all_activations = []
all_token_strs = []
for text in texts:
tokens = model.to_tokens(text)
token_strs = model.to_str_tokens(text)
with torch.no_grad():
logits, cache = model.run_with_cache(tokens, names_filter=resid_hooks)
layer_acts = {}
for layer in range(n_layers):
hook_key = f"blocks.{layer}.hook_resid_post"
if hook_key not in cache:
hook_key = f"blocks.{layer}.hook_resid_pre"
if hook_key not in cache:
available = [k for k in cache.keys() if f"blocks.{layer}" in k and "resid" in k]
if available:
hook_key = available[0]
else:
print(f"[extract_all_layer_activations] No resid hook found for layer {layer}, available keys: {[k for k in cache.keys() if f'blocks.{layer}' in k]}")
continue
resid = cache[hook_key][0]
if position == "last":
act = resid[-1, :].float().cpu().numpy()
elif position == "mean":
act = resid.mean(dim=0).float().cpu().numpy()
else:
act = resid[-1, :].float().cpu().numpy()
layer_acts[layer] = act
all_activations.append(layer_acts)
all_token_strs.append(token_strs)
return all_activations, all_token_strs
def extract_final_logits(model, text):
"""Get final-layer logits and top-k predictions for a text."""
tokens = model.to_tokens(text)
with torch.no_grad():
logits = model(tokens)
last_logits = logits[0, -1, :].float()
probs = torch.softmax(last_logits, dim=-1)
entropy = -torch.sum(probs * torch.log(probs + 1e-10)).item()
top_probs, top_indices = probs.topk(10)
top_tokens = []
for i in range(10):
tok_id = top_indices[i].item()
tok_str = model.tokenizer.decode([tok_id])
top_tokens.append({
"token": tok_str,
"prob": round(top_probs[i].item(), 4),
"id": tok_id
})
return {
"topTokens": top_tokens,
"entropy": round(entropy, 3),
"confidence": round(top_probs[0].item(), 4)
}
@app.route('/latent-language/analyze', methods=['POST'])
def latent_language_analyze():
"""
Detect at which layers a model recognizes input language.
Trains per-layer linear probes on Japanese vs English activations.
"""
try:
data = request.json
japanese_texts = data.get('japaneseTexts', [])
english_texts = data.get('englishTexts', [])
model_id = data.get('model', 'gpt2-small')
translation_pairs = data.get('translationPairs', [])
if len(japanese_texts) < 2 or len(english_texts) < 2:
return jsonify({"error": "Need at least 2 texts in each language"}), 400
from transformer_lens import HookedTransformer
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import cross_val_score
global _model, _saes, _current_model_id
base_model = get_base_model_id(model_id)
with _model_swap_lock:
if _model is None or get_base_model_id(_current_model_id) != base_model:
if _model is not None:
_cleanup_model_vram()
print(f"[LatentLang] Loading model: {base_model}")
_model = _load_hooked_transformer(base_model)
_saes = {}
_current_model_id = model_id
all_texts = english_texts + japanese_texts
labels = [0] * len(english_texts) + [1] * len(japanese_texts)
print(f"[LatentLang] Extracting activations for {len(all_texts)} texts...")
activations, _ = extract_all_layer_activations(_model, all_texts)
n_layers = _model.cfg.n_layers
y = np.array(labels)
layers_data = []
for layer in range(n_layers):
X = np.array([act[layer] for act in activations])
min_class_size = min(len(english_texts), len(japanese_texts))
if min_class_size >= 3:
cv_folds = min(5, min_class_size)
clf = LogisticRegression(max_iter=1000, random_state=42)
scores = cross_val_score(clf, X, y, cv=cv_folds, scoring='accuracy')
accuracy = float(scores.mean())
else:
clf = LogisticRegression(max_iter=1000, random_state=42)
clf.fit(X, y)
accuracy = float(clf.score(X, y))
clf_full = LogisticRegression(max_iter=1000, random_state=42)
clf_full.fit(X, y)
raw_weights = clf_full.coef_[0]
weight_magnitudes = np.abs(raw_weights)
top_neuron_indices = weight_magnitudes.argsort()[-10:][::-1].tolist()
top_neuron_weights = [round(float(raw_weights[i]), 4) for i in top_neuron_indices]
layers_data.append({
"layer": layer,
"accuracy": round(accuracy, 3),
"topNeurons": top_neuron_indices,
"topNeuronWeights": top_neuron_weights
})
similarity_data = []
if translation_pairs:
for pair in translation_pairs:
en_text = pair.get("english", "")
ja_text = pair.get("japanese", "")
if not en_text or not ja_text:
continue
pair_acts, _ = extract_all_layer_activations(_model, [en_text, ja_text])
en_acts = pair_acts[0]
ja_acts = pair_acts[1]
pair_similarities = []
for layer in range(n_layers):
en_vec = en_acts[layer]
ja_vec = ja_acts[layer]
cos_sim = float(np.dot(en_vec, ja_vec) / (np.linalg.norm(en_vec) * np.linalg.norm(ja_vec) + 1e-10))
pair_similarities.append({
"layer": layer,
"similarity": round(cos_sim, 4)
})
similarity_data.append({
"english": en_text,
"japanese": ja_text,
"layers": pair_similarities
})
return jsonify({
"model": model_id,
"nLayers": n_layers,
"nEnglish": len(english_texts),
"nJapanese": len(japanese_texts),
"probeResults": layers_data,
"translationSimilarity": similarity_data
})
except Exception as e:
print(f"[LatentLang] Error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/cross-lingual/analyze', methods=['POST'])
def cross_lingual_analyze():
"""
Compare activation patterns between parallel Japanese/English facts.
Measures cosine similarity at each layer to find where representations converge.
"""
try:
data = request.json
pairs = data.get('pairs', [])
model_id = data.get('model', 'gpt2-small')
if not pairs or len(pairs) < 1:
return jsonify({"error": "Need at least 1 translation pair"}), 400
from transformer_lens import HookedTransformer
global _model, _saes, _current_model_id
base_model = get_base_model_id(model_id)
with _model_swap_lock:
if _model is None or get_base_model_id(_current_model_id) != base_model:
if _model is not None:
_cleanup_model_vram()
print(f"[CrossLingual] Loading model: {base_model}")
_model = _load_hooked_transformer(base_model)
_saes = {}
_current_model_id = model_id
n_layers = _model.cfg.n_layers
pair_results = []
for pair in pairs:
en_text = pair.get("english", "")
ja_text = pair.get("japanese", "")
label = pair.get("label", "")
domain = pair.get("domain", "general")
if not en_text or not ja_text:
continue
print(f"[CrossLingual] Analyzing pair: {label}")
acts, token_strs = extract_all_layer_activations(_model, [en_text, ja_text])
en_acts = acts[0]
ja_acts = acts[1]
en_logits = extract_final_logits(_model, en_text)
ja_logits = extract_final_logits(_model, ja_text)
layer_data = []
for layer in range(n_layers):
en_vec = en_acts[layer]
ja_vec = ja_acts[layer]
cos_sim = float(np.dot(en_vec, ja_vec) / (np.linalg.norm(en_vec) * np.linalg.norm(ja_vec) + 1e-10))
l2_dist = float(np.linalg.norm(en_vec - ja_vec))
layer_data.append({
"layer": layer,
"cosineSimilarity": round(cos_sim, 4),
"l2Distance": round(l2_dist, 2)
})
convergence_layer = -1
max_sim = 0
for ld in layer_data:
if ld["cosineSimilarity"] > max_sim:
max_sim = ld["cosineSimilarity"]
convergence_layer = ld["layer"]
pair_results.append({
"english": en_text,
"japanese": ja_text,
"label": label,
"domain": domain,
"layers": layer_data,
"convergenceLayer": convergence_layer,
"maxSimilarity": round(max_sim, 4),
"englishPrediction": en_logits,
"japanesePrediction": ja_logits,
"englishTokens": len(token_strs[0]),
"japaneseTokens": len(token_strs[1])
})
avg_similarities = []
for layer in range(n_layers):
sims = [p["layers"][layer]["cosineSimilarity"] for p in pair_results if len(p["layers"]) > layer]
avg_sim = float(np.mean(sims)) if sims else 0
avg_similarities.append({
"layer": layer,
"avgSimilarity": round(avg_sim, 4)
})
return jsonify({
"model": model_id,
"nLayers": n_layers,
"nPairs": len(pair_results),
"pairs": pair_results,
"averageSimilarity": avg_similarities
})
except Exception as e:
print(f"[CrossLingual] Error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/japanese-bench/analyze', methods=['POST'])
def japanese_bench_analyze():
"""
Run domain-specific prompts in both Japanese and English,
compare model confidence, entropy, and top predictions.
"""
try:
data = request.json
prompts = data.get('prompts', [])
model_id = data.get('model', 'gpt2-small')
if not prompts:
return jsonify({"error": "Need at least 1 prompt pair"}), 400
from transformer_lens import HookedTransformer
global _model, _saes, _current_model_id
base_model = get_base_model_id(model_id)
with _model_swap_lock:
if _model is None or get_base_model_id(_current_model_id) != base_model:
if _model is not None:
_cleanup_model_vram()
print(f"[JapaneseBench] Loading model: {base_model}")
_model = _load_hooked_transformer(base_model)
_saes = {}
_current_model_id = model_id
n_layers = _model.cfg.n_layers
W_U = _model.W_U
b_U = _model.b_U if hasattr(_model, 'b_U') and _model.b_U is not None else None
results = []
for prompt_pair in prompts:
en_text = prompt_pair.get("english", "")
ja_text = prompt_pair.get("japanese", "")
domain = prompt_pair.get("domain", "general")
label = prompt_pair.get("label", "")
if not en_text or not ja_text:
continue
print(f"[JapaneseBench] Analyzing: {label} ({domain})")
en_logits = extract_final_logits(_model, en_text)
ja_logits = extract_final_logits(_model, ja_text)
acts, token_strs = extract_all_layer_activations(_model, [en_text, ja_text])
en_lens = []
ja_lens = []
for layer in range(n_layers):
for lang_idx, lens_list in [(0, en_lens), (1, ja_lens)]:
if layer not in acts[lang_idx]:
continue
resid = torch.tensor(acts[lang_idx][layer]).float()
if hasattr(_model, 'ln_final'):
resid_normed = _model.ln_final(resid.unsqueeze(0)).squeeze(0)
else:
resid_normed = resid
logits = resid_normed.float() @ W_U.float()
if b_U is not None:
logits = logits + b_U.float()
probs = torch.softmax(logits, dim=-1)
top_p, top_i = probs.topk(5)
top_toks = []
for k in range(5):
tok_id = top_i[k].item()
tok_str = _model.tokenizer.decode([tok_id])
top_toks.append({
"token": tok_str,
"prob": round(top_p[k].item(), 4)
})
entropy = -torch.sum(probs * torch.log(probs + 1e-10)).item()
lens_list.append({
"layer": layer,
"topTokens": top_toks,
"entropy": round(entropy, 3),
"confidence": round(top_p[0].item(), 4)
})
confidence_gap = en_logits["confidence"] - ja_logits["confidence"]
entropy_gap = ja_logits["entropy"] - en_logits["entropy"]
results.append({
"english": en_text,
"japanese": ja_text,
"domain": domain,
"label": label,
"englishPrediction": en_logits,
"japanesePrediction": ja_logits,
"englishLens": en_lens,
"japaneseLens": ja_lens,
"confidenceGap": round(confidence_gap, 4),
"entropyGap": round(entropy_gap, 3),
"englishTokenCount": len(token_strs[0]),
"japaneseTokenCount": len(token_strs[1])
})
domain_summaries = {}
for r in results:
d = r["domain"]
if d not in domain_summaries:
domain_summaries[d] = {"confidenceGaps": [], "entropyGaps": [], "count": 0}
domain_summaries[d]["confidenceGaps"].append(r["confidenceGap"])
domain_summaries[d]["entropyGaps"].append(r["entropyGap"])
domain_summaries[d]["count"] += 1
for d in domain_summaries:
domain_summaries[d]["avgConfidenceGap"] = round(float(np.mean(domain_summaries[d]["confidenceGaps"])), 4)
domain_summaries[d]["avgEntropyGap"] = round(float(np.mean(domain_summaries[d]["entropyGaps"])), 3)
del domain_summaries[d]["confidenceGaps"]
del domain_summaries[d]["entropyGaps"]
return jsonify({
"model": model_id,
"nLayers": n_layers,
"nPrompts": len(results),
"results": results,
"domainSummary": domain_summaries
})
except Exception as e:
print(f"[JapaneseBench] Error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/concept-transfer/analyze', methods=['POST'])
def concept_transfer_analyze():
"""
Causal cross-lingual knowledge transfer experiment.
Tests whether patching English activations into a Japanese forward pass
changes the model's predictions — revealing at which layers concepts
are language-agnostic vs language-specific.
"""
try:
data = request.json
pairs = data.get('pairs', [])
model_id = data.get('model', 'gpt2-small')
if not pairs or len(pairs) < 1:
return jsonify({"error": "Need at least 1 prompt pair"}), 400
from transformer_lens import HookedTransformer
global _model, _saes, _current_model_id
base_model = get_base_model_id(model_id)
with _model_swap_lock:
if _model is None or get_base_model_id(_current_model_id) != base_model:
if _model is not None:
_cleanup_model_vram()
print(f"[ConceptTransfer] Loading model: {base_model}")
_model = _load_hooked_transformer(base_model)
_saes = {}
_current_model_id = model_id
model = _model
n_layers = model.cfg.n_layers
pair_results = []
for pair in pairs:
en_text = pair.get("english", "")
ja_text = pair.get("japanese", "")
target = pair.get("target", "")
label = pair.get("label", en_text[:40])
if not en_text or not ja_text:
continue
print(f"[ConceptTransfer] Analyzing: {label}")
en_tokens = model.to_tokens(en_text)
ja_tokens = model.to_tokens(ja_text)
with torch.no_grad():
en_logits, en_cache = model.run_with_cache(en_tokens)
ja_logits, ja_cache = model.run_with_cache(ja_tokens)
en_last_logits = en_logits[0, -1, :]
ja_last_logits = ja_logits[0, -1, :]
en_probs = torch.softmax(en_last_logits, dim=-1)
ja_probs = torch.softmax(ja_last_logits, dim=-1)
target_token_id = None
if target:
target_ids = model.to_tokens(target, prepend_bos=False)[0]
target_token_id = target_ids[0].item()
else:
target_token_id = en_probs.argmax().item()
target_str = model.tokenizer.decode([target_token_id])
en_target_prob = en_probs[target_token_id].item()
ja_target_prob = ja_probs[target_token_id].item()
en_top5 = []
top_p, top_i = en_probs.topk(5)
for k in range(5):
en_top5.append({"token": model.tokenizer.decode([top_i[k].item()]), "prob": round(top_p[k].item(), 4)})
ja_top5 = []
top_p, top_i = ja_probs.topk(5)
for k in range(5):
ja_top5.append({"token": model.tokenizer.decode([top_i[k].item()]), "prob": round(top_p[k].item(), 4)})
layer_results = []
for layer in range(n_layers):
en_resid = en_cache[f"blocks.{layer}.hook_resid_post"]
def patch_hook(activation, hook, source_act=en_resid):
patched = activation.clone()
patched[0, -1, :] = source_act[0, -1, :]
return patched
hook_name = f"blocks.{layer}.hook_resid_post"
model.reset_hooks()
with torch.no_grad():
patched_logits = model.run_with_hooks(
ja_tokens,
fwd_hooks=[(hook_name, patch_hook)],
reset_hooks_end=True
)
patched_probs = torch.softmax(patched_logits[0, -1, :], dim=-1)
patched_target_prob = patched_probs[target_token_id].item()
transfer_effect = patched_target_prob - ja_target_prob
patched_top3 = []
top_p, top_i = patched_probs.topk(3)
for k in range(3):
patched_top3.append({"token": model.tokenizer.decode([top_i[k].item()]), "prob": round(top_p[k].item(), 4)})
en_vec = en_cache[f"blocks.{layer}.hook_resid_post"][0, -1, :].detach().float().cpu().numpy()
ja_vec = ja_cache[f"blocks.{layer}.hook_resid_post"][0, -1, :].detach().float().cpu().numpy()
cos_sim = float(np.dot(en_vec, ja_vec) / (np.linalg.norm(en_vec) * np.linalg.norm(ja_vec) + 1e-10))
layer_results.append({
"layer": layer,
"transferEffect": round(transfer_effect, 6),
"patchedTargetProb": round(patched_target_prob, 6),
"cosineSimilarity": round(cos_sim, 4),
"patchedTop3": patched_top3
})
peak_layer = max(layer_results, key=lambda x: x["transferEffect"])
del en_cache, ja_cache
if torch.cuda.is_available():
torch.cuda.empty_cache()
pair_results.append({
"english": en_text,
"japanese": ja_text,
"target": target_str.strip(),
"targetTokenId": target_token_id,
"label": label,
"englishTargetProb": round(en_target_prob, 6),
"japaneseTargetProb": round(ja_target_prob, 6),
"englishTop5": en_top5,
"japaneseTop5": ja_top5,
"layers": layer_results,
"peakTransferLayer": peak_layer["layer"],
"peakTransferEffect": round(peak_layer["transferEffect"], 6),
"englishTokenCount": en_tokens.shape[1],
"japaneseTokenCount": ja_tokens.shape[1]
})
avg_transfer = []
for layer in range(n_layers):
effects = [p["layers"][layer]["transferEffect"] for p in pair_results if len(p["layers"]) > layer]
sims = [p["layers"][layer]["cosineSimilarity"] for p in pair_results if len(p["layers"]) > layer]
avg_transfer.append({
"layer": layer,
"avgTransferEffect": round(float(np.mean(effects)), 6) if effects else 0,
"avgCosineSimilarity": round(float(np.mean(sims)), 4) if sims else 0
})
return jsonify({
"model": model_id,
"nLayers": n_layers,
"nPairs": len(pair_results),
"pairs": pair_results,
"averageTransfer": avg_transfer
})
except Exception as e:
print(f"[ConceptTransfer] Error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/token-trajectory/analyze', methods=['POST'])
def token_trajectory_analyze():
"""
Token Trajectory: compare how English vs Japanese representations
evolve through the model's layers. Shows where fragmented Japanese
byte-tokens converge into coherent representations (or don't)
relative to a single English token.
"""
try:
data = request.json
concepts = data.get('concepts', [])
model_id = data.get('model', 'gpt2-small')
aggregation = data.get('aggregation', 'mean')
if not concepts or len(concepts) < 1:
return jsonify({"error": "Need at least 1 concept pair"}), 400
from transformer_lens import HookedTransformer
global _model, _saes, _current_model_id
base_model = get_base_model_id(model_id)
with _model_swap_lock:
if _model is None or get_base_model_id(_current_model_id) != base_model:
if _model is not None:
_cleanup_model_vram()
print(f"[TokenTrajectory] Loading model: {base_model}")
_model = _load_hooked_transformer(base_model)
_saes = {}
_current_model_id = model_id
model = _model
n_layers = model.cfg.n_layers
W_U = model.W_U
concept_results = []
for concept in concepts:
en_text = concept.get("english", "")
ja_text = concept.get("japanese", "")
label = concept.get("label", en_text[:30])
if not en_text or not ja_text:
continue
print(f"[TokenTrajectory] Analyzing: {label} ({en_text} / {ja_text})")
en_tokens = model.to_tokens(en_text)
ja_tokens = model.to_tokens(ja_text)
en_token_count = en_tokens.shape[1]
ja_token_count = ja_tokens.shape[1]
en_token_strs = [model.tokenizer.decode([en_tokens[0, t].item()]) for t in range(en_token_count)]
ja_token_strs = [model.tokenizer.decode([ja_tokens[0, t].item()]) for t in range(ja_token_count)]
resid_hooks = [f"blocks.{l}.hook_resid_post" for l in range(n_layers)]
with torch.no_grad():
_, en_cache = model.run_with_cache(en_tokens, names_filter=resid_hooks)
_, ja_cache = model.run_with_cache(ja_tokens, names_filter=resid_hooks)
layer_results = []
for layer in range(n_layers):
en_resid = en_cache[f"blocks.{layer}.hook_resid_post"]
ja_resid = ja_cache[f"blocks.{layer}.hook_resid_post"]
en_vec = en_resid[0, -1, :]
if aggregation == "last":
ja_vec = ja_resid[0, -1, :]
else:
ja_content_start = 1
ja_vec = ja_resid[0, ja_content_start:, :].mean(dim=0)
cos_sim = torch.nn.functional.cosine_similarity(
en_vec.unsqueeze(0), ja_vec.unsqueeze(0)
).item()
en_logits = en_vec @ W_U
en_probs = torch.softmax(en_logits, dim=-1)
en_top_p, en_top_i = en_probs.topk(5)
en_top5 = [{"token": model.tokenizer.decode([en_top_i[k].item()]), "prob": round(en_top_p[k].item(), 4)} for k in range(5)]
ja_logits = ja_vec @ W_U
ja_probs = torch.softmax(ja_logits, dim=-1)
ja_top_p, ja_top_i = ja_probs.topk(5)
ja_top5 = [{"token": model.tokenizer.decode([ja_top_i[k].item()]), "prob": round(ja_top_p[k].item(), 4)} for k in range(5)]
shared_tokens = set()
en_top20_i = en_probs.topk(20).indices.tolist()
ja_top20_i = ja_probs.topk(20).indices.tolist()
overlap_count = len(set(en_top20_i) & set(ja_top20_i))
layer_results.append({
"layer": layer,
"cosineSimilarity": round(cos_sim, 4),
"englishTop5": en_top5,
"japaneseTop5": ja_top5,
"vocabOverlap20": overlap_count
})
first_sim = layer_results[0]["cosineSimilarity"]
last_sim = layer_results[-1]["cosineSimilarity"]
peak_sim = max(layer_results, key=lambda x: x["cosineSimilarity"])
min_sim = min(layer_results, key=lambda x: x["cosineSimilarity"])
convergence = last_sim - first_sim
del en_cache, ja_cache
if torch.cuda.is_available():
torch.cuda.empty_cache()
concept_results.append({
"english": en_text,
"japanese": ja_text,
"label": label,
"englishTokens": en_token_strs,
"japaneseTokens": ja_token_strs,
"englishTokenCount": en_token_count,
"japaneseTokenCount": ja_token_count,
"layers": layer_results,
"convergence": round(convergence, 4),
"peakSimilarityLayer": peak_sim["layer"],
"peakSimilarity": round(peak_sim["cosineSimilarity"], 4),
"minSimilarityLayer": min_sim["layer"],
"minSimilarity": round(min_sim["cosineSimilarity"], 4)
})
avg_similarity = []
for layer in range(n_layers):
sims = [c["layers"][layer]["cosineSimilarity"] for c in concept_results if len(c["layers"]) > layer]
overlaps = [c["layers"][layer]["vocabOverlap20"] for c in concept_results if len(c["layers"]) > layer]
avg_similarity.append({
"layer": layer,
"avgSimilarity": round(float(np.mean(sims)), 4) if sims else 0,
"avgVocabOverlap": round(float(np.mean(overlaps)), 1) if overlaps else 0
})
avg_convergence = round(float(np.mean([c["convergence"] for c in concept_results])), 4) if concept_results else 0
return jsonify({
"model": model_id,
"nLayers": n_layers,
"nConcepts": len(concept_results),
"aggregation": aggregation,
"concepts": concept_results,
"averageSimilarity": avg_similarity,
"averageConvergence": avg_convergence
})
except Exception as e:
print(f"[TokenTrajectory] Error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
# ============================================
# Feature Universality Endpoint
# ============================================
@app.route('/feature-universality/analyze', methods=['POST'])
def feature_universality_analyze():
"""
Cross-lingual Feature Universality: find SAE features that activate
for both English and Japanese versions of the same concept.
Universal features = language-agnostic concept encoding.
"""
try:
data = request.json
pairs = data.get('pairs', [])
model_id = data.get('model', 'gpt2-small')
top_k = data.get('topK', 20)
if not pairs or len(pairs) < 1:
return jsonify({"error": "Need at least 1 concept pair"}), 400
from transformer_lens import HookedTransformer
from sae_lens import SAE
model, sae = get_model_and_sae(model_id)
if sae is None:
return jsonify({"error": f"SAE not available for model {model_id}"}), 503
hook_name = get_hook_name(model_id)
n_features = sae.cfg.d_sae
pair_results = []
en_activation_matrix = []
jp_activation_matrix = []
for pair in pairs:
en_text = pair.get('english', '')
jp_text = pair.get('japanese', '')
label = pair.get('label', '')
if not en_text or not jp_text:
continue
en_tokens = model.to_tokens(en_text)
jp_tokens = model.to_tokens(jp_text)
en_token_strs = model.to_str_tokens(en_text)
jp_token_strs = model.to_str_tokens(jp_text)
with torch.no_grad():
_, en_cache = model.run_with_cache(en_tokens, names_filter=[hook_name])
_, jp_cache = model.run_with_cache(jp_tokens, names_filter=[hook_name])
en_acts = safe_get_cache(en_cache, hook_name)
jp_acts = safe_get_cache(jp_cache, hook_name)
en_feature_acts = sae.encode(en_acts)
jp_feature_acts = sae.encode(jp_acts)
en_max_acts = en_feature_acts[0].max(dim=0).values.detach().float().cpu().numpy()
jp_max_acts = jp_feature_acts[0].max(dim=0).values.detach().float().cpu().numpy()
del en_cache, jp_cache, en_acts, jp_acts, en_feature_acts, jp_feature_acts
if torch.cuda.is_available():
torch.cuda.empty_cache()
en_activation_matrix.append(en_max_acts)
jp_activation_matrix.append(jp_max_acts)
en_top_indices = np.argsort(en_max_acts)[-top_k:][::-1]
jp_top_indices = np.argsort(jp_max_acts)[-top_k:][::-1]
en_top_set = set(en_top_indices.tolist())
jp_top_set = set(jp_top_indices.tolist())
shared_features = en_top_set & jp_top_set
en_only = en_top_set - jp_top_set
jp_only = jp_top_set - en_top_set
both_active = (en_max_acts > 0) & (jp_max_acts > 0)
cosine_sim = 0.0
en_norm = np.linalg.norm(en_max_acts)
jp_norm = np.linalg.norm(jp_max_acts)
if en_norm > 0 and jp_norm > 0:
cosine_sim = float(np.dot(en_max_acts, jp_max_acts) / (en_norm * jp_norm))
shared_detail = []
for fi in sorted(shared_features):
shared_detail.append({
"feature": int(fi),
"englishActivation": round(float(en_max_acts[fi]), 4),
"japaneseActivation": round(float(jp_max_acts[fi]), 4),
"ratio": round(float(min(en_max_acts[fi], jp_max_acts[fi]) / max(en_max_acts[fi], jp_max_acts[fi])), 4) if max(en_max_acts[fi], jp_max_acts[fi]) > 0 else 0
})
shared_detail.sort(key=lambda x: min(x["englishActivation"], x["japaneseActivation"]), reverse=True)
pair_results.append({
"english": en_text,
"japanese": jp_text,
"label": label,
"englishTokens": list(en_token_strs),
"japaneseTokens": list(jp_token_strs),
"cosineSimilarity": round(cosine_sim, 4),
"sharedFeatureCount": len(shared_features),
"englishOnlyCount": len(en_only),
"japaneseOnlyCount": len(jp_only),
"totalActiveEnglish": int(np.sum(en_max_acts > 0)),
"totalActiveJapanese": int(np.sum(jp_max_acts > 0)),
"sharedFeatures": shared_detail[:top_k],
"englishOnlyFeatures": [{"feature": int(fi), "activation": round(float(en_max_acts[fi]), 4)} for fi in sorted(en_only, key=lambda x: en_max_acts[x], reverse=True)][:10],
"japaneseOnlyFeatures": [{"feature": int(fi), "activation": round(float(jp_max_acts[fi]), 4)} for fi in sorted(jp_only, key=lambda x: jp_max_acts[x], reverse=True)][:10]
})
if len(en_activation_matrix) < 1:
return jsonify({"error": "No valid pairs processed"}), 400
en_mat = np.array(en_activation_matrix)
jp_mat = np.array(jp_activation_matrix)
universal_features = []
n_pairs = len(en_activation_matrix)
for fi in range(n_features):
en_col = en_mat[:, fi]
jp_col = jp_mat[:, fi]
en_active = np.sum(en_col > 0)
jp_active = np.sum(jp_col > 0)
both_active_count = np.sum((en_col > 0) & (jp_col > 0))
if both_active_count >= max(1, n_pairs * 0.5):
en_mean = float(np.mean(en_col[en_col > 0])) if en_active > 0 else 0
jp_mean = float(np.mean(jp_col[jp_col > 0])) if jp_active > 0 else 0
correlation = 0.0
if n_pairs > 1 and np.std(en_col) > 0 and np.std(jp_col) > 0:
correlation = float(np.corrcoef(en_col, jp_col)[0, 1])
universal_features.append({
"feature": int(fi),
"bothActiveCount": int(both_active_count),
"englishMeanActivation": round(en_mean, 4),
"japaneseMeanActivation": round(jp_mean, 4),
"correlation": round(correlation, 4),
"universalityScore": round(float(both_active_count / n_pairs) * (1 + abs(correlation)) / 2, 4)
})
universal_features.sort(key=lambda x: x["universalityScore"], reverse=True)
en_specific = []
jp_specific = []
for fi in range(n_features):
en_col = en_mat[:, fi]
jp_col = jp_mat[:, fi]
en_active = np.sum(en_col > 0)
jp_active = np.sum(jp_col > 0)
if en_active >= max(1, n_pairs * 0.5) and jp_active == 0:
en_specific.append({
"feature": int(fi),
"activeCount": int(en_active),
"meanActivation": round(float(np.mean(en_col[en_col > 0])), 4)
})
elif jp_active >= max(1, n_pairs * 0.5) and en_active == 0:
jp_specific.append({
"feature": int(fi),
"activeCount": int(jp_active),
"meanActivation": round(float(np.mean(jp_col[jp_col > 0])), 4)
})
en_specific.sort(key=lambda x: x["meanActivation"], reverse=True)
jp_specific.sort(key=lambda x: x["meanActivation"], reverse=True)
avg_cosine = round(float(np.mean([p["cosineSimilarity"] for p in pair_results])), 4)
avg_shared = round(float(np.mean([p["sharedFeatureCount"] for p in pair_results])), 1)
return jsonify({
"model": model_id,
"nFeatures": int(n_features),
"nPairs": len(pair_results),
"saeHook": hook_name,
"pairs": pair_results,
"universalFeatures": universal_features[:50],
"englishSpecificFeatures": en_specific[:30],
"japaneseSpecificFeatures": jp_specific[:30],
"averageCosineSimilarity": avg_cosine,
"averageSharedFeatures": avg_shared,
"totalUniversalFound": len(universal_features),
"totalEnglishSpecific": len(en_specific),
"totalJapaneseSpecific": len(jp_specific)
})
except Exception as e:
print(f"[FeatureUniversality] Error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
# ============================================
# Tokenization Tax Endpoint
# ============================================
@app.route('/tokenization-tax/analyze', methods=['POST'])
def tokenization_tax_analyze():
"""
Tokenization Tax: quantify the representation cost of Japanese
multi-byte tokenization vs English single-token encoding.
Measures per-layer prediction confidence, entropy, and how many
extra layers Japanese needs to reach equivalent representation quality.
"""
try:
data = request.json
pairs = data.get('pairs', [])
model_id = data.get('model', 'gpt2-small')
if not pairs or len(pairs) < 1:
return jsonify({"error": "Need at least 1 concept pair"}), 400
from transformer_lens import HookedTransformer
global _model, _saes, _current_model_id
base_model = get_base_model_id(model_id)
with _model_swap_lock:
if _model is None or get_base_model_id(_current_model_id) != base_model:
if _model is not None:
_cleanup_model_vram()
print(f"[TokenizationTax] Loading model: {base_model}")
_model = _load_hooked_transformer(base_model)
_saes = {}
_current_model_id = model_id
model = _model
n_layers = model.cfg.n_layers
W_U = model.W_U.detach().float().cpu()
b_U = model.b_U.detach().float().cpu() if model.b_U is not None else None
pair_results = []
for pair in pairs:
en_text = pair.get('english', '')
jp_text = pair.get('japanese', '')
label = pair.get('label', '')
if not en_text or not jp_text:
continue
en_tokens = model.to_tokens(en_text)
jp_tokens = model.to_tokens(jp_text)
en_token_strs = model.to_str_tokens(en_text)
jp_token_strs = model.to_str_tokens(jp_text)
resid_hooks = [f"blocks.{l}.hook_resid_post" for l in range(n_layers)]
with torch.no_grad():
_, en_cache = model.run_with_cache(en_tokens, names_filter=resid_hooks)
_, jp_cache = model.run_with_cache(jp_tokens, names_filter=resid_hooks)
en_layers = []
jp_layers = []
for layer in range(n_layers):
en_resid = en_cache[f"blocks.{layer}.hook_resid_post"][0, -1].detach().float().cpu()
jp_resid = jp_cache[f"blocks.{layer}.hook_resid_post"][0, -1].detach().float().cpu()
en_logits = en_resid @ W_U
jp_logits = jp_resid @ W_U
if b_U is not None:
en_logits = en_logits + b_U
jp_logits = jp_logits + b_U
en_probs = torch.softmax(en_logits, dim=-1)
jp_probs = torch.softmax(jp_logits, dim=-1)
en_entropy = -torch.sum(en_probs * torch.log(en_probs + 1e-10)).item()
jp_entropy = -torch.sum(jp_probs * torch.log(jp_probs + 1e-10)).item()
en_top1_prob = en_probs.max().item()
jp_top1_prob = jp_probs.max().item()
en_top5_vals, en_top5_idx = torch.topk(en_probs, 5)
jp_top5_vals, jp_top5_idx = torch.topk(jp_probs, 5)
en_top5 = [{"token": model.tokenizer.decode([idx.item()]), "prob": round(val.item(), 6)} for val, idx in zip(en_top5_vals, en_top5_idx)]
jp_top5 = [{"token": model.tokenizer.decode([idx.item()]), "prob": round(val.item(), 6)} for val, idx in zip(jp_top5_vals, jp_top5_idx)]
en_layers.append({
"layer": layer,
"entropy": round(en_entropy, 4),
"top1Prob": round(en_top1_prob, 6),
"top5Cumulative": round(en_top5_vals.sum().item(), 6),
"topPredictions": en_top5
})
jp_layers.append({
"layer": layer,
"entropy": round(jp_entropy, 4),
"top1Prob": round(jp_top1_prob, 6),
"top5Cumulative": round(jp_top5_vals.sum().item(), 6),
"topPredictions": jp_top5
})
en_final_entropy = en_layers[-1]["entropy"]
jp_final_entropy = jp_layers[-1]["entropy"]
en_half_confidence_layer = n_layers - 1
jp_half_confidence_layer = n_layers - 1
final_en_top1 = en_layers[-1]["top1Prob"]
final_jp_top1 = jp_layers[-1]["top1Prob"]
for l in en_layers:
if l["top1Prob"] >= final_en_top1 * 0.5:
en_half_confidence_layer = l["layer"]
break
for l in jp_layers:
if l["top1Prob"] >= final_jp_top1 * 0.5:
jp_half_confidence_layer = l["layer"]
break
confidence_tax = jp_half_confidence_layer - en_half_confidence_layer
entropy_ratios = []
for i in range(n_layers):
if en_layers[i]["entropy"] > 0:
entropy_ratios.append(jp_layers[i]["entropy"] / en_layers[i]["entropy"])
else:
entropy_ratios.append(1.0)
avg_entropy_ratio = float(np.mean(entropy_ratios))
del en_cache, jp_cache
if torch.cuda.is_available():
torch.cuda.empty_cache()
pair_results.append({
"english": en_text,
"japanese": jp_text,
"label": label,
"englishTokenCount": len(en_token_strs),
"japaneseTokenCount": len(jp_token_strs),
"tokenCountRatio": round(len(jp_token_strs) / max(len(en_token_strs), 1), 2),
"englishTokens": list(en_token_strs),
"japaneseTokens": list(jp_token_strs),
"englishLayers": en_layers,
"japaneseLayers": jp_layers,
"englishHalfConfidenceLayer": en_half_confidence_layer,
"japaneseHalfConfidenceLayer": jp_half_confidence_layer,
"confidenceTax": confidence_tax,
"averageEntropyRatio": round(avg_entropy_ratio, 4),
"englishFinalEntropy": round(en_final_entropy, 4),
"japaneseFinalEntropy": round(jp_final_entropy, 4)
})
if not pair_results:
return jsonify({"error": "No valid pairs processed"}), 400
avg_layers = []
for layer in range(n_layers):
en_entropies = [p["englishLayers"][layer]["entropy"] for p in pair_results]
jp_entropies = [p["japaneseLayers"][layer]["entropy"] for p in pair_results]
en_top1s = [p["englishLayers"][layer]["top1Prob"] for p in pair_results]
jp_top1s = [p["japaneseLayers"][layer]["top1Prob"] for p in pair_results]
avg_layers.append({
"layer": layer,
"avgEnglishEntropy": round(float(np.mean(en_entropies)), 4),
"avgJapaneseEntropy": round(float(np.mean(jp_entropies)), 4),
"avgEnglishTop1": round(float(np.mean(en_top1s)), 6),
"avgJapaneseTop1": round(float(np.mean(jp_top1s)), 6),
"entropyGap": round(float(np.mean(jp_entropies)) - float(np.mean(en_entropies)), 4),
"confidenceGap": round(float(np.mean(en_top1s)) - float(np.mean(jp_top1s)), 6)
})
avg_confidence_tax = round(float(np.mean([p["confidenceTax"] for p in pair_results])), 2)
avg_token_ratio = round(float(np.mean([p["tokenCountRatio"] for p in pair_results])), 2)
avg_entropy_ratio_overall = round(float(np.mean([p["averageEntropyRatio"] for p in pair_results])), 4)
return jsonify({
"model": model_id,
"nLayers": n_layers,
"nPairs": len(pair_results),
"pairs": pair_results,
"averageLayers": avg_layers,
"averageConfidenceTax": avg_confidence_tax,
"averageTokenCountRatio": avg_token_ratio,
"averageEntropyRatio": avg_entropy_ratio_overall
})
except Exception as e:
print(f"[TokenizationTax] Error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/model-merge/analyze', methods=['POST'])
def model_merge_analyze():
"""
Analyze what happens to SAE features during model merging.
Simulates interpolation between two domain activation vectors.
"""
try:
data = request.json
domain_a_texts = data.get('domainA', [])
domain_b_texts = data.get('domainB', [])
model_id = data.get('model', 'gpt2-small')
alpha_steps = data.get('alphaSteps', [0.0, 0.25, 0.5, 0.75, 1.0])
if not domain_a_texts or not domain_b_texts:
return jsonify({"error": "Both domainA and domainB texts are required"}), 400
model, sae = get_model_and_sae(model_id)
hook_name = get_hook_name(model_id)
n_features = sae.W_dec.shape[0] if hasattr(sae, 'W_dec') else sae.W_enc.shape[1]
domain_a_vectors = []
domain_a_info = []
domain_b_vectors = []
domain_b_info = []
all_items = [(item, 'a') for item in domain_a_texts] + [(item, 'b') for item in domain_b_texts]
with torch.no_grad():
for item, domain_tag in all_items:
tokens = model.to_tokens(item['text'])
n_tok = tokens.shape[-1]
_, cache = model.run_with_cache(tokens)
activations = safe_get_cache(cache, hook_name)
feature_acts = sae.encode(activations)
max_pooled = feature_acts[0].max(dim=0).values.detach().float().cpu().numpy()
n_active = int(np.sum(max_pooled > 0.1))
info = {
"text": item['text'],
"label": item.get('label', ''),
"nTokens": int(n_tok),
"nActiveFeatures": n_active
}
if domain_tag == 'a':
domain_a_vectors.append(max_pooled)
domain_a_info.append(info)
else:
domain_b_vectors.append(max_pooled)
domain_b_info.append(info)
del cache, activations, feature_acts
if torch.cuda.is_available():
torch.cuda.empty_cache()
domain_a_mean = np.mean(np.stack(domain_a_vectors), axis=0)
domain_b_mean = np.mean(np.stack(domain_b_vectors), axis=0)
merged_vectors = []
for alpha in alpha_steps:
merged = alpha * domain_a_mean + (1.0 - alpha) * domain_b_mean
merged_vectors.append(merged)
merged_matrix = np.stack(merged_vectors)
per_feature_std = np.std(merged_matrix, axis=0)
per_feature_max = np.max(merged_matrix, axis=0)
stability = 1.0 - (per_feature_std / (per_feature_max + 1e-8))
mean_across_alphas = np.mean(merged_matrix, axis=0)
stable_mask = (stability > 0.8) & (mean_across_alphas > 0.1)
stable_indices = np.where(stable_mask)[0]
stable_sorted = stable_indices[np.argsort(-stability[stable_indices])][:30]
fragile_mask = (stability < 0.3) & (per_feature_max > 0.1)
fragile_indices = np.where(fragile_mask)[0]
fragile_sorted = fragile_indices[np.argsort(stability[fragile_indices])][:30]
domain_a_exclusive_mask = (domain_a_mean > 0.1) & (domain_b_mean <= 0.1)
domain_a_excl_indices = np.where(domain_a_exclusive_mask)[0]
domain_a_excl_sorted = domain_a_excl_indices[np.argsort(-domain_a_mean[domain_a_excl_indices])][:20]
domain_b_exclusive_mask = (domain_b_mean > 0.1) & (domain_a_mean <= 0.1)
domain_b_excl_indices = np.where(domain_b_exclusive_mask)[0]
domain_b_excl_sorted = domain_b_excl_indices[np.argsort(-domain_b_mean[domain_b_excl_indices])][:20]
active_mask = per_feature_max > 0.05
if np.any(active_mask):
merge_compatibility = float(np.mean(stability[active_mask]))
else:
merge_compatibility = 0.0
def cosine_sim(a, b):
dot = np.dot(a, b)
norm_a = np.linalg.norm(a)
norm_b = np.linalg.norm(b)
if norm_a < 1e-10 or norm_b < 1e-10:
return 0.0
return float(dot / (norm_a * norm_b))
alpha_results = []
for i, alpha in enumerate(alpha_steps):
merged = merged_vectors[i]
cos_a = cosine_sim(merged, domain_a_mean)
cos_b = cosine_sim(merged, domain_b_mean)
active_count = int(np.sum(merged > 0.1))
alpha_results.append({
"alpha": round(float(alpha), 4),
"cosineToDomainA": round(cos_a, 4),
"cosineToDomainB": round(cos_b, 4),
"activeFeatureCount": active_count
})
mid_alpha_idx = len(alpha_steps) // 2
mid_merged = merged_vectors[mid_alpha_idx]
stable_features_list = []
for idx in stable_sorted:
idx = int(idx)
stable_features_list.append({
"feature": idx,
"stability": round(float(stability[idx]), 4),
"domainAMean": round(float(domain_a_mean[idx]), 4),
"domainBMean": round(float(domain_b_mean[idx]), 4),
"mergedMean": round(float(mid_merged[idx]), 4)
})
fragile_features_list = []
for idx in fragile_sorted:
idx = int(idx)
fragile_features_list.append({
"feature": idx,
"stability": round(float(stability[idx]), 4),
"domainAMean": round(float(domain_a_mean[idx]), 4),
"domainBMean": round(float(domain_b_mean[idx]), 4),
"variance": round(float(per_feature_std[idx] ** 2), 4)
})
domain_a_excl_list = []
for idx in domain_a_excl_sorted:
idx = int(idx)
domain_a_excl_list.append({
"feature": idx,
"activation": round(float(domain_a_mean[idx]), 4)
})
domain_b_excl_list = []
for idx in domain_b_excl_sorted:
idx = int(idx)
domain_b_excl_list.append({
"feature": idx,
"activation": round(float(domain_b_mean[idx]), 4)
})
domain_a_active = int(np.sum(domain_a_mean > 0.1))
domain_b_active = int(np.sum(domain_b_mean > 0.1))
return jsonify({
"model": model_id,
"nFeatures": int(n_features),
"saeHook": hook_name,
"nDomainA": len(domain_a_texts),
"nDomainB": len(domain_b_texts),
"domainATexts": domain_a_info,
"domainBTexts": domain_b_info,
"alphaSteps": alpha_results,
"stableFeatures": stable_features_list,
"fragileFeatures": fragile_features_list,
"domainAExclusive": domain_a_excl_list,
"domainBExclusive": domain_b_excl_list,
"mergeCompatibility": round(float(merge_compatibility), 4),
"totalStable": int(np.sum(stable_mask)),
"totalFragile": int(np.sum(fragile_mask)),
"totalDomainAExclusive": int(np.sum(domain_a_exclusive_mask)),
"totalDomainBExclusive": int(np.sum(domain_b_exclusive_mask)),
"domainASummary": {
"totalActive": domain_a_active,
"meanActivation": round(float(np.mean(domain_a_mean[domain_a_mean > 0.1])) if domain_a_active > 0 else 0.0, 4)
},
"domainBSummary": {
"totalActive": domain_b_active,
"meanActivation": round(float(np.mean(domain_b_mean[domain_b_mean > 0.1])) if domain_b_active > 0 else 0.0, 4)
}
})
except Exception as e:
print(f"[ModelMerge] Error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/financial-circuits/analyze', methods=['POST'])
def financial_circuits_analyze():
"""
Map which SAE features encode financial concepts across English and Japanese.
"""
try:
data = request.json
financial_pairs = data.get('financialPairs', [])
baseline_pairs = data.get('baselinePairs', [])
model_id = data.get('model', 'gpt2-small')
top_k = data.get('topK', 20)
if not financial_pairs:
return jsonify({"error": "financialPairs are required"}), 400
if not baseline_pairs:
return jsonify({"error": "baselinePairs are required"}), 400
model, sae = get_model_and_sae(model_id)
hook_name = get_hook_name(model_id)
n_features = sae.W_dec.shape[0] if hasattr(sae, 'W_dec') else sae.W_enc.shape[1]
def cosine_sim(a, b):
dot = np.dot(a, b)
norm_a = np.linalg.norm(a)
norm_b = np.linalg.norm(b)
if norm_a < 1e-10 or norm_b < 1e-10:
return 0.0
return float(dot / (norm_a * norm_b))
all_texts = []
all_tags = []
for i, pair in enumerate(financial_pairs):
all_texts.append(pair['english'])
all_tags.append(('fin_en', i))
all_texts.append(pair['japanese'])
all_tags.append(('fin_jp', i))
for i, pair in enumerate(baseline_pairs):
all_texts.append(pair['english'])
all_tags.append(('base_en', i))
all_texts.append(pair['japanese'])
all_tags.append(('base_jp', i))
all_vectors = {}
all_str_tokens = {}
with torch.no_grad():
for text, tag in zip(all_texts, all_tags):
tokens = model.to_tokens(text)
str_tokens = model.to_str_tokens(text)
_, cache = model.run_with_cache(tokens, names_filter=[hook_name])
activations = safe_get_cache(cache, hook_name)
feature_acts = sae.encode(activations)
max_pooled = feature_acts[0].max(dim=0).values.detach().float().cpu().numpy()
all_vectors[tag] = max_pooled
all_str_tokens[tag] = [t.replace('\u0120', ' ').replace('\u010a', '\n') for t in str_tokens]
del cache, activations, feature_acts
if torch.cuda.is_available():
torch.cuda.empty_cache()
fin_en_vectors = []
fin_jp_vectors = []
financial_pair_results = []
for i, pair in enumerate(financial_pairs):
en_vec = all_vectors[('fin_en', i)]
jp_vec = all_vectors[('fin_jp', i)]
en_tokens = all_str_tokens[('fin_en', i)]
jp_tokens = all_str_tokens[('fin_jp', i)]
fin_en_vectors.append(en_vec)
fin_jp_vectors.append(jp_vec)
cos_sim = cosine_sim(en_vec, jp_vec)
en_top_k_indices = np.argsort(-en_vec)[:top_k]
jp_top_k_indices = np.argsort(-jp_vec)[:top_k]
shared_top_k = set(en_top_k_indices.tolist()) & set(jp_top_k_indices.tolist())
combined_indices = np.unique(np.concatenate([en_top_k_indices, jp_top_k_indices]))
top_financial = []
for idx in combined_indices[:top_k]:
idx = int(idx)
top_financial.append({
"feature": idx,
"englishActivation": round(float(en_vec[idx]), 4),
"japaneseActivation": round(float(jp_vec[idx]), 4)
})
top_financial.sort(key=lambda x: max(x['englishActivation'], x['japaneseActivation']), reverse=True)
top_financial = top_financial[:top_k]
financial_pair_results.append({
"english": pair['english'],
"japanese": pair['japanese'],
"label": pair.get('label', ''),
"englishTokens": en_tokens,
"japaneseTokens": jp_tokens,
"cosineSimilarity": round(cos_sim, 4),
"sharedFeatureCount": len(shared_top_k),
"topFinancialFeatures": top_financial,
"totalActiveEnglish": int(np.sum(en_vec > 0.1)),
"totalActiveJapanese": int(np.sum(jp_vec > 0.1))
})
base_en_vectors = []
base_jp_vectors = []
baseline_pair_results = []
for i, pair in enumerate(baseline_pairs):
en_vec = all_vectors[('base_en', i)]
jp_vec = all_vectors[('base_jp', i)]
base_en_vectors.append(en_vec)
base_jp_vectors.append(jp_vec)
cos_sim = cosine_sim(en_vec, jp_vec)
en_top_k_indices = np.argsort(-en_vec)[:top_k]
jp_top_k_indices = np.argsort(-jp_vec)[:top_k]
shared_top_k = set(en_top_k_indices.tolist()) & set(jp_top_k_indices.tolist())
baseline_pair_results.append({
"english": pair['english'],
"japanese": pair['japanese'],
"label": pair.get('label', ''),
"cosineSimilarity": round(cos_sim, 4),
"sharedFeatureCount": len(shared_top_k)
})
mean_fin_en = np.mean(np.stack(fin_en_vectors), axis=0)
mean_fin_jp = np.mean(np.stack(fin_jp_vectors), axis=0)
mean_base_en = np.mean(np.stack(base_en_vectors), axis=0)
mean_base_jp = np.mean(np.stack(base_jp_vectors), axis=0)
specificity = np.maximum(mean_fin_en, mean_fin_jp) - np.maximum(mean_base_en, mean_base_jp)
finance_specific_mask = specificity > 0.1
finance_specific_indices = np.where(finance_specific_mask)[0]
finance_sorted = finance_specific_indices[np.argsort(-specificity[finance_specific_indices])][:30]
finance_specific_list = []
for idx in finance_sorted:
idx = int(idx)
finance_specific_list.append({
"feature": idx,
"specificityScore": round(float(specificity[idx]), 4),
"financialMeanEN": round(float(mean_fin_en[idx]), 4),
"financialMeanJP": round(float(mean_fin_jp[idx]), 4),
"baselineMeanEN": round(float(mean_base_en[idx]), 4),
"baselineMeanJP": round(float(mean_base_jp[idx]), 4)
})
universal_mask = finance_specific_mask & (mean_fin_en > 0.1) & (mean_fin_jp > 0.1)
universal_indices = np.where(universal_mask)[0]
universal_sorted = universal_indices[np.argsort(-specificity[universal_indices])][:20]
universal_list = []
for idx in universal_sorted:
idx = int(idx)
universal_list.append({
"feature": idx,
"englishActivation": round(float(mean_fin_en[idx]), 4),
"japaneseActivation": round(float(mean_fin_jp[idx]), 4),
"specificityScore": round(float(specificity[idx]), 4)
})
en_only_mask = finance_specific_mask & (mean_fin_en > 0.1) & (mean_fin_jp < 0.05)
en_only_indices = np.where(en_only_mask)[0]
en_only_sorted = en_only_indices[np.argsort(-mean_fin_en[en_only_indices])][:15]
en_only_list = []
for idx in en_only_sorted:
idx = int(idx)
en_only_list.append({
"feature": idx,
"activation": round(float(mean_fin_en[idx]), 4),
"specificityScore": round(float(specificity[idx]), 4)
})
jp_only_mask = finance_specific_mask & (mean_fin_jp > 0.1) & (mean_fin_en < 0.05)
jp_only_indices = np.where(jp_only_mask)[0]
jp_only_sorted = jp_only_indices[np.argsort(-mean_fin_jp[jp_only_indices])][:15]
jp_only_list = []
for idx in jp_only_sorted:
idx = int(idx)
jp_only_list.append({
"feature": idx,
"activation": round(float(mean_fin_jp[idx]), 4),
"specificityScore": round(float(specificity[idx]), 4)
})
cross_lingual_overlap = cosine_sim(mean_fin_en, mean_fin_jp)
avg_fin_cos = round(float(np.mean([r['cosineSimilarity'] for r in financial_pair_results])), 4)
avg_base_cos = round(float(np.mean([r['cosineSimilarity'] for r in baseline_pair_results])), 4)
return jsonify({
"model": model_id,
"nFeatures": int(n_features),
"saeHook": hook_name,
"nFinancialPairs": len(financial_pairs),
"nBaselinePairs": len(baseline_pairs),
"financialPairResults": financial_pair_results,
"baselinePairResults": baseline_pair_results,
"financeSpecificFeatures": finance_specific_list,
"universalFinancialFeatures": universal_list,
"englishOnlyFinancial": en_only_list,
"japaneseOnlyFinancial": jp_only_list,
"crossLingualOverlap": round(float(cross_lingual_overlap), 4),
"avgFinancialCosineSim": avg_fin_cos,
"avgBaselineCosineSim": avg_base_cos,
"financialVsBaselineGap": round(avg_fin_cos - avg_base_cos, 4),
"totalFinanceSpecific": int(np.sum(finance_specific_mask)),
"totalUniversalFinancial": int(np.sum(universal_mask)),
"totalEnglishOnlyFinancial": int(np.sum(en_only_mask)),
"totalJapaneseOnlyFinancial": int(np.sum(jp_only_mask))
})
except Exception as e:
print(f"[FinancialCircuits] Error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/domain-benchmark/analyze', methods=['POST'])
def domain_benchmark_analyze():
"""
Analyze model prediction confidence across domains for English vs Japanese text.
Request body:
{
"domains": [
{
"domain": "Legal",
"englishTexts": ["The court ruled...", ...],
"japaneseTexts": ["裁判所は...", ...]
},
...
],
"model": "gpt2-small"
}
Response:
{
"model": "gpt2-small",
"domains": [
{
"domain": "Legal",
"englishAccuracy": 0.85,
"japaneseAccuracy": 0.62,
"englishEntropy": 2.1,
"japaneseEntropy": 3.4,
"gap": 0.23
},
...
],
"overallEnglishAccuracy": 0.82,
"overallJapaneseAccuracy": 0.60
}
"""
try:
data = request.json
domains = data.get('domains', [])
model_id = data.get('model', data.get('model_id', 'gpt2-small'))
if not domains:
return jsonify({"error": "No domains provided"}), 400
import torch.nn.functional as F
with _model_lock:
model, sae = get_model_and_sae(model_id)
domain_results = []
all_en_confidences = []
all_jp_confidences = []
with torch.no_grad():
for domain_entry in domains:
domain_name = domain_entry.get('domain', 'Unknown')
en_texts = domain_entry.get('englishTexts', [])
jp_texts = domain_entry.get('japaneseTexts', [])
en_confidences = []
en_entropies = []
for text in en_texts:
if not text or not text.strip():
continue
try:
tokens = model.to_tokens(text)
logits = model(tokens)
final_logits = logits[0, -1, :]
probs = F.softmax(final_logits, dim=-1)
top1_prob = probs.max().item()
log_probs = torch.log(probs + 1e-10)
entropy = -(probs * log_probs).sum().item()
en_confidences.append(top1_prob)
en_entropies.append(entropy)
except Exception as e:
print(f"[DomainBenchmark] Error processing EN text: {e}")
jp_confidences = []
jp_entropies = []
for text in jp_texts:
if not text or not text.strip():
continue
try:
tokens = model.to_tokens(text)
logits = model(tokens)
final_logits = logits[0, -1, :]
probs = F.softmax(final_logits, dim=-1)
top1_prob = probs.max().item()
log_probs = torch.log(probs + 1e-10)
entropy = -(probs * log_probs).sum().item()
jp_confidences.append(top1_prob)
jp_entropies.append(entropy)
except Exception as e:
print(f"[DomainBenchmark] Error processing JP text: {e}")
en_acc = float(np.mean(en_confidences)) if en_confidences else 0.0
jp_acc = float(np.mean(jp_confidences)) if jp_confidences else 0.0
en_ent = float(np.mean(en_entropies)) if en_entropies else 0.0
jp_ent = float(np.mean(jp_entropies)) if jp_entropies else 0.0
all_en_confidences.extend(en_confidences)
all_jp_confidences.extend(jp_confidences)
domain_results.append({
"domain": domain_name,
"englishAccuracy": round(en_acc, 4),
"japaneseAccuracy": round(jp_acc, 4),
"englishEntropy": round(en_ent, 4),
"japaneseEntropy": round(jp_ent, 4),
"gap": round(en_acc - jp_acc, 4)
})
print(f"[DomainBenchmark] {domain_name}: EN={en_acc:.4f}, JP={jp_acc:.4f}, gap={en_acc - jp_acc:.4f}")
overall_en = round(float(np.mean(all_en_confidences)), 4) if all_en_confidences else 0.0
overall_jp = round(float(np.mean(all_jp_confidences)), 4) if all_jp_confidences else 0.0
return jsonify({
"model": model_id,
"domains": domain_results,
"overallEnglishAccuracy": overall_en,
"overallJapaneseAccuracy": overall_jp
})
except Exception as e:
print(f"[DomainBenchmark] Error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/statistical_test', methods=['POST'])
def statistical_test():
"""Run formal statistical tests on experimental data.
Supports:
- paired_bootstrap: Bootstrap confidence intervals for paired measurements (e.g., EN vs JP similarities)
- permutation_test: Permutation test for group differences
- effect_size: Cohen's d effect size with bootstrap CI
- correlation: Pearson/Spearman correlation with significance
- one_sample: One-sample test against a reference value (e.g., is mean similarity > 0.5?)
- summary_stats: Descriptive statistics with bootstrap CIs for means
All tests use non-parametric methods (no normality assumption required).
"""
try:
data = request.json
if not data:
return jsonify({"error": "No data provided"}), 400
test_type = data.get("test", "summary_stats")
n_bootstrap = min(data.get("nBootstrap", 10000), 50000)
confidence_level = data.get("confidenceLevel", 0.95)
import scipy.stats as stats
def bootstrap_ci(values, n_boot=10000, ci=0.95):
values = np.array(values, dtype=float)
values = values[~np.isnan(values)]
if len(values) < 2:
return {"mean": float(np.mean(values)) if len(values) > 0 else None, "ci_lower": None, "ci_upper": None, "n": len(values)}
boot_means = np.array([np.mean(np.random.choice(values, size=len(values), replace=True)) for _ in range(n_boot)])
alpha = 1 - ci
ci_lower = float(np.percentile(boot_means, 100 * alpha / 2))
ci_upper = float(np.percentile(boot_means, 100 * (1 - alpha / 2)))
return {
"mean": float(np.mean(values)),
"std": float(np.std(values, ddof=1)),
"ci_lower": ci_lower,
"ci_upper": ci_upper,
"ci_level": ci,
"n": len(values),
"se": float(np.std(boot_means)),
}
result = {"test": test_type, "confidenceLevel": confidence_level}
if test_type == "summary_stats":
groups = data.get("groups", {})
if not groups:
values = data.get("values", [])
groups = {"data": values}
group_results = {}
for name, values in groups.items():
arr = np.array(values, dtype=float)
arr = arr[~np.isnan(arr)]
ci = bootstrap_ci(arr, n_bootstrap, confidence_level)
group_results[name] = {
**ci,
"median": float(np.median(arr)) if len(arr) > 0 else None,
"min": float(np.min(arr)) if len(arr) > 0 else None,
"max": float(np.max(arr)) if len(arr) > 0 else None,
"iqr": float(np.percentile(arr, 75) - np.percentile(arr, 25)) if len(arr) >= 4 else None,
}
result["groups"] = group_results
elif test_type == "paired_bootstrap":
group_a = np.array(data.get("groupA", []), dtype=float)
group_b = np.array(data.get("groupB", []), dtype=float)
label_a = data.get("labelA", "A")
label_b = data.get("labelB", "B")
min_len = min(len(group_a), len(group_b))
if min_len < 2:
return jsonify({"error": f"Need at least 2 paired observations, got {min_len}"}), 400
group_a, group_b = group_a[:min_len], group_b[:min_len]
diffs = group_a - group_b
boot_diffs = np.array([np.mean(np.random.choice(diffs, size=len(diffs), replace=True)) for _ in range(n_bootstrap)])
alpha = 1 - confidence_level
p_value = float(2 * min(np.mean(boot_diffs <= 0), np.mean(boot_diffs >= 0)))
p_value = min(p_value, 1.0)
pooled_std = float(np.sqrt((np.var(group_a, ddof=1) + np.var(group_b, ddof=1)) / 2))
cohens_d = float(np.mean(diffs) / pooled_std) if pooled_std > 0 else 0.0
result.update({
"labelA": label_a,
"labelB": label_b,
"n": min_len,
"meanDiff": float(np.mean(diffs)),
"diffCI": {
"lower": float(np.percentile(boot_diffs, 100 * alpha / 2)),
"upper": float(np.percentile(boot_diffs, 100 * (1 - alpha / 2))),
},
"pValue": p_value,
"significant": p_value < (1 - confidence_level),
"effectSize": cohens_d,
"effectMagnitude": "large" if abs(cohens_d) >= 0.8 else "medium" if abs(cohens_d) >= 0.5 else "small" if abs(cohens_d) >= 0.2 else "negligible",
"summaryA": bootstrap_ci(group_a, n_bootstrap, confidence_level),
"summaryB": bootstrap_ci(group_b, n_bootstrap, confidence_level),
"interpretation": f"{'Significant' if p_value < (1 - confidence_level) else 'No significant'} difference between {label_a} and {label_b} (mean diff = {float(np.mean(diffs)):.4f}, p = {p_value:.4f}, Cohen's d = {cohens_d:.3f} [{('large' if abs(cohens_d) >= 0.8 else 'medium' if abs(cohens_d) >= 0.5 else 'small' if abs(cohens_d) >= 0.2 else 'negligible')}], n = {min_len})",
})
elif test_type == "permutation_test":
group_a = np.array(data.get("groupA", []), dtype=float)
group_b = np.array(data.get("groupB", []), dtype=float)
label_a = data.get("labelA", "A")
label_b = data.get("labelB", "B")
n_perms = min(data.get("nPermutations", 10000), 50000)
if len(group_a) < 2 or len(group_b) < 2:
return jsonify({"error": "Need at least 2 observations per group"}), 400
observed_diff = float(np.mean(group_a) - np.mean(group_b))
combined = np.concatenate([group_a, group_b])
na = len(group_a)
perm_diffs = np.zeros(n_perms)
for i in range(n_perms):
np.random.shuffle(combined)
perm_diffs[i] = np.mean(combined[:na]) - np.mean(combined[na:])
p_value = float(np.mean(np.abs(perm_diffs) >= abs(observed_diff)))
pooled_std = float(np.sqrt(((len(group_a) - 1) * np.var(group_a, ddof=1) + (len(group_b) - 1) * np.var(group_b, ddof=1)) / (len(group_a) + len(group_b) - 2)))
cohens_d = float(observed_diff / pooled_std) if pooled_std > 0 else 0.0
result.update({
"labelA": label_a,
"labelB": label_b,
"nA": len(group_a),
"nB": len(group_b),
"observedDiff": observed_diff,
"pValue": p_value,
"significant": p_value < (1 - confidence_level),
"effectSize": cohens_d,
"effectMagnitude": "large" if abs(cohens_d) >= 0.8 else "medium" if abs(cohens_d) >= 0.5 else "small" if abs(cohens_d) >= 0.2 else "negligible",
"summaryA": bootstrap_ci(group_a, n_bootstrap, confidence_level),
"summaryB": bootstrap_ci(group_b, n_bootstrap, confidence_level),
"interpretation": f"Permutation test: {'Significant' if p_value < (1 - confidence_level) else 'No significant'} difference between {label_a} and {label_b} (observed diff = {observed_diff:.4f}, p = {p_value:.4f}, Cohen's d = {cohens_d:.3f}, nA = {len(group_a)}, nB = {len(group_b)})",
})
elif test_type == "effect_size":
group_a = np.array(data.get("groupA", []), dtype=float)
group_b = np.array(data.get("groupB", []), dtype=float)
if len(group_a) < 2 or len(group_b) < 2:
return jsonify({"error": "Need at least 2 observations per group"}), 400
pooled_std = float(np.sqrt(((len(group_a) - 1) * np.var(group_a, ddof=1) + (len(group_b) - 1) * np.var(group_b, ddof=1)) / (len(group_a) + len(group_b) - 2)))
cohens_d = float((np.mean(group_a) - np.mean(group_b)) / pooled_std) if pooled_std > 0 else 0.0
boot_ds = []
for _ in range(n_bootstrap):
ba = np.random.choice(group_a, size=len(group_a), replace=True)
bb = np.random.choice(group_b, size=len(group_b), replace=True)
ps = float(np.sqrt(((len(ba) - 1) * np.var(ba, ddof=1) + (len(bb) - 1) * np.var(bb, ddof=1)) / (len(ba) + len(bb) - 2)))
boot_ds.append(float((np.mean(ba) - np.mean(bb)) / ps) if ps > 0 else 0.0)
boot_ds = np.array(boot_ds)
alpha = 1 - confidence_level
result.update({
"cohensD": cohens_d,
"effectMagnitude": "large" if abs(cohens_d) >= 0.8 else "medium" if abs(cohens_d) >= 0.5 else "small" if abs(cohens_d) >= 0.2 else "negligible",
"ci": {
"lower": float(np.percentile(boot_ds, 100 * alpha / 2)),
"upper": float(np.percentile(boot_ds, 100 * (1 - alpha / 2))),
},
"nA": len(group_a),
"nB": len(group_b),
})
elif test_type == "correlation":
x = np.array(data.get("x", []), dtype=float)
y = np.array(data.get("y", []), dtype=float)
method = data.get("method", "pearson")
min_len = min(len(x), len(y))
if min_len < 3:
return jsonify({"error": f"Need at least 3 observations for correlation, got {min_len}"}), 400
x, y = x[:min_len], y[:min_len]
if method == "spearman":
r, p = stats.spearmanr(x, y)
else:
r, p = stats.pearsonr(x, y)
boot_rs = []
for _ in range(n_bootstrap):
idx = np.random.choice(len(x), size=len(x), replace=True)
if method == "spearman":
br, _ = stats.spearmanr(x[idx], y[idx])
else:
br, _ = stats.pearsonr(x[idx], y[idx])
boot_rs.append(br)
boot_rs = np.array(boot_rs)
alpha_val = 1 - confidence_level
result.update({
"method": method,
"r": float(r),
"pValue": float(p),
"significant": float(p) < (1 - confidence_level),
"ci": {
"lower": float(np.percentile(boot_rs, 100 * alpha_val / 2)),
"upper": float(np.percentile(boot_rs, 100 * (1 - alpha_val / 2))),
},
"n": min_len,
"rSquared": float(r ** 2),
"interpretation": f"{method.capitalize()} r = {float(r):.4f} (p = {float(p):.4f}), {'significant' if float(p) < (1 - confidence_level) else 'not significant'}, R² = {float(r**2):.4f}, n = {min_len}",
})
elif test_type == "one_sample":
values = np.array(data.get("values", []), dtype=float)
reference = data.get("reference", 0.0)
alternative = data.get("alternative", "two-sided")
values = values[~np.isnan(values)]
if len(values) < 2:
return jsonify({"error": "Need at least 2 observations"}), 400
observed_mean = float(np.mean(values))
ci = bootstrap_ci(values, n_bootstrap, confidence_level)
boot_means = np.array([np.mean(np.random.choice(values, size=len(values), replace=True)) for _ in range(n_bootstrap)])
centered = boot_means - observed_mean + reference
if alternative == "greater":
p_value = float(np.mean(centered >= observed_mean))
elif alternative == "less":
p_value = float(np.mean(centered <= observed_mean))
else:
p_value = float(2 * min(np.mean(centered >= observed_mean), np.mean(centered <= observed_mean)))
p_value = min(p_value, 1.0)
result.update({
"observedMean": observed_mean,
"reference": reference,
"alternative": alternative,
"pValue": p_value,
"significant": p_value < (1 - confidence_level),
"ci": {"lower": ci["ci_lower"], "upper": ci["ci_upper"]},
"n": len(values),
"interpretation": f"One-sample test: mean = {observed_mean:.4f} vs reference = {reference} ({alternative}), p = {p_value:.4f}, {'significant' if p_value < (1 - confidence_level) else 'not significant'}, n = {len(values)}",
})
else:
return jsonify({"error": f"Unknown test type: {test_type}. Supported: summary_stats, paired_bootstrap, permutation_test, effect_size, correlation, one_sample"}), 400
print(f"[StatisticalTest] {test_type} completed successfully")
return jsonify(result)
except Exception as e:
print(f"[StatisticalTest] Error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
_specter2_model = None
_specter2_tokenizer = None
def _load_specter2():
"""Lazy-load SPECTER2 model for paper embeddings. ~110MB, runs on CPU."""
global _specter2_model, _specter2_tokenizer
if _specter2_model is not None:
return _specter2_model, _specter2_tokenizer
print("[SPECTER2] Loading allenai/specter2_base...")
from transformers import AutoTokenizer, AutoModel
_specter2_tokenizer = AutoTokenizer.from_pretrained("allenai/specter2_base")
_specter2_model = AutoModel.from_pretrained("allenai/specter2_base")
_specter2_model.eval()
print("[SPECTER2] Model loaded successfully")
return _specter2_model, _specter2_tokenizer
@app.route('/specter2/embed', methods=['POST'])
def specter2_embed():
"""Embed one or more papers using local SPECTER2.
Input: { "papers": [{"title": "...", "abstract": "..."}] }
Returns: { "embeddings": [[...768 floats...], ...] }
"""
try:
data = request.json
papers = data.get("papers", [])
if not papers:
return jsonify({"error": "No papers provided"}), 400
if len(papers) > 32:
return jsonify({"error": "Max 32 papers per batch"}), 400
model, tokenizer = _load_specter2()
texts = []
for p in papers:
title = p.get("title", "")
abstract = p.get("abstract", "")
text = title + (tokenizer.sep_token + abstract if abstract else "")
texts.append(text)
inputs = tokenizer(
texts,
padding=True,
truncation=True,
max_length=512,
return_tensors="pt"
)
with torch.no_grad():
outputs = model(**inputs)
embeddings = outputs.last_hidden_state[:, 0, :]
embedding_list = embeddings.cpu().numpy().tolist()
return jsonify({"embeddings": embedding_list, "count": len(embedding_list), "dim": len(embedding_list[0])})
except Exception as e:
print(f"[SPECTER2] Error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/warmup', methods=['POST'])
def warmup():
"""Pre-load a model and run a test inference so subsequent requests are fast."""
model_id = request.json.get("model", "gpt2-small") if request.json else "gpt2-small"
try:
print(f"[SAE Service] Warmup request for {model_id}...")
model, sae = get_model_and_sae(model_id)
print(f"[SAE Service] {model_id} loaded, running test inference...")
import time
t0 = time.time()
with torch.no_grad():
tokens = model.to_tokens("The quick brown fox")
_, cache = model.run_with_cache(tokens)
elapsed = time.time() - t0
print(f"[SAE Service] Test inference complete in {elapsed:.1f}s — fully warmed up.")
return jsonify({"status": "ready", "model": model_id, "test_inference_ms": int(elapsed * 1000)})
except Exception as e:
print(f"[SAE Service] Warmup failed: {e}")
return jsonify({"status": "error", "error": str(e)}), 500
_openreviewer_model = None
_openreviewer_tokenizer = None
def _load_openreviewer():
"""Load the Llama-OpenReviewer-8B model for peer review generation.
Clears SAE models from VRAM first since the 8B model needs the full T4."""
global _openreviewer_model, _openreviewer_tokenizer
if _openreviewer_model is not None:
return _openreviewer_model, _openreviewer_tokenizer
_cleanup_model_vram()
device = _get_device()
print(f"[OpenReviewer] Loading maxidl/Llama-OpenReviewer-8B on {device}...")
from transformers import AutoModelForCausalLM, AutoTokenizer
hf_token = os.environ.get("HF_TOKEN")
_openreviewer_tokenizer = AutoTokenizer.from_pretrained(
"maxidl/Llama-OpenReviewer-8B",
token=hf_token,
)
_openreviewer_model = AutoModelForCausalLM.from_pretrained(
"maxidl/Llama-OpenReviewer-8B",
token=hf_token,
torch_dtype=torch.float16,
device_map=device,
)
print(f"[OpenReviewer] Model loaded on {device}")
if torch.cuda.is_available():
free_mb = torch.cuda.mem_get_info()[0] // (1024 * 1024)
print(f"[OpenReviewer] VRAM free after load: {free_mb} MB")
return _openreviewer_model, _openreviewer_tokenizer
def _unload_openreviewer():
"""Free OpenReviewer from VRAM so SAE models can load."""
global _openreviewer_model, _openreviewer_tokenizer
if _openreviewer_model is not None:
try:
_openreviewer_model.cpu()
except Exception:
pass
del _openreviewer_model
_openreviewer_model = None
if _openreviewer_tokenizer is not None:
del _openreviewer_tokenizer
_openreviewer_tokenizer = None
if torch.cuda.is_available():
torch.cuda.empty_cache()
import gc
gc.collect()
print("[OpenReviewer] Model unloaded from VRAM")
@app.route('/openreviewer/chat', methods=['POST'])
def openreviewer_chat():
"""OpenReviewer chat completions endpoint.
Expects OpenAI-compatible request body:
{
"messages": [
{"role": "system", "content": "..."},
{"role": "user", "content": "..."}
],
"max_tokens": 2048,
"temperature": 0.3
}
Returns OpenAI-compatible response:
{
"choices": [{"message": {"role": "assistant", "content": "..."}}],
"model": "maxidl/Llama-OpenReviewer-8B"
}
"""
try:
data = request.json or {}
messages = data.get("messages", [])
max_tokens = data.get("max_tokens", 2048)
temperature = data.get("temperature", 0.3)
if not messages:
return jsonify({"error": "No messages provided"}), 400
with _model_swap_lock:
model, tokenizer = _load_openreviewer()
if tokenizer.chat_template:
prompt_text = tokenizer.apply_chat_template(
messages, tokenize=False, add_generation_prompt=True
)
else:
parts = []
for m in messages:
role = m.get("role", "user")
content = m.get("content", "")
if role == "system":
parts.append(f"<<SYS>>\n{content}\n<</SYS>>\n\n")
elif role == "user":
parts.append(f"[INST] {content} [/INST]\n")
elif role == "assistant":
parts.append(f"{content}\n")
prompt_text = "".join(parts)
inputs = tokenizer(prompt_text, return_tensors="pt").to(model.device)
with torch.no_grad():
outputs = model.generate(
**inputs,
max_new_tokens=max_tokens,
temperature=max(temperature, 0.01),
do_sample=temperature > 0,
top_p=0.9,
pad_token_id=tokenizer.eos_token_id,
)
new_tokens = outputs[0][inputs["input_ids"].shape[1]:]
response_text = tokenizer.decode(new_tokens, skip_special_tokens=True)
print(f"[OpenReviewer] Generated {len(new_tokens)} tokens")
return jsonify({
"choices": [{"message": {"role": "assistant", "content": response_text}}],
"model": "maxidl/Llama-OpenReviewer-8B",
})
except torch.cuda.OutOfMemoryError:
_unload_openreviewer()
return jsonify({"error": "CUDA out of memory loading OpenReviewer. Try again after VRAM is freed."}), 503
except Exception as e:
print(f"[OpenReviewer] Error: {e}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/superposition-analysis/analyze', methods=['POST'])
def superposition_analysis_analyze():
"""
Profile MLP neurons in a chosen layer to expose polysemanticity / superposition.
Pipeline:
1. Forward each prompt with cache; gather `blocks.{L}.mlp.hook_post` -> [N_prompts, T, d_mlp].
2. Concatenate token-level activations into [N_total, d_mlp]; record (prompt_idx, pos, token_id) per row.
3. For each neuron compute max, mean, fire_rate (% tokens > 0).
4. Pick top-K neurons by max activation.
5. For each chosen neuron, take top-N rows by activation → record (token, context window, activation).
6. Polysemy score: take top-N tokens' unembedding rows W_U[:,token_id], compute mean pairwise
cosine similarity; polysemy_score = 1 − mean_cos. Coherence = mean_cos.
Request: { model, prompts:[..], layer, topK, examplesPerNeuron }
Response: { success, device, layer, totalNeurons, totalTokens, meanFireRate, meanPolysemy,
neurons:[{neuron,maxActivation,meanActivation,fireRate,polysemyScore,
semanticCoherence,examples:[{token,activation,context}]}] }
"""
try:
data = request.json or {}
prompts = [p for p in (data.get('prompts') or []) if isinstance(p, str) and p.strip()][:40]
if len(prompts) < 4:
return jsonify({"error": "at least 4 prompts required"}), 400
model_id = data.get('model', 'gpt2-small')
layer = int(data.get('layer') or 6)
top_k = max(3, min(20, int(data.get('topK') or 8)))
examples_per_neuron = max(3, min(10, int(data.get('examplesPerNeuron') or 5)))
if model_id in ACTIVATION_ONLY_MODELS:
model = _load_model_only(model_id)
else:
model, _ = get_model_and_sae(model_id)
device = next(model.parameters()).device
layer = max(0, min(model.cfg.n_layers - 1, layer))
# Collect activations from chosen layer's MLP post-activation
hook_name = f'blocks.{layer}.mlp.hook_post'
all_acts = [] # list of [T_i, d_mlp]
all_token_ids = [] # list of [T_i]
all_prompt_idx = [] # list of [T_i] indices into prompts
for pi, p in enumerate(prompts):
try:
tokens = model.to_tokens(p).to(device)
if tokens.shape[1] > 96:
tokens = tokens[:, :96]
with torch.no_grad():
_, cache = model.run_with_cache(tokens, names_filter=lambda n: n == hook_name)
acts = cache[hook_name][0].float() # [T, d_mlp]
all_acts.append(acts)
all_token_ids.append(tokens[0])
all_prompt_idx.extend([pi] * acts.shape[0])
del cache
except Exception as inner:
print(f"[SuperpositionAnalysis] skip prompt {pi}: {inner}")
if not all_acts:
return jsonify({"error": "no usable prompts"}), 400
acts = torch.cat(all_acts, dim=0) # [N_total, d_mlp]
token_ids = torch.cat(all_token_ids, dim=0) # [N_total]
prompt_idx = torch.tensor(all_prompt_idx, device=device)
n_total, d_mlp = acts.shape
max_per_neuron = acts.max(dim=0).values # [d_mlp]
mean_per_neuron = acts.mean(dim=0) # [d_mlp]
fire_per_neuron = (acts > 0).float().mean(dim=0) # [d_mlp]
mean_fire_rate = fire_per_neuron.mean().item()
# top-K neurons by max activation
top_neuron_ids = max_per_neuron.topk(top_k).indices.tolist()
# Unembedding for cosine: model.W_U is [d_model, vocab] in HookedTransformer; we want columns.
W_U = model.W_U.float() # [d_model, vocab]
neurons_out = []
polysemy_scores = []
for nid in top_neuron_ids:
col = acts[:, nid] # [N_total]
top_vals, top_pos = col.topk(min(examples_per_neuron, n_total))
examples = []
top_token_ids_for_neuron = []
for v, p in zip(top_vals.tolist(), top_pos.tolist()):
pi = prompt_idx[p].item()
# local position within prompt = p - start_offset
offset = sum(t.shape[0] for t in all_token_ids[:pi])
local_pos = p - offset
tok_seq = all_token_ids[pi]
tok_id = int(tok_seq[local_pos].item())
top_token_ids_for_neuron.append(tok_id)
# build context window with marked token
lo = max(0, local_pos - 4)
hi = min(tok_seq.shape[0], local_pos + 5)
pre = model.to_string(tok_seq[lo:local_pos]) if local_pos > lo else ""
tok_str = model.to_string(tok_seq[local_pos:local_pos+1])
post = model.to_string(tok_seq[local_pos+1:hi]) if hi > local_pos+1 else ""
examples.append({
"token": tok_str,
"activation": round(float(v), 4),
"context": f"{pre}«{tok_str}»{post}",
})
# Polysemy: mean pairwise cosine similarity of unembed rows of top tokens.
# Lower mean cos => tokens are spread out in embedding space => polysemantic.
tt = torch.tensor(top_token_ids_for_neuron, device=device)
embs = W_U[:, tt].t() # [N_top, d_model]
embs_n = embs / embs.norm(dim=-1, keepdim=True).clamp_min(1e-9)
sim_matrix = embs_n @ embs_n.t() # [N_top, N_top]
n_top = sim_matrix.shape[0]
if n_top > 1:
# take upper triangle excluding diagonal
mask = torch.triu(torch.ones_like(sim_matrix), diagonal=1).bool()
mean_cos = sim_matrix[mask].mean().item()
else:
mean_cos = 1.0
poly = max(0.0, 1.0 - mean_cos)
polysemy_scores.append(poly)
neurons_out.append({
"neuron": int(nid),
"maxActivation": round(float(max_per_neuron[nid].item()), 4),
"meanActivation": round(float(mean_per_neuron[nid].item()), 4),
"fireRate": round(float(fire_per_neuron[nid].item()), 4),
"polysemyScore": round(poly, 4),
"semanticCoherence": round(mean_cos, 4),
"examples": examples,
})
mean_polysemy = sum(polysemy_scores) / max(1, len(polysemy_scores))
return jsonify({
"success": True,
"device": str(device),
"layer": layer,
"totalNeurons": int(d_mlp),
"totalTokens": int(n_total),
"meanFireRate": round(mean_fire_rate, 4),
"meanPolysemy": round(mean_polysemy, 4),
"neurons": neurons_out,
})
except Exception as e:
print(f"[SuperpositionAnalysis] Error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/data-leakage-scanner/analyze', methods=['POST'])
def data_leakage_scanner_analyze():
"""
Detect probable training-data memorization for each candidate string.
Two signals per candidate:
1. Teacher-forcing perplexity, compared to a fixed-baseline perplexity (computed on a
neutral reference text on the same model). A candidate ppl significantly below
baseline indicates the model finds it abnormally predictable.
2. Exact-completion test: split the candidate at the midpoint (token-wise), feed the
prefix, greedily generate `len(suffix)` tokens, count exact-token matches against
the held-out suffix.
Combined leakage score = 0.5*ppl_norm + 0.5*exact_ratio, where
ppl_norm = clip(1 - ppl/baseline, 0, 1) (capped so score in [0,1]).
Verdicts: >=0.6 likely-memorized, >=0.35 suspicious, else safe.
Request: { model, candidates:[..] }
Response: { success, device, baselinePerplexity,
candidates:[{candidate,tokens,meanNll,perplexity,prefixTokens,suffixTokens,
exactMatchTokens,exactMatchRatio,generatedSuffix,expectedSuffix,
leakageScore,verdict}] }
"""
try:
data = request.json or {}
candidates = [c for c in (data.get('candidates') or []) if isinstance(c, str) and c.strip()][:20]
if not candidates:
return jsonify({"error": "candidates required"}), 400
model_id = data.get('model', 'gpt2-small')
if model_id in ACTIVATION_ONLY_MODELS:
model = _load_model_only(model_id)
else:
model, _ = get_model_and_sae(model_id)
device = next(model.parameters()).device
def compute_nll(text):
tokens = model.to_tokens(text).to(device)
if tokens.shape[1] < 2:
return None, 0
if tokens.shape[1] > 256:
tokens = tokens[:, :256]
with torch.no_grad():
logits = model(tokens)
log_probs = torch.log_softmax(logits[0, :-1, :].float(), dim=-1)
targets = tokens[0, 1:]
token_lp = log_probs.gather(-1, targets.unsqueeze(-1)).squeeze(-1)
return -token_lp.mean().item(), tokens.shape[1] - 1
# Baseline: fixed neutral reference text. Average a few sentences for stability.
baseline_refs = [
"The garden contained several plants of various sizes and colors arranged along the gravel path.",
"She decided to take the longer route home because she wanted some time to think things over.",
"The committee will reconvene next week to review the proposed changes to the operating procedures.",
]
baseline_nlls = []
for ref in baseline_refs:
n, _ = compute_nll(ref)
if n is not None:
baseline_nlls.append(n)
baseline_mean_nll = sum(baseline_nlls) / max(1, len(baseline_nlls)) if baseline_nlls else 5.0
baseline_ppl = float(torch.tensor(baseline_mean_nll).exp().item())
results = []
for cand in candidates:
try:
tokens = model.to_tokens(cand).to(device)
if tokens.shape[1] < 4:
results.append({
"candidate": cand, "tokens": int(tokens.shape[1]),
"meanNll": 0.0, "perplexity": 0.0,
"prefixTokens": 0, "suffixTokens": 0,
"exactMatchTokens": 0, "exactMatchRatio": 0.0,
"generatedSuffix": "", "expectedSuffix": "",
"leakageScore": 0.0, "verdict": "too-short",
})
continue
if tokens.shape[1] > 256:
tokens = tokens[:, :256]
# 1) NLL / perplexity on full candidate
with torch.no_grad():
logits_full = model(tokens)
log_probs = torch.log_softmax(logits_full[0, :-1, :].float(), dim=-1)
targets = tokens[0, 1:]
token_lp = log_probs.gather(-1, targets.unsqueeze(-1)).squeeze(-1)
mean_nll = -token_lp.mean().item()
ppl = float(torch.tensor(mean_nll).exp().item())
del logits_full, log_probs
# 2) Exact-completion test
T = tokens.shape[1]
split = T // 2
prefix_ids = tokens[:, :split]
suffix_ids = tokens[0, split:]
gen_len = suffix_ids.shape[0]
cur = prefix_ids.clone()
generated = []
with torch.no_grad():
for _ in range(gen_len):
out = model(cur)
nxt = out[0, -1, :].argmax().item()
generated.append(nxt)
cur = torch.cat([cur, torch.tensor([[nxt]], device=device)], dim=1)
exact_matches = sum(1 for a, b in zip(generated, suffix_ids.tolist()) if a == b)
exact_ratio = exact_matches / max(1, gen_len)
expected_suffix = model.to_string(suffix_ids)
generated_suffix = model.to_string(torch.tensor(generated, device=device))
# 3) Leakage score
ppl_norm = max(0.0, min(1.0, 1.0 - ppl / max(baseline_ppl, 1e-9)))
leakage = 0.5 * ppl_norm + 0.5 * exact_ratio
if leakage >= 0.6:
verdict = "likely-memorized"
elif leakage >= 0.35:
verdict = "suspicious"
else:
verdict = "safe"
results.append({
"candidate": cand,
"tokens": int(T),
"meanNll": round(mean_nll, 4),
"perplexity": round(ppl, 3),
"prefixTokens": int(split),
"suffixTokens": int(gen_len),
"exactMatchTokens": int(exact_matches),
"exactMatchRatio": round(exact_ratio, 4),
"generatedSuffix": generated_suffix,
"expectedSuffix": expected_suffix,
"leakageScore": round(leakage, 4),
"verdict": verdict,
})
except Exception as inner:
print(f"[DataLeakageScanner] skip candidate: {inner}")
results.append({
"candidate": cand, "tokens": 0, "meanNll": 0.0, "perplexity": 0.0,
"prefixTokens": 0, "suffixTokens": 0, "exactMatchTokens": 0, "exactMatchRatio": 0.0,
"generatedSuffix": "", "expectedSuffix": "",
"leakageScore": 0.0, "verdict": "error",
})
return jsonify({
"success": True,
"device": str(device),
"baselinePerplexity": round(baseline_ppl, 3),
"candidates": results,
})
except Exception as e:
print(f"[DataLeakageScanner] Error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/multilingual-parity/analyze', methods=['POST'])
def multilingual_parity_analyze():
"""
Compute per-language mean perplexity and a parity ratio.
For each (language, prompts) pack:
- For each prompt, tokenize, forward pass, gather log-prob of each next token (teacher-forcing),
mean negative log-likelihood = -mean(log P(token_t | tokens_<t)) over t=1..T-1.
- Perplexity = exp(mean_nll).
parityRatio = max_lang_ppl / min_lang_ppl. 1.0 == perfect parity.
Request: { model, languages: [{language, prompts:[..]}] }
Response: { success, device, perLanguage:[{language,prompts,meanNll,meanPerplexity,meanTokens}],
parityRatio, bestLanguage, worstLanguage, totalPrompts }
"""
try:
data = request.json or {}
languages = data.get('languages') or []
if len(languages) < 2:
return jsonify({"error": "at least 2 languages required"}), 400
model_id = data.get('model', 'gpt2-small')
if model_id in ACTIVATION_ONLY_MODELS:
model = _load_model_only(model_id)
else:
model, _ = get_model_and_sae(model_id)
device = next(model.parameters()).device
per_language = []
total_prompts = 0
for pack in languages[:12]:
lang = (pack.get('language') or '').strip()
prompts = [p for p in (pack.get('prompts') or []) if isinstance(p, str) and p.strip()][:30]
if not lang or not prompts:
continue
nlls = []
tok_counts = []
for p in prompts:
try:
tokens = model.to_tokens(p).to(device)
if tokens.shape[1] < 2:
continue
if tokens.shape[1] > 256:
tokens = tokens[:, :256]
with torch.no_grad():
logits = model(tokens) # [1, T, V]
# Teacher-forcing nll on positions 1..T-1
log_probs = torch.log_softmax(logits[0, :-1, :].float(), dim=-1) # [T-1, V]
targets = tokens[0, 1:] # [T-1]
token_lp = log_probs.gather(-1, targets.unsqueeze(-1)).squeeze(-1) # [T-1]
nll = -token_lp.mean().item()
nlls.append(nll)
tok_counts.append(tokens.shape[1] - 1)
total_prompts += 1
del logits, log_probs
except Exception as inner:
print(f"[MultilingualParity] skip {lang!r} prompt: {inner}")
if not nlls:
continue
mean_nll = sum(nlls) / len(nlls)
mean_tok = sum(tok_counts) / len(tok_counts)
per_language.append({
"language": lang,
"prompts": len(nlls),
"meanNll": round(mean_nll, 4),
"meanPerplexity": round(float(torch.tensor(mean_nll).exp().item()), 3),
"meanTokens": round(mean_tok, 1),
})
if len(per_language) < 2:
return jsonify({"error": "fewer than 2 languages produced valid results"}), 400
ppls = [(l['language'], l['meanPerplexity']) for l in per_language]
best = min(ppls, key=lambda x: x[1])
worst = max(ppls, key=lambda x: x[1])
parity = worst[1] / max(best[1], 1e-9)
return jsonify({
"success": True,
"device": str(device),
"perLanguage": per_language,
"parityRatio": round(parity, 3),
"bestLanguage": best[0],
"worstLanguage": worst[0],
"totalPrompts": total_prompts,
})
except Exception as e:
print(f"[MultilingualParity] Error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/latency-profiler/analyze', methods=['POST'])
def latency_profiler_analyze():
"""
Profile end-to-end and per-block forward-pass latency.
Pipeline:
1. Tokenize prompt; do 1 warmup pass (always discarded).
2. For each trial:
a. Time prefill (forward on full prompt) with cuda.synchronize() if CUDA.
b. Greedily generate `generatedTokens` extra tokens; time the whole loop.
c. Hook every `blocks.{l}` to measure per-block time on a single forward pass:
install pre-/post-hooks recording perf_counter; do one extra forward.
3. Aggregate: mean ± std end-to-end ms, mean per-layer ms, per-stage breakdown
(prefill / generation / per-token-mean / overhead).
Request: { model, prompt, generatedTokens, trials }
Response: { success, device, prompt, promptTokens, generatedTokens, trials,
totalMeanMs, totalStdMs, tokensPerSec,
perLayer:[{layer, meanMs, stdMs}],
stages:[{stage, meanMs, pct}] }
"""
try:
import time
data = request.json or {}
prompt = (data.get('prompt') or '').strip()
if not prompt:
return jsonify({"error": "prompt is required"}), 400
gen_tokens = int(data.get('generatedTokens') or 20)
gen_tokens = max(1, min(50, gen_tokens))
trials = int(data.get('trials') or 5)
trials = max(1, min(15, trials))
model_id = data.get('model', 'gpt2-small')
if model_id in ACTIVATION_ONLY_MODELS:
model = _load_model_only(model_id)
else:
model, _ = get_model_and_sae(model_id)
device = next(model.parameters()).device
n_layers = model.cfg.n_layers
is_cuda = device.type == 'cuda'
def sync():
if is_cuda:
torch.cuda.synchronize()
tokens = model.to_tokens(prompt).to(device)
prompt_len = tokens.shape[1]
if prompt_len > 64:
return jsonify({"error": f"prompt too long ({prompt_len} tokens); max 64 for latency profiling"}), 400
# Warmup (discarded).
with torch.no_grad():
_ = model(tokens)
sync()
prefill_times = []
gen_times = []
total_times = []
for _ in range(trials):
sync(); t0 = time.perf_counter()
with torch.no_grad():
_ = model(tokens)
sync(); t1 = time.perf_counter()
prefill_times.append((t1 - t0) * 1000.0)
sync(); t0 = time.perf_counter()
cur = tokens.clone()
with torch.no_grad():
for _ in range(gen_tokens):
out = model(cur)
next_id = out[0, -1, :].argmax().item()
cur = torch.cat([cur, torch.tensor([[next_id]], device=device)], dim=1)
sync(); t1 = time.perf_counter()
gen_times.append((t1 - t0) * 1000.0)
total_times.append(prefill_times[-1] + gen_times[-1])
# Per-layer profiling via hooks (single forward pass per trial).
per_layer_acc = [[] for _ in range(n_layers)]
for _ in range(trials):
layer_starts = [0.0] * n_layers
layer_durations = [0.0] * n_layers
handles = []
def make_pre(l_idx):
def _pre(module, inputs):
sync()
layer_starts[l_idx] = time.perf_counter()
return _pre
def make_post(l_idx):
def _post(module, inputs, outputs):
sync()
layer_durations[l_idx] = (time.perf_counter() - layer_starts[l_idx]) * 1000.0
return _post
for l in range(n_layers):
block = model.blocks[l]
handles.append(block.register_forward_pre_hook(make_pre(l)))
handles.append(block.register_forward_hook(make_post(l)))
with torch.no_grad():
_ = model(tokens)
for h in handles:
h.remove()
for l in range(n_layers):
per_layer_acc[l].append(layer_durations[l])
def mean_std(xs):
if not xs:
return 0.0, 0.0
m = sum(xs) / len(xs)
v = sum((x - m) ** 2 for x in xs) / max(1, len(xs))
return m, v ** 0.5
per_layer = []
for l in range(n_layers):
m, s = mean_std(per_layer_acc[l])
per_layer.append({"layer": l, "meanMs": round(m, 4), "stdMs": round(s, 4)})
total_mean, total_std = mean_std(total_times)
prefill_mean, _ = mean_std(prefill_times)
gen_mean, _ = mean_std(gen_times)
per_token_mean = gen_mean / gen_tokens if gen_tokens > 0 else 0.0
overhead = max(0.0, total_mean - prefill_mean - gen_mean)
total_for_pct = max(1e-9, total_mean)
stages = [
{"stage": "prefill", "meanMs": round(prefill_mean, 3), "pct": round(prefill_mean / total_for_pct * 100, 2)},
{"stage": "generation total", "meanMs": round(gen_mean, 3), "pct": round(gen_mean / total_for_pct * 100, 2)},
{"stage": "per-gen-token avg", "meanMs": round(per_token_mean, 3), "pct": round(per_token_mean / total_for_pct * 100, 2)},
{"stage": "overhead", "meanMs": round(overhead, 3), "pct": round(overhead / total_for_pct * 100, 2)},
]
tokens_per_sec = (gen_tokens / (gen_mean / 1000.0)) if gen_mean > 0 else 0.0
return jsonify({
"success": True,
"device": str(device),
"prompt": prompt,
"promptTokens": prompt_len,
"generatedTokens": gen_tokens,
"trials": trials,
"totalMeanMs": round(total_mean, 3),
"totalStdMs": round(total_std, 3),
"tokensPerSec": round(tokens_per_sec, 2),
"perLayer": per_layer,
"stages": stages,
})
except Exception as e:
print(f"[LatencyProfiler] Error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/drift-monitor/analyze', methods=['POST'])
def drift_monitor_analyze():
"""
Measure representational drift between two prompt distributions (e.g., baseline vs canary).
Algorithm:
1. For each prompt in set A and set B, run forward pass with cache; per layer take the
last-token residual stream (a sentence summary vector). Average within set → mean_a[l], mean_b[l].
2. Per-layer drift = cosine_distance(mean_a[l], mean_b[l]) and L2 distance.
3. Output drift = JS divergence between the mean next-token softmax of set A and set B.
4. Top shifted tokens = top-K tokens by |mean_prob_a - mean_prob_b|.
Request: { model, promptsA, promptsB, labelA, labelB }
Response: { success, labelA, labelB, nA, nB, nLayers,
perLayerCosineDistance[], perLayerL2Distance[],
meanCosineDistance, outputJSDivergence,
topShiftedTokens:[{token, deltaProb, probA, probB}] }
"""
try:
data = request.json or {}
prompts_a = [p for p in (data.get('promptsA') or []) if isinstance(p, str) and p.strip()][:50]
prompts_b = [p for p in (data.get('promptsB') or []) if isinstance(p, str) and p.strip()][:50]
if not prompts_a or not prompts_b:
return jsonify({"error": "Need at least 1 prompt in each set"}), 400
label_a = (data.get('labelA') or 'A')[:64]
label_b = (data.get('labelB') or 'B')[:64]
model_id = data.get('model', 'gpt2-small')
if model_id in ACTIVATION_ONLY_MODELS:
model = _load_model_only(model_id)
else:
model, _ = get_model_and_sae(model_id)
device = next(model.parameters()).device
n_layers = model.cfg.n_layers
d_model = model.cfg.d_model
def encode_set(prompts):
"""Returns (per_layer_mean[L, d_model], mean_probs[vocab], per_prompt_probs[N, vocab])."""
sums = [torch.zeros(d_model, device=device) for _ in range(n_layers)]
counts = 0
prob_sum = None
per_prompt_probs = [] # store each prompt's softmax for per-prompt JS
for p in prompts:
try:
tokens = model.to_tokens(p).to(device)
if tokens.shape[1] > 128:
tokens = tokens[:, :128]
with torch.no_grad():
logits, cache = model.run_with_cache(tokens)
for l in range(n_layers):
last_resid = cache[f'blocks.{l}.hook_resid_post'][0, -1, :]
sums[l] = sums[l] + last_resid
probs = torch.softmax(logits[0, -1, :].float(), dim=-1)
per_prompt_probs.append(probs.detach().clone())
prob_sum = probs if prob_sum is None else prob_sum + probs
counts += 1
del logits, cache
except Exception as inner:
print(f"[DriftMonitor] skip prompt: {inner}")
continue
if counts == 0:
return None, None, None, 0
means = torch.stack([s / counts for s in sums], dim=0) # [L, d]
mean_probs = prob_sum / counts
return means, mean_probs, per_prompt_probs, counts
mean_a, probs_a, prompts_probs_a, n_a = encode_set(prompts_a)
mean_b, probs_b, prompts_probs_b, n_b = encode_set(prompts_b)
if mean_a is None or mean_b is None:
return jsonify({"error": "All prompts failed to encode"}), 400
cos = torch.nn.functional.cosine_similarity(mean_a, mean_b, dim=-1) # [L]
cos_dist = (1.0 - cos).cpu().tolist()
l2 = torch.norm(mean_a - mean_b, dim=-1).cpu().tolist()
mean_cos = sum(cos_dist) / len(cos_dist)
eps = 1e-12
def js_div(p, q):
m = 0.5 * (p + q)
kl_pm = (p * (p.clamp_min(eps).log() - m.clamp_min(eps).log())).sum().item()
kl_qm = (q * (q.clamp_min(eps).log() - m.clamp_min(eps).log())).sum().item()
return 0.5 * (kl_pm + kl_qm)
# JS of mean softmaxes — kept for reference but uninformative when
# averaging many peaked distributions (both means converge to the English
# unigram marginal regardless of topic, so this is near-zero by design).
js_of_means = js_div(probs_a, probs_b)
# MEANINGFUL drift metric: average per-prompt JS divergence to the OTHER
# set's centroid. Because each prompt's softmax is sharply peaked on a
# different token, this captures real distributional drift.
per_prompt_js_a = [js_div(p, probs_b) for p in prompts_probs_a]
per_prompt_js_b = [js_div(p, probs_a) for p in prompts_probs_b]
all_js = per_prompt_js_a + per_prompt_js_b
js = sum(all_js) / len(all_js) if all_js else 0.0
# Top shifted tokens.
delta = (probs_a - probs_b).cpu()
top_k = 12
abs_delta = delta.abs()
top_idx = torch.topk(abs_delta, top_k).indices.tolist()
top_shifted = []
for tid in top_idx:
try:
tok_str = model.to_string(torch.tensor([tid]))
except Exception:
tok_str = f"[id={tid}]"
top_shifted.append({
"token": tok_str,
"deltaProb": round(float(delta[tid].item()), 6),
"probA": round(float(probs_a[tid].item()), 6),
"probB": round(float(probs_b[tid].item()), 6),
})
return jsonify({
"success": True,
"labelA": label_a,
"labelB": label_b,
"nA": n_a,
"nB": n_b,
"nLayers": n_layers,
"perLayerCosineDistance": [round(x, 6) for x in cos_dist],
"perLayerL2Distance": [round(x, 4) for x in l2],
"meanCosineDistance": round(mean_cos, 6),
"outputJSDivergence": round(float(js), 6),
"outputJSDivergenceOfMeans": round(float(js_of_means), 6),
"topShiftedTokens": top_shifted,
})
except Exception as e:
print(f"[DriftMonitor] Error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/robustness-testing/analyze', methods=['POST'])
def robustness_testing_analyze():
"""
Measure representational stability under input perturbations.
Algorithm:
1. Run clean prompt through HookedTransformer with cache; record per-layer
residual-stream activations (resid_post for each block).
2. For k = 1..nSamples: generate a perturbed version of the prompt
(char-noise / swap-adjacent / delete) at noiseRate, run with cache.
3. For each layer, compute mean cosine distance between clean and perturbed
residual streams (averaged over token positions where both are valid).
Take min(clean_len, pert_len) positions.
4. Compute per-sample mean layer distance and output-KL between clean and
perturbed final next-token softmax.
5. Aggregate to perLayerMeanDistance and meanOverallDistance.
Request: { model, prompt, perturbationType, noiseRate, nSamples }
Response: { success, prompt, perturbationType, noiseRate, nSamples, nLayers,
perLayerMeanDistance[], meanOverallDistance, meanOutputKL,
samples[{perturbed, meanLayerDistance, outputKL}] }
"""
try:
import random
import string
data = request.json or {}
prompt = (data.get('prompt') or '').strip()
if not prompt:
return jsonify({"error": "prompt is required"}), 400
perturbation_type = data.get('perturbationType', 'char')
if perturbation_type not in ('char', 'swap', 'delete'):
perturbation_type = 'char'
noise_rate = float(data.get('noiseRate') or 0.1)
noise_rate = max(0.01, min(0.5, noise_rate))
n_samples = int(data.get('nSamples') or 5)
n_samples = max(1, min(20, n_samples))
model_id = data.get('model', 'gpt2-small')
if model_id in ACTIVATION_ONLY_MODELS:
model = _load_model_only(model_id)
else:
model, _ = get_model_and_sae(model_id)
device = next(model.parameters()).device
n_layers = model.cfg.n_layers
def perturb(s: str, ptype: str, rate: float) -> str:
chars = list(s)
n_changes = max(1, int(len(chars) * rate))
indices = random.sample(range(len(chars)), min(n_changes, len(chars)))
if ptype == 'char':
for i in indices:
chars[i] = random.choice(string.ascii_lowercase + ' ')
elif ptype == 'swap':
for i in indices:
if i + 1 < len(chars):
chars[i], chars[i + 1] = chars[i + 1], chars[i]
elif ptype == 'delete':
for i in sorted(indices, reverse=True):
if len(chars) > 1:
del chars[i]
return ''.join(chars)
# Clean forward pass with cache for residual streams.
clean_tokens = model.to_tokens(prompt).to(device)
if clean_tokens.shape[1] > 128:
return jsonify({"error": f"prompt too long ({clean_tokens.shape[1]} tokens); max 128 for robustness testing"}), 400
with torch.no_grad():
clean_logits, clean_cache = model.run_with_cache(clean_tokens)
clean_resids = [clean_cache[f'blocks.{l}.hook_resid_post'][0] for l in range(n_layers)] # each [T, d]
clean_T = clean_resids[0].shape[0]
clean_probs = torch.softmax(clean_logits[0, -1, :].float(), dim=-1)
per_layer_sums = [0.0] * n_layers
per_layer_counts = [0] * n_layers
sample_results = []
all_kls = []
for s_idx in range(n_samples):
perturbed = perturb(prompt, perturbation_type, noise_rate)
try:
pert_tokens = model.to_tokens(perturbed).to(device)
with torch.no_grad():
pert_logits, pert_cache = model.run_with_cache(pert_tokens)
pert_T = pert_tokens.shape[1]
T = min(clean_T, pert_T)
sample_layer_dists = []
for l in range(n_layers):
pert_resid = pert_cache[f'blocks.{l}.hook_resid_post'][0][:T]
clean_resid = clean_resids[l][:T]
cos_sim = torch.nn.functional.cosine_similarity(clean_resid, pert_resid, dim=-1)
cos_dist = (1.0 - cos_sim).mean().item()
per_layer_sums[l] += cos_dist
per_layer_counts[l] += 1
sample_layer_dists.append(cos_dist)
pert_probs = torch.softmax(pert_logits[0, -1, :].float(), dim=-1)
kl = torch.nn.functional.kl_div(
pert_probs.clamp_min(1e-12).log(),
clean_probs,
reduction='sum',
).item()
all_kls.append(kl)
mean_layer_dist = sum(sample_layer_dists) / len(sample_layer_dists)
sample_results.append({
"perturbed": perturbed,
"meanLayerDistance": round(mean_layer_dist, 6),
"outputKL": round(float(kl), 6),
})
del pert_logits, pert_cache
except Exception as inner:
print(f"[RobustnessTesting] sample {s_idx} skipped: {inner}")
continue
per_layer_mean = [
round(per_layer_sums[l] / per_layer_counts[l], 6) if per_layer_counts[l] > 0 else 0.0
for l in range(n_layers)
]
mean_overall = sum(per_layer_mean) / len(per_layer_mean) if per_layer_mean else 0.0
mean_kl = sum(all_kls) / len(all_kls) if all_kls else 0.0
return jsonify({
"success": True,
"prompt": prompt,
"perturbationType": perturbation_type,
"noiseRate": noise_rate,
"nSamples": len(sample_results),
"nLayers": n_layers,
"perLayerMeanDistance": per_layer_mean,
"meanOverallDistance": round(mean_overall, 6),
"meanOutputKL": round(mean_kl, 6),
"samples": sample_results,
})
except Exception as e:
print(f"[RobustnessTesting] Error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/bias-probing/analyze', methods=['POST'])
def bias_probing_analyze():
"""
Behavioral bias probe via paired-prompt next-token logit/prob deltas.
For each pair {promptA, promptB} and each target token (first sub-token):
- Compute next-token logit and softmax prob at last position of promptA → (logitA, probA).
- Same for promptB → (logitB, probB).
- deltaProb = probA - probB; deltaLogit = logitA - logitB.
Aggregate per-target across pairs (mean deltaProb, mean |deltaProb|, mean deltaLogit)
and compute global mean |deltaProb|.
Request: { model, pairs:[{promptA,promptB}], targets:[str], labelA, labelB }
Response: { success, labelA, labelB, total, meanAbsDeltaProb,
perTarget:[{target, meanDeltaProb, meanAbsDeltaProb, meanDeltaLogit}],
pairs:[{pairIndex, promptA, promptB, results:[{target,probA,probB,deltaProb,logitA,logitB,deltaLogit}]}] }
"""
try:
data = request.json or {}
pairs_in = data.get('pairs') or []
targets_in = data.get('targets') or []
if not isinstance(pairs_in, list) or len(pairs_in) < 1 or not isinstance(targets_in, list) or len(targets_in) < 1:
return jsonify({"error": "Need at least 1 pair and 1 target"}), 400
pairs_in = pairs_in[:50]
targets_in = targets_in[:50]
label_a = (data.get('labelA') or 'A')[:64]
label_b = (data.get('labelB') or 'B')[:64]
model_id = data.get('model', 'gpt2-small')
if model_id in ACTIVATION_ONLY_MODELS:
model = _load_model_only(model_id)
else:
model, _ = get_model_and_sae(model_id)
# Resolve target tokens (first sub-token of each).
target_specs = []
for t in targets_in:
try:
ids = model.to_tokens(t, prepend_bos=False)[0]
if ids.numel() == 0:
continue
tid = int(ids[0].item())
target_specs.append({"text": t, "id": tid})
except Exception:
continue
if not target_specs:
return jsonify({"error": "No targets could be tokenized"}), 400
# Cache distributions per unique prompt to avoid duplicate forward passes.
prompt_cache = {}
def get_dist(prompt: str):
if prompt in prompt_cache:
return prompt_cache[prompt]
tokens = model.to_tokens(prompt)
with torch.no_grad():
logits = model(tokens)
last_logits = logits[0, -1, :]
probs = torch.softmax(last_logits, dim=-1)
prompt_cache[prompt] = (last_logits, probs)
return prompt_cache[prompt]
pair_results = []
per_target_acc = {ts["text"]: {"deltaProb": [], "deltaLogit": []} for ts in target_specs}
all_abs_deltas = []
for pi, p in enumerate(pairs_in):
pa = (p.get('promptA') or '').strip()
pb = (p.get('promptB') or '').strip()
if not pa or not pb:
continue
try:
logits_a, probs_a = get_dist(pa)
logits_b, probs_b = get_dist(pb)
except Exception as inner:
print(f"[BiasProbing] skip pair {pi}: {inner}")
continue
results = []
for ts in target_specs:
tid = ts["id"]
la = float(logits_a[tid].item())
lb = float(logits_b[tid].item())
pa_prob = float(probs_a[tid].item())
pb_prob = float(probs_b[tid].item())
d_prob = pa_prob - pb_prob
d_logit = la - lb
results.append({
"target": ts["text"],
"probA": round(pa_prob, 6),
"probB": round(pb_prob, 6),
"deltaProb": round(d_prob, 6),
"logitA": round(la, 4),
"logitB": round(lb, 4),
"deltaLogit": round(d_logit, 4),
})
per_target_acc[ts["text"]]["deltaProb"].append(d_prob)
per_target_acc[ts["text"]]["deltaLogit"].append(d_logit)
all_abs_deltas.append(abs(d_prob))
pair_results.append({
"pairIndex": pi,
"promptA": pa,
"promptB": pb,
"results": results,
})
per_target = []
for ts in target_specs:
arr = per_target_acc[ts["text"]]["deltaProb"]
arrl = per_target_acc[ts["text"]]["deltaLogit"]
if not arr:
continue
per_target.append({
"target": ts["text"],
"meanDeltaProb": round(sum(arr) / len(arr), 6),
"meanAbsDeltaProb": round(sum(abs(x) for x in arr) / len(arr), 6),
"meanDeltaLogit": round(sum(arrl) / len(arrl), 4),
})
total = sum(len(p["results"]) for p in pair_results)
mean_abs = sum(all_abs_deltas) / len(all_abs_deltas) if all_abs_deltas else 0.0
return jsonify({
"success": True,
"labelA": label_a,
"labelB": label_b,
"total": total,
"meanAbsDeltaProb": round(mean_abs, 6),
"perTarget": per_target,
"pairs": pair_results,
})
except Exception as e:
print(f"[BiasProbing] Error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/gradient-attribution/analyze', methods=['POST'])
def gradient_attribution_analyze():
"""
Integrated-gradients attribution of a target token's logit back to each
input token's embedding.
Algorithm:
1. Tokenize prompt; capture input embeddings via hook_embed.
2. Resolve target_token_id (user-supplied first token of `target`, else
argmax of next-token distribution).
3. For k = 1..steps: alpha = k/steps; replace embeddings with
(alpha * embeds) using a forward hook; compute target_logit at last
position; backprop to get d(target_logit)/d(scaled_embeds); accumulate.
4. avg_grad = sum / steps. Integrated grad attribution = embeds * avg_grad
(zero baseline). Per-token signed attribution = sum over embed dim.
Saliency = abs sum.
Request: { model, prompt, target?, steps }
Response: { success, prompt, targetToken, targetProb, topPredicted{token,prob},
steps, tokens:[{position,token,attribution,saliency}], maxAbsAttribution }
"""
try:
data = request.json or {}
prompt = (data.get('prompt') or '').strip()
if not prompt:
return jsonify({"error": "prompt is required"}), 400
steps = int(data.get('steps') or 20)
steps = max(1, min(50, steps))
model_id = data.get('model', 'gpt2-small')
user_target = data.get('target')
if model_id in ACTIVATION_ONLY_MODELS:
model = _load_model_only(model_id)
else:
model, _ = get_model_and_sae(model_id)
device = next(model.parameters()).device
model.eval()
tokens = model.to_tokens(prompt).to(device)
seq_len = tokens.shape[1]
if seq_len > 64:
return jsonify({"error": f"prompt too long ({seq_len} tokens); max 64 for gradient attribution"}), 400
token_strs = [model.to_string([int(t)]) for t in tokens[0]]
# Capture clean embeddings
clean_emb = {}
def cap_hook(act, hook):
clean_emb['e'] = act.detach().clone()
return act
with torch.no_grad():
logits_clean = model.run_with_hooks(tokens, fwd_hooks=[('hook_embed', cap_hook)])
last_logits = logits_clean[0, -1, :]
probs = torch.softmax(last_logits, dim=-1)
top_id = int(torch.argmax(probs).item())
top_prob = float(probs[top_id].item())
top_str = model.to_string([top_id])
if user_target is not None and len(user_target) > 0:
tgt_ids = model.to_tokens(user_target, prepend_bos=False)[0]
if tgt_ids.numel() == 0:
return jsonify({"error": "target tokenized to empty sequence"}), 400
target_id = int(tgt_ids[0].item())
else:
target_id = top_id
target_str = model.to_string([target_id])
target_prob = float(probs[target_id].item())
embeds = clean_emb['e'] # [1, T, d_model]
# Integrated gradients along straight-line path from zero baseline to embeds.
accum = torch.zeros_like(embeds)
for k in range(1, steps + 1):
alpha = k / steps
# Build a fresh tensor each step that requires grad.
scaled = (alpha * embeds).clone().detach().requires_grad_(True)
captured_for_grad = [scaled]
def replace_hook(act, hook):
return captured_for_grad[0]
logits = model.run_with_hooks(
tokens,
fwd_hooks=[('hook_embed', replace_hook)],
)
target_logit = logits[0, -1, target_id]
grad = torch.autograd.grad(target_logit, scaled, retain_graph=False, create_graph=False)[0]
accum = accum + grad.detach()
del scaled, logits, target_logit, grad
avg_grad = accum / steps
ig = embeds * avg_grad # [1, T, d_model]
per_token_signed = ig.sum(dim=-1)[0] # [T]
per_token_saliency = ig.abs().sum(dim=-1)[0] # [T]
per_token_signed_l = per_token_signed.detach().cpu().tolist()
per_token_saliency_l = per_token_saliency.detach().cpu().tolist()
max_abs = max((abs(x) for x in per_token_signed_l), default=1.0)
token_results = []
for i in range(seq_len):
token_results.append({
"position": i,
"token": token_strs[i],
"attribution": round(per_token_signed_l[i], 6),
"saliency": round(per_token_saliency_l[i], 6),
})
return jsonify({
"success": True,
"prompt": prompt,
"targetToken": target_str,
"targetProb": round(target_prob, 6),
"topPredicted": {"token": top_str, "prob": round(top_prob, 6)},
"steps": steps,
"tokens": token_results,
"maxAbsAttribution": round(float(max_abs), 6),
})
except Exception as e:
print(f"[GradientAttribution] Error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/confidence-calibration/analyze', methods=['POST'])
def confidence_calibration_analyze():
"""
Compute Expected Calibration Error (ECE) and reliability bins for a model.
Algorithm:
1. For each {prompt, target} pair:
- Forward pass on prompt; take next-token distribution at last position.
- target_first_token_id = first token of `target` when tokenized.
- predictedToken = argmax of distribution; predictedProb = its prob.
- targetProb = prob assigned to target_first_token_id.
- correct = (predictedToken_id == target_first_token_id)
2. Bin examples by predictedProb into nBins equal-width bins on [0,1].
For each bin: avgConfidence = mean(predictedProb), accuracy = mean(correct).
3. ECE = sum_bins (n_bin / N) * |avgConfidence - accuracy|
Request: { model, pairs: [{prompt, target}], nBins }
Response: { success, total, ece, meanConfidence, accuracy, bins, examples }
"""
try:
data = request.json or {}
pairs = data.get('pairs') or []
if not isinstance(pairs, list) or len(pairs) < 2:
return jsonify({"error": "pairs must be a list of at least 2 {prompt, target} objects"}), 400
pairs = pairs[:100]
n_bins = int(data.get('nBins') if data.get('nBins') is not None else 10)
n_bins = max(2, min(50, n_bins))
model_id = data.get('model', 'gpt2-small')
if model_id in ACTIVATION_ONLY_MODELS:
model = _load_model_only(model_id)
else:
model, _ = get_model_and_sae(model_id)
examples = []
skipped = 0
for p in pairs:
prompt = (p.get('prompt') or '').strip()
target = p.get('target') or ''
if not prompt or not target:
skipped += 1
continue
try:
target_ids = model.to_tokens(target, prepend_bos=False)[0]
if target_ids.numel() == 0:
skipped += 1
continue
target_first_id = int(target_ids[0].item())
target_first_str = model.to_string([target_first_id])
prompt_tokens = model.to_tokens(prompt)
with torch.no_grad():
logits = model(prompt_tokens)
last_logits = logits[0, -1, :]
probs = torch.softmax(last_logits, dim=-1)
pred_id = int(torch.argmax(probs).item())
pred_prob = float(probs[pred_id].item())
target_prob = float(probs[target_first_id].item())
pred_str = model.to_string([pred_id])
examples.append({
"prompt": prompt,
"target": target,
"predictedToken": pred_str,
"predictedProb": round(pred_prob, 6),
"targetProb": round(target_prob, 6),
"correct": pred_id == target_first_id,
})
except Exception as inner:
print(f"[ConfidenceCalibration] skipped pair due to error: {inner}")
skipped += 1
continue
n = len(examples)
if n == 0:
return jsonify({"error": "No pairs could be scored"}), 400
bins = []
ece = 0.0
for i in range(n_bins):
low = i / n_bins
high = (i + 1) / n_bins
in_bin = [e for e in examples if (e["predictedProb"] > low or i == 0) and e["predictedProb"] <= high]
count = len(in_bin)
if count > 0:
avg_conf = sum(e["predictedProb"] for e in in_bin) / count
acc = sum(1 for e in in_bin if e["correct"]) / count
ece += (count / n) * abs(avg_conf - acc)
else:
avg_conf = 0.0
acc = 0.0
bins.append({
"binIndex": i,
"binLow": round(low, 4),
"binHigh": round(high, 4),
"count": count,
"avgConfidence": round(avg_conf, 6),
"accuracy": round(acc, 6),
})
mean_conf = sum(e["predictedProb"] for e in examples) / n
accuracy = sum(1 for e in examples if e["correct"]) / n
return jsonify({
"success": True,
"total": n,
"skipped": skipped,
"ece": round(ece, 6),
"meanConfidence": round(mean_conf, 6),
"accuracy": round(accuracy, 6),
"bins": bins,
"examples": examples,
})
except Exception as e:
print(f"[ConfidenceCalibration] Error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/hallucination-detector/analyze', methods=['POST'])
def hallucination_detector_analyze():
"""
Detect overconfident hallucinations by comparing final-layer output probability
against per-layer logit-lens support for the chosen token.
Algorithm:
1. Greedy-generate up to maxTokens new tokens from the prompt.
2. For each generated token at position i:
- p_out = softmax(W_U · ln_final(resid_final[i]))[t_i]
- For each layer L: p_L = softmax(W_U · ln_final(resid_L[i]))[t_i]
- support_score = mean(p_L for L in last 50% of layers)
- divergence = p_out - support_score
- flagged = (p_out >= confThreshold) and (support_score < supportThreshold)
Request:
{ model, prompt, maxTokens, confThreshold, supportThreshold }
Response:
{ success, prompt, generatedTokens, nLayers, analysis: [...], flaggedCount, meanDivergence }
"""
try:
data = request.json or {}
prompt = data.get('prompt', '').strip()
if not prompt:
return jsonify({"error": "prompt is required"}), 400
model_id = data.get('model', 'gpt2-small')
max_tokens = int(data.get('maxTokens', 8))
max_tokens = max(1, min(20, max_tokens))
conf_threshold = float(data.get('confThreshold') if data.get('confThreshold') is not None else 0.6)
support_threshold = float(data.get('supportThreshold') if data.get('supportThreshold') is not None else 0.3)
if model_id in ACTIVATION_ONLY_MODELS:
model = _load_model_only(model_id)
else:
model, _ = get_model_and_sae(model_id)
n_layers = model.cfg.n_layers
W_U = model.W_U
b_U = model.b_U if hasattr(model, 'b_U') and model.b_U is not None else None
prompt_tokens = model.to_tokens(prompt)
current_tokens = prompt_tokens
prompt_len = current_tokens.shape[1]
analysis = []
generated_strs = []
resid_filter = lambda name: name.endswith("hook_resid_post")
for step in range(max_tokens):
with torch.no_grad():
_, cache = model.run_with_cache(current_tokens, names_filter=resid_filter)
# Final layer prediction at the LAST position
final_resid = cache[f"blocks.{n_layers - 1}.hook_resid_post"][0, -1, :]
if hasattr(model, 'ln_final'):
final_normed = model.ln_final(final_resid.unsqueeze(0)).squeeze(0)
else:
final_normed = final_resid
final_logits = final_normed.float() @ W_U.float()
if b_U is not None:
final_logits = final_logits + b_U.float()
final_probs = torch.softmax(final_logits, dim=-1)
chosen_id = int(final_probs.argmax().item())
p_out = float(final_probs[chosen_id].item())
chosen_str = model.tokenizer.decode([chosen_id])
# Per-layer logit lens prob for the chosen token
layer_probs = []
for layer in range(n_layers):
resid = cache[f"blocks.{layer}.hook_resid_post"][0, -1, :]
if hasattr(model, 'ln_final'):
resid_normed = model.ln_final(resid.unsqueeze(0)).squeeze(0)
else:
resid_normed = resid
logits = resid_normed.float() @ W_U.float()
if b_U is not None:
logits = logits + b_U.float()
probs = torch.softmax(logits, dim=-1)
p_layer = float(probs[chosen_id].item())
if np.isnan(p_layer) or np.isinf(p_layer):
p_layer = 0.0
layer_probs.append(round(p_layer, 4))
del cache
if torch.cuda.is_available():
torch.cuda.empty_cache()
# Support score = mean over last half of layers
half = max(1, n_layers // 2)
late_probs = layer_probs[-half:]
support_score = float(sum(late_probs) / len(late_probs))
divergence = p_out - support_score
flagged = bool(p_out >= conf_threshold and support_score < support_threshold)
analysis.append({
"position": step,
"token": chosen_str,
"outputProb": round(p_out, 4),
"supportScore": round(support_score, 4),
"divergence": round(divergence, 4),
"flagged": flagged,
"layerProbs": layer_probs,
})
generated_strs.append(chosen_str)
# Append chosen token and continue
new_tok = torch.tensor([[chosen_id]], device=current_tokens.device, dtype=current_tokens.dtype)
current_tokens = torch.cat([current_tokens, new_tok], dim=1)
flagged_count = sum(1 for a in analysis if a["flagged"])
mean_div = float(sum(a["divergence"] for a in analysis) / len(analysis)) if analysis else 0.0
return jsonify({
"success": True,
"prompt": prompt,
"generatedTokens": generated_strs,
"nLayers": n_layers,
"analysis": analysis,
"flaggedCount": flagged_count,
"meanDivergence": round(mean_div, 4),
})
except Exception as e:
print(f"[HallucinationDetector] Error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/layer-ablation/analyze', methods=['POST'])
def layer_ablation_analyze():
"""
For a prompt + two candidate tokens (A, B), zero-ablate each layer's
attention output AND each layer's MLP output (one at a time) and measure
how much the logit difference between A and B drops.
This is a coarser-grained version of Direct Logit Attribution — DLA
decomposes the contribution of every individual head/MLP, while this
measures the CAUSAL impact of removing each layer's component entirely
(which captures effects beyond linear projection: downstream feedback,
second-order effects, etc.)
Cost: 2 * n_layers + 1 forward passes. On GPT-2 small that's 25 forwards.
Request: { prompt, tokenA, tokenB }
Response: { success, model, device, baseline:{logitA,logitB,logitDiff},
layers:[{layer, attnAblated:{logitDiff, dropFromBaseline},
mlpAblated:{logitDiff, dropFromBaseline}}] }
"""
try:
data = request.json or {}
prompt = data.get('prompt')
a = data.get('tokenA')
b = data.get('tokenB')
if not prompt or not a or not b:
return jsonify({"error": "prompt, tokenA, tokenB required"}), 400
model = _load_hooked_transformer('gpt2')
device = next(model.parameters()).device
def first_id(s):
ids = model.to_tokens(s, prepend_bos=False)[0]
if ids.numel() == 0:
raise ValueError(f"empty: {s!r}")
return int(ids[0].item())
a_id = first_id(a)
b_id = first_id(b)
def zero_hook(activation, hook):
return torch.zeros_like(activation)
with torch.no_grad():
tokens = model.to_tokens(prompt) # [1, seq] with BOS
n_layers = int(model.cfg.n_layers)
# Baseline
base_logits = model(tokens)[0, -1, :].float().cpu()
base_a = float(base_logits[a_id].item())
base_b = float(base_logits[b_id].item())
base_diff = base_a - base_b
layer_results = []
for layer in range(n_layers):
# Attn ablation
with model.hooks(fwd_hooks=[(f"blocks.{layer}.hook_attn_out", zero_hook)]):
attn_logits = model(tokens)[0, -1, :].float().cpu()
attn_a = float(attn_logits[a_id].item())
attn_b = float(attn_logits[b_id].item())
attn_diff = attn_a - attn_b
# MLP ablation
with model.hooks(fwd_hooks=[(f"blocks.{layer}.hook_mlp_out", zero_hook)]):
mlp_logits = model(tokens)[0, -1, :].float().cpu()
mlp_a = float(mlp_logits[a_id].item())
mlp_b = float(mlp_logits[b_id].item())
mlp_diff = mlp_a - mlp_b
layer_results.append({
"layer": layer,
"attnAblated": {
"logitA": round(attn_a, 4),
"logitB": round(attn_b, 4),
"logitDiff": round(attn_diff, 4),
"dropFromBaseline": round(base_diff - attn_diff, 4),
},
"mlpAblated": {
"logitA": round(mlp_a, 4),
"logitB": round(mlp_b, 4),
"logitDiff": round(mlp_diff, 4),
"dropFromBaseline": round(base_diff - mlp_diff, 4),
},
})
return jsonify({
"success": True,
"model": "gpt2",
"device": str(device),
"nLayers": n_layers,
"tokenA": {"input": a, "used": model.to_string([a_id]), "id": a_id},
"tokenB": {"input": b, "used": model.to_string([b_id]), "id": b_id},
"baseline": {
"logitA": round(base_a, 4),
"logitB": round(base_b, 4),
"logitDiff": round(base_diff, 4),
},
"layers": layer_results,
})
except Exception as e:
print(f"[LayerAblation] Error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/logit-difference/analyze', methods=['POST'])
def logit_difference_analyze():
"""
Compute the logit difference between two candidate next-tokens for a given
prompt. This is THE foundational metric used in mech-interp literature
(Wang et al. 2022 IOI, Anthropic transformer-circuits), because:
- It's a clean scalar per prompt
- Differences cancel out the unembedding bias
- Sign tells you which token wins, magnitude tells you by how much
Returns logits, probabilities, ranks, and the difference for both tokens
at the FINAL position (i.e., the next-token prediction).
Request: { prompt, tokenA, tokenB } // tokenA/B are literal strings
Response: { success, model, device, prompt, promptTokens, lastTokenIdx,
tokenA: {used, id, logit, prob, rank}, tokenB: {...},
logitDiff, probDiff, top5Predictions:[{token, logit, prob, rank}] }
"""
try:
data = request.json or {}
prompt = data.get('prompt')
a = data.get('tokenA')
b = data.get('tokenB')
if not prompt or not a or not b:
return jsonify({"error": "prompt, tokenA, tokenB required"}), 400
model = _load_hooked_transformer('gpt2')
device = next(model.parameters()).device
def first_id(s):
ids = model.to_tokens(s, prepend_bos=False)[0]
if ids.numel() == 0:
raise ValueError(f"token tokenized to empty: {s!r}")
return int(ids[0].item()), ids.numel() > 1, model.to_string([int(ids[0].item())])
a_id, a_multi, a_used = first_id(a)
b_id, b_multi, b_used = first_id(b)
with torch.no_grad():
tokens = model.to_tokens(prompt) # [1, seq] with BOS
logits = model(tokens) # [1, seq, vocab]
last_logits = logits[0, -1, :].float().cpu() # [vocab]
probs = torch.softmax(last_logits, dim=-1)
# Ranks: argsort descending then find positions
sorted_idx = torch.argsort(last_logits, descending=True)
rank_map = torch.empty_like(sorted_idx)
rank_map[sorted_idx] = torch.arange(len(sorted_idx))
# Top-5 alternatives
top_logits, top_idx = torch.topk(last_logits, 5)
prompt_token_strs = [model.to_string([int(t.item())]) for t in tokens[0]]
last_idx = int(tokens.shape[1] - 1)
a_logit = float(last_logits[a_id].item())
b_logit = float(last_logits[b_id].item())
a_prob = float(probs[a_id].item())
b_prob = float(probs[b_id].item())
a_rank = int(rank_map[a_id].item()) + 1 # 1-indexed
b_rank = int(rank_map[b_id].item()) + 1
top5 = []
for log_val, i in zip(top_logits.tolist(), top_idx.tolist()):
r = int(rank_map[i].item()) + 1
top5.append({
"token": model.to_string([int(i)]),
"tokenId": int(i),
"logit": round(float(log_val), 4),
"prob": round(float(probs[int(i)].item()), 5),
"rank": r,
})
return jsonify({
"success": True,
"model": "gpt2",
"device": str(device),
"prompt": prompt,
"promptTokens": prompt_token_strs,
"lastTokenIdx": last_idx,
"tokenA": {
"input": a, "used": a_used, "id": a_id, "multi": a_multi,
"logit": round(a_logit, 4), "prob": round(a_prob, 5), "rank": a_rank,
},
"tokenB": {
"input": b, "used": b_used, "id": b_id, "multi": b_multi,
"logit": round(b_logit, 4), "prob": round(b_prob, 5), "rank": b_rank,
},
"logitDiff": round(a_logit - b_logit, 4),
"probDiff": round(a_prob - b_prob, 5),
"winner": "A" if a_logit > b_logit else "B",
"top5Predictions": top5,
})
except Exception as e:
print(f"[LogitDifference] Error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/vector-arithmetic/analyze', methods=['POST'])
def vector_arithmetic_analyze():
"""
Compute v = E(a) - E(b) + E(c) and return the K nearest vocab tokens
to v by cosine similarity in GPT-2's input embedding space.
The classic word2vec-style analogy probe: e.g.
" king" - " man" + " woman" -> ?
" Paris" - " France" + " Germany" -> ?
" walking" - " walk" + " run" -> ?
We exclude the three input tokens themselves from the results so the
user sees only NEW candidates.
Request: { a, b, c, k? } // strings, e.g. " king", " man", " woman"
Response: { success, model, device, dModel, vocabSize,
tokens:{a,b,c} with id+used_str, neighbors:[{token, tokenId, similarity}] }
"""
try:
data = request.json or {}
a = data.get('a'); b = data.get('b'); c = data.get('c')
if not a or not b or not c:
return jsonify({"error": "a, b, c required"}), 400
k = int(data.get('k', 15))
k = max(1, min(40, k))
model = _load_hooked_transformer('gpt2')
device = next(model.parameters()).device
def first_id(s):
ids = model.to_tokens(s, prepend_bos=False)[0]
if ids.numel() == 0:
raise ValueError(f"token tokenized to empty: {s!r}")
return int(ids[0].item()), ids.numel() > 1
a_id, a_multi = first_id(a)
b_id, b_multi = first_id(b)
c_id, c_multi = first_id(c)
with torch.no_grad():
W_E = model.W_E # [vocab, d_model]
v = (W_E[a_id].float() - W_E[b_id].float() + W_E[c_id].float())
v_norm = v / (v.norm() + 1e-9)
W_norm = W_E.float() / (W_E.float().norm(dim=-1, keepdim=True) + 1e-9)
sims = (W_norm @ v_norm).cpu()
# Pull more than k so we can filter the 3 input tokens out
top_vals, top_idx = torch.topk(sims, k + 5)
skip = {a_id, b_id, c_id}
neighbors = []
for s, i in zip(top_vals.tolist(), top_idx.tolist()):
if int(i) in skip:
continue
neighbors.append({
"token": model.to_string([int(i)]),
"tokenId": int(i),
"similarity": round(float(s), 5),
})
if len(neighbors) >= k:
break
return jsonify({
"success": True,
"model": "gpt2",
"device": str(device),
"dModel": int(W_E.shape[1]),
"vocabSize": int(W_E.shape[0]),
"tokens": {
"a": {"used": model.to_string([a_id]), "id": a_id, "multi": a_multi},
"b": {"used": model.to_string([b_id]), "id": b_id, "multi": b_multi},
"c": {"used": model.to_string([c_id]), "id": c_id, "multi": c_multi},
},
"neighbors": neighbors,
})
except Exception as e:
print(f"[VectorArithmetic] Error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/vocab-neighbors/analyze', methods=['POST'])
def vocab_neighbors_analyze():
"""
Find the K nearest vocab tokens to an input token in embedding space.
No forward pass needed — we operate directly on GPT-2's input embedding
matrix W_E (shape [vocab_size, d_model]). For the input token, we compute
cosine similarity against every other vocab token and return the top-K.
Useful for:
- Exploring what GPT-2 "thinks" is similar (e.g., " king" → " queen", " prince")
- Finding tokenization quirks (capitalization splits, leading-space variants)
- Sanity-checking that semantic structure exists in raw embeddings
Request: { token, k? } // token is the literal string, e.g. " king"
Response: { success, model, device, inputToken, inputTokenId, dModel,
vocabSize, neighbors:[{token, tokenId, similarity}] }
"""
try:
data = request.json or {}
raw = data.get('token')
if raw is None or raw == "":
return jsonify({"error": "token required"}), 400
k = int(data.get('k', 20))
k = max(1, min(50, k))
model = _load_hooked_transformer('gpt2')
device = next(model.parameters()).device
# Tokenize. If multi-token, use the first piece and report the actual
# piece used so the user understands what was queried.
ids = model.to_tokens(raw, prepend_bos=False)[0]
if ids.numel() == 0:
return jsonify({"error": "token tokenized to empty sequence"}), 400
tok_id = int(ids[0].item())
used_str = model.to_string([tok_id])
with torch.no_grad():
W_E = model.W_E # [vocab, d_model]
v = W_E[tok_id].float()
v_norm = v / (v.norm() + 1e-9)
# Normalize all rows (vocab is ~50k for GPT-2 — fine on GPU)
W_norm = W_E.float() / (W_E.float().norm(dim=-1, keepdim=True) + 1e-9)
sims = (W_norm @ v_norm).cpu() # [vocab]
# Get top K+1 (will include the token itself at position 0)
top_vals, top_idx = torch.topk(sims, k + 1)
neighbors = []
for s, i in zip(top_vals.tolist(), top_idx.tolist()):
if int(i) == tok_id:
continue # skip self
neighbors.append({
"token": model.to_string([int(i)]),
"tokenId": int(i),
"similarity": round(float(s), 5),
})
if len(neighbors) >= k:
break
return jsonify({
"success": True,
"model": "gpt2",
"device": str(device),
"inputToken": used_str,
"inputTokenId": tok_id,
"multiTokenInput": ids.numel() > 1,
"originalPieces": [model.to_string([int(t.item())]) for t in ids] if ids.numel() > 1 else None,
"dModel": int(W_E.shape[1]),
"vocabSize": int(W_E.shape[0]),
"neighbors": neighbors,
})
except Exception as e:
print(f"[VocabNeighbors] Error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/token-surprisal/analyze', methods=['POST'])
def token_surprisal_analyze():
"""
Per-token surprisal (negative log probability) under the model.
For every token at position p (p >= 1), we compute -log P(token_p | tokens<p)
using the model's logits at position p-1. Lower = the model expected this
token; higher = the model was surprised by it.
Also returns the model's top-1 expected token at each position so users
can see what the model was predicting instead.
Request: { prompt }
Response: { success, model, device, tokens, surprisals (bits),
expectedTokens, predictedCorrectly, totalLogProb,
avgSurprisal, perplexity, maxSurprisal }
"""
try:
data = request.json or {}
prompt = (data.get('prompt') or "").strip()
if not prompt:
return jsonify({"error": "prompt required"}), 400
if len(prompt) > 1500:
prompt = prompt[:1500]
model = _load_hooked_transformer('gpt2')
device = next(model.parameters()).device
toks = model.to_tokens(prompt)
# Cap for performance and rendering
if toks.shape[1] > 80:
toks = toks[:, :80]
token_strs = [model.to_string([int(t.item())]) for t in toks[0]]
n = toks.shape[1]
with torch.no_grad():
logits = model(toks) # [1, n, vocab]
log_probs = torch.log_softmax(logits[0].float(), dim=-1) # [n, vocab]
# For position p in 1..n-1, surprisal of toks[p] = -log_probs[p-1, toks[p]]
# in bits = nats / ln(2)
ln2 = float(torch.log(torch.tensor(2.0)).item())
surprisals = [None] # position 0 has no previous context
expected_tokens = [None]
predicted_correctly = [None]
for p in range(1, n):
target_id = int(toks[0, p].item())
nats = -float(log_probs[p - 1, target_id].item())
bits = nats / ln2
top1_id = int(log_probs[p - 1].argmax().item())
surprisals.append(round(bits, 5))
expected_tokens.append(model.to_string([top1_id]))
predicted_correctly.append(top1_id == target_id)
valid_surprisals = [s for s in surprisals if s is not None]
avg_surprisal = sum(valid_surprisals) / len(valid_surprisals) if valid_surprisals else 0.0
perplexity = float(2.0 ** avg_surprisal) if avg_surprisal else 0.0
max_surprisal = max(valid_surprisals) if valid_surprisals else 0.0
total_log_prob = -sum(valid_surprisals) * ln2 # in nats, total log P(sequence)
return jsonify({
"success": True,
"model": "gpt2",
"device": str(device),
"tokens": token_strs,
"surprisals": surprisals,
"expectedTokens": expected_tokens,
"predictedCorrectly": predicted_correctly,
"avgSurprisal": round(avg_surprisal, 5),
"perplexity": round(perplexity, 4),
"maxSurprisal": round(max_surprisal, 5),
"totalLogProb": round(total_log_prob, 5),
})
except Exception as e:
print(f"[TokenSurprisal] Error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/attention-pattern/analyze', methods=['POST'])
def attention_pattern_analyze():
"""
Visualize the attention pattern of a single (layer, head) on a prompt.
Returns the post-softmax attention matrix [seq_len, seq_len] with the
corresponding token labels for both axes, so the frontend can render a
heatmap of how each query position attends to each key position for the
chosen head.
Request: { prompt, layer, head }
Response: { success, model, device, layer, head, nLayers, nHeads,
tokens:[str], pattern:[[float]], maxAttention,
topAttended:[{from, to, weight}] (top 10 off-diagonal pairs) }
"""
try:
data = request.json or {}
prompt = (data.get('prompt') or "").strip()
layer = int(data.get('layer', 5))
head = int(data.get('head', 5))
if not prompt:
return jsonify({"error": "prompt required"}), 400
if len(prompt) > 600:
prompt = prompt[:600]
model = _load_hooked_transformer('gpt2')
device = next(model.parameters()).device
n_layers = model.cfg.n_layers
n_heads = model.cfg.n_heads
layer = max(0, min(layer, n_layers - 1))
head = max(0, min(head, n_heads - 1))
toks = model.to_tokens(prompt)
# Cap sequence length for visualization (40x40 matrix is the max useful render).
if toks.shape[1] > 40:
toks = toks[:, :40]
token_strs = [model.to_string([int(t.item())]) for t in toks[0]]
pattern_name = f'blocks.{layer}.attn.hook_pattern'
with torch.no_grad():
_, cache = model.run_with_cache(toks, names_filter=[pattern_name])
patt = cache[pattern_name][0, head].float().cpu() # [q, k]
# Convert to lists, also find top off-diagonal attended pairs.
# Skip the BOS attention sink (j==0) — attention heads use the BOS token
# as a "do nothing" target when they have nothing relevant to attend to,
# and that uninteresting behaviour dominates raw rankings.
seq_len = patt.shape[0]
rows = [[round(float(patt[i, j].item()), 5) for j in range(seq_len)] for i in range(seq_len)]
pairs = []
for i in range(seq_len):
for j in range(seq_len):
if i != j and j != 0: # skip self and BOS sink
pairs.append((float(patt[i, j].item()), i, j))
pairs.sort(reverse=True)
top_attended = [{"fromIdx": p[1], "fromToken": token_strs[p[1]],
"toIdx": p[2], "toToken": token_strs[p[2]],
"weight": round(p[0], 5)} for p in pairs[:10]]
del cache
if device.type == "cuda":
torch.cuda.empty_cache()
return jsonify({
"success": True,
"model": "gpt2",
"device": str(device),
"layer": layer,
"head": head,
"nLayers": n_layers,
"nHeads": n_heads,
"tokens": token_strs,
"pattern": rows,
"maxAttention": round(float(patt.max().item()), 5),
"topAttended": top_attended,
})
except Exception as e:
print(f"[AttentionPattern] Error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/direct-logit-attribution/analyze', methods=['POST'])
def direct_logit_attribution_analyze():
"""
Direct Logit Attribution (DLA) — decompose the model's prediction at the
final token into per-attention-head and per-MLP-layer contributions by
projecting each component's output (at the last position) onto the
unembedding direction for a target token.
This is the standard mechanistic-interpretability technique for asking
"which components actually caused the model to predict token X?".
Algorithm:
1. Tokenize the prompt and run forward with cache (z + mlp_out).
2. Pick a target token (top-1 prediction by default, or user-supplied
word — first BPE token taken).
3. unembed_dir = W_U[:, target_token]
4. For each layer l and head h:
per_head_out[l,h] = z[l, last, h] @ W_O[l, h] # [d_model]
contribution[l,h] = per_head_out[l,h] · unembed_dir
5. For each layer l:
mlp_contribution[l] = mlp_out[l, last] · unembed_dir
6. Return ranked head contributions + per-layer MLP contributions +
per-layer aggregated attention contribution.
Note: contributions are raw dot products (not LN-scaled), so absolute
magnitudes don't equal final logit deltas exactly, but the relative
ranking — which is what users want — is preserved.
Request: { prompt, targetToken? (string, optional) }
Response: { success, model, device, prompt, targetToken, targetTokenId,
targetLogit, targetProb, baselineTop5,
headContributions:[{layer, head, contribution}], // ranked desc
mlpContributions:[{layer, contribution}],
attnLayerContributions:[{layer, contribution}],
totalAttn, totalMlp }
"""
try:
data = request.json or {}
prompt = (data.get('prompt') or "").strip()
target_word = (data.get('targetToken') or "").strip()
if not prompt:
return jsonify({"error": "prompt required"}), 400
if len(prompt) > 1000:
prompt = prompt[:1000]
model = _load_hooked_transformer('gpt2')
device = next(model.parameters()).device
toks = model.to_tokens(prompt)
n_layers = model.cfg.n_layers
n_heads = model.cfg.n_heads
# Names we need: per-layer z (head outputs pre-W_O), mlp_out, ln_final scale (optional).
z_names = [f'blocks.{l}.attn.hook_z' for l in range(n_layers)]
mlp_names = [f'blocks.{l}.hook_mlp_out' for l in range(n_layers)]
names = z_names + mlp_names
with torch.no_grad():
logits, cache = model.run_with_cache(toks, names_filter=names)
last_logits = logits[0, -1] # [vocab]
top5_vals, top5_idx = torch.topk(last_logits.softmax(dim=-1), 5)
# Choose target token
if target_word:
target_ids = model.to_tokens(target_word, prepend_bos=False)[0]
target_id = int(target_ids[0].item()) if target_ids.numel() > 0 else int(top5_idx[0].item())
else:
target_id = int(top5_idx[0].item())
unembed_dir = model.W_U[:, target_id].float() # [d_model]
head_contribs = torch.zeros(n_layers, n_heads, device=device)
mlp_contribs = torch.zeros(n_layers, device=device)
for l in range(n_layers):
z_l = cache[z_names[l]][0, -1].float() # [head, d_head]
W_O_l = model.W_O[l].float() # [head, d_head, d_model]
# per-head output at last pos: [head, d_model]
per_head = torch.einsum('h d, h d m -> h m', z_l, W_O_l)
head_contribs[l] = per_head @ unembed_dir
mlp_contribs[l] = cache[mlp_names[l]][0, -1].float() @ unembed_dir
head_contribs_cpu = head_contribs.cpu()
mlp_contribs_cpu = mlp_contribs.cpu()
head_list = []
for l in range(n_layers):
for h in range(n_heads):
head_list.append({
"layer": l, "head": h,
"contribution": round(float(head_contribs_cpu[l, h].item()), 5),
})
head_list.sort(key=lambda x: abs(x['contribution']), reverse=True)
mlp_list = [{"layer": l, "contribution": round(float(mlp_contribs_cpu[l].item()), 5)} for l in range(n_layers)]
attn_layer_list = [{"layer": l, "contribution": round(float(head_contribs_cpu[l].sum().item()), 5)} for l in range(n_layers)]
target_logit = float(last_logits[target_id].item())
target_prob = float(last_logits.softmax(dim=-1)[target_id].item())
target_str = model.to_string([target_id])
top5 = [{"token": model.to_string([int(i.item())]),
"tokenId": int(i.item()),
"prob": round(float(p.item()), 5)} for p, i in zip(top5_vals, top5_idx)]
del cache
if device.type == "cuda":
torch.cuda.empty_cache()
return jsonify({
"success": True,
"model": "gpt2",
"device": str(device),
"prompt": prompt,
"targetToken": target_str,
"targetTokenId": target_id,
"targetLogit": round(target_logit, 5),
"targetProb": round(target_prob, 5),
"baselineTop5": top5,
"headContributions": head_list,
"mlpContributions": mlp_list,
"attnLayerContributions": attn_layer_list,
"totalAttn": round(float(head_contribs_cpu.sum().item()), 5),
"totalMlp": round(float(mlp_contribs_cpu.sum().item()), 5),
})
except Exception as e:
print(f"[DLA] Error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/induction-heads/analyze', methods=['POST'])
def induction_heads_analyze():
"""
Detect induction heads — attention heads that implement the
"[A][B]...[A] → predict [B]" pattern, the mechanism Anthropic identified
as foundational to in-context learning.
Algorithm (per Anthropic 2022, "In-context Learning and Induction Heads"):
1. Build B random sequences of unique tokens of length seq_len, then
concatenate each with itself to form a "repeated" sequence of length
2*seq_len. Position p in the second half corresponds to the same token
as position p - seq_len in the first half. The induction prediction
at position p is "the token that came AFTER position (p - seq_len)
in the first half" — i.e., the token at position p - seq_len + 1.
2. Run forward with cache, collecting attention patterns from
blocks.{l}.attn.hook_pattern for every layer.
3. For each (layer, head): the induction score is the mean attention
weight on the diagonal that maps query position q (in the second half)
to key position (q - seq_len + 1) (the "prev-token-after-match"
position in the first half), averaged over query positions in the
second half and across batches.
4. Top induction heads are the ones with the largest score.
Request: { batchSize, seqLen }
Response: { success, model, device, nLayers, nHeads, batchSize, seqLen,
topHeads:[{layer, head, score}], // sorted desc
heatmap:[L][H], // raw scores
meanScore, maxScore }
"""
try:
data = request.json or {}
batch = max(1, min(int(data.get('batchSize', 4)), 16))
seq_len = max(8, min(int(data.get('seqLen', 25)), 64))
model = _load_hooked_transformer('gpt2')
device = next(model.parameters()).device
n_layers = model.cfg.n_layers
n_heads = model.cfg.n_heads
vocab = model.cfg.d_vocab
# Build random unique-token sequences and repeat them.
# Avoid special tokens — sample from middle of vocab to skip BOS/EOS.
rand = torch.randint(low=1000, high=vocab - 1000, size=(batch, seq_len), device=device)
repeated = torch.cat([rand, rand], dim=1) # [batch, 2*seq_len]
pattern_names = [f'blocks.{l}.attn.hook_pattern' for l in range(n_layers)]
with torch.no_grad():
_, cache = model.run_with_cache(repeated, names_filter=pattern_names)
# induction score per (layer, head):
# mean over batch and over query positions q in [seq_len, 2*seq_len-1]
# of attn_pattern[batch, head, q, q - seq_len + 1].
scores = torch.zeros(n_layers, n_heads, device=device)
for l in range(n_layers):
patt = cache[pattern_names[l]] # [batch, n_heads, q, k]
# gather diag: for q in seq_len..2*seq_len-1, k = q - seq_len + 1
qs = torch.arange(seq_len, 2 * seq_len, device=device)
ks = qs - seq_len + 1
# patt[:, :, qs, ks] — fancy indexing
diag = patt[:, :, qs, ks] # [batch, heads, seq_len]
scores[l] = diag.mean(dim=(0, 2))
scores_cpu = scores.cpu()
flat = []
for l in range(n_layers):
for h in range(n_heads):
flat.append({"layer": l, "head": h, "score": round(float(scores_cpu[l, h].item()), 5)})
flat.sort(key=lambda x: x['score'], reverse=True)
del cache
if device.type == "cuda":
torch.cuda.empty_cache()
return jsonify({
"success": True,
"model": "gpt2",
"device": str(device),
"nLayers": n_layers,
"nHeads": n_heads,
"batchSize": batch,
"seqLen": seq_len,
"topHeads": flat[:12],
"heatmap": [[round(float(scores_cpu[l, h].item()), 5) for h in range(n_heads)] for l in range(n_layers)],
"meanScore": round(float(scores_cpu.mean().item()), 5),
"maxScore": round(float(scores_cpu.max().item()), 5),
})
except Exception as e:
print(f"[InductionHeads] Error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/activation-steering/analyze', methods=['POST'])
def activation_steering_analyze():
"""
Activation Steering — classic mechanistic-interpretability technique.
1. Build a "concept direction" by averaging the residual-stream activation
at a chosen layer (last token position) over POSITIVE prompts that
exemplify the target concept, then subtracting the mean over NEGATIVE
prompts. The result is a single d_model-dim vector that points from
"not concept" toward "concept" in activation space.
2. Run the TEST prompt through the model twice:
- Baseline: no intervention.
- Steered: register a forward hook on blocks.{layer}.hook_resid_post
that adds (coefficient * steering_vector) at every position.
3. Compare next-token distributions: top-5 token shifts, KL divergence,
and per-token probability changes for the most-affected tokens.
Request: { positivePrompts:[..], negativePrompts:[..], testPrompt,
layer (0..n-1), coefficient (default 6.0) }
Response: { success, device, model, layer, coefficient,
steeringVectorNorm, baselineTop5, steeredTop5,
klBaselineToSteered, topShifts:[{token, baselineProb, steeredProb, delta}] }
"""
try:
data = request.json or {}
pos = [p for p in (data.get('positivePrompts') or []) if isinstance(p, str) and p.strip()][:30]
neg = [p for p in (data.get('negativePrompts') or []) if isinstance(p, str) and p.strip()][:30]
test_prompt = (data.get('testPrompt') or '').strip()
coef = float(data.get('coefficient', 6.0))
layer = int(data.get('layer', 6))
if len(pos) < 2 or len(neg) < 2:
return jsonify({"error": "need at least 2 positive AND 2 negative prompts"}), 400
if not test_prompt:
return jsonify({"error": "testPrompt is required"}), 400
model = _load_hooked_transformer('gpt2')
device = next(model.parameters()).device
n_layers = model.cfg.n_layers
if layer < 0 or layer >= n_layers:
return jsonify({"error": f"layer must be in [0, {n_layers - 1}]"}), 400
hook_name = f'blocks.{layer}.hook_resid_post'
def mean_resid(prompts):
acc = None
n = 0
for p in prompts:
try:
toks = model.to_tokens(p).to(device)
if toks.shape[1] > 64:
toks = toks[:, :64]
with torch.no_grad():
_, cache = model.run_with_cache(toks, names_filter=[hook_name])
v = cache[hook_name][0, -1, :].float()
acc = v if acc is None else acc + v
n += 1
del cache
except Exception as inner:
print(f"[ActivationSteering] skip: {inner}")
continue
return (acc / n) if (acc is not None and n > 0) else None
v_pos = mean_resid(pos)
v_neg = mean_resid(neg)
if v_pos is None or v_neg is None:
return jsonify({"error": "all prompts failed to encode"}), 400
steering_vec = (v_pos - v_neg).to(next(model.parameters()).dtype)
sv_norm = float(torch.norm(steering_vec).item())
# Baseline forward
test_toks = model.to_tokens(test_prompt).to(device)
if test_toks.shape[1] > 64:
test_toks = test_toks[:, :64]
with torch.no_grad():
base_logits = model(test_toks)
base_probs = torch.softmax(base_logits[0, -1, :].float(), dim=-1)
# Steered forward — hook adds coef * steering_vec at every position.
def steer_hook(activation, hook):
# activation: [batch, pos, d_model] in model dtype
return activation + (coef * steering_vec)
with torch.no_grad():
steered_logits = model.run_with_hooks(test_toks, fwd_hooks=[(hook_name, steer_hook)])
steered_probs = torch.softmax(steered_logits[0, -1, :].float(), dim=-1)
# Top-5 each side
def top5(probs):
vals, idxs = probs.topk(5)
return [
{"token": model.to_string(torch.tensor([int(i)])), "prob": round(float(v.item()), 4)}
for v, i in zip(vals, idxs)
]
baseline_top5 = top5(base_probs)
steered_top5 = top5(steered_probs)
# KL(steered || baseline)
eps = 1e-12
kl = (steered_probs * (steered_probs.clamp_min(eps).log() - base_probs.clamp_min(eps).log())).sum().item()
# Top shifts by absolute delta
delta = (steered_probs - base_probs).cpu()
abs_top = torch.topk(delta.abs(), 12).indices.tolist()
top_shifts = []
for tid in abs_top:
top_shifts.append({
"token": model.to_string(torch.tensor([tid])),
"baselineProb": round(float(base_probs[tid].item()), 5),
"steeredProb": round(float(steered_probs[tid].item()), 5),
"delta": round(float(delta[tid].item()), 5),
})
return jsonify({
"success": True,
"device": str(device),
"model": "gpt2",
"layer": layer,
"nLayers": n_layers,
"coefficient": coef,
"steeringVectorNorm": round(sv_norm, 4),
"nPositive": len(pos),
"nNegative": len(neg),
"baselineTop5": baseline_top5,
"steeredTop5": steered_top5,
"klBaselineToSteered": round(float(kl), 6),
"topShifts": top_shifts,
})
except Exception as e:
print(f"[ActivationSteering] Error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/quantization-impact/analyze', methods=['POST'])
def quantization_impact_analyze():
"""
Compare a model in fp32 (full precision) vs fp16 (half precision) to measure
the quality / latency tradeoff that production deployments care about.
Pipeline:
1. Load gpt2 fresh in fp32, deep-copy and cast to fp16 — both share weights but
differ only in numerical precision.
2. Warm up both models with one forward pass each (CUDA tensor-core JIT).
3. For each prompt:
- Time forward pass on fp32 and fp16 separately (CUDA-synchronized).
- Take final-position next-token logits, softmax in fp32 for fair comparison.
- Compute top-1 agreement, top-5 Jaccard, KL(fp16 ‖ fp32).
4. Aggregate: mean KL, top-1 agreement rate, mean Jaccard, mean latency speedup,
theoretical memory savings (fp32 → fp16 = 2× reduction).
Request: { prompts:[..] }
Response: { success, device, model, paramCount, nPrompts,
fp32MeanLatencyMs, fp16MeanLatencyMs, speedup,
memBytesFp32, memBytesFp16, memReductionPct,
top1AgreementRate, meanTop5Jaccard, meanKL,
rows:[{prompt, top1Fp32, top1Fp16, agree, top5Jaccard, kl,
latencyFp32Ms, latencyFp16Ms}] }
"""
try:
import copy
import time
from transformer_lens import HookedTransformer
data = request.json or {}
prompts = [p for p in (data.get('prompts') or []) if isinstance(p, str) and p.strip()][:25]
if len(prompts) < 1:
return jsonify({"error": "at least 1 prompt required"}), 400
device_str = "cuda" if torch.cuda.is_available() else "cpu"
# Fresh load both copies — bypass the cached singleton so we don't corrupt
# the global model precision. Use .to(device, dtype=...) which converts
# parameters AND buffers (.half() alone misses some HookedTransformer buffers
# like attn.mask / attn.IGNORE that get registered as fp32).
model_fp32 = HookedTransformer.from_pretrained('gpt2', dtype=torch.float32).to(device_str)
model_fp32.eval()
model_fp16 = HookedTransformer.from_pretrained('gpt2', dtype=torch.float16).to(device_str)
model_fp16.eval()
device = next(model_fp32.parameters()).device
param_count = sum(p.numel() for p in model_fp32.parameters())
mem_fp32 = param_count * 4
mem_fp16 = param_count * 2
# Warmup
with torch.no_grad():
warm = model_fp32.to_tokens("warmup")
_ = model_fp32(warm.to(device))
_ = model_fp16(warm.to(device))
if device.type == "cuda":
torch.cuda.synchronize()
rows = []
kls = []
agree_n = 0
jaccards = []
lat32s = []
lat16s = []
for p in prompts:
try:
tokens = model_fp32.to_tokens(p).to(device)
if tokens.shape[1] > 64:
tokens = tokens[:, :64]
with torch.no_grad():
if device.type == "cuda":
torch.cuda.synchronize()
t0 = time.perf_counter()
l32 = model_fp32(tokens)
if device.type == "cuda":
torch.cuda.synchronize()
t1 = time.perf_counter()
l16 = model_fp16(tokens)
if device.type == "cuda":
torch.cuda.synchronize()
t2 = time.perf_counter()
lat32 = (t1 - t0) * 1000.0
lat16 = (t2 - t1) * 1000.0
lat32s.append(lat32)
lat16s.append(lat16)
p32 = torch.softmax(l32[0, -1, :].float(), dim=-1)
p16 = torch.softmax(l16[0, -1, :].float(), dim=-1)
top1_32 = int(p32.argmax().item())
top1_16 = int(p16.argmax().item())
agree = (top1_32 == top1_16)
if agree:
agree_n += 1
top5_32 = set(p32.topk(5).indices.tolist())
top5_16 = set(p16.topk(5).indices.tolist())
inter = len(top5_32 & top5_16)
union = len(top5_32 | top5_16)
jacc = inter / union if union else 0.0
jaccards.append(jacc)
eps = 1e-12
kl = (p16 * (p16.clamp_min(eps).log() - p32.clamp_min(eps).log())).sum().item()
kls.append(kl)
rows.append({
"prompt": p[:200],
"top1Fp32": model_fp32.to_string(torch.tensor([top1_32])),
"top1Fp16": model_fp16.to_string(torch.tensor([top1_16])),
"agree": agree,
"top5Jaccard": round(jacc, 4),
"kl": round(float(kl), 6),
"latencyFp32Ms": round(lat32, 2),
"latencyFp16Ms": round(lat16, 2),
})
del l32, l16, p32, p16
except Exception as inner:
print(f"[QuantizationImpact] skip prompt: {inner}")
continue
if not rows:
return jsonify({"error": "no usable prompts"}), 400
mean_lat32 = sum(lat32s) / len(lat32s)
mean_lat16 = sum(lat16s) / len(lat16s)
speedup = mean_lat32 / mean_lat16 if mean_lat16 > 0 else 1.0
# Free memory
del model_fp32, model_fp16
if device.type == "cuda":
torch.cuda.empty_cache()
return jsonify({
"success": True,
"device": str(device),
"model": "gpt2",
"paramCount": param_count,
"nPrompts": len(rows),
"fp32MeanLatencyMs": round(mean_lat32, 2),
"fp16MeanLatencyMs": round(mean_lat16, 2),
"speedup": round(speedup, 3),
"memBytesFp32": mem_fp32,
"memBytesFp16": mem_fp16,
"memReductionPct": round(50.0, 1),
"top1AgreementRate": round(agree_n / len(rows), 4),
"meanTop5Jaccard": round(sum(jaccards) / len(jaccards), 4),
"meanKL": round(sum(kls) / len(kls), 6),
"rows": rows,
})
except Exception as e:
print(f"[QuantizationImpact] Error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@app.route('/knowledge-distillation/analyze', methods=['POST'])
def knowledge_distillation_analyze():
"""
Compare a teacher model and a student (distilled) model on the same prompts.
Pipeline:
1. Load teacher (gpt2 full, 12 layers) and student (distilgpt2, 6 layers) — both
share the GPT-2 BPE tokenizer / 50257 vocab so logits align.
2. For each prompt, run a forward pass on each model and take next-token logits at
the final position. Convert to probability distributions in fp32.
3. Per-prompt metrics:
- top-1 agreement: argmax_t == argmax_s
- top-5 jaccard: |T5 ∩ S5| / |T5 ∪ S5|
- KL(student || teacher) = Σ s * (log s − log t) — the canonical distillation loss
4. Aggregate: mean KL, top-1 agreement rate, mean top-5 jaccard.
Request: { prompts:[..] } (always uses gpt2 vs distilgpt2)
Response: { success, device, teacher, student, nPrompts,
meanKL, top1AgreementRate, meanTop5Jaccard,
rows:[{prompt, teacherTop1, studentTop1, agree, top5Jaccard, kl,
teacherTop3:[{tok,prob}], studentTop3:[{tok,prob}]}] }
"""
try:
data = request.json or {}
prompts = [p for p in (data.get('prompts') or []) if isinstance(p, str) and p.strip()][:25]
if len(prompts) < 1:
return jsonify({"error": "at least 1 prompt required"}), 400
teacher = _load_hooked_transformer('gpt2')
student = _load_hooked_transformer('distilgpt2')
device = next(teacher.parameters()).device
rows = []
kls = []
agree_n = 0
jaccards = []
def top3(probs, model):
vals, ids = probs.topk(3)
out = []
for v, i in zip(vals.tolist(), ids.tolist()):
try:
tok = model.to_string(torch.tensor([i]))
except Exception:
tok = f"[id={i}]"
out.append({"tok": tok, "prob": round(float(v), 4)})
return out
for p in prompts:
try:
t_tokens = teacher.to_tokens(p).to(device)
s_tokens = student.to_tokens(p).to(device)
if t_tokens.shape[1] > 64:
t_tokens = t_tokens[:, :64]
if s_tokens.shape[1] > 64:
s_tokens = s_tokens[:, :64]
with torch.no_grad():
t_logits = teacher(t_tokens)
s_logits = student(s_tokens)
t_probs = torch.softmax(t_logits[0, -1, :].float(), dim=-1)
s_probs = torch.softmax(s_logits[0, -1, :].float(), dim=-1)
t_top1 = int(t_probs.argmax().item())
s_top1 = int(s_probs.argmax().item())
agree = (t_top1 == s_top1)
if agree:
agree_n += 1
t_top5 = set(t_probs.topk(5).indices.tolist())
s_top5 = set(s_probs.topk(5).indices.tolist())
inter = len(t_top5 & s_top5)
union = len(t_top5 | s_top5)
jacc = inter / union if union else 0.0
jaccards.append(jacc)
eps = 1e-12
kl = (s_probs * (s_probs.clamp_min(eps).log() - t_probs.clamp_min(eps).log())).sum().item()
kls.append(kl)
rows.append({
"prompt": p[:200],
"teacherTop1": teacher.to_string(torch.tensor([t_top1])),
"studentTop1": student.to_string(torch.tensor([s_top1])),
"agree": agree,
"top5Jaccard": round(jacc, 4),
"kl": round(float(kl), 6),
"teacherTop3": top3(t_probs, teacher),
"studentTop3": top3(s_probs, student),
})
del t_logits, s_logits, t_probs, s_probs
except Exception as inner:
print(f"[KnowledgeDistillation] skip prompt: {inner}")
continue
if not rows:
return jsonify({"error": "no usable prompts"}), 400
return jsonify({
"success": True,
"device": str(device),
"teacher": "gpt2 (12 layers, 124M)",
"student": "distilgpt2 (6 layers, 82M)",
"nPrompts": len(rows),
"meanKL": round(sum(kls) / len(kls), 6),
"top1AgreementRate": round(agree_n / len(rows), 4),
"meanTop5Jaccard": round(sum(jaccards) / len(jaccards), 4),
"rows": rows,
})
except Exception as e:
print(f"[KnowledgeDistillation] Error: {str(e)}")
import traceback
traceback.print_exc()
return jsonify({"error": str(e)}), 500
if __name__ == '__main__':
port = int(os.environ.get('SAE_SERVICE_PORT', 7860))
print(f"[SAE Service] Starting on port {port}")
print(f"[SAE Service] Models will be loaded on first request or via /warmup")
app.run(host='0.0.0.0', port=port, debug=False, threaded=False)