benchmark.py

#!/usr/bin/env python3
"""Benchmark: 20 úloh — A/B ThinkingCap vs base Qwen3.6-27B."""
import json, time, sys, requests
from pathlib import Path
from datetime import datetime

URL = "http://192.168.110.67:8000/v1/chat/completions"
MODEL = "rtx6000-impl"
MAX_TOKENS = 16384
RESULTS = Path(file).parent / "benchmark_results.json"

Sampling recipe ze serveru: neposílat temp/top_p atd., necháme defaulty vLLM

HEADERS = {"Content-Type": "application/json"}

TASKS = [
{
"id": 1,
"category": "Kódování",
"prompt": "Napiš Python funkci validuj_iban_cz(iban: str) -> bool pro český IBAN (CZ + 22 znaků, mod-97 kontrola dle ISO 13616) včetně normalizace mezer a malých písmen. Přidej 6 unit testů (3 platné, 3 neplatné).\nKontrola: CZ65 0800 0000 1920 0014 5399 → True; změna jedné číslice → False.",
"check": "mod97, CZ prefix, 6 testů"
},
{
"id": 2,
"category": "Kódování",
"prompt": "Napiš Python třídu LRUCacheTTL(kapacita, ttl_s) — LRU vytěsňování + expirace položek po TTL. Metody get/put O(1). Bez externích knihoven. Unit testy s mockovaným časem.\nKontrola: get po expiraci vrací None; nejstarší se vytěsní při přeplnění.",
"check": "O(1), get/put, TTL, LRU"
},
{
"id": 3,
"category": "Kódování",
"prompt": "Napiš robustní CSV parser v Pythonu (bez modulu csv): uvozovky, escapované uvozovky (""), čárky i \n uvnitř polí, BOM, CRLF. Vrať list[list[str]].\nKontrola: 'a,"b,""c""\nd",e' → [['a','b,"c"\nd','e']].",
"check": "quotes, escaped quotes, newlines, BOM, CRLF"
},
{
"id": 4,
"category": "Kódování",
"prompt": "Regulární výraz (Python, komentovaný re.VERBOSE) pro česká telefonní čísla: volitelně +420/00420, mezery/pomlčky mezi trojicemi, celkem 9 číslic. Uveď 8 testovacích případů.\nKontrola: "+420 601 123 456" ✅, "601123456" ✅, "60112345" ❌.",
"check": "re.VERBOSE, +420/00420, 9 číslic, 8 testů"
},
{
"id": 5,
"category": "Kódování",
"prompt": "Máš SQL tabulky objednavky(id, zakaznik_id, datum, castka) a zakaznici(id, jmeno, region). Napiš dotaz: pro každý region top 3 zákazníci dle součtu částek za rok 2025, včetně pořadí. Pak vysvětli, jaké indexy dotaz potřebuje a proč.\nKontrola: window funkce (RANK/ROW_NUMBER) + PARTITION BY region; index (datum) a (zakaznik_id).",
"check": "window function, PARTITION BY, index recommendation"
},
{
"id": 6,
"category": "Kódování",
"prompt": "Napiš bash skript pro rotaci logů: adresář /var/log/aplikace, soubory *.log starší 7 dní zagzipovat, starší 30 dní smazat, s dry-run přepínačem a ošetřením mezer v názvech. Shellcheck-clean.\nKontrola: find -print0 | while read -d '', žádné neuvozené proměnné.",
"check": "find -print0, gzip, rm, dry-run, spaces safe"
},
{
"id": 7,
"category": "Kódování",
"prompt": "Napiš PowerShell funkci Test-DdaReady pro Hyper-V host: ověří SR-IOV/ACS, vypne kartu dle PCI ID (parametr), vrátí objekt se stavem (Připraveno/Chybí kroky + důvody). Bez skutečného spuštění — jen kód.\nKontrola: Get-PnpDevice dle InstanceId, Disable-PnpDevice -Confirm:$false, Dismount-VMHostAssignableDevice.",
"check": "Get-PnpDevice, Disable-PnpDevice, Dismount-VMHostAssignableDevice"
},
{
"id": 8,
"category": "Kódování",
"prompt": "Refaktoruj tento kód (uveď ho modelu doslovně):\n\ndef zprac(d):\n r=[]\n for i in range(len(d)):\n if d[i]!=None:\n if type(d[i])==str:\n if d[i].strip()!="":\n r.append(d[i].strip().lower())\n else:\n if type(d[i])==int or type(d[i])==float:\n if d[i]>0: r.append(str(d[i]))\n return r\n\nCíl: idiomatický Python, type hints, docstring, zachovat chování 1:1, přidej testy dokazující ekvivalenci.\nKontrola: isinstance, comprehension/guard klauzule, chování identické (i pořadí podmínek pro bool — pozor, bool je subclass int!).",
"check": "isinstance, type hints, docstring, bool subclass int"
},
{
"id": 9,
"category": "Algoritmy",
"prompt": "Batoh (0/1 knapsack): kapacita 15 kg, předměty [(3,40),(5,70),(7,90),(2,20),(4,55),(6,80)] (kg, Kč). Najdi optimum dynamickým programováním, ukaž tabulku a rekonstruuj vybrané předměty.\nKontrola: optimum 225 Kč (3+5+4+2=14 kg? ověř — model musí ukázat výpočet; správnost = konzistentní DP tabulka + rekonstrukce).",
"check": "DP table, reconstruction, optimal value"
},
{
"id": 10,
"category": "Algoritmy",
"prompt": "Graf: města A–F, silnice s časy A-B:4, A-C:2, B-C:1, B-D:5, C-D:8, C-E:10, D-E:2, D-F:6, E-F:3. Najdi nejkratší cestu A→F Dijkstrou, ukaž tabulku vzdáleností krok po kroku.\nKontrola: A→C→B→D→E→F? model musí ukázat kroky; správná délka = ověř ručně dle jeho tabulky (očekávám A-C-B-D-E-F = 2+1+5+2+3=13).",
"check": "Dijkstra steps, path A-F, distance"
},
{
"id": 11,
"category": "Algoritmy",
"prompt": "Kolika způsoby lze rozměnit 100 Kč na mince 1, 2, 5, 10, 20, 50 Kč? Vyřeš DP, uveď mezivýsledky pro 10, 20, 50 Kč a finální číslo.\nKontrola: konzistence mezivýsledků (10 Kč = 10 způsobů? model ukáže); finální číslo musí sedět s jeho mezikroky.",
"check": "DP change-making, intermediate results, final count"
},
{
"id": 12,
"category": "Algoritmy",
"prompt": "Letadlo odlétá z Prahy 23:45 SELČ 28.3.2026 (noc přechodu na letní čas je 29.3.), letí 11 h 20 min do Tokia (UTC+9). Kolik je v Tokiu při příletu a kolik v Praze? Pozor na změnu času během letu.\nKontrola: SELČ už platí (změna 29.3. 02:00 SEČ→SELČ — ale 28.3. večer je ještě SEČ! Chyták: 23:45 28.3. = SEČ/UTC+1; přílet 11:05 UTC 29.3. = 20:05 Tokio; Praha v tu chvíli 13:05 SELČ). Správně = model si všimne přechodu.",
"check": "SEČ vs SELČ trap, UTC+1 departure, arrival times"
},
{
"id": 13,
"category": "Čeština",
"prompt": "Shrň následující text do 3 vět a vytěž strukturovaně (JSON: strany, předmět, částka, lhůty): „Smlouva o dílo uzavřená mezi objednatelem XYZ s.r.o., IČO 12345678, se sídlem v Liberci, a zhotovitelem ABC a.s. Předmětem je dodávka a instalace serverové infrastruktury v hodnotě 2 450 000 Kč bez DPH. Zhotovitel se zavazuje dílo dokončit do 90 dnů od podpisu, záruka činí 24 měsíců. Smluvní pokuta za prodlení je 0,05 % z ceny díla za každý den."\nKontrola: JSON validní, částka 2450000, lhůty 90 dnů/24 měsíců/0,05 %, diakritika bezchybná.",
"check": "valid JSON, amount 2450000, deadlines, diacritics"
},
{
"id": 14,
"category": "Čeština",
"prompt": "Oprav diakritiku a pravopis (text uveď modelu doslovně): „Vazeny pane rediteli, dovolte mi abych Vas informoval ze nase spolecnost uspesne dokoncila migraci datoveho centra. Behem cele akce nedoslo k zadnemu vypadku sluzeb a vsechna data byla prenesena bez ztraty. Dekuji vsem clenum tymu za jejich usili."\nKontrola: 100 % správná diakritika (Vážený, řediteli, abych Vás, že naše společnost úspěšně, datového, Během celé, nedošlo k žádnému výpadku služeb, přenesena, Děkuji všem členům týmu, úsilí) + čárka za „mi".",
"check": "100% diacritics, comma after mi"
},
{
"id": 15,
"category": "Čeština",
"prompt": "Napiš formální email jednateli firmy: žádost o schválení nákupu GPU serveru za 800 000 Kč, argumentace úsporou cloudu (současné náklady 45 000 Kč/měsíc), návratnost, rizika, 150–200 slov, formální čeština.\nKontrola: návratnost ~18 měsíců zmíněna a spočtena správně, struktura (oslovení/věc/argumenty/závěr), bezchybná čeština.",
"check": "ROI ~18 months, structure, formal Czech, word count"
},
{
"id": 16,
"category": "Strukturovaný výstup",
"prompt": "Navrhni JSON Schema (draft 2020-12) pro konfiguraci vLLM služby: model_path (povinné, regex ^/ai_models/), served_name, port (1024–65535), dtype (enum bfloat16/auto), max_model_len (násobek 1024, max 262144), speculative (objekt: method enum [mtp], num_tokens 1–5). Přidej validní a nevalidní příklad.\nKontrola: schema syntakticky validní, required pole, multipleOf 1024, oba příklady konzistentní se schématem.",
"check": "valid schema, required, multipleOf, valid+invalid examples"
},
{
"id": 17,
"category": "Strukturovaný výstup",
"prompt": "Převeď na docker-compose.yml: kontejner litellm (image ghcr.io/berriai/litellm:main-latest), port 4000:4000, mount ./config.yaml:/app/config.yaml, env LITELLM_MASTER_KEY ze souboru .env, restart unless-stopped, healthcheck curl /health po 30 s, síť ai-net (externí). Přidej komentáře česky.\nKontrola: validní YAML, env_file/environment správně, healthcheck syntaxe, external network.",
"check": "valid YAML, env_file, healthcheck, external network"
},
{
"id": 18,
"category": "Diagnostika",
"prompt": "Diagnostikuj (traceback uveď modelu doslovně): „FileNotFoundError: [Errno 2] No such file or directory: 'ninja'" v subprocess.run volaném z flashinfer/jit/core.py build_and_load při startu vLLM. Systém Ubuntu, driver OK, nvcc nainstalován. Co je příčina, jak opravit, jak ověřit?\nKontrola: chybí ninja-build balík (ne pip ninja!), apt install ninja-build, which ninja; bonus: zmínka build-essential.",
"check": "apt install ninja-build, not pip ninja, which ninja"
},
{
"id": 19,
"category": "Diagnostika",
"prompt": "Analyzuj log a urči příčinu (uveď doslovně): „Maximum concurrency for 262,144 tokens per request: 3.61x" a poté klient dostává 400 Bad Request s hláškou o překročení max_model_len při požadavku s max_tokens=163840 a promptem 5000 tokenů. Vysvětli vztah context window × max_tokens a navrhni správné hodnoty.\nKontrola: prompt+max_tokens ≤ max_model_len; 5000+163840 > 163840 (pokud server na 163840) resp. vysvětlení dvou limitů; doporučení max_tokens 8–16k.",
"check": "context window explanation, two limits, recommendation"
},
{
"id": 20,
"category": "Diagnostika",
"prompt": "Code review: najdi VŠECHNY chyby (uveď kód doslovně):\n\nimport threading\ncitac = 0\ndef pridej():\n global citac\n for _ in range(100000):\n citac += 1\nvlakna = [threading.Thread(target=pridej) for _ in range(5)]\nfor v in vlakna: v.start()\nprint(f"Vysledek: {citac}")\n\nKontrola: (1) chybí join před print, (2) race condition citac += 1 není atomické → Lock, (3) GIL nezaručuje atomicitu složené operace; bonus: očekávaný výsledek < 500000.",
"check": "join, race condition/Lock, GIL explanation, <500000"
},
]

def run_task(task: dict, model_name: str) -> dict:
"""Run a single task against the model and return result."""
payload = {
"model": MODEL,
"messages": [{"role": "user", "content": task["prompt"]}],
"max_tokens": MAX_TOKENS,
}

t0 = time.monotonic()
try:
    resp = requests.post(URL, headers=HEADERS, json=payload, timeout=300)
    elapsed = time.monotonic() - t0

    if resp.status_code != 200:
        return {
            "error": f"HTTP {resp.status_code}",
            "error_body": resp.text[:500],
            "elapsed": round(elapsed, 2),
        }

    data = resp.json()
    choice = data["choices"][0]
    content = choice["message"].get("content", "")

    # Parse reasoning vs content
    reasoning = ""
    content_clean = content
    if "<think>" in content and "</think>" in content:
        reasoning = content[content.index("<think>") + 7:content.index("</think>")].strip()
        content_clean = content[content.index("</think>") + 8:].strip()
    elif "<think>" in content:
        # unclosed think tag
        reasoning = content[content.index("<think>") + 7:].strip()
        content_clean = ""

    usage = data.get("usage", {})
    return {
        "completion_tokens": usage.get("completion_tokens", 0),
        "prompt_tokens": usage.get("prompt_tokens", 0),
        "total_tokens": usage.get("total_tokens", 0),
        "finish_reason": choice.get("finish_reason", "unknown"),
        "reasoning_len": len(reasoning),
        "content_len": len(content_clean),
        "elapsed": round(elapsed, 2),
        "content": content_clean,
        "reasoning": reasoning,
    }

except Exception as e:
    elapsed = time.monotonic() - t0
    return {
        "error": str(e),
        "elapsed": round(elapsed, 2),
    }

def run_task_with_meta(task: dict, model_name: str) -> dict:
"""Wrapper that attaches task metadata to the result."""
res = run_task(task, model_name)
res["task_id"] = task["id"]
res["category"] = task["category"]
res["check"] = task["check"]
return res

def run_benchmark(model_name: str):
"""Run all tasks with max 5 parallel workers."""
from concurrent.futures import ThreadPoolExecutor, as_completed

print(f"\n{'='*70}")
print(f"BENCHMARK: {model_name}")
print(f"Model: {MODEL} @ {URL}")
print(f"max_tokens: {MAX_TOKENS}")
print(f"{'='*70}\n")

results = []
total_tokens = 0
total_time = 0
finished = 0

t0 = time.monotonic()

with ThreadPoolExecutor(max_workers=5) as pool:
    futures = {
        pool.submit(run_task_with_meta, task, model_name): task
        for task in TASKS
    }
    for future in as_completed(futures):
        task = futures[future]
        res = future.result()
        finished += 1
        tid = task["id"]
        cat = task["category"]

        if "error" in res:
            print(f"[{tid}/20] {cat} — ERROR: {res['error']} ({finished}/20)")
            results.append(res)
            continue

        tokens = res["completion_tokens"]
        total_tokens += tokens
        total_time += res["elapsed"]

        print(f"[{tid}/20] {cat} — ✅ {tokens} tok, {res['elapsed']}s, "
              f"think={res['reasoning_len']}B, "
              f"content={res['content_len']}B, "
              f"finish={res['finish_reason']} ({finished}/20)")

        results.append(res)

total_wall = time.monotonic() - t0
# Sort by task_id
results.sort(key=lambda r: r["task_id"])

print(f"\n{'='*70}")
print(f"SUMMARY: {len(results)} tasks, {total_tokens} tokens, "
      f"{total_time:.1f}s (sum), {total_wall:.1f}s (wall)")
print(f"{'='*70}\n")

return {
    "model": model_name,
    "timestamp": datetime.now().isoformat(),
    "total_tokens": total_tokens,
    "total_time": round(total_time, 2),
    "total_wall": round(total_wall, 2),
    "results": results,
}

def save_run(data: dict, model_name: str):
"""Save results to file."""
run_file = Path(file).parent / f"benchmark_{model_name}.json"
with open(run_file, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
print(f"Saved: {run_file}")

def print_summary(tc_data: dict, base_data: dict):
"""Print comparison table."""
tc = {r["task_id"]: r for r in tc_data["results"]}
base = {r["task_id"]: r for r in base_data["results"]}

print(f"\n{'='*120}")
print(f"{'#':>2} | {'TC ✅/❌':>6} | {'TC tok':>7} | {'TC s':>6} | {'Base ✅/❌':>8} | {'Base tok':>8} | {'Base s':>7} | Pozn.")
print(f"{'-'*120}")

for tid in range(1, 21):
    tc_r = tc.get(tid, {})
    base_r = base.get(tid, {})

    tc_ok = "✅" if "error" not in tc_r and tc_r.get("finish_reason") == "stop" else "❌"
    base_ok = "✅" if "error" not in base_r and base_r.get("finish_reason") == "stop" else "❌"

    tc_tok = tc_r.get("completion_tokens", 0)
    base_tok = base_r.get("completion_tokens", 0)
    tc_time = tc_r.get("elapsed", 0)
    base_time = base_r.get("elapsed", 0)

    print(f"{tid:>2} | {tc_ok:>6} | {tc_tok:>7} | {tc_time:>6.1f} | {base_ok:>8} | {base_tok:>8} | {base_time:>7.1f} |")

print(f"{'='*120}")
tc_pass = sum(1 for r in tc_data["results"] if "error" not in r and r.get("finish_reason") == "stop")
base_pass = sum(1 for r in base_data["results"] if "error" not in r and r.get("finish_reason") == "stop")
print(f"TC pass: {tc_pass}/20, tokens: {tc_data['total_tokens']}, time: {tc_data['total_time']}s")
print(f"Base pass: {base_pass}/20, tokens: {base_data['total_tokens']}, time: {base_data['total_time']}s")
print(f"Token diff: {tc_data['total_tokens'] - base_data['total_tokens']:+d}")
print(f"Time diff: {tc_data['total_time'] - base_data['total_time']:+.1f}s")

if name == "main":
model_name = sys.argv[1] if len(sys.argv) > 1 else "TC"

print(f"Starting benchmark run for: {model_name}")
data = run_benchmark(model_name)
save_run(data, model_name)

# Check if both runs exist and print comparison
tc_file = Path(__file__).parent / "benchmark_TC.json"
base_file = Path(__file__).parent / "benchmark_base.json"

if tc_file.exists() and base_file.exists():
    with open(tc_file) as f:
        tc_data = json.load(f)
    with open(base_file) as f:
        base_data = json.load(f)
    print_summary(tc_data, base_data)

Thank you so much for contributing!