Quantization & Hardware Sizing

# script_id: day_092_quantization_hardware/rounding_toy
# rounding_toy.py: what quantization does to numbers, with no ML library at all.
# Install: nothing (stdlib only). Env: none. Services: none.
import random


def quantize(weights: list[float], bits: int) -> list[float]:
    """Symmetric 'absmax' rounding: map each weight to one of 2**bits levels and back."""
    qmax = 2 ** (bits - 1) - 1                    # 4 bits -> integers -7..7, 8 bits -> -127..127
    scale = max(abs(w) for w in weights) / qmax   # one scale per block, stored next to the integers
    return [round(w / scale) * scale for w in weights]


def mean_abs_error(a: list[float], b: list[float]) -> float:
    return sum(abs(x - y) for x, y in zip(a, b)) / len(a)


tiny = [0.12, -0.53, 0.91, -0.07]                 # the hand-worked example from the text
print("4-bit:", [round(x, 3) for x in quantize(tiny, 4)])
print("8-bit:", [round(x, 3) for x in quantize(tiny, 8)])

random.seed(0)
block = [random.gauss(0, 0.02) for _ in range(256)]   # real weights are small and bell-shaped
for bits in (8, 4, 2):
    err = mean_abs_error(block, quantize(block, bits))
    print(f"{bits}-bit  levels={2 ** bits:>3}  mean error={err:.5f}  ({err / 0.02:.0%} of a typical weight)")

4-bit: [0.13, -0.52, 0.91, -0.13]
8-bit: [0.122, -0.53, 0.91, -0.072]
8-bit  levels=256  mean error=0.00014  (1% of a typical weight)
4-bit  levels= 16  mean error=0.00257  (13% of a typical weight)
2-bit  levels=  4  mean error=0.01473  (74% of a typical weight)

Qwen3-4B-Instruct-2507-Q4_K_M.gguf
│     │  │        │    │  │ └─ M = medium mix (S/M/L trade a little size for a little quality)
│     │  │        │    │  └─── K = K-quant, the current rounding scheme
│     │  │        │    └────── Q4 = about 4 bits per weight
│     │  │        └─────────── release tag
│     │  └──────────────────── instruction-tuned (chat), not a base model
│     └─────────────────────── 4 billion parameters
└───────────────────────────── model family

# script_id: day_092_quantization_hardware/fits
# fits.py: the sizing rule of thumb as code. Install: nothing (stdlib). Env: none. Services: none.
import re

# Bits per weight as measured on Llama 3.1 8B in llama.cpp's tools/quantize/README.md.
# K-quants mix widths, so "Q4" stores ~4.9 bits/weight, not 4. Other models land within a few %.
BITS_PER_WEIGHT = {"F16": 16.0, "Q8_0": 8.50, "Q6_K": 6.56, "Q5_K_M": 5.70,
                   "Q4_K_M": 4.89, "Q4_K_S": 4.67, "Q3_K_M": 4.00, "Q2_K": 3.16}
KV_GB_PER_1K_TOKENS = 0.15   # F16 KV cache of a typical 3-8B model with grouped-query attention
RUNTIME_GB = 1.0             # compute buffers, runtime, tokenizer: rule of thumb
HEADROOM = 0.75              # leave ~25% of RAM for the OS, browser and IDE


def decode_quant(name: str) -> str:
    """'Qwen3-4B-Instruct-2507-Q4_K_M.gguf' or 'llama3.2:3b-instruct-q4_K_M' -> 'Q4_K_M'."""
    m = re.search(r"(BF16|F16|FP16|Q\d_K(?:_[SML])?|Q\d_\d)", name, re.IGNORECASE)
    if not m:
        raise ValueError(f"no quant level in {name!r}")
    level = m.group(1).upper()
    return "F16" if level in ("BF16", "FP16") else level   # BF16 and F16 are both 16 bits


def estimate_memory_gb(params_b: float, bits: float, ctx_tokens: int = 8192, *,
                       kv_gb_per_1k: float = KV_GB_PER_1K_TOKENS, runtime_gb: float = RUNTIME_GB) -> float:
    """weights + KV cache + runtime, in GB (10**9 bytes)."""
    weights = params_b * bits / 8          # 1B params at 8 bits = 1 GB
    kv_cache = ctx_tokens / 1000 * kv_gb_per_1k
    return weights + kv_cache + runtime_gb


if __name__ == "__main__":
    print(decode_quant("Qwen3-4B-Instruct-2507-Q4_K_M.gguf"), decode_quant("llama3.2:3b-instruct-q8_0"))
    machines = (8, 16, 24)
    print(f"{'model':<12}{'needs GB':>9}" + "".join(f"{f'{ram} GB':>7}" for ram in machines))
    for params_b in (4, 8, 20):
        for level in ("Q4_K_M", "Q8_0"):
            need = estimate_memory_gb(params_b, BITS_PER_WEIGHT[level])
            marks = "".join(f"{'yes' if need <= ram * HEADROOM else 'no':>7}" for ram in machines)
            print(f"{params_b:>2}B {level:<8}{need:>9.1f}{marks}")

Q4_K_M Q8_0
model        needs GB   8 GB  16 GB  24 GB
 4B Q4_K_M        4.7    yes    yes    yes
 4B Q8_0          6.5     no    yes    yes
 8B Q4_K_M        7.1     no    yes    yes
 8B Q8_0         10.7     no    yes    yes
20B Q4_K_M       14.5     no     no    yes
20B Q8_0         23.5     no     no     no

# script_id: day_092_quantization_hardware/ollama_ps
# ollama_ps.py: check the rule of thumb against what Ollama actually loaded.
# Install: uv add ollama   Env: none
# Services: Ollama running with a model loaded (e.g. `ollama run llama3.2:3b-instruct-q4_K_M "hi"`).
import ollama

from fits import BITS_PER_WEIGHT, decode_quant, estimate_memory_gb


def params_billions(size: str) -> float:
    """Ollama reports '3.2B' or '268.10M'."""
    return float(size[:-1]) / (1000 if size.endswith("M") else 1)


try:
    loaded = ollama.ps().models
except ConnectionError as err:           # the ollama library raises the builtin ConnectionError
    loaded = []
    print(f"{err}\nEstimate only: 3.2B at Q4_K_M, 8K context ≈ "
          f"{estimate_memory_gb(3.2, BITS_PER_WEIGHT['Q4_K_M']):.1f} GB")

for m in loaded:
    level = decode_quant(m.details.quantization_level)
    if level not in BITS_PER_WEIGHT:         # e.g. Q4_0: add its bits-per-weight to the table first
        print(f"{m.model}: no bits-per-weight entry for {level}")
        continue
    ctx = m.context_length or 8192
    est = estimate_memory_gb(params_billions(m.details.parameter_size), BITS_PER_WEIGHT[level], ctx)
    on_gpu = m.size_vram / m.size if m.size else 0.0
    print(f"{m.model}: estimate {est:.1f} GB · loaded {m.size / 1e9:.1f} GB · "
          f"{on_gpu:.0%} on GPU · context {ctx}")

llama3.2:3b-instruct-q4_K_M: estimate 4.2 GB · loaded 3.9 GB · 100% on GPU · context 8192

ollama pull llama3.2:3b-instruct-q8_0      # 3.42 GB
ollama pull llama3.2:3b-instruct-q4_K_M    # 2.02 GB
ollama list                                # confirm both tags are present

# script_id: day_092_quantization_hardware/schemas
# schemas.py: IssueRecord (compact copy of Day 15/20) + the frozen extraction prompt.
# Install: uv add pydantic   Env: ORBIT_DATA (optional, path to datasets/orbit). Services: none.
import json
import os
from pathlib import Path
from typing import Literal

from pydantic import BaseModel, ConfigDict, Field, ValidationError

DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
Component = Literal["core", "scheduler", "flows", "broker-redis", "broker-postgres", "broker-sqs",
                    "cli", "dashboard", "retries", "results", "observability", "auth", "docs"]


class IssueRecord(BaseModel):
    """The facts Atlas extracts from one issue. Fields mirror eval/extraction_gold.jsonl."""
    model_config = ConfigDict(extra="forbid")
    type: Literal["bug", "feature", "question", "docs"]
    components: list[Component]
    severity: Literal["S1", "S2", "S3", "S4"] | None
    orbit_version: str | None
    python_version: str | None
    error_codes: list[str]
    summary: str = Field(min_length=1)


def parse_record(text: str) -> IssueRecord:
    """Small models often wrap JSON in ``` fences; strip them, then let Pydantic judge."""
    body = text.strip().removeprefix("```json").removeprefix("```").removesuffix("```")
    return IssueRecord.model_validate_json(body)


def format_errors(err: ValidationError) -> str:
    """Day 15's helper: one short line per failing field."""
    return "; ".join(f"{'.'.join(map(str, e['loc'])) or '(root)'}: {e['msg']}" for e in err.errors())

# script_id: day_092_quantization_hardware/schemas
# Frozen here; Days 94-96 copy it verbatim. Day 16's instructions + the JSON-shape paragraph.
SYSTEM_PROMPT_EXTRACT = """You extract facts from GitHub issues filed against Orbit, an open-source
Python background-jobs library. Fill every field of the schema from the issue text only.

Rules:
- Use null when a fact is not stated. Never guess a version from dates or context.
- severity is for bugs only; use null for features, questions and docs.
- error_codes: every ORB-NNNN code that appears anywhere, including logs and comments. [] if none.
- components: usually one; list two only when the issue really spans both.
- summary: one or two sentences in English, even when the issue is in another language.

Component glossary:
core = queues, workers, leases, job execution | scheduler = cron schedules and delays
flows = Flows, steps and step state | broker-redis, broker-postgres, broker-sqs = that broker
cli = the orbit command | dashboard = the web UI | retries = retry policy and backoff
results = the result backend | observability = metrics and tracing
auth = dashboard login and OIDC | docs = the documentation itself

Reply with one JSON object and nothing else: no prose, no code fences. Use exactly these keys:
type ("bug", "feature", "question" or "docs"), components (list of glossary names),
severity ("S1" outage or data loss, "S2" major breakage, "S3" has a workaround, "S4" cosmetic, or null),
orbit_version (string or null), python_version (string or null), error_codes (list of strings),
summary (string)."""

if __name__ == "__main__":
    gold = [json.loads(line) for line in (DATA_DIR / "eval" / "extraction_gold.jsonl").open(encoding="utf-8")]
    records = [IssueRecord.model_validate({k: v for k, v in g.items() if k != "issue_id"}) for g in gold]
    print(f"{len(records)} gold records validate · prompt is {len(SYSTEM_PROMPT_EXTRACT)} chars")

30 gold records validate · prompt is 1405 chars

# script_id: day_092_quantization_hardware/first_compare
# first_compare.py: one Orbit issue through two quant builds of the same model.
# Install: uv add "openai>=3.46,<4" pydantic   Env: ORBIT_DATA (optional)
# Services: Ollama running, both llama3.2 tags pulled (bash block above). Imports schemas.py.
import json

from openai import OpenAI

from schemas import DATA_DIR, SYSTEM_PROMPT_EXTRACT, parse_record

client = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")   # key required but ignored
issue = json.loads((DATA_DIR / "issues" / "issues.jsonl").open(encoding="utf-8").readline())
text = f"# {issue['title']}\n\n{issue['body']}"

for tag in ("llama3.2:3b-instruct-q8_0", "llama3.2:3b-instruct-q4_K_M"):
    resp = client.responses.create(model=tag, instructions=SYSTEM_PROMPT_EXTRACT, input=text, temperature=0)
    record = parse_record(resp.output_text)          # raises ValidationError if the JSON is broken
    print(tag, record.model_dump(exclude={"summary"}))

llama3.2:3b-instruct-q8_0 {'type': 'bug', 'components': ['cli'], 'severity': 'S3', 'orbit_version': '2.8.0', 'python_version': None, 'error_codes': []}
llama3.2:3b-instruct-q4_K_M {'type': 'bug', 'components': ['cli', 'core'], 'severity': 'S3', 'orbit_version': '2.8.0', 'python_version': None, 'error_codes': []}

# script_id: day_092_quantization_hardware/quant_eval
# quant_eval.py: run the R1 extraction eval on several quant builds of ONE model.
# Install: uv add "openai>=3.46,<4" pydantic   Env: ORBIT_DATA, ATLAS_LOCAL_BASE_URL (optional)
# Services: Ollama on :11434 with the QUANT_VARIANTS tags pulled. Imports schemas.py.
# Compact versions of Day 20's runner and Day 91's make_client() (no cross-lesson imports).
import json, math, os, time
from dataclasses import dataclass

from openai import OpenAI

from schemas import DATA_DIR, SYSTEM_PROMPT_EXTRACT, IssueRecord, ValidationError, parse_record

LOCAL_BASE_URL = os.environ.get("ATLAS_LOCAL_BASE_URL", "http://localhost:11434/v1")
FIELDS = ("type", "components", "severity", "orbit_version", "python_version", "error_codes")


@dataclass(frozen=True)
class QuantVariant:
    name: str        # label for the report
    model: str       # the exact tag or file you would ship
    base_url: str
    size_gb: float   # download size, for the table


QUANT_VARIANTS = [
    QuantVariant("Q8_0", "llama3.2:3b-instruct-q8_0", LOCAL_BASE_URL, 3.42),
    QuantVariant("Q4_K_M", "llama3.2:3b-instruct-q4_K_M", LOCAL_BASE_URL, 2.02),
]


def read_jsonl(path) -> list[dict]:
    return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip()]


def load_cases() -> list[dict]:
    """The 30 gold records joined to their issues. Eval-only: never few-shot, never training data."""
    issues = {i["id"]: i for i in read_jsonl(DATA_DIR / "issues" / "issues.jsonl")}
    return [{"issue": issues[g["issue_id"]], "gold": g} for g in read_jsonl(DATA_DIR / "eval" / "extraction_gold.jsonl")]

# script_id: day_092_quantization_hardware/quant_eval
_clients: dict[str, OpenAI] = {}


def client_for(base_url: str) -> OpenAI:
    """Day 91's make_client("local"), cached per server. Local servers ignore the key; the SDK needs one."""
    if base_url not in _clients:
        _clients[base_url] = OpenAI(base_url=base_url, api_key="local", timeout=300, max_retries=0)
    return _clients[base_url]


def issue_to_input(issue: dict, max_chars: int = 12_000) -> str:
    """Title, body and comments; never the labels (they're the answer key). Same as Day 20."""
    parts = [f"# {issue['title']}", issue["body"]]
    parts += [f"--- comment by {c['author']} ---\n{c['body']}" for c in issue["comments"]]
    return "\n\n".join(parts)[:max_chars]


def run_one(v: QuantVariant, issue: dict) -> tuple[str, float, int]:
    """One extraction call -> (raw text, seconds, output tokens)."""
    start = time.perf_counter()
    resp = client_for(v.base_url).responses.create(
        model=v.model, instructions=SYSTEM_PROMPT_EXTRACT, input=issue_to_input(issue),
        temperature=0, max_output_tokens=800)          # temperature 0: compare quants, not luck
    return resp.output_text, time.perf_counter() - start, resp.usage.output_tokens if resp.usage else 0


def score(record: IssueRecord | None, gold: dict, issue_text: str) -> tuple[dict[str, bool], int]:
    """Per-field match (lists as sets) + the number of ORB codes that aren't in the issue."""
    if record is None:                                  # invalid JSON scores zero on every field
        return {f: False for f in FIELDS}, 0
    got = record.model_dump()
    hits = {f: set(got[f]) == set(gold[f]) if isinstance(gold[f], list) else got[f] == gold[f]
            for f in FIELDS}
    return hits, sum(code not in issue_text for code in record.error_codes)

# script_id: day_092_quantization_hardware/quant_eval
@dataclass
class VariantReport:
    variant: QuantVariant
    valid_json_rate: float
    field_accuracy: float          # mean over FIELDS
    per_field: dict[str, float]
    invented_codes: int
    tokens_per_s: float            # end to end, including prompt processing
    p95_s: float


def p95(xs: list[float]) -> float:
    return sorted(xs)[max(0, math.ceil(0.95 * len(xs)) - 1)]     # nearest-rank percentile


def eval_variant(v: QuantVariant, cases: list[dict]) -> VariantReport:
    run_one(v, cases[0]["issue"])                  # warm-up: don't time the model load
    valid = invented = tokens = 0
    secs: list[float] = []
    hits = dict.fromkeys(FIELDS, 0)
    for case in cases:
        text, seconds, n_tokens = run_one(v, case["issue"])
        secs.append(seconds)
        tokens += n_tokens
        try:
            record = parse_record(text)
            valid += 1
        except ValidationError:
            record = None
        field_hits, bad_codes = score(record, case["gold"], issue_to_input(case["issue"]))
        invented += bad_codes
        for f in FIELDS:
            hits[f] += field_hits[f]
    per_field = {f: hits[f] / len(cases) for f in FIELDS}
    return VariantReport(v, valid / len(cases), sum(per_field.values()) / len(FIELDS), per_field,
                         invented, tokens / sum(secs), p95(secs))

# script_id: day_092_quantization_hardware/quant_eval
def print_table(reports: list[VariantReport]) -> None:
    print(f"{'variant':<8}{'GB':>6}{'valid JSON':>12}{'field acc':>11}{'invented':>10}{'tok/s':>8}{'p95 s':>7}")
    for r in reports:
        print(f"{r.variant.name:<8}{r.variant.size_gb:>6.2f}{r.valid_json_rate:>12.0%}"
              f"{r.field_accuracy:>11.0%}{r.invented_codes:>10}{r.tokens_per_s:>8.1f}{r.p95_s:>7.1f}")
    for r in reports:
        print(f"  {r.variant.name}: " + " · ".join(f"{f} {a:.2f}" for f, a in r.per_field.items()))


if __name__ == "__main__":
    cases = load_cases()
    print_table([eval_variant(v, cases) for v in QUANT_VARIANTS])

variant     GB  valid JSON  field acc  invented   tok/s  p95 s
Q8_0      3.42         97%        71%         0    24.8    6.9
Q4_K_M    2.02         93%        67%         1    38.5    4.6
  Q8_0: type 0.83 · components 0.63 · severity 0.53 · orbit_version 0.80 · python_version 0.87 · error_codes 0.60
  Q4_K_M: type 0.80 · components 0.57 · severity 0.50 · orbit_version 0.77 · python_version 0.87 · error_codes 0.53

ollama pull llama3.2:3b-instruct-q2_K      # 1.36 GB

# fragment
QUANT_VARIANTS.append(QuantVariant("Q2_K", "llama3.2:3b-instruct-q2_K", LOCAL_BASE_URL, 1.36))

# script_id: day_092_quantization_hardware/break_it
# break_it.py: the three ways a heavily quantized model breaks R1, with real Pydantic errors.
# Install: uv add pydantic   Env: ORBIT_DATA (optional). Services: none. Imports schemas.py.
import json

from schemas import DATA_DIR, ValidationError, format_errors, parse_record

issue = next(i for i in map(json.loads, (DATA_DIR / "issues" / "issues.jsonl").open(encoding="utf-8"))
             if i["id"] == 25)                     # Spanish issue quoting ORB-1202
issue_text = issue["title"] + issue["body"] + "".join(c["body"] for c in issue["comments"])

q2_outputs = [   # typical Q2_K replies: cut-off JSON, made-up labels, an invented code
    '{"type": "question", "components": ["core"], "severity": null, "orbit_version": "2.8.0", '
    '"python_version": "3.11.9", "error_codes": ["ORB-1202"], "summary": "Payload too large',
    '{"type": "question", "components": ["payloads"], "severity": "low", "orbit_version": "2.8.0", '
    '"python_version": "3.11.9", "error_codes": ["ORB-1202"], "summary": "Asks about the limit."}',
    '```json\n{"type": "question", "components": ["core"], "severity": null, "orbit_version": "2.8.0", '
    '"python_version": "3.11.9", "error_codes": ["ORB-1202", "ORB-1203"], "summary": "Asks about the limit."}\n```',
]
for n, text in enumerate(q2_outputs, 1):
    try:
        record = parse_record(text)
    except ValidationError as err:
        print(f"output {n}: INVALID  {format_errors(err)}")
        continue
    invented = [code for code in record.error_codes if code not in issue_text]
    print(f"output {n}: valid JSON · invented codes: {invented or 'none'}")

output 1: INVALID  (root): Invalid JSON: EOF while parsing a string at line 1 column 175
output 2: INVALID  components.0: Input should be 'core', 'scheduler', 'flows', 'broker-redis', 'broker-postgres', 'broker-sqs', 'cli', 'dashboard', 'retries', 'results', 'observability', 'auth' or 'docs'; severity: Input should be 'S1', 'S2', 'S3' or 'S4'
output 3: valid JSON · invented codes: ['ORB-1203']

variant     GB  valid JSON  field acc  invented   tok/s  p95 s
Q8_0      3.42         97%        71%         0    24.8    6.9
Q4_K_M    2.02         93%        67%         1    38.5    4.6
Q2_K      1.36         70%        44%         6    45.1    5.2

# 1. Tools: llama-quantize, llama-bench, llama-server (Windows: winget install llama.cpp)
brew install llama.cpp

# 2. The converter lives in the repo, not in the brew package
git clone https://github.com/ggml-org/llama.cpp
cd llama.cpp && uv venv && source .venv/bin/activate
uv pip install -r requirements.txt --index-strategy unsafe-best-match

# 3. Download from Hugging Face and convert to a 16-bit GGUF in one step
python convert_hf_to_gguf.py --remote Qwen/Qwen3-4B-Instruct-2507 \
    --outtype bf16 --outfile Qwen3-4B-Instruct-2507-BF16.gguf

# 4. Quantize, always from the 16-bit file, never from another quant
llama-quantize Qwen3-4B-Instruct-2507-BF16.gguf Qwen3-4B-Instruct-2507-Q8_0.gguf Q8_0
llama-quantize Qwen3-4B-Instruct-2507-BF16.gguf Qwen3-4B-Instruct-2507-Q4_K_M.gguf Q4_K_M
ls -lh *.gguf && shasum -a 256 *Q4_K_M.gguf     # size check + the checksum you'll pin

# 5. Benchmark: prompt processing (pp512) and generation (tg128), 5 repetitions each
llama-bench -m Qwen3-4B-Instruct-2507-Q8_0.gguf -m Qwen3-4B-Instruct-2507-Q4_K_M.gguf -p 512 -n 128

# 6. Serve one file at a time on an explicit port (current builds default to 9931, not 8080)
llama-server -m Qwen3-4B-Instruct-2507-Q4_K_M.gguf --port 8080 -c 8192

# script_id: day_092_quantization_hardware/llamacpp_eval
# llamacpp_eval.py: the same R1 eval against YOUR quants, served by llama-server.
# Install: uv add "openai>=3.46,<4" pydantic   Env: ORBIT_DATA (optional). Imports quant_eval.py.
# Services: llama-server -m <file>.gguf --port 8080 (one file at a time; bash block above).
# Usage: python llamacpp_eval.py Q4_K_M, then restart the server with the Q8_0 file and pass Q8_0.
import sys
import urllib.request

from quant_eval import QuantVariant, eval_variant, load_cases, print_table

URL = "http://localhost:8080"
SIZES_GB = {"Q4_K_M": 2.5, "Q8_0": 4.3}     # approximate; use your `ls -lh` numbers

level = sys.argv[1] if len(sys.argv) > 1 else "Q4_K_M"
try:
    urllib.request.urlopen(f"{URL}/health", timeout=5)
except OSError as err:
    raise SystemExit(f"llama-server isn't answering on {URL} ({err}). Start it first.")

# llama-server answers with the one model it loaded; the name is for your report (⚠ verify on your build)
variant = QuantVariant(level, f"Qwen3-4B-Instruct-2507-{level}.gguf", f"{URL}/v1", SIZES_GB[level])
print_table([eval_variant(variant, load_cases())])

variant     GB  valid JSON  field acc  invented   tok/s  p95 s
Q4_K_M    2.50         97%        72%         0    31.0    5.3