"""Sovereign AI - GPU Fleet Sizing: calculation engine. Pure Python (standard library only) so it runs both in CPython (pytest) and in the browser via Pyodide. It is a line-for-line port of the formulas in the "GPU Fleet Sizing Calculator" Google Sheet (tabs: Inputs, Model sizing, Fleet plan, Sharing structure, Cost & chargeback, Scenarios, Notes & sources), generalised to MANY models and MANY serving pools. Rules: * No hidden constants. Every number the maths uses is a named input field in one of the dataclasses below, and each field carries metadata: label, unit, tag (CITED / ASSUMPTION / EXAMPLE / FORMULA) and a source note. * Only unit conversions (1e9 bytes per GB, 1e12 FLOPs per TFLOP, 3600 s per hour, 86 400 s per day, 24 h per day, 1000 GB/s per TB/s, 1000 W per kW) appear as literals; those are definitions, not assumptions. * Default example: a large enterprise (for example a Nordic bank) with 80 H200 GPUs and 5,000 developers. It is ONE example customer, not the product. """ from __future__ import annotations import json import math from dataclasses import asdict, dataclass, field, fields from typing import Any, Dict, List, Optional # --------------------------------------------------------------------------- # Unit conversions (definitions, not assumptions) # --------------------------------------------------------------------------- BYTES_PER_GB = 1e9 FLOPS_PER_TFLOP = 1e12 PARAMS_PER_B = 1e9 GBPS_PER_TBPS = 1000.0 SECONDS_PER_HOUR = 3600.0 SECONDS_PER_DAY = 86400.0 HOURS_PER_DAY = 24.0 W_PER_KW = 1000.0 MS_PER_S = 1000.0 TOKENS_PER_MTOK = 1e6 FLOPS_PER_PARAM_PER_TOKEN = 2.0 # forward pass: one multiply + one add per weight def inp(default: Any, label: str, unit: str, tag: str, note: str, **extra: Any): """Declare a named input with its unit and source note.""" meta = {"label": label, "unit": unit, "tag": tag, "note": note} meta.update(extra) if isinstance(default, (list, dict)): return field(default_factory=lambda d=default: json.loads(json.dumps(d)), metadata=meta) return field(default=default, metadata=meta) # --------------------------------------------------------------------------- # Inputs # --------------------------------------------------------------------------- @dataclass class Hardware: gpus: int = inp(80, "GPU count", "GPUs", "EXAMPLE", "Example customer fleet: 80x H200 (10 x 8-GPU HGX/DGX nodes)") hbm_gb: float = inp(141.0, "HBM per GPU", "GB", "CITED", "NVIDIA H200 spec: 141 GB HBM3e [S1]") hbm_bw_tbps: float = inp(4.8, "HBM bandwidth per GPU", "TB/s", "CITED", "NVIDIA H200 spec: 4.8 TB/s [S1]") bf16_tflops: float = inp(989.5, "BF16 dense Tensor TFLOPS", "TFLOPS", "CITED", "NVIDIA lists 1,979 BF16 'with sparsity' [S1]; dense = half") fp8_tflops: float = inp(1979.0, "FP8 dense Tensor TFLOPS", "TFLOPS", "CITED", "NVIDIA lists 3,958 FP8 'with sparsity' [S1]; dense = half") nvlink_domain: int = inp(8, "GPUs per NVLink domain (node)", "GPUs", "CITED", "HGX/DGX H200 = 8 GPUs on NVLink/NVSwitch [S1][S2]") mem_util: float = inp(0.90, "Serving engine GPU memory utilisation", "fraction", "ASSUMPTION", "vLLM --gpu-memory-utilization (default 0.9; GLM-5 recipe uses 0.85 [S3])") runtime_overhead_gb: float = inp(6.0, "Runtime/activation overhead per GPU", "GB", "ASSUMPTION", "CUDA graphs, activations, comm buffers per GPU") mig_slices_per_gpu: int = inp(7, "MIG slices per GPU", "slices", "CITED", "H200 SXM: up to 7 MIG instances [S1] (1g.18gb)") mig_slice_gb: float = inp(18.0, "Memory per MIG slice", "GB", "CITED", "H200 SXM: 7 MIGs @ 18 GB [S1]") gpu_tdp_w: float = inp(700.0, "GPU board power (TDP)", "W", "CITED", "H200 SXM: up to 700 W [S1] (reference)") power_per_gpu_w: float = inp(1275.0, "Power per GPU incl. server share", "W", "CITED", "DGX H200 ~10.2 kW max / 8 GPUs = 1,275 W [S2] (max, not average)") @dataclass class Efficiency: min_full_ctx_seqs: int = inp(16, "Min full-context sequences a replica must hold", "sequences", "ASSUMPTION", "Sizing rule used to choose GPUs per replica") mbu: float = inp(0.5, "MBU - memory-bandwidth utilisation in decode", "fraction", "ASSUMPTION", "0.4-0.7 typical; includes TP/EP communication losses. Validate on your hardware") mfu: float = inp(0.4, "MFU - compute efficiency (prefill & compute cap)", "fraction", "ASSUMPTION", "Validate with a benchmark on your hardware") spec_decode_gain: float = inp(1.0, "Speculative decoding speed-up (MTP / EAGLE)", "x", "ASSUMPTION", "1.0 = off. 1.3-2x is commonly reported; validate") prefix_hit: float = inp(0.5, "Prefix-cache hit rate on prompts", "fraction", "ASSUMPTION", "Share of prompt tokens served from prefix cache (agents re-send context)") utilisation: float = inp(0.5, "Average utilisation of capacity over 24h", "fraction", "ASSUMPTION", "Used for daily quotas and cost per token") @dataclass class Demand: users: int = inp(5000, "Developers & engineers", "users", "EXAMPLE", "Example customer: a large enterprise (for example a Nordic bank) with 5,000 developers") concurrency: float = inp(0.05, "Avg concurrent active users (share of users)", "fraction", "ASSUMPTION", "5% = 250 users generating at the same moment at peak; measure from gateway logs") target_tps: float = inp(3000.0, "Target aggregate output tok/s (primary model)", "tok/s", "EXAMPLE", "Test target for the frontier model") @dataclass class Workload: """Per-request shape. Used for the primary model and as the default for pools.""" ctx: int = inp(32768, "Context length (max per sequence)", "tokens", "EXAMPLE", "Serving max_model_len; GLM-5 supports ~200K [S3]") prompt: int = inp(3000, "Avg prompt (input) tokens per request", "tokens", "ASSUMPTION", "Coding agents send big contexts; chat ~1k") output: int = inp(500, "Avg output tokens per request", "tokens", "ASSUMPTION", "") batch_cap: int = inp(64, "Max decode batch per replica (latency SLO knob)", "sequences", "ASSUMPTION", "Larger = more tok/s per replica but fewer tok/s per user") @dataclass class Sharing: inference_share: float = inp(0.70, "Inference share of fleet", "fraction", "EXAMPLE", "Inference = production + interactive + batch") reserve_share: float = inp(0.10, "Reserve / failover share of fleet", "fraction", "EXAMPLE", "One 8-GPU node held for failover/maintenance") prod_share_of_inf: float = inp(0.30, "Production apps share of inference GPUs", "fraction", "EXAMPLE", "Rounded down to whole replicas") batch_share_of_inf: float = inp(0.15, "Batch / eval share of inference GPUs", "fraction", "EXAMPLE", "Rounded down to whole replicas") mig_gpus: int = inp(4, "GPUs partitioned with MIG for notebooks", "GPUs", "EXAMPLE", "Taken from the training share") teams_prod: int = inp(25, "Teams using production tier", "teams", "EXAMPLE", "") teams_interactive: int = inp(100, "Teams using interactive tier", "teams", "EXAMPLE", "~50 devs per team") teams_batch: int = inp(40, "Teams using batch / eval tier", "teams", "EXAMPLE", "") teams_training: int = inp(10, "Teams using fine-tune / training tier", "teams", "EXAMPLE", "") @dataclass class Cost: pue: float = inp(1.2, "Data-centre PUE", "x", "EXAMPLE", "Nordic free-cooling DCs often 1.1-1.3") elec_eur_kwh: float = inp(0.10, "Electricity cost", "EUR/kWh", "EXAMPLE", "Check actual contract; Statistics Finland table 13rb [S9]") gpu_eur_hour: float = inp(2.00, "GPU amortised cost per GPU-hour (excl. power)", "EUR/GPU-h", "EXAMPLE", "Capex + support + DC space amortised; replace with finance figure") hours_per_month: float = inp(730.0, "Hours per month", "h", "FORMULA", "8,760 / 12") @dataclass class ModelSpec: name: str = inp("GLM-5.x (744B-A40B) FP8", "Model name", "text", "CITED", "Z.ai GLM-5 family [S3][S4][S6]") total_b: float = inp(744.0, "Total parameters", "B params", "CITED", "GLM-5: 744B total [S3][S4]") active_b: float = inp(40.0, "Active parameters per token", "B params", "CITED", "GLM-5: 40B active [S3]. Dense model: = total") experts: int = inp(256, "Routed experts per MoE layer", "experts", "CITED", "GLM-5 n_routed_experts=256 [S5]. Dense: 0") experts_per_tok: int = inp(8, "Experts activated per token", "experts", "CITED", "GLM-5 num_experts_per_tok=8 [S5]. Dense: 0") weight_bytes: float = inp(1.0, "Weight precision", "bytes/param", "INPUT", "FP8 = 1, BF16 = 2, INT4/FP4 = 0.5") weight_overhead: float = inp(0.03, "Checkpoint overhead", "fraction", "ASSUMPTION", "Non-quantised layers, scales, MTP head") layers: int = inp(78, "Layers", "layers", "CITED", "GLM-5 num_hidden_layers=78 [S5]") kv_heads: int = inp(1, "KV heads (GQA; MLA latent = 1)", "heads", "CITED", "GLM-5 uses MLA latent KV [S5]") head_dim: int = inp(576, "Head dim (MLA: kv_lora_rank + rope)", "dims", "CITED", "GLM-5: 512 + 64 = 576 [S5]") kv_factor: int = inp(1, "K+V factor", "x", "INPUT", "GQA: 2 (K and V). MLA: 1") kv_bytes: float = inp(2.0, "KV-cache bytes per element", "bytes", "ASSUMPTION", "BF16 KV = 2, FP8 KV = 1") extra_bytes: float = inp(128.0, "Extra cached bytes per token per layer", "bytes", "ASSUMPTION", "GLM-5 DSA indexer keys (index_head_dim 128 [S5] at ~1 byte). 0 for normal models") sparse_topk: int = inp(2048, "Sparse-attention top-k tokens read (0 = dense)", "tokens", "CITED", "GLM-5 index_topk=2048 [S5]. 0 for dense attention") @dataclass class Pool: """A serving pool: one model, its own workload shape and GPU allocation.""" name: str = inp("Frontier pool", "Pool name", "text", "EXAMPLE", "") model: str = inp("GLM-5.x (744B-A40B) FP8", "Model (by name)", "text", "EXAMPLE", "") gpus: int = inp(24, "GPUs allocated", "GPUs", "EXAMPLE", "") target_tps: float = inp(3000.0, "Target output tok/s (0 = none)", "tok/s", "EXAMPLE", "") ctx: int = inp(32768, "Context length", "tokens", "EXAMPLE", "") prompt: int = inp(3000, "Avg prompt tokens", "tokens", "ASSUMPTION", "") output: int = inp(500, "Avg output tokens", "tokens", "ASSUMPTION", "") batch_cap: int = inp(64, "Max decode batch", "sequences", "ASSUMPTION", "") # Preset model library (Scenarios tab A/B/C). PRESET_MODELS: List[Dict[str, Any]] = [ dict(name="GLM-5.x (744B-A40B) FP8", total_b=744.0, active_b=40.0, experts=256, experts_per_tok=8, weight_bytes=1.0, weight_overhead=0.03, layers=78, kv_heads=1, head_dim=576, kv_factor=1, kv_bytes=2.0, extra_bytes=128.0, sparse_topk=2048), dict(name="Llama-3.3-70B dense FP8", total_b=70.6, active_b=70.6, experts=0, experts_per_tok=0, weight_bytes=1.0, weight_overhead=0.02, layers=80, kv_heads=8, head_dim=128, kv_factor=2, kv_bytes=1.0, extra_bytes=0.0, sparse_topk=0), dict(name="Qwen3-Coder-30B-A3B FP8", total_b=30.5, active_b=3.3, experts=128, experts_per_tok=8, weight_bytes=1.0, weight_overhead=0.03, layers=48, kv_heads=4, head_dim=128, kv_factor=2, kv_bytes=2.0, extra_bytes=0.0, sparse_topk=0), ] # Default pools = the sheet's "D: Mixed fleet" (frontier model + small coding model share inference GPUs). PRESET_POOLS: List[Dict[str, Any]] = [ dict(name="Frontier (production + reasoning)", model="GLM-5.x (744B-A40B) FP8", gpus=24, target_tps=3000.0, ctx=32768, prompt=3000, output=500, batch_cap=64), dict(name="Coding assistant (interactive)", model="Qwen3-Coder-30B-A3B FP8", gpus=32, target_tps=0.0, ctx=32768, prompt=3000, output=500, batch_cap=64), ] SOURCES: List[Dict[str, str]] = [ dict(id="S1", item="H200 SXM: 141 GB HBM3e, 4.8 TB/s, BF16 1,979 / FP8 3,958 TFLOPS (with sparsity; dense = half), TDP up to 700 W, up to 7 MIGs @ 18 GB, NVLink 900 GB/s", url="https://www.nvidia.com/en-us/data-center/h200/", status="CITED (NVIDIA: preliminary specifications)"), dict(id="S2", item="DGX H200: 8x H200, 1,128 GB total, system power ~10.2 kW max", url="https://www.nvidia.com/en-us/data-center/dgx-h200/", status="CITED"), dict(id="S3", item="GLM-5 model card: 744B total / 40B active, DSA sparse attention, vLLM TP 8, MTP speculative decoding", url="https://huggingface.co/zai-org/GLM-5", status="CITED"), dict(id="S4", item="Z.ai GLM-5 announcement: 744B params, 40B active; GLM-5.1/5.2/5.3 also listed as 744B-A40B", url="https://z.ai/blog/glm-5", status="CITED"), dict(id="S5", item="GLM-5 config.json: 78 layers, 256 routed + 1 shared expert, 8 experts/token, MLA kv_lora_rank 512, rope dim 64, index_topk 2048, index_head_dim 128", url="https://huggingface.co/zai-org/GLM-5/blob/main/config.json", status="CITED"), dict(id="S6", item="Hugging Face Transformers docs: GLM-5 family (GlmMoeDsa) 744B MoE, 40B active, 200K context", url="https://huggingface.co/docs/transformers/model_doc/glm_moe_dsa", status="CITED"), dict(id="S7", item="Qwen3-Coder-30B-A3B-Instruct: 30.5B total, 3.3B active, 48 layers, 4 KV heads, head_dim 128, 128 experts, 8 active", url="https://huggingface.co/Qwen/Qwen3-Coder-30B-A3B-Instruct", status="CITED"), dict(id="S8", item="Llama-3.3-70B-Instruct config: 80 layers, 8 KV heads, head_dim 128; ~70.6B params", url="https://huggingface.co/meta-llama/Llama-3.3-70B-Instruct", status="CITED"), dict(id="S9", item="Statistics Finland table 13rb: electricity price by consumer type (EXAMPLE 0.10 EUR/kWh not taken from table)", url="https://pxdata.stat.fi/PxWeb/pxweb/en/StatFin/StatFin__ehi/13rb.px/", status="EXAMPLE - replace with contract price"), ] METHOD_NOTES: List[str] = [ "ASSUMPTION: MBU 0.5 and MFU 0.4 are planning estimates, not measurements. Benchmark each model on one 8x H200 node (vLLM/SGLang benchmark_serving with your prompt/output mix) and feed the measured tok/s back via MBU.", "METHOD: decode is memory-bandwidth bound: step time = (weight bytes touched + KV bytes read) / (GPUs x bandwidth x MBU). For MoE the share of experts touched per step is 1-(1-k/E)^batch, so large batches read close to ALL parameters, not just the active ones.", "METHOD: batching gain is capped by compute: tok/s <= GPUs x dense TFLOPS x MFU / (2 x active params). Prefill time per output token = (prompt/output) x (1 - prefix hit) x 2 x active / compute.", "METHOD: GPUs per replica = smallest of 1/2/4/8 (then multiples of 8) whose usable HBM holds weights + the minimum number of full-context sequences.", "NOT MODELLED: TP/EP communication beyond the MBU factor, queueing, prefill/decode disaggregation, time-to-first-token SLOs, reasoning-token overhead.", "EXAMPLE: tier splits, team counts, PUE, electricity price and GPU-hour cost are placeholders to replace with real figures.", "Power uses DGX H200 max system power as an upper bound; average draw in inference is usually lower.", ] @dataclass class Config: hardware: Hardware = field(default_factory=Hardware) efficiency: Efficiency = field(default_factory=Efficiency) demand: Demand = field(default_factory=Demand) workload: Workload = field(default_factory=Workload) sharing: Sharing = field(default_factory=Sharing) cost: Cost = field(default_factory=Cost) models: List[ModelSpec] = field(default_factory=lambda: [ModelSpec(**m) for m in PRESET_MODELS]) pools: List[Pool] = field(default_factory=lambda: [Pool(**p) for p in PRESET_POOLS]) primary_model: str = "GLM-5.x (744B-A40B) FP8" # -- (de)serialisation ------------------------------------------------- @staticmethod def _load(cls, data: Optional[Dict[str, Any]]): obj = cls() if not data: return obj for f in fields(cls): if f.name in data and data[f.name] is not None and data[f.name] != "": cur = getattr(obj, f.name) v = data[f.name] if isinstance(cur, bool): v = bool(v) elif isinstance(cur, int): v = int(float(v)) elif isinstance(cur, float): v = float(v) else: v = str(v) setattr(obj, f.name, v) return obj @classmethod def from_dict(cls, d: Optional[Dict[str, Any]]) -> "Config": d = d or {} c = cls() for name, kls in (("hardware", Hardware), ("efficiency", Efficiency), ("demand", Demand), ("workload", Workload), ("sharing", Sharing), ("cost", Cost)): setattr(c, name, cls._load(kls, d.get(name))) if d.get("models"): c.models = [cls._load(ModelSpec, m) for m in d["models"]] if "pools" in d and d["pools"] is not None: c.pools = [cls._load(Pool, p) for p in d["pools"]] if d.get("primary_model"): c.primary_model = str(d["primary_model"]) return c def to_dict(self) -> Dict[str, Any]: return asdict(self) # --------------------------------------------------------------------------- # Excel-equivalent helpers # --------------------------------------------------------------------------- def _int(x: float) -> int: # Excel INT (floor) return int(math.floor(x + 1e-12)) def _roundup(x: float) -> int: # Excel ROUNDUP(x, 0) for x >= 0 return int(math.ceil(x - 1e-12)) def _div(a: float, b: float) -> float: return a / b if b else 0.0 def gpus_per_replica(raw: float, nvlink_domain: int) -> int: """IF(raw<=1,1,IF(raw<=2,2,IF(raw<=4,4,IF(raw<=8,8,CEILING(raw,8))))) (8 = NVLink domain).""" size = 1 while size < nvlink_domain: if raw <= size: return size size *= 2 if raw <= nvlink_domain: return nvlink_domain return int(math.ceil(raw / nvlink_domain - 1e-12) * nvlink_domain) # --------------------------------------------------------------------------- # Core: size one model replica for one workload (sheet tab "Model sizing") # --------------------------------------------------------------------------- def size_model(m: ModelSpec, wl: Workload, hw: Hardware, eff: Efficiency) -> Dict[str, Any]: r: Dict[str, Any] = {"model": m.name} r["weight_gb"] = m.total_b * m.weight_bytes * (1 + m.weight_overhead) r["usable_gb_per_gpu"] = hw.hbm_gb * hw.mem_util - hw.runtime_overhead_gb r["kv_bytes_per_token"] = m.layers * (m.kv_heads * m.head_dim * m.kv_factor * m.kv_bytes + m.extra_bytes) r["kv_gb_full_ctx"] = r["kv_bytes_per_token"] * wl.ctx / BYTES_PER_GB r["mem_needed_gb"] = r["weight_gb"] + eff.min_full_ctx_seqs * r["kv_gb_full_ctx"] r["raw_gpus"] = _div(r["mem_needed_gb"], r["usable_gb_per_gpu"]) gpr = gpus_per_replica(r["raw_gpus"], hw.nvlink_domain) r["gpus_per_replica"] = gpr r["kv_free_gb"] = gpr * r["usable_gb_per_gpu"] - r["weight_gb"] r["max_seqs_full_ctx"] = max(0, _int(_div(r["kv_free_gb"], r["kv_gb_full_ctx"]))) r["avg_ctx"] = wl.prompt + wl.output / 2 r["kv_gb_avg"] = r["kv_bytes_per_token"] * (wl.prompt + wl.output) / BYTES_PER_GB r["max_seqs_avg"] = max(0, _int(_div(r["kv_free_gb"], r["kv_gb_avg"]))) bsz = max(1, min(wl.batch_cap, r["max_seqs_avg"])) r["batch"] = bsz E, k = m.experts, m.experts_per_tok routed = 0.0 if E <= k else (m.total_b - m.active_b) / (1 - k / E) r["routed_b"] = routed r["always_b"] = m.total_b - routed frac = 0.0 if routed == 0 else 1 - (1 - k / E) ** bsz r["expert_touch_frac"] = frac wfac = m.weight_bytes * (1 + m.weight_overhead) r["weight_read_gb"] = (r["always_b"] + routed * frac) * wfac attn_tokens = r["avg_ctx"] if m.sparse_topk == 0 else min(r["avg_ctx"], m.sparse_topk) r["kv_read_gb"] = m.layers * (attn_tokens * m.kv_heads * m.head_dim * m.kv_factor * m.kv_bytes + r["avg_ctx"] * m.extra_bytes) / BYTES_PER_GB r["eff_bw_gbps"] = gpr * hw.hbm_bw_tbps * GBPS_PER_TBPS * eff.mbu r["step_ms"] = (r["weight_read_gb"] + bsz * r["kv_read_gb"]) / r["eff_bw_gbps"] * MS_PER_S r["mem_tps"] = bsz / (r["step_ms"] / MS_PER_S) * eff.spec_decode_gain tfl = hw.fp8_tflops if m.weight_bytes <= 1 else hw.bf16_tflops r["tflops_used"] = tfl compute = gpr * tfl * FLOPS_PER_TFLOP * eff.mfu r["compute_tps"] = compute / (FLOPS_PER_PARAM_PER_TOKEN * m.active_b * PARAMS_PER_B) r["decode_tps"] = min(r["mem_tps"], r["compute_tps"]) r["bottleneck"] = "Memory-bandwidth bound" if r["mem_tps"] < r["compute_tps"] else "Compute bound" r["prefill_s_per_out_tok"] = (_div(wl.prompt, wl.output) * (1 - eff.prefix_hit) * FLOPS_PER_PARAM_PER_TOKEN * m.active_b * PARAMS_PER_B / compute) r["replica_tps"] = 1 / (1 / r["decode_tps"] + r["prefill_s_per_out_tok"]) r["per_stream_tps"] = r["replica_tps"] / bsz b1_frac = 0.0 if E == 0 else k / E r["single_stream_tps"] = r["eff_bw_gbps"] / ((r["always_b"] + routed * b1_frac) * wfac + r["kv_read_gb"]) * eff.spec_decode_gain r["multi_node"] = gpr > hw.nvlink_domain r["fits_one_node"] = "Multi-node: replica spans nodes - check IB" if r["multi_node"] else "Fits in one NVLink node" return r def cost_per_gpu_hour(hw: Hardware, cost: Cost) -> Dict[str, float]: kw = hw.power_per_gpu_w / W_PER_KW * cost.pue p = kw * cost.elec_eur_kwh return {"power_kw_per_gpu": kw, "power_eur_per_gpu_h": p, "all_in_eur_per_gpu_h": cost.gpu_eur_hour + p} def cost_per_mtok(gpr: int, replica_tps: float, allin: float, util: float) -> float: return _div(allin * gpr, replica_tps * SECONDS_PER_HOUR * util) * TOKENS_PER_MTOK def pass_fail(ok: bool) -> str: return "PASS" if ok else "FAIL" # --------------------------------------------------------------------------- # Fleet-level computations # --------------------------------------------------------------------------- def _find_model(cfg: Config, name: str) -> ModelSpec: for m in cfg.models: if m.name == name: return m if not cfg.models: raise ValueError("At least one model is required") return cfg.models[0] def fleet_plan(cfg: Config, ms: Dict[str, Any]) -> Dict[str, Any]: hw, d, sh = cfg.hardware, cfg.demand, cfg.sharing gpr, rtps, bsz = ms["gpus_per_replica"], ms["replica_tps"], ms["batch"] f: Dict[str, Any] = {} f["inference_gpus"] = infg = _int(hw.gpus * sh.inference_share) f["replicas_fit"] = repfit = _int(infg / gpr) f["agg_tps"] = agg = repfit * rtps f["check_inference_share"] = pass_fail(agg >= d.target_tps) f["replicas_needed"] = repneed = _roundup(_div(d.target_tps, rtps)) f["gpus_needed"] = gneed = repneed * gpr f["gpus_needed_n1"] = (repneed + 1) * gpr f["tps_delivered"] = tpsneed = repneed * rtps f["check_target"] = pass_fail(tpsneed >= d.target_tps and gneed <= hw.gpus) f["headroom"] = _div(tpsneed, d.target_tps) - 1 if d.target_tps else 0.0 f["headroom_all"] = _div(agg, d.target_tps) - 1 if d.target_tps else 0.0 f["gpus_left"] = hw.gpus - gneed f["gpus_left_n1"] = hw.gpus - f["gpus_needed_n1"] f["active_users"] = actu = d.users * d.concurrency f["users_per_replica"] = _div(bsz, d.concurrency) f["replicas_for_concurrency"] = _roundup(_div(actu, bsz)) f["check_concurrency"] = pass_fail(repfit * bsz >= actu) f["tps_per_active_user_target"] = _div(d.target_tps, actu) f["tps_per_active_user_all"] = _div(agg, actu) f["per_stream_tps"] = ms["per_stream_tps"] return f def sharing_structure(cfg: Config, ms: Dict[str, Any], fp: Dict[str, Any], allin: float) -> Dict[str, Any]: hw, sh, eff, cost = cfg.hardware, cfg.sharing, cfg.efficiency, cfg.cost gpr, rtps = ms["gpus_per_replica"], ms["replica_tps"] infg = fp["inference_gpus"] g_prod = _int(infg * sh.prod_share_of_inf / gpr) * gpr g_batch = _int(infg * sh.batch_share_of_inf / gpr) * gpr g_inter = infg - g_prod - g_batch g_res = _int(hw.gpus * sh.reserve_share) g_train = hw.gpus - infg - g_res - sh.mig_gpus g_mig = sh.mig_gpus def inf_tier(key, name, g, pri, pre, sched, teams): reps = _int(g / gpr) cap = reps * rtps return dict(key=key, tier=name, gpus=g, priority=pri, preemptible=pre, stack=sched, replicas=reps, capacity_tps=cap, teams=teams, quota_per_team_day=_div(cap * SECONDS_PER_DAY * eff.utilisation, teams), quota_unit="output tokens/day") tiers = [ inf_tier("prod", "Production apps (customer-facing, SLA)", g_prod, 1, "No", "vLLM or SGLang replicas behind an LLM gateway (auth, rate limits, routing); Kubernetes; guaranteed quota", sh.teams_prod), inf_tier("inter", "Interactive devs / coding agents", g_inter, 2, "No (soft limits)", "vLLM/SGLang behind the same gateway; per-team token budgets; prefix caching; mix frontier + small coding model", sh.teams_interactive), inf_tier("batch", "Batch / eval", g_batch, 3, "Yes", "vLLM offline/batch API or Kueue jobs; soaks up idle capacity at night; doubles as warm failover", sh.teams_batch), dict(key="train", tier="Fine-tune / training (full GPUs)", gpus=g_train, priority=4, preemptible="Yes (checkpointing)", stack="Kueue or Volcano on Kubernetes (gang scheduling), or Slurm; whole 8-GPU nodes", replicas=None, capacity_tps=None, teams=sh.teams_training, quota_per_team_day=_div(g_train * HOURS_PER_DAY * eff.utilisation, sh.teams_training), quota_unit="GPU-hours/day"), dict(key="mig", tier="Notebooks / experiments (MIG)", gpus=g_mig, priority=5, preemptible="Yes", stack="MIG 1g.18gb slices via NVIDIA GPU Operator; JupyterHub; slice = 1 user", replicas=None, capacity_tps=None, teams=None, slices=g_mig * hw.mig_slices_per_gpu, quota_per_team_day=hw.mig_slice_gb, quota_unit="GB per slice"), dict(key="res", tier="Reserve / failover", gpus=g_res, priority=6, preemptible="n/a (lends preemptibly)", stack="Held idle or lent to batch/training as preemptible; reclaimed for failover and maintenance", replicas=None, capacity_tps=None, teams=None, quota_per_team_day=None, quota_unit=""), ] teams_of = {"prod": sh.teams_prod, "inter": sh.teams_interactive, "batch": sh.teams_batch, "train": sh.teams_training} for t in tiers: t["share_of_fleet"] = _div(t["gpus"], hw.gpus) t["cost_per_month"] = t["gpus"] * allin * cost.hours_per_month tm = teams_of.get(t["key"]) t["cost_per_team_month"] = _div(t["cost_per_month"], tm) if tm else None t["eur_per_mtok"] = (cost_per_mtok(gpr, rtps, allin, eff.utilisation) if t["key"] in ("prod", "inter", "batch") else None) t["valid"] = t["gpus"] >= 0 total = sum(t["gpus"] for t in tiers) return {"tiers": tiers, "total_gpus": total, "total_check": "OK: = fleet" if total == hw.gpus else "CHECK: <> fleet", "all_non_negative": all(t["valid"] for t in tiers)} def cost_summary(cfg: Config, ms: Dict[str, Any]) -> Dict[str, Any]: hw, cost, eff = cfg.hardware, cfg.cost, cfg.efficiency c = cost_per_gpu_hour(hw, cost) allin = c["all_in_eur_per_gpu_h"] c["replica_eur_h"] = allin * ms["gpus_per_replica"] c["eur_per_mtok"] = cost_per_mtok(ms["gpus_per_replica"], ms["replica_tps"], allin, eff.utilisation) c["eur_per_mtok_100"] = cost_per_mtok(ms["gpus_per_replica"], ms["replica_tps"], allin, 1.0) c["fleet_kw_it"] = hw.gpus * hw.power_per_gpu_w / W_PER_KW c["fleet_kw_pue"] = c["fleet_kw_it"] * cost.pue c["gpu_boards_kw"] = hw.gpus * hw.gpu_tdp_w / W_PER_KW c["kwh_month"] = c["fleet_kw_pue"] * cost.hours_per_month c["elec_eur_month"] = c["kwh_month"] * cost.elec_eur_kwh c["fleet_eur_month"] = hw.gpus * allin * cost.hours_per_month return c def size_pool(cfg: Config, pool: Pool, allin: float) -> Dict[str, Any]: m = _find_model(cfg, pool.model) wl = Workload(ctx=pool.ctx, prompt=pool.prompt, output=pool.output, batch_cap=pool.batch_cap) ms = size_model(m, wl, cfg.hardware, cfg.efficiency) gpr, rtps = ms["gpus_per_replica"], ms["replica_tps"] reps = _int(pool.gpus / gpr) agg = reps * rtps out = dict(pool=pool.name, model=m.name, gpus=pool.gpus, target_tps=pool.target_tps, ctx=pool.ctx, gpus_per_replica=gpr, replica_tps=rtps, replicas=reps, agg_tps=agg, idle_gpus=pool.gpus - reps * gpr, concurrent_seqs=reps * ms["batch"], per_stream_tps=ms["per_stream_tps"], kv_gb_full_ctx=ms["kv_gb_full_ctx"], max_seqs_full_ctx=ms["max_seqs_full_ctx"], bottleneck=ms["bottleneck"], multi_node=ms["multi_node"], eur_per_mtok=cost_per_mtok(gpr, rtps, allin, cfg.efficiency.utilisation)) if pool.target_tps > 0: need = _roundup(pool.target_tps / rtps) out.update(replicas_needed=need, gpus_needed=need * gpr, gpus_needed_n1=(need + 1) * gpr, check=pass_fail(agg >= pool.target_tps)) else: out.update(replicas_needed=None, gpus_needed=None, gpus_needed_n1=None, check="n/a") out["sizing"] = ms return out def tps_curve(ms: Dict[str, Any], max_gpus: int) -> List[List[float]]: gpr = ms["gpus_per_replica"] pts = [[0, 0.0]] n = 1 while n * gpr <= max_gpus: pts.append([n * gpr, n * ms["replica_tps"]]) n += 1 return pts def compute(cfg: Config) -> Dict[str, Any]: hw, eff = cfg.hardware, cfg.efficiency primary = _find_model(cfg, cfg.primary_model) ms = size_model(primary, cfg.workload, hw, eff) fp = fleet_plan(cfg, ms) cs = cost_summary(cfg, ms) allin = cs["all_in_eur_per_gpu_h"] share = sharing_structure(cfg, ms, fp, allin) infg = fp["inference_gpus"] scenarios = [] for m in cfg.models: s = size_model(m, cfg.workload, hw, eff) reps = _int(infg / s["gpus_per_replica"]) need = _roundup(_div(cfg.demand.target_tps, s["replica_tps"])) scenarios.append(dict( model=m.name, gpus_allocated=infg, gpus_per_replica=s["gpus_per_replica"], weight_gb=s["weight_gb"], kv_bytes_per_token=s["kv_bytes_per_token"], replica_tps=s["replica_tps"], per_stream_tps=s["per_stream_tps"], single_stream_tps=s["single_stream_tps"], batch=s["batch"], bottleneck=s["bottleneck"], replicas_fit=reps, agg_tps=reps * s["replica_tps"], check=pass_fail(reps * s["replica_tps"] >= cfg.demand.target_tps), replicas_needed=need, gpus_needed=need * s["gpus_per_replica"], concurrent_seqs=reps * s["batch"], eur_per_mtok=cost_per_mtok(s["gpus_per_replica"], s["replica_tps"], allin, eff.utilisation), curve=tps_curve(s, hw.gpus))) pools = [size_pool(cfg, p, allin) for p in cfg.pools] pool_gpus = sum(p.gpus for p in cfg.pools) pools_summary = dict( gpus_allocated=pool_gpus, inference_gpus=infg, unallocated=infg - pool_gpus, check=pass_fail(pool_gpus <= infg), agg_tps=sum(p["agg_tps"] for p in pools), concurrent_seqs=sum(p["concurrent_seqs"] for p in pools), concurrency_check=pass_fail(sum(p["concurrent_seqs"] for p in pools) >= fp["active_users"]), targets_met=all(p["check"] != "FAIL" for p in pools)) t = {x["key"]: x["gpus"] for x in share["tiers"]} rec = [ f"{primary.name}: {fp['gpus_needed']} GPUs ({fp['replicas_needed']} replica(s) x {ms['gpus_per_replica']} GPUs) " f"give {fp['tps_delivered']:,.0f} tok/s vs target {cfg.demand.target_tps:,.0f} -> {fp['check_target']}", f"With N+1 failover: {fp['gpus_needed_n1']} GPUs incl. one spare replica; {fp['gpus_left_n1']} GPUs left for everything else", f"Per-user speed: ~{ms['per_stream_tps']:.1f} tok/s per stream at batch {ms['batch']}; " f"{fp['users_per_replica']:,.0f} registered users per replica at {cfg.demand.concurrency:.0%} concurrency", f"Split: Prod {t['prod']} | Interactive {t['inter']} | Batch {t['batch']} | Training {t['train']} | " f"MIG notebooks {t['mig']} ({t['mig'] * hw.mig_slices_per_gpu} slices) | Reserve {t['res']}", ] return dict(primary=primary.name, model_sizing=ms, fleet=fp, sharing=share, cost=cs, scenarios=scenarios, pools=pools, pools_summary=pools_summary, recommendations=rec, sources=SOURCES, notes=METHOD_NOTES) # --------------------------------------------------------------------------- # JSON API used by the browser UI (Pyodide) # --------------------------------------------------------------------------- def _meta(cls) -> List[Dict[str, Any]]: out = [] for f in fields(cls): d = dict(f.metadata) d["key"] = f.name d["default"] = f.default if f.default is not None and not callable(f.default) else None d["type"] = "text" if f.type in ("str", str) else ("int" if f.type in ("int", int) else "float") out.append(d) return out def schema_json() -> str: return json.dumps({ "hardware": _meta(Hardware), "efficiency": _meta(Efficiency), "demand": _meta(Demand), "workload": _meta(Workload), "sharing": _meta(Sharing), "cost": _meta(Cost), "model": _meta(ModelSpec), "pool": _meta(Pool), }) def defaults_json() -> str: return json.dumps(Config().to_dict()) def compute_json(cfg_json: str) -> str: try: cfg = Config.from_dict(json.loads(cfg_json) if cfg_json else {}) return json.dumps({"ok": True, "result": compute(cfg)}) except Exception as e: # surfaced in the UI rather than crashing the page return json.dumps({"ok": False, "error": f"{type(e).__name__}: {e}"}) if __name__ == "__main__": r = compute(Config()) print(json.dumps({k: r[k] for k in ("fleet", "recommendations")}, indent=2))