File size: 4,728 Bytes
f5b54bc | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 | #!/usr/bin/env python3
"""Live ball-occupancy + synthetic k* sweep.
Synthetic (no weights)::
python examples/occupancy.py --sweep
Live store (HF weights or a local Memory)::
python examples/occupancy.py --live
python examples/occupancy.py --live --limit 306 --model thefinalboss/palimpseste-max
The first query whose Hamming ball exceeds k*_exact ≈ 0.02 D is the first
predictable exact-recall failure. Soft cleanup holds longer (~0.03 D).
"""
from __future__ import annotations
import argparse
import json
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
from palimseste.capacity import ( # noqa: E402
DEFAULT_D,
DEFAULT_M,
format_report,
format_sweep,
kstar_exact,
kstar_soft,
kstar_theory,
occupancy,
occupancy_from_addresses,
sweep_exact_soft,
)
def cmd_sweep(args: argparse.Namespace) -> int:
rows = sweep_exact_soft(D=args.D, M=args.M, reps=args.reps)
print(format_sweep(rows, args.D, args.M))
if args.json:
print(json.dumps(rows, indent=2))
return 0
def cmd_live(args: argparse.Namespace) -> int:
try:
from palimseste.hf import HFPalimpsesteLM
from palimseste.phi import Phi
except ImportError as e:
print(
"live path needs the full palimseste package "
"(phi.py / memory.py / hf.py).\n"
f"Import failed: {e}\n"
"Run --sweep until you drop capacity.py into the HF tree.",
file=sys.stderr,
)
return 2
model_id = args.model
print(f"loading {model_id} …", file=sys.stderr)
lm = HFPalimpsesteLM.from_pretrained(model_id)
mem = lm.mem
phi = getattr(lm, "phi", None) or Phi()
D = int(getattr(next(iter(mem.traces)).address, "D", DEFAULT_D))
print(
f"|M|={len(mem.traces)} D={D} radius={phi.config.radius} "
f"k*_exact={kstar_exact(D)} k*_soft={kstar_soft(D)} "
f"k*_theory={kstar_theory(D)}",
file=sys.stderr,
)
if args.queries:
queries = []
tags = []
# encode each line as the model would — fall back to learner if present
encode = getattr(lm, "encode", None) or getattr(lm, "embed", None)
if encode is None:
print("model has no encode(); probing stored addresses instead", file=sys.stderr)
rep = occupancy_from_addresses(mem, phi, limit=args.limit)
else:
for line in Path(args.queries).read_text(encoding="utf-8").splitlines():
line = line.strip()
if not line:
continue
queries.append(encode(line))
tags.append(line[:80])
if args.limit and len(queries) >= args.limit:
break
rep = occupancy(mem, phi, queries, tags=tags)
else:
rep = occupancy_from_addresses(mem, phi, limit=args.limit)
print(format_report(rep))
if args.json:
print(json.dumps(rep.as_dict(), indent=2))
return 1 if rep.n_over_exact else 0
def main() -> int:
p = argparse.ArgumentParser(description="PALIMPSESTE capacity / occupancy")
sub = p.add_subparsers(dest="cmd")
ps = sub.add_parser("sweep", help="synthetic exact/soft codebook sweep")
ps.add_argument("--D", type=int, default=8000)
ps.add_argument("--M", type=int, default=DEFAULT_M)
ps.add_argument("--reps", type=int, default=60)
ps.add_argument("--json", action="store_true")
pl = sub.add_parser("live", help="histogram |N_r(q)| on a loaded store")
pl.add_argument("--model", default="thefinalboss/palimpseste-max")
pl.add_argument("--limit", type=int, default=306)
pl.add_argument("--queries", default=None, help="text file, one query per line")
pl.add_argument("--json", action="store_true")
# also allow --sweep flag without subcommand
p.add_argument("--sweep", action="store_true")
p.add_argument("--live", action="store_true")
p.add_argument("--D", type=int, default=8000)
p.add_argument("--M", type=int, default=DEFAULT_M)
p.add_argument("--reps", type=int, default=60)
p.add_argument("--model", default="thefinalboss/palimpseste-max")
p.add_argument("--limit", type=int, default=306)
p.add_argument("--queries", default=None)
p.add_argument("--json", action="store_true")
args = p.parse_args()
if args.cmd == "sweep" or (args.sweep and args.cmd is None):
return cmd_sweep(args)
if args.cmd == "live" or args.live:
return cmd_live(args)
# default: sweep, cheap
return cmd_sweep(args)
if __name__ == "__main__":
raise SystemExit(main())
|