legacy_numpy.py · 可读全文
这里保留原文的中文、公式标记、代码和链接。可直接交给 Agent 读取，也可选取下方原文复制。此入口返回 HTML；原始 .md / .py 文件仍保留，完整候选包同时包含 UTF-8 原件。
#!/usr/bin/env python3
"""Lecture 05 R2.1: small, offline, NumPy-only experiments A--G.
The saved results distinguish constructed computations, optimization smoke tests,
and a small trained LM comparison. No pretrained model or cloud notebook is used.
Run from the repository root: python labs/r21_numpy.py
"""
from __future__ import annotations
import argparse
import hashlib
import importlib.util
import itertools
import json
import math
import os
import platform
import re
import sys
import time
from collections import Counter, defaultdict
from datetime import datetime, timezone
from pathlib import Path
# Keep the small CPU timing exercise reproducible; do not overwrite user settings.
for _name in ("OPENBLAS_NUM_THREADS", "OMP_NUM_THREADS", "MKL_NUM_THREADS"):
os.environ.setdefault(_name, "1")
import numpy as np
VERSION = "l5-r21-numpy-1.0"
BASE_SEED = 20261013
ATOL = 1e-10
_CLASSROOM = None
def classroom_parameters():
"""Use the teaching source; the generated sidecar keeps downloaded labs runnable."""
global _CLASSROOM
if _CLASSROOM is not None:
return _CLASSROOM
source = Path(__file__).resolve().parents[1] / "content" / "r21" / "attention_transformer.py"
snapshot = Path(__file__).with_name("classroom-parameters.json")
if source.exists():
spec = importlib.util.spec_from_file_location("l5_classroom_numbers", source)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
_CLASSROOM = {"source_path": "content/r21/attention_transformer.py",
"source_sha256": hashlib.sha256(source.read_bytes()).hexdigest(),
"READ": module.READ, "BLOCK": module.BLOCK}
snapshot.write_text(json.dumps(_CLASSROOM, ensure_ascii=False, indent=2) + "\n")
else:
_CLASSROOM = json.loads(snapshot.read_text())
return _CLASSROOM
def softmax(x):
z = x - np.max(x, axis=-1, keepdims=True)
z = np.exp(z)
return z / z.sum(axis=-1, keepdims=True)
def sigmoid(x):
return 1 / (1 + np.exp(-np.clip(x, -40, 40)))
def norm(grads):
return float(np.sqrt(sum(np.sum(g * g) for g in grads.values())))
def clip(grads, threshold):
before = norm(grads)
scale = min(1.0, threshold / max(before, 1e-30))
return {k: v * scale for k, v in grads.items()}, before, before * scale
class Adam:
def __init__(self, params, lr=0.01):
self.lr, self.t = lr, 0
self.m = {k: np.zeros_like(v) for k, v in params.items()}
self.v = {k: np.zeros_like(v) for k, v in params.items()}
def step(self, params, grads):
self.t += 1
for k in params:
self.m[k] = .9 * self.m[k] + .1 * grads[k]
self.v[k] = .999 * self.v[k] + .001 * grads[k] ** 2
m = self.m[k] / (1 - .9 ** self.t)
v = self.v[k] / (1 - .999 ** self.t)
params[k] -= self.lr * m / (np.sqrt(v) + 1e-8)
def parameter_count(params):
return sum(v.size for v in params.values())
def digest_rows(rows):
return hashlib.sha256(json.dumps(rows, separators=(",", ":")).encode()).hexdigest()
class Recurrent:
"""One-layer RNN/LSTM with embeddings and a shared output head; manual BPTT."""
def __init__(self, inputs, outputs, embed=8, hidden=24, kind="rnn", seed=0):
rng = np.random.default_rng(seed)
self.hidden, self.kind = hidden, kind
gate_width = hidden * (4 if kind == "lstm" else 1)
self.p = {
"E": rng.normal(0, .15, (inputs, embed)),
"Wx": rng.normal(0, 1 / math.sqrt(embed), (embed, gate_width)),
"Wh": rng.normal(0, .45 / math.sqrt(hidden), (hidden, gate_width)),
"b": np.zeros(gate_width),
"Wy": rng.normal(0, 1 / math.sqrt(hidden), (hidden, outputs)),
"by": np.zeros(outputs),
}
if kind == "lstm":
self.p["b"][hidden:2 * hidden] = 1.0 # explicit forget-gate bias
def forward(self, x):
p = self.p
b, steps = x.shape
h = np.zeros((b, self.hidden))
c = np.zeros_like(h)
states, cache = [], []
for t in range(steps):
e, hp, cp = p["E"][x[:, t]], h, c
a = e @ p["Wx"] + hp @ p["Wh"] + p["b"]
if self.kind == "rnn":
h = np.tanh(a)
cache.append((e, hp, h))
else:
ai, af, ag, ao = np.split(a, 4, axis=-1)
i, f, g, o = sigmoid(ai), sigmoid(af), np.tanh(ag), sigmoid(ao)
c = f * cp + i * g
h = o * np.tanh(c)
cache.append((e, hp, cp, i, f, g, o, c))
states.append(h)
hs = np.stack(states, axis=1)
return hs @ p["Wy"] + p["by"], hs, cache
def loss_grad(self, x, targets, mask=None):
logits, hs, cache = self.forward(x)
probs = softmax(logits)
if mask is None:
mask = np.ones(targets.shape)
denom = mask.sum()
picked = np.take_along_axis(probs, targets[..., None], axis=-1)[..., 0]
loss = float(-(np.log(np.maximum(picked, 1e-300)) * mask).sum() / denom)
dl = probs.copy()
bidx, tidx = np.indices(targets.shape)
dl[bidx, tidx, targets] -= 1
dl *= mask[..., None] / denom
p = self.p
grads = {k: np.zeros_like(v) for k, v in p.items()}
grads["Wy"] = hs.reshape(-1, self.hidden).T @ dl.reshape(-1, dl.shape[-1])
grads["by"] = dl.sum(axis=(0, 1))
direct = dl @ p["Wy"].T
dh_next = np.zeros_like(hs[:, 0])
dc_next = np.zeros_like(dh_next)
path = np.zeros(x.shape[1])
for t in reversed(range(x.shape[1])):
dh = direct[:, t] + dh_next
path[t] = np.linalg.norm(dh)
if self.kind == "rnn":
e, hp, h = cache[t]
da = dh * (1 - h * h)
else:
e, hp, cp, i, f, g, o, c = cache[t]
tc = np.tanh(c)
dc = dc_next + dh * o * (1 - tc * tc)
di, df, dg, do = dc * g, dc * cp, dc * i, dh * tc
da = np.concatenate((di * i * (1 - i), df * f * (1 - f),
dg * (1 - g * g), do * o * (1 - o)), axis=-1)
dc_next = dc * f
grads["Wx"] += e.T @ da
grads["Wh"] += hp.T @ da
grads["b"] += da.sum(axis=0)
np.add.at(grads["E"], x[:, t], da @ p["Wx"].T)
dh_next = da @ p["Wh"].T
return loss, grads, {"hidden_gradient_l2_by_position": path, "probs": probs}
class FixedWindow:
def __init__(self, inputs, outputs, window=2, embed=8, hidden=24, seed=0):
rng = np.random.default_rng(seed)
self.window = window
self.p = {"E": rng.normal(0, .15, (inputs, embed)),
"W": rng.normal(0, 1 / math.sqrt(window * embed), (window * embed, hidden)),
"b": np.zeros(hidden), "U": rng.normal(0, 1 / math.sqrt(hidden), (hidden, outputs)),
"c": np.zeros(outputs)}
def forward(self, contexts):
z = self.p["E"][contexts].reshape(len(contexts), -1)
h = np.tanh(z @ self.p["W"] + self.p["b"])
return h @ self.p["U"] + self.p["c"], (z, h)
def loss_grad(self, contexts, targets):
logits, (z, h) = self.forward(contexts)
probs = softmax(logits)
loss = float(-np.log(probs[np.arange(len(targets)), targets]).mean())
dl = probs.copy()
dl[np.arange(len(targets)), targets] -= 1
dl /= len(targets)
da = (dl @ self.p["U"].T) * (1 - h * h)
grads = {"U": h.T @ dl, "c": dl.sum(0), "W": z.T @ da, "b": da.sum(0),
"E": np.zeros_like(self.p["E"])}
dz = (da @ self.p["W"].T).reshape(len(targets), self.window, -1)
np.add.at(grads["E"], contexts.reshape(-1), dz.reshape(-1, dz.shape[-1]))
return loss, grads
def route_a():
sentences = [
"she deposited money at the bank", "she sat on the bank of the river",
"the bank accepted cash and money", "the river covered the grassy bank",
"cash arrived at the bank today", "birds rested on the river bank",
"she read a book near the river", "she paid cash for the book",
]
docs = [re.findall(r"[a-z]+", s.lower()) for s in sentences]
vocab = sorted(set(itertools.chain.from_iterable(docs)))
ids = {w: i for i, w in enumerate(vocab)}
reports, embeddings = {}, {}
for window in (1, 3):
counts = np.zeros((len(vocab), len(vocab)))
pairs = []
for doc in docs:
for i, word in enumerate(doc):
for j in range(max(0, i - window), min(len(doc), i + window + 1)):
if i != j:
counts[ids[word], ids[doc[j]]] += 1
pairs.append([word, doc[j]])
total = counts.sum()
ratio = counts * total / np.maximum(counts.sum(1)[:, None] * counts.sum(0)[None, :], 1)
ppmi = np.maximum(np.log(np.maximum(ratio, 1e-300)), 0)
u, s, _ = np.linalg.svd(ppmi, full_matrices=False)
emb = u[:, :4] * np.sqrt(s[:4])
embeddings[window] = emb
reports[str(window)] = {"training_pair_count": len(pairs), "count_shape": counts.shape,
"bank_cooccurrence": {w: int(counts[ids['bank'], j]) for j, w in enumerate(vocab) if counts[ids['bank'], j]},
"first_12_pairs": pairs[:12]}
emb = embeddings[3]
contexts = []
reps = []
for doc in docs[:2]:
i = doc.index("bank")
neighbors = [doc[j] for j in range(max(0, i - 3), min(len(doc), i + 4)) if j != i]
contexts.append(neighbors)
# This is an explicit, untrained contextual composition, not a language model.
reps.append(np.concatenate((emb[ids["bank"]], emb[[ids[w] for w in neighbors]].mean(0))))
fixed_bank_vectors = [emb[ids[doc[doc.index("bank")]]].copy() for doc in docs[:2]]
return {"status": "constructed_count_SVD_and_untrained_context_composition", "seed": None,
"sentences": sentences, "tokenization": "lowercase regex [a-z]+; punctuation discarded",
"vocabulary": vocab, "documents": len(docs), "tokens": sum(map(len, docs)),
"windows": reports, "svd_dimensions": 4, "fixed_bank_vectors": fixed_bank_vectors,
"fixed_bank_max_abs_difference": float(np.max(np.abs(fixed_bank_vectors[0] - fixed_bank_vectors[1]))),
"context_windows": contexts, "context_vectors": reps,
"context_vector_l2_difference": float(np.linalg.norm(reps[0] - reps[1])),
"interpretation": "Different constructed vectors establish context dependence, not learned word-sense accuracy."}
def lm_dataset():
sentences = [[s, v, "a", obj, "to", r, "near", "the", place, "<EOS>"]
for s, v, obj, r, place in itertools.product(
["amy", "ben", "cao", "dia"], ["gave", "sent", "showed", "lent"],
["book", "key", "cup", "map"], ["eli", "fay", "gus", "han"],
["river", "school", "office", "park"])]
rng = np.random.default_rng(BASE_SEED)
order = rng.permutation(len(sentences))
# Vocab is fixed by the published synthetic grammar, not inferred from test labels.
vocab = sorted(set(itertools.chain.from_iterable(sentences)))
ids = {w: i for i, w in enumerate(vocab)}
encoded = np.array([[ids[w] for w in s] for s in sentences], dtype=np.int64)
splits = {"train": encoded[order[:600]], "validation": encoded[order[600:700]], "test": encoded[order[700:800]]}
return splits, vocab
def fixed_contexts(targets, bos, k=2):
pad = np.pad(targets, ((0, 0), (k, 0)), constant_values=bos)
return np.stack([pad[:, t:t+k] for t in range(targets.shape[1])], axis=1).reshape(-1, k)
def route_b():
splits, vocab = lm_dataset()
v, bos, k = len(vocab), len(vocab), 2
counts = defaultdict(lambda: np.zeros(v))
for context, y in zip(fixed_contexts(splits["train"], bos, k), splits["train"].reshape(-1)):
counts[tuple(context)][y] += 1
alpha = .1
def ngram_prob(context):
c = counts.get(tuple(context), np.zeros(v))
return (c + alpha) / (c.sum() + alpha * v)
fixed = FixedWindow(v + 1, v, seed=BASE_SEED + 1)
rnn = Recurrent(v + 1, v, seed=BASE_SEED + 1)
models = {"fixed_window": fixed, "rnn": rnn}
metrics, curves = {}, {}
def model_loss(name, model, y):
if name == "fixed_window":
return model.loss_grad(fixed_contexts(y, bos), y.reshape(-1))[0]
x = np.concatenate((np.full((len(y), 1), bos), y[:, :-1]), axis=1)
return model.loss_grad(x, y)[0]
for name, model in models.items():
opt = Adam(model.p, .015)
rng = np.random.default_rng(BASE_SEED + 2)
curve = [{"epoch": 0, "train_loss": model_loss(name, model, splits["train"]),
"validation_loss": model_loss(name, model, splits["validation"])}]
started = time.perf_counter()
grad_max = 0.0
for epoch in range(1, 13):
order = rng.permutation(len(splits["train"]))
for start in range(0, len(order), 32):
y = splits["train"][order[start:start + 32]]
if name == "fixed_window":
_, grads = model.loss_grad(fixed_contexts(y, bos), y.reshape(-1))
else:
x = np.concatenate((np.full((len(y), 1), bos), y[:, :-1]), axis=1)
_, grads, _ = model.loss_grad(x, y)
grads, pre, _ = clip(grads, 1.0)
grad_max = max(grad_max, pre)
opt.step(model.p, grads)
if epoch in (1, 4, 8, 12):
curve.append({"epoch": epoch, "train_loss": model_loss(name, model, splits["train"]),
"validation_loss": model_loss(name, model, splits["validation"])})
elapsed = time.perf_counter() - started
loss = model_loss(name, model, splits["test"])
metrics[name] = {"test_nll_nats_per_token": loss, "test_perplexity": math.exp(loss),
"parameter_count": parameter_count(model.p), "updates": opt.t,
"epochs": 12, "wall_seconds_including_curve_evaluation": elapsed,
"max_unclipped_gradient_l2": grad_max}
curves[name] = curve
ng_losses = []
for context, target in zip(fixed_contexts(splits["test"], bos), splits["test"].reshape(-1)):
ng_losses.append(-math.log(ngram_prob(context)[target]))
ng_loss = float(np.mean(ng_losses))
metrics["trigram_add_0.1"] = {"test_nll_nats_per_token": ng_loss, "test_perplexity": math.exp(ng_loss),
"observed_contexts": len(counts), "stored_count_cells": len(counts) * v,
"updates": "one counting pass"}
def generate(name, temperature, seed):
rng = np.random.default_rng(seed)
prefix, output = [], []
for _ in range(20):
if name == "trigram_add_0.1":
dist = ngram_prob(([bos] * k + prefix)[-k:])
elif name == "fixed_window":
context = np.array([([bos] * k + prefix)[-k:]])
dist = softmax(fixed.forward(context)[0])[0]
else:
dist = softmax(rnn.forward(np.array([[bos] + prefix]))[0])[0, -1]
nxt = int(np.argmax(dist)) if temperature == 0 else int(rng.choice(v, p=softmax(np.log(dist) / temperature)))
output.append(vocab[nxt])
if vocab[nxt] == "<EOS>":
break
prefix.append(nxt)
return {"tokens": output, "terminated_with_eos": output[-1] == "<EOS>"}
generation = {name: {"greedy": generate(name, 0, BASE_SEED + 3),
"temperature_0.8": generate(name, .8, BASE_SEED + 3)} for name in metrics}
return {"status": "actual_small_synthetic_LM_training_single_seed", "seed_data": BASE_SEED,
"seed_initialization": BASE_SEED + 1, "seed_minibatches": BASE_SEED + 2,
"seed_sampling": BASE_SEED + 3, "vocabulary": vocab, "bos_input_only_id": bos,
"tokenization": "published grammar tokens; no subword model; EOS counted, BOS not a target",
"grammar_total_unique_sentences": 1024, "splits": {name: {"sentences": len(y), "predicted_tokens": y.size,
"sha256": digest_rows(y.tolist())} for name, y in splits.items()},
"window": 2, "embed": 8, "hidden": 24, "batch_sentences": 32,
"optimizer": "Adam lr=0.015 beta=(0.9,0.999) eps=1e-8; global gradient clipping at 1.0",
"metrics": metrics, "learning_curves": curves, "generation": generation,
"interpretation": "One grammar, one split, one initialization; no architecture ranking or natural-language claim."}
def delayed_bit_data(noise_length, template_counts=(96, 32, 32), seed=0, evidence_position=0):
"""Split unique distractor templates BEFORE making bit-0/bit-1 paired examples."""
rng, templates = np.random.default_rng(seed), set()
need = sum(template_counts)
if not 0 <= evidence_position <= noise_length:
raise ValueError("Evidence must be inserted before the QUERY token")
if 8 ** noise_length < need:
raise ValueError("Not enough unique distractor templates for disjoint splits")
ordered = []
while len(ordered) < need:
suffix = tuple(rng.integers(2, 10, noise_length).tolist())
if suffix not in templates:
templates.add(suffix)
ordered.append(suffix)
result, start = {}, 0
for name, n in zip(("train", "validation", "test"), template_counts):
group = ordered[start:start+n]
x = np.array([list(s[:evidence_position]) + [bit] + list(s[evidence_position:]) + [10]
for s in group for bit in (0, 1)])
y = np.array([bit for _ in group for bit in (0, 1)])
result[name] = (x, y, group)
start += n
return result
def classify_loss(model, x, y):
targets = np.zeros_like(x)
targets[:, -1] = y
mask = np.zeros_like(x, dtype=float)
mask[:, -1] = 1
loss, grads, aux = model.loss_grad(x, targets, mask)
pred = aux["probs"][:, -1].argmax(-1)
return loss, grads, {**aux, "accuracy": float((pred == y).mean()),
"both_correct_pair_fraction": float(np.all((pred == y).reshape(-1, 2), axis=1).mean())}
def gradient_check(model, x, y, seed=0, samples_per_tensor=6):
_, analytical, _ = classify_loss(model, x, y)
rng, checks = np.random.default_rng(seed), []
epsilon = 1e-5
for name, p in model.p.items():
for flat in rng.choice(p.size, min(samples_per_tensor, p.size), replace=False):
index = np.unravel_index(flat, p.shape)
original = p[index]
p[index] = original + epsilon
plus = classify_loss(model, x, y)[0]
p[index] = original - epsilon
minus = classify_loss(model, x, y)[0]
p[index] = original
numerical = (plus - minus) / (2 * epsilon)
exact = analytical[name][index]
checks.append({"parameter": name, "index": index, "analytic": exact, "numerical": numerical,
"absolute_error": abs(exact - numerical),
"scaled_error": abs(exact - numerical) / max(1, abs(exact), abs(numerical))})
maximum = max(c["absolute_error"] for c in checks)
assert maximum < 1e-7, (model.kind, maximum)
return {"epsilon": epsilon, "absolute_tolerance": 1e-7, "checked_elements": len(checks),
"max_absolute_error": maximum, "samples": checks}
def route_c():
seed = BASE_SEED + 4
data = delayed_bit_data(8, seed=seed)
split_sets = {n: set(t) for n, (_, _, t) in data.items()}
assert not (split_sets["train"] & split_sets["test"])
assert not (split_sets["validation"] & split_sets["test"])
assert not (split_sets["train"] & split_sets["validation"])
checks = {}
for gap in (3, 8, 32, 64):
d = delayed_bit_data(gap, (96, 32, 32), seed)
x, y, _ = d["test"]
suffix_groups = defaultdict(list)
for row, label in zip(x[:, -min(4, gap+1):], y):
suffix_groups[tuple(row)].append(int(label))
best_suffix_accuracy = sum(max(Counter(labels).values()) for labels in suffix_groups.values()) / len(y)
assert best_suffix_accuracy == .5
checks[str(gap)] = {"sequence_length": x.shape[1], "query_minus_relevant_position": gap+1,
"same_suffix_opposite_label_pairs": len(y)//2,
"best_deterministic_suffix_only_accuracy": best_suffix_accuracy}
small = delayed_bit_data(3, (2, 1, 1), seed)
gx, gy, _ = small["train"]
results = {}
for kind in ("rnn", "lstm"):
gcmodel = Recurrent(11, 2, embed=3, hidden=5, kind=kind, seed=seed)
gc = gradient_check(gcmodel, gx, gy, seed)
model = Recurrent(11, 2, embed=4, hidden=8, kind=kind, seed=seed)
x, y, _ = data["train"]
opt = Adam(model.p, .01)
initial, _, aux0 = classify_loss(model, x, y)
curve, grad_norms = [], []
started = time.perf_counter()
# Only a smoke run: 12 full-batch steps, NOT a capability comparison.
for step in range(12):
loss, grads, aux = classify_loss(model, x, y)
grads, before, after = clip(grads, .1)
grad_norms.append({"step": step, "before": before, "after": after})
opt.step(model.p, grads)
curve.append(loss)
final, _, aux1 = classify_loss(model, x, y)
tx, ty, _ = data["test"]
tloss, _, taux = classify_loss(model, tx, ty)
results[kind] = {"parameter_count": parameter_count(model.p), "gradient_check": gc,
"smoke_updates": 12, "initial_train_loss": initial, "final_train_loss": final,
"train_loss_before_each_update": curve, "test_loss": tloss,
"test_accuracy_smoke_only": taux["accuracy"],
"test_both_correct_pair_fraction_smoke_only": taux["both_correct_pair_fraction"],
"initial_hidden_gradient_l2_by_position": aux0["hidden_gradient_l2_by_position"],
"final_hidden_gradient_l2_by_position": aux1["hidden_gradient_l2_by_position"],
"gradient_clipping": grad_norms, "wall_seconds": time.perf_counter() - started}
return {"status": "generator_leakage_checks_BPTT_gradient_checks_and_short_optimization_smoke",
"seed": seed, "input_ids": {"bit_0": 0, "bit_1": 1, "independent_noise": "2..9", "QUERY": 10},
"smoke_noise_length": 8, "loss_locations": "QUERY only; no label inside the input",
"splits": {n: {"paired_templates": len(t), "examples": len(y), "label_mean": float(y.mean()),
"sha256": digest_rows(x.tolist())} for n, (x, y, t) in data.items()},
"suffix_checks": checks, "models": results,
"interpretation": "Finite gradients and optimization plumbing only. These 12 updates do not test RNN/LSTM long-memory superiority."}
TREE = {(): {"a": .6, "b": .4}, ("a",): {"x": .55, "<EOS>": .45},
("b",): {"y": .99, "<EOS>": .01}, ("a", "x"): {"<EOS>": 1.0},
("b", "y"): {"<EOS>": 1.0}}
def beam_search(tree, width=2, max_steps=6, length_alpha=0.0):
"""Keep finished hypotheses, never expand EOS; explicit max-step failure."""
beams, trace = [((), 0.0)], []
def score(item):
tokens, logp = item
return logp / (((5 + len(tokens)) / 6) ** length_alpha)
for step in range(max_steps):
candidates = []
for tokens, logp in beams:
if tokens and tokens[-1] == "<EOS>":
candidates.append((tokens, logp))
continue
probs = tree[tokens]
assert abs(sum(probs.values()) - 1) < 1e-12
candidates.extend((tokens + (token,), logp + math.log(prob)) for token, prob in probs.items() if prob > 0)
beams = sorted(candidates, key=lambda item: (-score(item), item[0]))[:width]
trace.append([{"tokens": s, "log_probability": l, "probability": math.exp(l), "score": score((s, l))} for s, l in beams])
if all(s[-1] == "<EOS>" for s, _ in beams):
break
finished = [item for item in beams if item[0][-1] == "<EOS>"]
if not finished:
raise RuntimeError("No EOS-completed beam; increase max_steps and report truncation")
best = max(finished, key=score)
return {"tokens": best[0], "probability": math.exp(best[1]), "log_probability": best[1],
"score": score(best), "width": width, "length_alpha": length_alpha, "trace": trace}
def exhaustive_paths(tree, prefix=(), probability=1.0):
if prefix and prefix[-1] == "<EOS>":
return [{"tokens": prefix, "probability": probability}]
return list(itertools.chain.from_iterable(exhaustive_paths(tree, prefix + (tok,), probability * p)
for tok, p in tree[prefix].items()))
def route_d():
greedy, beam = beam_search(TREE, 1), beam_search(TREE, 2)
all_paths = exhaustive_paths(TREE)
assert greedy["tokens"] == ("a", "x", "<EOS>")
assert beam["tokens"] == ("b", "y", "<EOS>")
assert abs(beam["probability"] - max(x["probability"] for x in all_paths)) < ATOL
fail_tree = {(): {"a": .4, "b": .35, "c": .25}, ("a",): {"<EOS>": .5, "x": .5},
("b",): {"<EOS>": .5, "x": .5}, ("c",): {"<EOS>": 1.0},
("a", "x"): {"<EOS>": 1.0}, ("b", "x"): {"<EOS>": 1.0}}
failed_beam = beam_search(fail_tree, 2)
fail_best = max(exhaustive_paths(fail_tree), key=lambda r: r["probability"])
assert failed_beam["probability"] < fail_best["probability"]
source = ["a", "c", "b"]
target = source[::-1] + ["<EOS>"]
decoder_input = ["<BOS>"] + target[:-1]
return {"status": "constructed_probability_tree_search_and_teacher_forcing_alignment",
"seed": None, "tree": {" ".join(k) or "<BOS>": v for k, v in TREE.items()},
"greedy": greedy, "beam_2": beam, "exhaustive_paths": all_paths,
"length_penalty_alpha_0.6": beam_search(TREE, 2, length_alpha=.6),
"beam_failure_counterexample": {"tree": {" ".join(k) or "<BOS>": v for k, v in fail_tree.items()},
"beam_2": failed_beam, "exhaustive_best": fail_best},
"seq2seq_example": {"source": source, "target": target, "decoder_input": decoder_input,
"prediction_rows": [{"t": t, "input": i, "gold_target": y} for t, (i, y) in enumerate(zip(decoder_input, target))]},
"interpretation": "Search is exact on this finite tree only when checked against enumeration; seq2seq training remains a design."}
def attention(q, k, v, allowed=None, scale=None):
scores = q @ k.T / (math.sqrt(q.shape[-1]) if scale is None else scale)
if allowed is not None:
if not np.all(allowed.any(-1)):
raise ValueError("Each query must have at least one allowed key")
scores = np.where(allowed, scores, -np.inf)
weights = softmax(scores)
return weights @ v, weights
def attention_reference(q, k, v, allowed=None, scale=None):
rows = []
for i in range(len(q)):
valid = list(range(len(k))) if allowed is None else np.flatnonzero(allowed[i]).tolist()
divisor = math.sqrt(q.shape[-1]) if scale is None else scale
scores = [sum(float(a) * float(b) for a, b in zip(q[i], k[j])) / divisor for j in valid]
exps = [math.exp(s - max(scores)) for s in scores]
denominator = sum(exps)
rows.append([sum(exps[a] / denominator * float(v[j, f]) for a, j in enumerate(valid)) for f in range(v.shape[-1])])
return np.array(rows)
def position_encoding(length, width, offset=0):
positions = np.arange(offset, offset + length)[:, None]
inv = 10000.0 ** (-np.arange(0, width, 2) / width)
out = np.zeros((length, width))
out[:, 0::2] = np.sin(positions * inv)
out[:, 1::2] = np.cos(positions * inv)
return out
def route_e():
classroom = classroom_parameters()
read_example = classroom["READ"]
q = np.array([read_example["q"], read_example["q2"], [1., 1.]])
k = np.array(read_example["K"], dtype=float)
v = np.array(read_example["V"], dtype=float)
y, a = attention(q, k, v, scale=1)
qc = q.copy()
qc[0] = read_example["q2"]
yq, aq = attention(qc, k, v, scale=1)
vc = np.array(read_example["V_changed"], dtype=float)
yv, av = attention(q, k, vc, scale=1)
causal = np.tri(3, dtype=bool)
ym, am = attention(q, k, v, causal, scale=1)
vf = v.copy()
vf[1:] += 1000
ymf, _ = attention(q, k, vf, causal, scale=1)
reference_error = np.max(np.abs(ym - attention_reference(q, k, v, causal, scale=1)))
classroom_errors = {name: float(np.max(np.abs(actual[0] - np.array(read_example[name]["output"])[0])))
for name, actual in (("base", y), ("query_changed", yq), ("value_changed", yv))}
scaled_y, scaled_a = attention(q, k, v, causal)
scaled_error = float(np.max(np.abs(scaled_y - attention_reference(q, k, v, causal))))
assert max(classroom_errors.values()) < ATOL and scaled_error < ATOL
assert reference_error < ATOL
assert np.max(np.abs(a - av)) == 0
assert np.max(np.abs((yv - y) - a[:, 1, None] * (vc[1] - v[1]))) < ATOL
assert np.max(np.abs(ymf[0] - ym[0])) == 0
rng = np.random.default_rng(BASE_SEED + 5)
x = rng.normal(size=(4, 4))
wq, wk, wv = [rng.normal(size=(4, 3)) / 2 for _ in range(3)]
p = np.array([2, 0, 3, 1])
read = lambda z, mask=None: attention(z @ wq, z @ wk, z @ wv, mask)[0]
error = float(np.max(np.abs(read(x[p]) - read(x)[p])))
pe = position_encoding(4, 4)
position_delta = float(np.max(np.abs(read(x[p] + pe) - read(x + pe)[p])))
causal_delta = float(np.max(np.abs(read(x[p], np.tri(4, dtype=bool)) - read(x, np.tri(4, dtype=bool))[p])))
assert error < ATOL
assert position_delta > 1e-5 and causal_delta > 1e-5
return {"status": "shared_classroom_READ_forward_and_independent_scalar_reference", "seed_permutation": BASE_SEED + 5,
"classroom_source": classroom["source_path"], "classroom_source_sha256": classroom["source_sha256"],
"basic_READ_score_divisor": 1.0, "classroom_reference_max_abs_errors": classroom_errors,
"Q": q, "K": k, "V": v, "softmax_axis": "keys / last axis", "weights": a, "output": y,
"query_changed": qc, "query_changed_weights": aq, "query_changed_output": yq,
"value_changed": vc, "value_changed_weights_max_abs_delta": float(np.max(np.abs(a-av))), "value_changed_output": yv,
"causal_allowed": causal, "causal_weights": am, "causal_output": ym,
"scaled_Transformer_variant": {"score_divisor": math.sqrt(2), "weights": scaled_a, "output": scaled_y,
"scalar_reference_max_abs_error": scaled_error},
"scalar_reference_max_abs_error": float(reference_error),
"masked_future_value_perturbation_query_0_max_abs_delta": float(np.max(np.abs(ymf[0]-ym[0]))),
"permutation": p, "no_position_full_attention_equivariance_max_abs_error": error,
"fixed_absolute_position_permutation_max_abs_delta": position_delta,
"fixed_causal_graph_permutation_max_abs_delta": causal_delta, "absolute_tolerance": ATOL,
"interpretation": "Full unpositioned self-attention is equivariant; fixed positions or a fixed causal graph change the experiment."}
def layer_norm(x, gamma, beta, eps=1e-5):
return (x - x.mean(-1, keepdims=True)) / np.sqrt(x.var(-1, keepdims=True) + eps) * gamma + beta
def row_norm_reference(row, gamma, beta, eps=1e-5):
mean = sum(float(v) for v in row) / len(row)
variance = sum((float(v) - mean) ** 2 for v in row) / len(row)
return np.array([(float(v) - mean) / math.sqrt(variance + eps) * float(g) + float(b) for v, g, b in zip(row, gamma, beta)])
def make_block(rng, width=8, ff=16):
p = {key: rng.normal(0, 1/math.sqrt(width), (width, width)) for key in ("Wq", "Wk", "Wv", "Wo")}
p.update({"W1": rng.normal(0, 1/math.sqrt(width), (width, ff)), "b1": np.zeros(ff),
"W2": rng.normal(0, 1/math.sqrt(ff), (ff, width)), "b2": np.zeros(width),
"g1": np.ones(width), "bN1": np.zeros(width), "g2": np.ones(width), "bN2": np.zeros(width)})
return p
def multihead(x, p, heads=2, causal=False):
steps, width = x.shape
dk = width // heads
q, k, v = [(x @ p[key]).reshape(steps, heads, dk).transpose(1, 0, 2) for key in ("Wq", "Wk", "Wv")]
scores = (q @ k.transpose(0, 2, 1)) / math.sqrt(dk)
if causal:
scores = np.where(np.tri(steps, dtype=bool)[None], scores, -np.inf)
a = softmax(scores)
per_head = a @ v
concat = per_head.transpose(1, 0, 2).reshape(steps, width)
return concat @ p["Wo"], {"Q": q, "K": k, "V": v, "weights": a,
"head_outputs": per_head, "concat": concat}
def route_f():
classroom = classroom_parameters()
block = classroom["BLOCK"]
x = np.array(block["X"], dtype=float)
p = {"Wq": np.concatenate([block["WQ1"], block["WQ2"]], axis=1),
"Wk": np.concatenate([block["WK1"], block["WK2"]], axis=1),
"Wv": np.concatenate([block["WV1"], block["WV2"]], axis=1),
"Wo": np.array(block["WO"], dtype=float), "W1": np.array(block["W1"], dtype=float),
"W2": np.array(block["W2"], dtype=float), "b1": np.zeros(6), "b2": np.zeros(4),
"g1": np.ones(4), "bN1": np.zeros(4), "g2": np.ones(4), "bN2": np.zeros(4)}
att, trace = multihead(x, p)
residual_1 = x + att
n1 = layer_norm(residual_1, p["g1"], p["bN1"])
ff_hidden = np.maximum(n1 @ p["W1"] + p["b1"], 0)
ff = ff_hidden @ p["W2"] + p["b2"]
residual_2 = n1 + ff
y = layer_norm(residual_2, p["g2"], p["bN2"])
# Independent head and per-position paths, not a comparison against the same call.
refs = [attention_reference(trace["Q"][h], trace["K"][h], trace["V"][h]) for h in range(2)]
att_ref = np.concatenate(refs, axis=-1) @ p["Wo"]
n1_ref = np.array([row_norm_reference(row, p["g1"], p["bN1"]) for row in x + att_ref])
ff_ref = np.array([np.maximum(row @ p["W1"] + p["b1"], 0) @ p["W2"] + p["b2"] for row in n1_ref])
y_ref = np.array([row_norm_reference(row, p["g2"], p["bN2"]) for row in n1_ref + ff_ref])
err = float(np.max(np.abs(y - y_ref)))
assert err < ATOL
classroom_error = float(np.max(np.abs(y[0] - np.array(block["z"]))))
assert classroom_error < ATOL
z = n1.copy()
z[0] += .5
ffz = np.maximum(z @ p["W1"] + p["b1"], 0) @ p["W2"] + p["b2"]
assert np.max(np.abs(ffz[1:] - ff[1:])) == 0
shifted = layer_norm(residual_1 + np.arange(1, len(x)+1)[:, None], p["g1"], p["bN1"])
return {"status": "shared_classroom_BLOCK_untrained_post_norm_forward", "seed": None,
"classroom_source": classroom["source_path"], "classroom_source_sha256": classroom["source_sha256"],
"classroom_token_0_reference_max_abs_error": classroom_error,
"config": {"tokens": 3, "d_model": 4, "heads": 2, "d_k": 2, "d_v": 2, "d_ff": 6,
"batch_axis": "omitted for a single sequence", "dropout": 0, "LayerNorm_epsilon": 1e-5},
"parameter_shapes": {key: v.shape for key, v in p.items()}, "parameter_count": parameter_count(p),
"activation_shapes": {key: a.shape for key, a in {"input": x, **trace, "attention_output": att,
"residual_1": residual_1, "norm_1": n1, "ff_hidden": ff_hidden,
"ff_output": ff, "residual_2": residual_2, "block_output": y}.items()},
"tracked_token_index": 0,
"token_0_trace": {"input": x[0], "head_0": trace["head_outputs"][0, 0], "head_1": trace["head_outputs"][1, 0],
"concat": trace["concat"][0], "attention_output": att[0], "residual_1": residual_1[0],
"norm_1": n1[0], "ff_hidden": ff_hidden[0], "ff_output": ff[0],
"residual_2": residual_2[0], "output": y[0]},
"output": y, "independent_reference_max_abs_error": err,
"LayerNorm_output_feature_means": y.mean(-1), "LayerNorm_output_feature_variances": y.var(-1),
"LayerNorm_per_row_constant_shift_max_abs_error": float(np.max(np.abs(n1-shifted))),
"FFN_other_positions_after_one_position_change_max_abs_delta": float(np.max(np.abs(ffz[1:]-ff[1:]))),
"absolute_tolerance": ATOL, "interpretation": "Traceable computation and shape checks, not a trained Transformer or translation score."}
class TinyDecoder:
"""Two pre-norm causal blocks; fixed random weights, no training/dropout.
full() computes every prefix position with batched matrix operations.
incremental() independently computes only the new position, loops over heads,
and normalizes a row with a scalar reference implementation.
"""
def __init__(self, seed=0, width=8, heads=2, layers=2, vocab=17):
rng = np.random.default_rng(seed)
self.width, self.heads, self.vocab = width, heads, vocab
self.embedding = rng.normal(0, .5, (vocab, width))
self.blocks = [make_block(rng, width, 2 * width) for _ in range(layers)]
self.final_g = rng.uniform(.8, 1.2, width)
self.final_b = rng.normal(0, .03, width)
self.output = rng.normal(0, 1/math.sqrt(width), (width, vocab))
def full(self, tokens, position_offset=0):
x = self.embedding[np.asarray(tokens)] + position_encoding(len(tokens), self.width, position_offset)
caches = []
for p in self.blocks:
n = layer_norm(x, p["g1"], p["bN1"])
a, tr = multihead(n, p, self.heads, causal=True)
caches.append({"K": tr["K"].copy(), "V": tr["V"].copy()})
x = x + a
n = layer_norm(x, p["g2"], p["bN2"])
x = x + np.maximum(n @ p["W1"] + p["b1"], 0) @ p["W2"] + p["b2"]
return layer_norm(x, self.final_g, self.final_b) @ self.output, caches
def incremental(self, token, caches, position):
d = self.width // self.heads
# Separate scalar absolute-position calculation from the vectorized full path.
pe = np.array([math.sin(position / 10000 ** (j / self.width)) if j % 2 == 0
else math.cos(position / 10000 ** ((j-1) / self.width)) for j in range(self.width)])
x = self.embedding[token] + pe
new_caches = []
for p, cache in zip(self.blocks, caches):
n = row_norm_reference(x, p["g1"], p["bN1"])
q, kn, vn = [(n @ p[key]).reshape(self.heads, d) for key in ("Wq", "Wk", "Wv")]
ks = np.concatenate((cache["K"], kn[:, None]), axis=1)
vs = np.concatenate((cache["V"], vn[:, None]), axis=1)
# The new row may attend to all cached positions and itself; no square mask.
heads = [attention_reference(q[h:h+1], ks[h], vs[h])[0] for h in range(self.heads)]
x = x + np.concatenate(heads) @ p["Wo"]
n = row_norm_reference(x, p["g2"], p["bN2"])
x = x + np.maximum(n @ p["W1"] + p["b1"], 0) @ p["W2"] + p["b2"]
new_caches.append({"K": ks, "V": vs})
return row_norm_reference(x, self.final_g, self.final_b) @ self.output, new_caches
def empty_cache(self):
shape = (self.heads, 0, self.width // self.heads)
return [{"K": np.empty(shape), "V": np.empty(shape)} for _ in self.blocks]
def timing_summary(samples):
return {"median_ms": float(np.median(samples) * 1000), "min_ms": min(samples) * 1000,
"max_ms": max(samples) * 1000, "repeats": len(samples)}
def route_g():
seed = BASE_SEED + 7
model = TinyDecoder(seed)
tokens = np.random.default_rng(seed + 1).integers(0, model.vocab, 48).tolist()
full_logits, _ = model.full(tokens)
cases = []
maximum = 0.0
# Include no-prefill, several prefill sizes, and a nonzero absolute position offset.
for prefill, offset in ((0, 0), (1, 0), (7, 0), (16, 0), (7, 11)):
if prefill:
_, caches = model.full(tokens[:prefill], offset)
else:
caches = model.empty_cache()
errors = []
for t in range(prefill, len(tokens)):
incremental_logits, caches = model.incremental(tokens[t], caches, t + offset)
fresh_logits, _ = model.full(tokens[:t+1], offset)
error = float(np.max(np.abs(incremental_logits - fresh_logits[-1])))
errors.append(error)
maximum = max(maximum, max(errors))
cases.append({"prefill_tokens": prefill, "absolute_position_offset": offset,
"decode_steps": len(errors), "max_absolute_logits_error": max(errors),
"errors_by_step": errors})
assert maximum < ATOL
prefix_logits, prefix_cache = model.full(tokens[:16])
prefix_invariance = float(np.max(np.abs(full_logits[:16] - prefix_logits)))
assert prefix_invariance < ATOL
correct, _ = model.incremental(tokens[16], prefix_cache, 16)
wrong_position, _ = model.incremental(tokens[16], prefix_cache, 0)
position_failure = float(np.max(np.abs(correct - wrong_position)))
changed = tokens[:17].copy()
changed[0] = (changed[0] + 1) % model.vocab
changed_logits, _ = model.full(changed)
stale_failure = float(np.max(np.abs(correct - changed_logits[-1])))
assert position_failure > 1e-5 and stale_failure > 1e-5
# Actual autoregressive choices: start with the same prefix, choose independently.
prompt = tokens[:7]
slow_tokens, fast_tokens = prompt.copy(), prompt.copy()
fast_logits, cache = model.full(prompt)
last_fast = fast_logits[-1]
for _ in range(8):
slow_logits, _ = model.full(slow_tokens)
slow_tokens.append(int(np.argmax(slow_logits[-1])))
fast_token = int(np.argmax(last_fast))
fast_tokens.append(fast_token)
last_fast, cache = model.incremental(fast_token, cache, len(fast_tokens)-1)
assert slow_tokens == fast_tokens
# Timed work uses predetermined tokens, so both branches process identical input.
prefill_samples, cached_samples, uncached_samples = [], [], []
for repeat in range(6):
started = time.perf_counter()
_, caches = model.full(tokens[:16])
prefill_time = time.perf_counter() - started
started = time.perf_counter()
for t in range(16, 48):
_, caches = model.incremental(tokens[t], caches, t)
cached_time = time.perf_counter() - started
started = time.perf_counter()
for t in range(16, 48):
model.full(tokens[:t+1])
uncached_time = time.perf_counter() - started
if repeat: # one warm-up, then five measurements
prefill_samples.append(prefill_time)
cached_samples.append(cached_time)
uncached_samples.append(uncached_time)
payload_bytes = sum(a.nbytes for c in caches for a in c.values())
context_task_checks = []
for length in (32, 128, 512):
for fraction in (.1, .5, .9):
position = int((length - 2) * fraction)
tasks = delayed_bit_data(length-2, (16, 8, 8), seed+2, evidence_position=position)
x, labels, _ = tasks["test"]
assert np.array_equal(x[0::2, -4:], x[1::2, -4:])
assert np.all(labels[0::2] != labels[1::2])
assert np.all(x[:, position] == labels)
context_task_checks.append({"sequence_length": length, "evidence_position_zero_based": position,
"distance_to_query": length-1-position, "test_paired_examples": len(labels),
"identical_last_4_tokens_opposite_labels": True,
"test_sha256": digest_rows(x.tolist())})
return {"status": "untrained_decoder_equivalence_and_CPU_microtiming_not_language_ability",
"implementation_version": VERSION, "seed_weights": seed, "seed_tokens": seed+1,
"config": {"layers": 2, "d_model": 8, "heads": 2, "d_k": 4, "d_ff": 16, "vocab": 17,
"architecture": "pre-norm causal decoder; separate final LayerNorm and vocab projection",
"position": "absolute sinusoidal, float64", "dtype": "float64", "eval": True, "dropout": 0.0,
"padding": False, "batch_size": 1, "prefix_immutable": True},
"input_token_ids": tokens, "cases": cases, "max_absolute_logits_error": maximum,
"absolute_tolerance": ATOL, "relative_tolerance": 0.0,
"old_position_logits_after_future_append_max_abs_error": prefix_invariance,
"wrong_position_negative_control_max_abs_delta": position_failure,
"changed_prefix_stale_cache_negative_control_max_abs_delta": stale_failure,
"greedy_generated_ids_after_prompt": fast_tokens[len(prompt):], "greedy_paths_equal": slow_tokens == fast_tokens,
"timing": {"clock": "time.perf_counter", "warmups": 1, "prefill_tokens": 16, "decode_tokens": 32,
"prefill": timing_summary(prefill_samples), "cached_decode_total": timing_summary(cached_samples),
"noncached_decode_total": timing_summary(uncached_samples),
"note": "Python scalar cached reference versus vectorized NumPy full-prefix; not an optimized cache speed benchmark."},
"cache_payload_bytes_at_48_tokens": payload_bytes,
"cache_payload_formula_bytes": 2 * 2 * 48 * 8 * 8,
"decode_projected_token_layers": {"cached": 32 * 2, "noncached": sum(range(17, 49)) * 2},
"decode_attention_score_elements": {"cached": sum(range(17, 49)) * 2 * 2,
"noncached": sum(n*n for n in range(17, 49)) * 2 * 2},
"effective_context_generator_checks": context_task_checks,
"effective_context_ability_test": "GENERATOR CHECKED; MODEL TRAINING/EVALUATION PENDING: evaluate a trained retrieval model on held-out balanced paired templates by evidence position; this decoder is random and untrained."}
def json_default(obj):
if isinstance(obj, np.ndarray):
return obj.tolist()
if isinstance(obj, np.generic):
return obj.item()
raise TypeError(type(obj).__name__)
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--routes", default="ABCDEFG", help="Subset of A B C D E F G, default all")
parser.add_argument("--output", type=Path, default=Path(__file__).parent / "results" / "r21-results.json")
args = parser.parse_args()
route_funcs = {"A": route_a, "B": route_b, "C": route_c, "D": route_d, "E": route_e, "F": route_f, "G": route_g}
chosen = list(dict.fromkeys(args.routes.upper().replace(",", "").replace(" ", "")))
if any(name not in route_funcs for name in chosen):
parser.error("--routes must contain only A..G")
started = time.perf_counter()
cpu = "unknown"
cpuinfo = Path("/proc/cpuinfo")
if cpuinfo.exists():
matches = re.findall(r"model name\s*:\s*(.+)", cpuinfo.read_text())
if matches:
cpu = matches[0]
result = {"metadata": {"implementation_version": VERSION, "generated_at_utc": datetime.now(timezone.utc).isoformat(),
"python": sys.version, "numpy": np.__version__, "platform": platform.platform(), "cpu": cpu,
"dtype": "float64", "torch_installed_or_used": False,
"thread_environment": {key: os.environ.get(key) for key in ("OPENBLAS_NUM_THREADS", "OMP_NUM_THREADS", "MKL_NUM_THREADS")},
"source_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
"run_scope": "local CPU only; no Deepnote API calls, no cloud runs, no publication"}, "routes": {}}
for name in chosen:
start = time.perf_counter()
result["routes"][name] = route_funcs[name]()
result["routes"][name]["route_wall_seconds"] = time.perf_counter() - start
print(f"{name}: {result['routes'][name]['status']} ({result['routes'][name]['route_wall_seconds']:.3f}s)", flush=True)
result["metadata"]["total_wall_seconds"] = time.perf_counter() - started
args.output.parent.mkdir(parents=True, exist_ok=True)
args.output.write_text(json.dumps(result, ensure_ascii=False, indent=2, default=json_default, allow_nan=False) + "\n")
print(f"Saved {args.output}")
if __name__ == "__main__":
main()