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 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)