#!/usr/bin/env python3 """Check a tiered-architecture design against its own measurements, and summarise it. Purpose: read the measurements.json a reader wrote for this part's project and answer three questions a design document should not be allowed to dodge. Does every decision cite a measurement that exists in the file? Is every measurement filled in rather than left as null? And does the arithmetic agree with the design: how long would one request's key-value cache take to cross each link that is marked as carrying per-request traffic, at the throughput that link was measured at? Prints the answers and appends one summary line to the lab notebook. Platform: all. Pure Python standard library: no pip install and no server needed. Minimum memory: 1 GB. It reads a JSON file. Assumes: Python 3.9 or later; a measurements.json copied from measurements-example.json and filled in; the measured link throughput figures from Part 18's lab. The script checks structure and arithmetic. It cannot check whether your reasoning is good, which is what the written document is for. Usage: python3 summarise-architecture.py --file measurements.json --labbook labbook.md python3 summarise-architecture.py --file measurements.json --strict exit non-zero if any check fails, which is what to run before you call it done """ from __future__ import annotations import argparse import json import sys import time from pathlib import Path SCHEMA = "part-22/tiered-architecture/1" REQUIRED_TOP = [ "schema", "requirement", "model", "machines", "links", "tiers", "routing", "measurements", "decisions", "one_more_machine", ] REQUIRED_MACHINE = ["name", "track", "device_memory_gb", "roles"] REQUIRED_LINK = ["from", "to", "class", "carries"] REQUIRED_MEASUREMENT = ["label", "configuration", "concurrency", "ttft_p50_s", "tpot_p50_s", "output_tokens_per_s"] KNOWN_ROLES = {"prefill", "decode", "router", "cache", "storage", "agent"} KNOWN_CARRIES = {"per-token", "per-request", "load-time", "client"} KNOWN_TRACKS = {"spark", "strix", "mac", "nvidia"} def human_bytes(n: float) -> str: for unit in ("B", "KB", "MB", "GB", "TB"): if abs(n) < 1000 or unit == "TB": return f"{n:.0f} {unit}" if unit == "B" else f"{n:.2f} {unit}" n /= 1000 return f"{n:.2f} TB" def main() -> int: parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) parser.add_argument("--file", default="measurements.json", help="the design's measurements file") parser.add_argument("--labbook", default="labbook.md", help="notebook to append to") parser.add_argument("--lab", default="part-22/project-a-tiered-inference-architecture") parser.add_argument("--strict", action="store_true", help="exit non-zero when any check fails") parser.add_argument("--print-only", action="store_true", help="write nothing") args = parser.parse_args() path = Path(args.file) try: design = json.loads(path.read_text(encoding="utf-8")) except OSError as exc: print(f"Could not read {path}: {exc}", file=sys.stderr) return 2 except json.JSONDecodeError as exc: print(f"{path} is not valid JSON: {exc}", file=sys.stderr) return 2 problems: list[str] = [] notes: list[str] = [] # --- structure --------------------------------------------------------------------- if design.get("schema") != SCHEMA: problems.append(f'schema should be "{SCHEMA}", found {design.get("schema")!r}') for key in REQUIRED_TOP: if key not in design: problems.append(f"missing top-level key: {key}") machines = design.get("machines") or [] if not machines: problems.append("no machines listed; a design with no machines is not a design") names = set() roles_present = set() for i, machine in enumerate(machines): for key in REQUIRED_MACHINE: if key not in machine: problems.append(f"machine {i}: missing {key}") name = machine.get("name") if name in names: problems.append(f"two machines are called {name!r}") names.add(name) if machine.get("track") not in KNOWN_TRACKS: problems.append(f"machine {name!r}: track should be one of " f"{sorted(KNOWN_TRACKS)}, found {machine.get('track')!r}") for role in machine.get("roles") or []: if role not in KNOWN_ROLES: problems.append(f"machine {name!r}: unknown role {role!r}; " f"Part 18 names {sorted(KNOWN_ROLES)}") roles_present.add(role) for role in ("prefill", "decode", "router"): if role not in roles_present: problems.append(f"no machine holds the {role} role; every design needs one, " f"even if one machine holds all three") links = design.get("links") or [] for i, link in enumerate(links): for key in REQUIRED_LINK: if key not in link: problems.append(f"link {i}: missing {key}") if link.get("carries") not in KNOWN_CARRIES: problems.append(f"link {i}: carries should be one of {sorted(KNOWN_CARRIES)}, " f"found {link.get('carries')!r}; that annotation is the whole " f"point of the link list") for end in ("from", "to"): other = link.get(end) if other not in names and other != "clients": notes.append(f"link {i}: {end} is {other!r}, which is not a machine in " f"this design (use \"clients\" for the outside world)") measurements = design.get("measurements") or [] by_label: dict[str, dict] = {} for i, m in enumerate(measurements): for key in REQUIRED_MEASUREMENT: if key not in m: problems.append(f"measurement {i}: missing {key}") label = m.get("label") if label: by_label[label] = m for key in ("ttft_p50_s", "tpot_p50_s", "output_tokens_per_s"): if m.get(key) is None: problems.append(f"measurement {label!r}: {key} is still null; the design " f"is not finished until every number is real") decisions = design.get("decisions") or [] if not decisions: problems.append("no decisions listed") for i, d in enumerate(decisions): evidence = d.get("evidence") or [] if not evidence: problems.append(f"decision {i}: no evidence cited; every decision in this " f"project has to point at a measurement") for label in evidence: if label not in by_label: problems.append(f"decision {i}: cites measurement {label!r}, which is not " f"in this file") if not str(d.get("because", "")).strip(): problems.append(f"decision {i}: no reason given") if not str(design.get("one_more_machine", "")).strip(): problems.append("one_more_machine is empty; a design that cannot say what a " "second or third machine would change has not been thought about") # --- arithmetic -------------------------------------------------------------------- model = design.get("model") or {} per_token = model.get("kv_bytes_per_token") prompt_tokens = (design.get("requirement") or {}).get("typical_prompt_tokens") payload = None if isinstance(per_token, int) and isinstance(prompt_tokens, int): payload = per_token * prompt_tokens print(f"==> {path}") print(f" machines {len(machines)}, links {len(links)}, " f"measurements {len(measurements)}, decisions {len(decisions)}") if payload: print(f" one typical request's key-value cache: {human_bytes(payload)} " f"({per_token} bytes per token x {prompt_tokens} tokens)") print("\n==> Links marked as carrying per-request traffic") per_request = [x for x in links if x.get("carries") == "per-request"] if not per_request: print(" None. This design does not move a key-value cache between machines,") print(" which is a perfectly good answer and should be stated as a decision.") for link in per_request: gbps = link.get("measured_gbps") label = f"{link.get('from')} to {link.get('to')} ({link.get('class')})" if not payload: print(f" {label}: no payload arithmetic (set kv_bytes_per_token and " f"typical_prompt_tokens)") continue if not isinstance(gbps, (int, float)) or gbps <= 0: problems.append(f"link {label}: carries per-request traffic but has no " f"measured_gbps; Part 18's lab is where that number comes from") print(f" {label}: not measured") continue seconds = payload / (gbps * 1e9 / 8) print(f" {label}: about {seconds:.2f} s to move one request's cache " f"at the measured {gbps} gigabits per second") budget = (design.get("requirement") or {}).get("ttft_budget_s") if isinstance(budget, (int, float)) and seconds > budget: print(f" That is longer than the whole time-to-first-token budget of " f"{budget} s. Say so in the document.") # --- report ------------------------------------------------------------------------ print() if notes: print("==> Notes") for note in notes: print(f" - {note}") print() if problems: print(f"==> {len(problems)} thing(s) to fix") for problem in problems: print(f" - {problem}") else: print("==> Every check passed. Every decision cites a measurement that exists,") print(" every measurement has a number in it, and every per-request link has a") print(" measured throughput. The reasoning is still yours to defend.") record = { "lab": args.lab, "record": "architecture", "file": str(path), "machines": [m.get("name") for m in machines], "roles": sorted(roles_present), "links_per_request": len(per_request), "measurement_labels": sorted(by_label), "decisions": len(decisions), "kv_bytes_per_request": payload, "problems": problems, "passed": not problems, "recorded_at": time.strftime("%Y-%m-%dT%H:%M:%S%z"), } if not args.print_only: with open(args.labbook, "a", encoding="utf-8") as handle: handle.write(json.dumps(record, sort_keys=True) + "\n") print(f"\n Appended one architecture line to {args.labbook}") if problems and args.strict: return 1 return 0 if __name__ == "__main__": sys.exit(main())