| """The full comparison: our diffusion model against classical solvers and the |
| published neural results, on a benchmark whose difficulty is stated rather than |
| assumed. |
| |
| Published Sudoku numbers are reported on different puzzle distributions, and |
| quoting them side by side without saying so is exactly the error that cost this |
| project its SATNet claim. So: |
| |
| * classical baselines are MEASURED on our test set, not cited; |
| * our model is reported STRATIFIED BY CLUE COUNT, so a reader can compare at |
| whatever difficulty a published number used; |
| * published numbers appear only in a clearly separated block, with their |
| distribution named. |
| |
| Three classical baselines, each measuring something different: |
| |
| propagation naked + hidden singles to fixpoint, NO search. The "free" solver. |
| backtracking full search. Solves everything, so the informative quantity is |
| its COST - how many search nodes it needs - which is the honest |
| yardstick for "how hard is this puzzle really". |
| greedy-MRV one pass of most-constrained-variable assignment without |
| backtracking: what a single confident sweep achieves. |
| |
| For the diffusion model the quantity of interest is board accuracy at a small |
| fixed number of forward passes, which is the comparison that makes the |
| generative framing worth anything. |
| """ |
| import argparse, glob, json, os, sys |
| from collections import defaultdict |
|
|
| import numpy as np |
| import torch |
|
|
| sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) |
| from model import RecurrentDenoiser, Transformer |
| import diffusion as dfn |
| from sudoku import CELLS, GRID, SudokuTokenizer |
| from gen_hard_sudoku import candidates, count_solutions, PEERS |
|
|
| |
| PUBLISHED = [ |
| ("SATNet (Wang et al. ICML 2019)", "SATNet easy split, ~36 clues", 0.983), |
| ("SATNet, evaluated on RRN hard", "RRN hard, 17 clues", 0.061), |
| ("RRN (Palm et al. NeurIPS 2018)", "RRN hard, 17 clues", 0.967), |
| ("Recurrent Transformer (Yang et al. 2023)", "RRN hard, 17 clues", 0.995), |
| ("Recurrent Transformer", "SATNet easy split", 1.000), |
| ] |
| PARAMS = {"Recurrent Transformer (Yang et al. 2023)": "211k", |
| "RRN (Palm et al. NeurIPS 2018)": "201k", |
| "SATNet (Wang et al. ICML 2019)": "618k"} |
|
|
|
|
| def units(): |
| us = [] |
| for r in range(GRID): |
| us.append([r * GRID + c for c in range(GRID)]) |
| for c in range(GRID): |
| us.append([r * GRID + c for r in range(GRID)]) |
| for br in range(0, GRID, 3): |
| for bc in range(0, GRID, 3): |
| us.append([(br + dr) * GRID + bc + dc |
| for dr in range(3) for dc in range(3)]) |
| return us |
|
|
|
|
| UNITS = units() |
|
|
|
|
| def propagate(grid): |
| g = [int(v) for v in grid] |
| while True: |
| cand = candidates(g) |
| if cand is None: |
| return None |
| changed = False |
| for i in range(CELLS): |
| if g[i] == 0 and bin(cand[i]).count("1") == 1: |
| g[i] = cand[i].bit_length() |
| changed = True |
| if not changed: |
| for unit in UNITS: |
| for d in range(9): |
| bit = 1 << d |
| spots = [i for i in unit if g[i] == 0 and cand[i] & bit] |
| if len(spots) == 1: |
| g[spots[0]] = d + 1 |
| changed = True |
| if not changed: |
| return g |
|
|
|
|
| def greedy_mrv(grid): |
| """Most-constrained-variable, one sweep, no backtracking.""" |
| g = [int(v) for v in grid] |
| for _ in range(CELLS): |
| cand = candidates(g) |
| if cand is None: |
| return None |
| best, bi = 10, -1 |
| for i in range(CELLS): |
| if g[i] == 0: |
| n = bin(cand[i]).count("1") |
| if n < best: |
| best, bi = n, i |
| if bi < 0: |
| return g |
| g[bi] = cand[bi].bit_length() if best == 1 else \ |
| [d + 1 for d in range(9) if cand[bi] & (1 << d)][0] |
| return g |
|
|
|
|
| def backtrack_cost(grid): |
| """Solve by full search; return (solution, number of search nodes).""" |
| g = [int(v) for v in grid] |
| nodes = [0] |
|
|
| def rec(): |
| nodes[0] += 1 |
| cand = candidates(g) |
| if cand is None: |
| return False |
| best, bi = 10, -1 |
| for i in range(CELLS): |
| if g[i] == 0: |
| n = bin(cand[i]).count("1") |
| if n < best: |
| best, bi = n, i |
| if n == 1: |
| break |
| if bi < 0: |
| return True |
| for d in range(9): |
| if cand[bi] & (1 << d): |
| g[bi] = d + 1 |
| if rec(): |
| return True |
| g[bi] = 0 |
| return False |
|
|
| ok = rec() |
| return (np.array(g) if ok else None), nodes[0] |
|
|
|
|
| p = argparse.ArgumentParser() |
| p.add_argument("--hard", default="data/sudoku_hard") |
| p.add_argument("--runs", default="runs") |
| p.add_argument("--pattern", default="sudh-*") |
| p.add_argument("--eval_bs", type=int, default=200) |
| p.add_argument("--out", default="figures/sudoku_compare.json") |
| a = p.parse_args() |
|
|
| P = np.load(os.path.join(a.hard, "test_puz.npy")) |
| S = np.load(os.path.join(a.hard, "test_sol.npy")) |
| clues = (P > 0).sum(1) |
| print(f"benchmark: {len(P)} puzzles, clues {clues.min()}-{clues.max()} " |
| f"(mean {clues.mean():.1f})\n", flush=True) |
|
|
| res = {"n": int(len(P)), "clues": {"min": int(clues.min()), "max": int(clues.max()), |
| "mean": float(clues.mean())}} |
|
|
| print("=" * 88) |
| print("CLASSICAL BASELINES, measured on THIS test set") |
| print("=" * 88) |
| prop_ok = np.array([(lambda r: r is not None and all(v > 0 for v in r) |
| and np.array_equal(r, S[i]))(propagate(P[i])) |
| for i in range(len(P))]) |
| mrv_ok = np.array([(lambda r: r is not None and all(v > 0 for v in r) |
| and np.array_equal(r, S[i]))(greedy_mrv(P[i])) |
| for i in range(len(P))]) |
| bt = [backtrack_cost(P[i]) for i in range(len(P))] |
| bt_ok = np.array([g is not None and np.array_equal(g, S[i]) for i, (g, _) in enumerate(bt)]) |
| nodes = np.array([n for _, n in bt]) |
| res["classical"] = {"propagation": float(prop_ok.mean()), |
| "greedy_mrv": float(mrv_ok.mean()), |
| "backtracking": float(bt_ok.mean()), |
| "backtrack_nodes_mean": float(nodes.mean()), |
| "backtrack_nodes_median": float(np.median(nodes))} |
| print(f" propagation only (no search) : {prop_ok.mean()*100:6.2f}%") |
| print(f" greedy MRV, one sweep : {mrv_ok.mean()*100:6.2f}%") |
| print(f" full backtracking search : {bt_ok.mean()*100:6.2f}% " |
| f"(mean {nodes.mean():.0f} search nodes, median {np.median(nodes):.0f})") |
| print(f"\n >>> propagation leaves {(1-prop_ok.mean())*100:.1f}% unsolved, so this") |
| print(f" >>> benchmark discriminates. Search solves all of it, at a cost.") |
|
|
| print() |
| print("=" * 88) |
| print("OUR FROM-SCRATCH MASKED DIFFUSION MODEL") |
| print("=" * 88) |
| dev = "cuda" if torch.cuda.is_available() else "cpu" |
| tok = SudokuTokenizer() |
| runs = defaultdict(list) |
| for d in sorted(glob.glob(os.path.join(a.runs, a.pattern))): |
| if os.path.exists(os.path.join(d, "result.json")): |
| r = json.load(open(os.path.join(d, "result.json"))) |
| tag = ("recurrent-" if r["args"].get("recurrent") else "feedforward-") \ |
| + r["args"]["mode"] |
| runs[tag].append((d, r)) |
|
|
| if not runs: |
| print(" (no trained models yet)") |
| else: |
| Pt = torch.from_numpy(P) |
| St = torch.from_numpy(S - 1) |
| blank = Pt == 0 |
|
|
| def load(d): |
| cfg = json.load(open(os.path.join(d, "result.json")))["args"] |
| if cfg.get("recurrent"): |
| m = RecurrentDenoiser(len(tok), cfg["d"], cfg["layers"], cfg["heads"], |
| cfg["pe"], max_len=CELLS + 8, |
| recurrences=cfg.get("R_infer") or cfg["R"]).to(dev) |
| else: |
| m = Transformer(len(tok), cfg["d"], cfg["layers"], cfg["heads"], |
| cfg["pe"], causal=False, max_len=CELLS + 8).to(dev) |
| m.load_state_dict(torch.load(os.path.join(d, "model.pt"), map_location=dev)) |
| m.eval() |
| return m |
|
|
| @torch.no_grad() |
| def run(m, steps): |
| out = [] |
| for i in range(0, len(Pt), a.eval_bs): |
| gold = St[i:i + a.eval_bs].to(dev) |
| bl = blank[i:i + a.eval_bs].to(dev) |
| x = torch.where(bl, torch.full_like(gold, tok.mask), gold) |
| y, _ = dfn.fixed_k_decode(m, x, tok, dev, steps, fillable=bl) |
| out.append(y.cpu().numpy() + 1) |
| return np.concatenate(out) |
|
|
| res["diffusion"] = {} |
| for mode, lst in sorted(runs.items()): |
| npar = json.load(open(os.path.join(lst[0][0], "result.json")))["args"] |
| print(f"\n {mode} ({len(lst)} seeds, d={npar['d']} layers={npar['layers']}" |
| + (f" R={npar.get('R')}→{npar.get('R_infer') or npar.get('R')}" |
| if npar.get('recurrent') else "") + ")") |
| print(f" {'passes':>8}{'board':>10}{'vs propagation':>17}") |
| per_pass = {} |
| for K in (1, 4, 8, 16, 32, 61): |
| accs, preds = [], None |
| for d, _ in lst: |
| m = load(d) |
| pr = run(m, K) |
| accs.append(float((pr == S).all(axis=1).mean())) |
| preds = pr if preds is None else preds |
| del m |
| torch.cuda.empty_cache() |
| mean = float(np.mean(accs)) |
| per_pass[K] = {"board": mean, "per_seed": accs} |
| if K == 61: |
| per_pass[K]["by_clue"] = { |
| int(c): float((preds[clues == c] == S[clues == c]).all(axis=1).mean()) |
| for c in sorted(set(clues.tolist())) |
| if (clues == c).sum() >= 20} |
| print(f" {K:>8}{mean*100:>9.2f}%{mean/max(prop_ok.mean(),1e-9):>15.1f}x") |
| res["diffusion"][mode] = per_pass |
|
|
| best_mode = max(res["diffusion"], key=lambda m: res["diffusion"][m][61]["board"]) |
| bc = res["diffusion"][best_mode][61].get("by_clue", {}) |
| if bc: |
| print(f"\n stratified by clue count (mode={best_mode}, 61 passes):") |
| print(f" {'clues':>7}{'n':>6}{'board':>10}") |
| for c in sorted(bc): |
| print(f" {c:>7}{int((clues==c).sum()):>6}{bc[c]*100:>9.2f}%") |
|
|
| print() |
| print("=" * 88) |
| print("PUBLISHED NEURAL RESULTS — different puzzle distributions, NOT directly comparable") |
| print("=" * 88) |
| for name, dist, acc in PUBLISHED: |
| pm = PARAMS.get(name, "") |
| print(f" {acc*100:6.1f}% {name:<42} [{dist}]" + (f" {pm} params" if pm else "")) |
| print(f"\n Our benchmark is {clues.min()}-{clues.max()} clues (mean {clues.mean():.1f}).") |
| print(" The 17-clue RRN split is strictly harder than ours; the SATNet split,") |
| print(" at ~36 clues, is strictly easier and is fully solved by propagation alone.") |
| print(" The stratified table above is what permits a matched-difficulty comparison.") |
|
|
| os.makedirs(os.path.dirname(a.out) or ".", exist_ok=True) |
| json.dump(res, open(a.out, "w"), indent=2) |
| print(f"\nwrote {a.out}") |
|
|
