| """Weight-space programming with Beam: Beam-1.0-501B writes the weights of a small neural network |
| that plays SUPERBEAM, with no training (no data, no loss, no gradients), in the spirit of |
| https://rikiyat.github.io/blog/weight-space-programming/ . |
| |
| The network: 8 features -> 16 ReLU -> 6 action logits, action = argmax. Features are read from the |
| game's registers/RAM each turn (see `features`). Beam gets the game rules, the feature definitions and |
| the network shape, and replies with the weights as JSON. Each candidate is scored on TRAINING levels |
| (seeds 0..TRAIN_LEVELS-1); Beam gets per-level results as feedback and may revise. The candidate with |
| the best training score is then evaluated once on the 20 held-out TEST levels used by the leaderboard |
| (seeds 1000000..1000019, difficulty 0.5, one attempt per level). |
| |
| modal run scripts/beam_weights.py [--rounds 4] |
| -> results/beam_weights.json (all rounds: Beam's reply, weights, scores) and the test result. |
| |
| The Hugging Face token stays in the Modal secret `huggingface-secret` and is never printed. |
| """ |
| import json |
| import os |
| import re |
| import sys |
| import time |
| import urllib.error |
| import urllib.request |
|
|
| import modal |
| import numpy as np |
|
|
| ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) |
| sys.path.insert(0, ROOT) |
| OUT = os.path.join(ROOT, "results", "beam_weights.json") |
| MODEL = "reflection/Beam-1.0-501B-A23B:baseten" |
| TRAIN_LEVELS = 10 |
| TEST_SEEDS = list(range(1000000, 1000020)) |
| N_IN, N_HID, N_OUT = 8, 16, 6 |
| ACTIONS = ["noop", "left", "right", "jump", "right+jump", "left+jump"] |
| FEATURES = [ |
| ("standing", "1 if you stand on ground or on a pipe this frame, else 0 (in the air)"), |
| ("height", "how high your feet are: 0 on ground level, 8 on a pipe top, 16 on a tall pipe top (px)"), |
| ("gap_dist", "px from your front (right) edge to the start of the next gap in the ground ahead; 0 if the gap is already under your front edge; 32 if no gap within 32 px"), |
| ("gap_width", "width of that gap in px (8 or 16); 0 if none"), |
| ("wall_dist", "px from your front edge to the next block whose top is higher than your feet (a pipe seen from the ground, or a tall pipe seen from a pipe); 32 if none within 32 px"), |
| ("wall_height", "how many px higher that block's top is than your feet (8 or 16); 0 if none"), |
| ("walker_dist", "px from your front edge to the nearest walker ahead of you (0 if touching); 32 if none within 32 px"), |
| ("vy", "vertical speed in px per frame: negative while rising, positive while falling, 0 when standing"), |
| ] |
|
|
| app = modal.App("beamboot-beam-weights") |
|
|
|
|
| @app.function(image=modal.Image.debian_slim(python_version="3.12").pip_install("numpy"), |
| secrets=[modal.Secret.from_name("huggingface-secret")], timeout=600) |
| def beam_chat(messages, model=MODEL, max_tokens=32000): |
| token = (os.environ.get("HF_TOKEN") or os.environ.get("HUGGING_FACE_HUB_TOKEN") |
| or os.environ.get("HUGGINGFACE_TOKEN") or os.environ.get("HUGGINGFACEHUB_API_TOKEN")) |
| if not token: |
| raise RuntimeError("Hugging Face token not found in the Modal secret") |
| payload = {"model": model, "messages": messages, "max_tokens": max_tokens} |
| req = urllib.request.Request("https://router.huggingface.co/v1/chat/completions", |
| data=json.dumps(payload).encode(), |
| headers={"Authorization": "Bearer " + token, "Content-Type": "application/json"}) |
| t0 = time.time() |
| try: |
| with urllib.request.urlopen(req, timeout=580) as response: |
| out = json.load(response) |
| except urllib.error.HTTPError as error: |
| raise RuntimeError("HF router HTTP %d: %s" % (error.code, error.read(800).decode(errors="replace"))) from None |
| out["_latency_s"] = round(time.time() - t0, 1) |
| return out |
|
|
|
|
| |
| def features(env): |
| """The 8 features, read from the emulator's registers and RAM.""" |
| regs = env.backend.regs() |
| x, y, cam = regs[0xA], regs[0xB], regs[0xD] |
| wx = cam * 8 + x |
| front = wx + 8 |
| standing = 1.0 if regs[0xE] == y and y in (18, 10, 2) else 0.0 |
| feet = y + 6 |
| height = float(max(0, 24 - feet)) |
| lv = env.live_level() |
| ncol = len(lv) |
|
|
| def top(c): |
| if c >= ncol: |
| return 24 |
| t = lv[c] & 3 |
| return None if t == 0 else {1: 24, 2: 16, 3: 8}[t] |
|
|
| gap_dist, gap_width = 32.0, 0.0 |
| wall_dist, wall_height = 32.0, 0.0 |
| c0 = max(0, wx // 8) |
| for c in range(c0, min(ncol, c0 + 7)): |
| left = c * 8 |
| if left + 8 <= front - 8: |
| continue |
| d = max(0, left - front) |
| if d > 32: |
| break |
| t = top(c) |
| if t is None and gap_dist == 32.0: |
| w = 0 |
| while c + w < ncol and top(c + w) is None: |
| w += 1 |
| gap_dist, gap_width = float(d), float(8 * w) |
| if t is not None and t < feet and wall_dist == 32.0 and left >= front - 1: |
| wall_dist, wall_height = float(d), float(feet - t) |
| walker_dist = 32.0 |
| et = env.sym["ETAB"] |
| for e in range(2): |
| alive, col, off = (env.backend.ram(et + 4 * e + i) for i in range(3)) |
| if not alive: |
| continue |
| d = col * 8 + off - front |
| if -8 <= d <= 32: |
| walker_dist = min(walker_dist, float(max(0, d))) |
| vy = 0.0 if standing else float(regs[0xC] - 8) |
| return np.array([standing, height, gap_dist, gap_width, wall_dist, wall_height, walker_dist, vy]) |
|
|
|
|
| def act(net, f): |
| h = np.maximum(0.0, net["W1"] @ f + net["b1"]) |
| return int(np.argmax(net["W2"] @ h + net["b2"])) |
|
|
|
|
| def play(net, seed, max_frames=600): |
| from beamboot.superbeam_gym import SuperBeamEnv |
| env = SuperBeamEnv(level=seed, difficulty=0.5, frame_skip=4, max_frames=max_frames) |
| env.reset() |
| term = trunc = False |
| info, last_f = {}, None |
| while not (term or trunc): |
| last_f = features(env) |
| _, _, term, trunc, info = env.step(act(net, last_f)) |
| cause = None |
| if info.get("dead"): |
| cause = ("fell into a gap" if last_f[2] <= 8 else "hit a walker" if last_f[6] <= 8 else "died") |
| elif not info.get("won"): |
| cause = "ran out of time (stuck: e.g. walking into a pipe wall, or standing still)" |
| return {"seed": seed, "won": bool(info.get("won")), "progress": round(float(info.get("best_progress", 0)), 3), |
| "coins": int(info.get("coins", 0)), "cause": cause, |
| "where": None if last_f is None else dict(zip([n for n, _ in FEATURES], [round(float(v), 1) for v in last_f]))} |
|
|
|
|
| def score(net, seeds): |
| res = [play(net, s) for s in seeds] |
| return {"won": sum(r["won"] for r in res), "levels": len(res), |
| "mean_progress": round(float(np.mean([r["progress"] for r in res])), 3), "detail": res} |
|
|
|
|
| def parse_net(text): |
| blocks = re.findall(r"```(?:json)?\s*(\{.*?\})\s*```", text, re.S) or re.findall(r"(\{\s*\"W1\".*\})", text, re.S) |
| if not blocks: |
| raise ValueError("no JSON object with W1/b1/W2/b2 found") |
| d = json.loads(blocks[-1]) |
| net = {k: np.array(d[k], dtype=float) for k in ("W1", "b1", "W2", "b2")} |
| shapes = {"W1": (N_HID, N_IN), "b1": (N_HID,), "W2": (N_OUT, N_HID), "b2": (N_OUT,)} |
| for k, s in shapes.items(): |
| if net[k].shape != s: |
| raise ValueError("%s has shape %s, expected %s" % (k, net[k].shape, s)) |
| return net |
|
|
|
|
| SYSTEM = """You are a weight-space programmer. You write the weights of a small neural network directly, by reasoning about what each unit should compute. There is no training, no data and no gradient descent: the numbers you write are the final network.""" |
|
|
|
|
| def task_prompt(): |
| sys.path.insert(0, os.path.join(ROOT, "scripts")) |
| from beam_play import SYSTEM as RULES |
| feats = "\n".join(" x[%d] %s: %s" % (i, n, d) for i, (n, d) in enumerate(FEATURES)) |
| acts = "\n".join(" %d %s" % (i, a) for i, a in enumerate(ACTIONS)) |
| rules = RULES.split("Actions:")[0].strip() |
| return f"""Write the weights of a neural network that plays SUPERBEAM well. |
| |
| THE GAME (these are the rules a player is given): |
| {rules} |
| |
| THE NETWORK. Every 4 frames it reads 8 numbers x[0..7] and picks one action: |
| h = relu(W1 @ x + b1) # W1: 16 x 8, b1: 16 |
| logits = W2 @ h + b2 # W2: 6 x 16, b2: 6 |
| action = argmax(logits) |
| Inputs (read from the game's memory; distances in pixels): |
| {feats} |
| Actions (index: name): |
| {acts} |
| |
| Design it like a program: give each hidden unit a job (for example "a gap is close", "a wall is close |
| and I stand on the ground", "a walker is close", "I am in the air"), set its weights and bias so it |
| switches on only in that situation (ReLU), then wire the units to the action logits so that the right |
| action wins. Keep "right" as the default so the player keeps moving. Unused units can be all zeros. |
| |
| Reply with your reasoning, then ONE JSON object in a ```json block with keys "W1" (16 lists of 8 numbers), |
| "b1" (16 numbers), "W2" (6 lists of 16 numbers), "b2" (6 numbers). Numbers only, no expressions.""" |
|
|
|
|
| def feedback(sc): |
| lines = ["Your network on %d training levels: won %d, mean progress %.0f%%." % (sc["levels"], sc["won"], 100 * sc["mean_progress"])] |
| for r in sc["detail"]: |
| if r["won"]: |
| lines.append(" level %d: won" % r["seed"]) |
| else: |
| lines.append(" level %d: %s at %.0f%% of the level; inputs at that moment: %s" |
| % (r["seed"], r["cause"], 100 * r["progress"], json.dumps(r["where"]))) |
| lines.append("Improve the weights to win more levels. Reply with reasoning, then the full JSON (all four arrays).") |
| return "\n".join(lines) |
|
|
|
|
| @app.local_entrypoint() |
| def main(rounds: int = 4): |
| train = list(range(TRAIN_LEVELS)) |
| messages = [{"role": "system", "content": SYSTEM}, {"role": "user", "content": task_prompt()}] |
| log, best = [], None |
| for r in range(rounds): |
| out = beam_chat.remote(messages) |
| msg = out["choices"][0]["message"] |
| text = msg.get("content") or "" |
| reasoning = msg.get("reasoning_content") or msg.get("reasoning") or "" |
| finish = out["choices"][0].get("finish_reason") |
| entry = {"round": r + 1, "latency_s": out.get("_latency_s"), "finish_reason": finish, |
| "usage": out.get("usage"), "reply": text, "reasoning_chars": len(reasoning)} |
| print("round %d: finish=%s content=%d chars reasoning=%d chars" % (r + 1, finish, len(text), len(reasoning)), flush=True) |
| try: |
| try: |
| net = parse_net(text) |
| except ValueError: |
| net = parse_net(reasoning) |
| entry["parsed_from"] = "reasoning_content" |
| sc = score(net, train) |
| entry.update(weights={k: v.tolist() for k, v in net.items()}, train=sc) |
| print("round %d: train won %d/%d, mean progress %.0f%% (%.0fs)" % (r + 1, sc["won"], sc["levels"], 100 * sc["mean_progress"], out.get("_latency_s", 0)), flush=True) |
| if best is None or (sc["won"], sc["mean_progress"]) > (best[1]["won"], best[1]["mean_progress"]): |
| best = (net, sc, r + 1) |
| follow = feedback(sc) |
| except Exception as e: |
| entry["error"] = str(e) |
| print("round %d: %s" % (r + 1, e), flush=True) |
| follow = "I could not use that: %s. Reply with the full JSON object in a ```json block." % e |
| log.append(entry) |
| json.dump({"model": MODEL, "rounds": log}, open(OUT, "w"), indent=1) |
| if best and best[1]["won"] == best[1]["levels"]: |
| break |
| turn = {"role": "assistant", "content": text} |
| if reasoning: |
| turn["reasoning_content"] = reasoning |
| messages += [turn, {"role": "user", "content": follow}] |
| result = {"model": MODEL, "network": "8 -> 16 ReLU -> 6, weights written by Beam, no training", |
| "rounds": log} |
| if best: |
| test = score(best[0], TEST_SEEDS) |
| result.update(best_round=best[2], best_train=best[1], test=test, |
| best_weights={k: v.tolist() for k, v in best[0].items()}) |
| print("TEST (20 held-out levels): won %d/20, mean progress %.0f%% (round %d weights)" |
| % (test["won"], 100 * test["mean_progress"], best[2]), flush=True) |
| json.dump(result, open(OUT, "w"), indent=1) |
| print("wrote", OUT) |
|
|
