diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 0a1fafc..4185fef 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -23,7 +23,7 @@ jobs: - name: Smoke test calc.py CLI run: | - echo '{"params_b":70,"active_params_b":70,"quant":"int4","queries_per_week":1000000,"avg_tokens_per_query":800,"api_cost_per_query_usd":0.002,"traffic_pattern":"business","gpu":{"name":"H100 80GB","vram_gb":80,"usd_per_hr":2.90,"bf16_tflops":989}}' | python3 calc.py inference + echo '{"params_b":70,"active_params_b":70,"quant":"int4","queries_per_week":1000000,"avg_tokens_per_query":800,"api_cost_per_query_usd":0.002,"traffic_pattern":"business","gpu":{"name":"H100 80GB","vram_gb":80,"usd_per_hr":2.90,"bf16_tflops":989}}' | python3 scripts/calc.py inference - name: Run unittest suite run: python3 -m unittest discover tests/ -v diff --git a/README.md b/README.md index 477d65e..32e5ee3 100644 --- a/README.md +++ b/README.md @@ -49,7 +49,7 @@ flowchart LR User["User prompt + attached context"] --> Agent Agent["Agent reads SKILL.md"] -->|"WebFetch"| Web["Runpod / OpenAI / lmarena"] Agent -->|"WebFetch"| AssumeRepo["sister repo ASSUMPTIONS.md"] - Agent -->|"python3 calc.py inference|finetune"| Calc["calc.py (stdlib only)"] + Agent -->|"python3 scripts/calc.py inference|finetune"| Calc["calc.py (stdlib only)"] Calc -->|"JSON result + derivation"| Agent Agent -->|"markdown report"| User ``` @@ -57,7 +57,7 @@ flowchart LR 1. **Context** — scan user message, open files, attachments for volume, model, traffic shape. 2. **Live data** — fetch GPU prices (Runpod/Lambda/Modal), API prices (models.dev or vendor page), quality (lmarena.ai). 3. **Clarify** — ask up to 2 questions per round for low-confidence high-leverage fields. -4. **Engine** — call `python3 calc.py inference` or `finetune` with JSON on stdin. +4. **Engine** — call `python3 scripts/calc.py inference` or `finetune` with JSON on stdin. 5. **Scenario matrix** — 3 traffic patterns × 2 quality bars = 6 invocations. 6. **Report** — headline, matrix, derivation, assumptions table, flip conditions, cited URLs. diff --git a/SKILL.md b/SKILL.md index d272d50..d29d779 100644 --- a/SKILL.md +++ b/SKILL.md @@ -33,7 +33,7 @@ training, non-LLM workloads). ## Hard rules - **Never compute VRAM, GPU-hours, or dollar costs in your head.** Always call - `python3 calc.py` with JSON on stdin and parse JSON from stdout. + `python3 scripts/calc.py` with JSON on stdin and parse JSON from stdout. - **Never invent GPU prices.** Fetch from vendor pages (Runpod, Lambda, Modal, Together, Fireworks, DeepInfra) and cite the URL + fetch timestamp in the report. @@ -232,13 +232,13 @@ Run the deterministic script from the repo root. Never substitute mental math. ### Inference ```bash -echo '' | python3 calc.py inference +echo '' | python3 scripts/calc.py inference ``` Example payload (70B INT4, business traffic, H100): ```bash -echo '{"params_b":70,"active_params_b":70,"quant":"int4","queries_per_week":1000000,"avg_tokens_per_query":800,"api_cost_per_query_usd":0.002,"traffic_pattern":"business","gpu":{"name":"H100 80GB","vram_gb":80,"usd_per_hr":2.90,"bf16_tflops":989}}' | python3 calc.py inference +echo '{"params_b":70,"active_params_b":70,"quant":"int4","queries_per_week":1000000,"avg_tokens_per_query":800,"api_cost_per_query_usd":0.002,"traffic_pattern":"business","gpu":{"name":"H100 80GB","vram_gb":80,"usd_per_hr":2.90,"bf16_tflops":989}}' | python3 scripts/calc.py inference ``` Expected output keys: `fits`, `vram_needed_gb`, `vram_headroom_gb`, @@ -254,13 +254,13 @@ Expected output keys: `fits`, `vram_needed_gb`, `vram_headroom_gb`, ### Fine-tune ```bash -echo '' | python3 calc.py finetune +echo '' | python3 scripts/calc.py finetune ``` Example payload (QLoRA on 65B, Guanaco-scale dataset): ```bash -echo '{"active_params_b":65,"total_params_b":65,"method":"qlora","num_examples":10000,"tokens_per_example":500,"epochs":3,"experiments_multiplier":1.0,"prep_cost_usd":0,"gpu":{"name":"H100 80GB","vram_gb":80,"usd_per_hr":2.90,"bf16_tflops":989,"gpus_per_node":8}}' | python3 calc.py finetune +echo '{"active_params_b":65,"total_params_b":65,"method":"qlora","num_examples":10000,"tokens_per_example":500,"epochs":3,"experiments_multiplier":1.0,"prep_cost_usd":0,"gpu":{"name":"H100 80GB","vram_gb":80,"usd_per_hr":2.90,"bf16_tflops":989,"gpus_per_node":8}}' | python3 scripts/calc.py finetune ``` Expected output keys: `total_tokens`, `method_flops`, `effective_flops_per_sec`, @@ -309,7 +309,7 @@ engine invocations (inference path) unless the user only asked about fine-tuning For each of the 6 cells, call: ```bash -echo '' | python3 calc.py inference +echo '' | python3 scripts/calc.py inference ``` Present results as a markdown table: @@ -419,7 +419,7 @@ If Phase 1 yields no volume, no model, and no cost signal: ### `calc.py` missing or errors on exit 1 Report the error JSON to the user. Do not fall back to manual calculation. -Suggest verifying `python3 calc.py` exists and Python ≥ 3.10 is available. +Suggest verifying `python3 scripts/calc.py` exists and Python ≥ 3.10 is available. ### Quality Elo gap > 100 points diff --git a/calc.py b/scripts/calc.py similarity index 56% rename from calc.py rename to scripts/calc.py index 807619b..e3a4516 100755 --- a/calc.py +++ b/scripts/calc.py @@ -11,6 +11,7 @@ """ import json +import math import sys FLOPS_PER_TOKEN_PER_PARAM = 6 @@ -28,30 +29,97 @@ TRAFFIC_PATTERN_HOURS = { "uniform": 168, "business": 50, + "business_hours": 50, "bursty": 20, "always_warm": 168, } +ALLOWED_TRAFFIC_PATTERNS = { + "always_warm", + "business_hours", + "business", + "bursty", + "cold_per_query", + "uniform", +} + + +def _exit_bad(message, field): + print(json.dumps({"error": message, "field": field})) + sys.exit(2) + def _require(d, key): - if key not in d: - raise KeyError(key) + if key not in d or d[key] is None: + _exit_bad(f"missing field {key}", key) return d[key] +def _require_positive(d, key): + value = _require(d, key) + if ( + not isinstance(value, (int, float)) + or isinstance(value, bool) + or not math.isfinite(value) + or value <= 0 + ): + _exit_bad(f"{key} must be a positive finite number", key) + return float(value) + + +def _require_numeric(d, key): + value = _require(d, key) + if ( + not isinstance(value, (int, float)) + or isinstance(value, bool) + or not math.isfinite(value) + ): + _exit_bad(f"{key} must be a finite number", key) + return float(value) + + +def _require_non_negative(d, key): + value = _require_numeric(d, key) + if value < 0: + _exit_bad(f"{key} must be a non-negative finite number", key) + return value + + def compute_inference(inp: dict) -> dict: - params_b = _require(inp, "params_b") + warnings = [] + + params_b = _require_positive(inp, "params_b") quant = _require(inp, "quant") - queries_per_week = _require(inp, "queries_per_week") - api_cost_per_query_usd = _require(inp, "api_cost_per_query_usd") + queries_per_week = _require_positive(inp, "queries_per_week") + api_cost_per_query_usd = _require_positive(inp, "api_cost_per_query_usd") traffic_pattern = _require(inp, "traffic_pattern") gpu = _require(inp, "gpu") - gpu_vram_gb = _require(gpu, "vram_gb") - gpu_usd_per_hr = _require(gpu, "usd_per_hr") + if not isinstance(gpu, dict): + _exit_bad("gpu must be an object", "gpu") + gpu_vram_gb = _require_positive(gpu, "vram_gb") + gpu_usd_per_hr = _require_positive(gpu, "usd_per_hr") + + if inp.get("active_params_b") is not None: + active_params_b = inp.get("active_params_b") + total_params_b = inp.get("total_params_b", params_b) + if ( + not isinstance(active_params_b, (int, float)) + or isinstance(active_params_b, bool) + or not math.isfinite(active_params_b) + ): + _exit_bad("active_params_b must be a finite number", "active_params_b") + if float(active_params_b) > float(total_params_b): + _exit_bad("active_params_b cannot exceed total_params_b", "active_params_b") if quant not in BYTES_PER_PARAM_INFERENCE: raise ValueError(f"unknown quant {quant}") + if traffic_pattern not in ALLOWED_TRAFFIC_PATTERNS: + _exit_bad( + "traffic_pattern must be one of always_warm|business_hours|business|bursty|cold_per_query|uniform", + "traffic_pattern", + ) + bytes_per_param = BYTES_PER_PARAM_INFERENCE[quant] vram_needed_gb = params_b * bytes_per_param * VRAM_OVERHEAD_FACTOR vram_headroom_gb = gpu_vram_gb - vram_needed_gb @@ -60,7 +128,7 @@ def compute_inference(inp: dict) -> dict: if traffic_pattern in TRAFFIC_PATTERN_HOURS: billed_hours_per_week = TRAFFIC_PATTERN_HOURS[traffic_pattern] elif traffic_pattern == "cold_per_query": - billed_hours_per_week = inp.get("hot_hours_per_week", 0) + billed_hours_per_week = _require_positive(inp, "hot_hours_per_week") else: raise ValueError(f"unknown traffic_pattern {traffic_pattern}") @@ -70,8 +138,31 @@ def compute_inference(inp: dict) -> dict: savings_pct = (weekly_savings_usd / api_weekly_usd * 100.0) if api_weekly_usd > 0 else 0.0 if not fits: - verdict = "infeasible" - elif selfhost_weekly_usd < api_weekly_usd: + warnings.append( + f"Self-host infeasible: needs {vram_needed_gb:.1f} GB, GPU has {gpu_vram_gb} GB." + ) + derivation = [ + {"step": "vram_needed_gb", "formula": "params_b * bytes_per_param[quant] * 1.2", "value": vram_needed_gb}, + {"step": "billed_hours", "formula": "pattern -> hours/week", "value": billed_hours_per_week}, + {"step": "selfhost_weekly_usd", "formula": "billed_hours * usd_per_hr", "value": 0}, + {"step": "api_weekly_usd", "formula": "queries_per_week * api_cost_per_query_usd", "value": api_weekly_usd}, + ] + return { + "fits": False, + "infeasible": True, + "vram_needed_gb": round(vram_needed_gb, 4), + "vram_headroom_gb": round(vram_headroom_gb, 4), + "billed_hours_per_week": billed_hours_per_week, + "selfhost_weekly_usd": 0, + "api_weekly_usd": round(api_weekly_usd, 4), + "weekly_savings_usd": 0, + "savings_pct": 0, + "verdict": "infeasible", + "warnings": warnings, + "derivation": derivation, + } + + if selfhost_weekly_usd < api_weekly_usd: verdict = "selfhost_wins" else: verdict = "api_wins" @@ -85,6 +176,7 @@ def compute_inference(inp: dict) -> dict: return { "fits": fits, + "infeasible": False, "vram_needed_gb": round(vram_needed_gb, 4), "vram_headroom_gb": round(vram_headroom_gb, 4), "billed_hours_per_week": billed_hours_per_week, @@ -93,24 +185,40 @@ def compute_inference(inp: dict) -> dict: "weekly_savings_usd": round(weekly_savings_usd, 4), "savings_pct": round(savings_pct, 4), "verdict": verdict, + "warnings": warnings, "derivation": derivation, } def compute_finetune(inp: dict) -> dict: - active_params_b = _require(inp, "active_params_b") - total_params_b = _require(inp, "total_params_b") + warnings = [] + + active_params_b = _require_positive(inp, "active_params_b") + total_params_b = _require_positive(inp, "total_params_b") + if active_params_b > total_params_b: + _exit_bad("active_params_b cannot exceed total_params_b", "active_params_b") + method = _require(inp, "method") - num_examples = _require(inp, "num_examples") - tokens_per_example = _require(inp, "tokens_per_example") - epochs = _require(inp, "epochs") - experiments_multiplier_in = _require(inp, "experiments_multiplier") - prep_cost_usd = _require(inp, "prep_cost_usd") + num_examples = _require_positive(inp, "num_examples") + tokens_per_example = _require_positive(inp, "tokens_per_example") + epochs = _require_positive(inp, "epochs") + prep_cost_usd = _require_non_negative(inp, "prep_cost_usd") gpu = _require(inp, "gpu") - gpu_vram_gb = _require(gpu, "vram_gb") - gpu_usd_per_hr = _require(gpu, "usd_per_hr") - gpu_bf16_tflops = _require(gpu, "bf16_tflops") - gpus_per_node = gpu.get("gpus_per_node", 8) + if not isinstance(gpu, dict): + _exit_bad("gpu must be an object", "gpu") + gpu_vram_gb = _require_positive(gpu, "vram_gb") + gpu_usd_per_hr = _require_positive(gpu, "usd_per_hr") + gpu_bf16_tflops = _require_positive(gpu, "bf16_tflops") + + if "gpus_per_node" in gpu: + gpus_per_node = _require_positive(gpu, "gpus_per_node") + else: + gpus_per_node = 8 + + if "experiments_multiplier" in inp: + experiments_multiplier_in = _require_numeric(inp, "experiments_multiplier") + else: + experiments_multiplier_in = 1.0 if method not in FT_METHODS: raise ValueError(f"unknown method {method}") @@ -140,7 +248,14 @@ def compute_finetune(inp: dict) -> dict: hours_with_cluster = single_gpu_hours * cluster_overhead single_run_gpu_cost_usd = hours_with_cluster * gpu_usd_per_hr - experiments_multiplier = max(1.0, experiments_multiplier_in) + + experiments_multiplier = experiments_multiplier_in + if experiments_multiplier < 1.0: + warnings.append( + f"experiments_multiplier {experiments_multiplier_in} clamped to 1.0" + ) + experiments_multiplier = 1.0 + gpu_cost_total_usd = single_run_gpu_cost_usd * experiments_multiplier total_capex_usd = gpu_cost_total_usd + prep_cost_usd @@ -171,36 +286,34 @@ def compute_finetune(inp: dict) -> dict: "gpu_cost_total_usd": round(gpu_cost_total_usd, 4), "prep_cost_usd": prep_cost_usd, "total_capex_usd": round(total_capex_usd, 4), + "warnings": warnings, "derivation": derivation, } def main(): if len(sys.argv) < 2: - print(json.dumps({"error": "missing subcommand"})) + print(json.dumps({"error": "missing subcommand", "field": None})) sys.exit(2) sub = sys.argv[1] if sub not in ("inference", "finetune"): - print(json.dumps({"error": f"unknown subcommand {sub}"})) + print(json.dumps({"error": f"unknown subcommand {sub}", "field": None})) sys.exit(2) try: inp = json.load(sys.stdin) except json.JSONDecodeError as e: - print(json.dumps({"error": f"invalid JSON: {e}"})) + print(json.dumps({"error": f"invalid JSON: {e}", "field": None})) sys.exit(2) try: if sub == "inference": result = compute_inference(inp) else: result = compute_finetune(inp) - except KeyError as e: - print(json.dumps({"error": f"missing field {e.args[0]}"})) - sys.exit(2) except ValueError as e: - print(json.dumps({"error": str(e)})) + print(json.dumps({"error": str(e), "field": None})) sys.exit(2) except Exception as e: - print(json.dumps({"error": str(e)})) + print(json.dumps({"error": str(e), "field": None})) sys.exit(1) print(json.dumps(result)) sys.exit(0) diff --git a/tests/test_calc.py b/tests/test_calc.py index cacf6dc..e70b5cf 100644 --- a/tests/test_calc.py +++ b/tests/test_calc.py @@ -10,7 +10,7 @@ from pathlib import Path REPO_ROOT = Path(__file__).resolve().parent.parent -CALC = REPO_ROOT / "calc.py" +CALC = REPO_ROOT / "scripts" / "calc.py" def run(subcmd, payload): @@ -103,12 +103,153 @@ def test_unknown_quant(self): rc, r = run("inference", { "params_b": 70, "active_params_b": 70, "quant": "fp8", "queries_per_week": 1, "avg_tokens_per_query": 1, - "api_cost_per_query_usd": 0, "traffic_pattern": "business", + "api_cost_per_query_usd": 0.001, "traffic_pattern": "business", "gpu": {"vram_gb": 80, "usd_per_hr": 1}, }) self.assertEqual(rc, 2) self.assertIn("unknown quant", r["error"]) +INFERENCE_BASE = { + "params_b": 70, + "active_params_b": 70, + "quant": "int4", + "queries_per_week": 1_000_000, + "avg_tokens_per_query": 800, + "api_cost_per_query_usd": 0.002, + "traffic_pattern": "business", + "gpu": {"name": "H100 80GB", "vram_gb": 80, "usd_per_hr": 2.90, "bf16_tflops": 989}, +} + +FINETUNE_BASE = { + "active_params_b": 65, + "total_params_b": 65, + "method": "qlora", + "num_examples": 10000, + "tokens_per_example": 500, + "epochs": 3, + "experiments_multiplier": 1.0, + "prep_cost_usd": 0, + "gpu": {"name": "H100 80GB", "vram_gb": 80, "usd_per_hr": 2.90, "bf16_tflops": 989, "gpus_per_node": 8}, +} + + +class TestHardenedValidation(unittest.TestCase): + def test_null_queries_per_week_inference(self): + payload = dict(INFERENCE_BASE) + payload["queries_per_week"] = None + rc, r = run("inference", payload) + self.assertEqual(rc, 2) + self.assertIn("error", r) + self.assertEqual(r["field"], "queries_per_week") + + def test_negative_queries_per_week_inference(self): + payload = dict(INFERENCE_BASE) + payload["queries_per_week"] = -1000 + rc, r = run("inference", payload) + self.assertEqual(rc, 2) + self.assertIn("error", r) + self.assertEqual(r["field"], "queries_per_week") + + def test_zero_bf16_tflops_finetune(self): + payload = dict(FINETUNE_BASE) + payload["gpu"] = dict(FINETUNE_BASE["gpu"]) + payload["gpu"]["bf16_tflops"] = 0 + rc, r = run("finetune", payload) + self.assertEqual(rc, 2) + self.assertEqual(r["field"], "bf16_tflops") + + def test_negative_usd_per_hr_inference(self): + payload = dict(INFERENCE_BASE) + payload["gpu"] = dict(INFERENCE_BASE["gpu"]) + payload["gpu"]["usd_per_hr"] = -1 + rc, r = run("inference", payload) + self.assertEqual(rc, 2) + self.assertEqual(r["field"], "usd_per_hr") + + def test_cold_per_query_without_hot_hours(self): + payload = dict(INFERENCE_BASE) + payload["traffic_pattern"] = "cold_per_query" + rc, r = run("inference", payload) + self.assertEqual(rc, 2) + self.assertEqual(r["field"], "hot_hours_per_week") + + def test_invalid_traffic_pattern(self): + payload = dict(INFERENCE_BASE) + payload["traffic_pattern"] = "weekends_only" + rc, r = run("inference", payload) + self.assertEqual(rc, 2) + self.assertEqual(r["field"], "traffic_pattern") + + def test_active_exceeds_total_params(self): + payload = dict(INFERENCE_BASE) + payload["active_params_b"] = 100 + payload["params_b"] = 70 + rc, r = run("inference", payload) + self.assertEqual(rc, 2) + self.assertEqual(r["field"], "active_params_b") + + def test_string_for_numeric_field(self): + payload = dict(INFERENCE_BASE) + payload["queries_per_week"] = "100k" + rc, r = run("inference", payload) + self.assertEqual(rc, 2) + self.assertEqual(r["field"], "queries_per_week") + + def test_nan_value(self): + payload = dict(INFERENCE_BASE) + payload["queries_per_week"] = None + rc, r = run("inference", payload) + self.assertEqual(rc, 2) + self.assertEqual(r["field"], "queries_per_week") + + def test_finetune_zero_num_examples(self): + payload = dict(FINETUNE_BASE) + payload["num_examples"] = 0 + rc, r = run("finetune", payload) + self.assertEqual(rc, 2) + self.assertEqual(r["field"], "num_examples") + + def test_finetune_zero_epochs(self): + payload = dict(FINETUNE_BASE) + payload["epochs"] = 0 + rc, r = run("finetune", payload) + self.assertEqual(rc, 2) + self.assertEqual(r["field"], "epochs") + + +class TestInfeasibleOutput(unittest.TestCase): + def test_infeasible_zeroes_savings(self): + rc, r = run("inference", { + "params_b": 405, + "active_params_b": 405, + "quant": "fp16", + "queries_per_week": 1_000_000, + "avg_tokens_per_query": 800, + "api_cost_per_query_usd": 0.002, + "traffic_pattern": "business", + "gpu": {"name": "H100 80GB", "vram_gb": 80, "usd_per_hr": 2.90, "bf16_tflops": 989}, + }) + self.assertEqual(rc, 0) + self.assertTrue(r["infeasible"]) + self.assertEqual(r["savings_pct"], 0) + self.assertEqual(r["weekly_savings_usd"], 0) + self.assertGreater(len(r["warnings"]), 0) + + +class TestWarnings(unittest.TestCase): + def test_experiments_multiplier_clamp(self): + payload = dict(FINETUNE_BASE) + payload["experiments_multiplier"] = 0.5 + rc, r = run("finetune", payload) + self.assertEqual(rc, 0) + self.assertTrue(any("experiments_multiplier 0.5 clamped to 1.0" in w for w in r["warnings"])) + + def test_no_warnings_on_clean_input(self): + rc, r = run("inference", INFERENCE_BASE) + self.assertEqual(rc, 0) + self.assertEqual(r["warnings"], []) + + if __name__ == "__main__": unittest.main()