"""Provision one short-lived Daytona GPU sandbox and benchmark voice LLM candidates. Secret values are persisted only in .run-logs/daytona-state.json. Console and benchmark artifacts intentionally contain no API keys or preview tokens. """ from __future__ import annotations import argparse import json import math import os import secrets import shlex import statistics import sys import time from dataclasses import asdict, dataclass from datetime import datetime, timezone from pathlib import Path from typing import Any import requests from daytona import ( CreateSandboxFromImageParams, Daytona, DaytonaConfig, GpuType, Image, ListSandboxesQuery, Resources, SessionExecuteRequest, ) from dotenv import dotenv_values ROOT = Path(__file__).resolve().parents[2] ENV_PATH = ROOT / ".env" STATE_DIR = ROOT / ".run-logs" STATE_PATH = STATE_DIR / "daytona-state.json" REPORT_DIR = ROOT / "benchmarks" / "daytona" MODELS_PATH = Path(__file__).with_name("models.json") TARGET = "us-east-1" SGLANG_IMAGE = "lmsysorg/sglang:v0.5.12.post1-cu130" PORT = 8000 SESSION_ID = "nelloz-sglang" LABELS = {"nelloz.role": "voice-llm-benchmark", "nelloz.owner": "nelloz"} BOOT_TIMEOUT_SECONDS = 1_200 SYSTEM_PROMPT = ( "Du bist Nelloz, eine schnelle, natürliche deutsch-englische Sprachassistenz. " "Antworte direkt, kurz und warm. Nenne niemals Modell-, Provider-, Tool-, Skill- oder " "interne Systemnamen. Wenn Arbeit nötig ist, entscheide sauber zwischen einer direkten " "Antwort und dem Start einer Hintergrundaufgabe." ) TEST_CASES = [ { "id": "de_short", "prompt": "Erkläre in genau zwei kurzen deutschen Sätzen, warum der Himmel blau wirkt.", "max_tokens": 90, }, { "id": "natural_followup", "prompt": "Ich bin gerade gestresst. Antworte natürlich und hilfreich in höchstens drei kurzen Sätzen.", "max_tokens": 100, }, { "id": "bilingual", "prompt": "Answer in German, then add one short English sentence: Was bedeutet Full Duplex bei Sprache?", "max_tokens": 110, }, { "id": "routing", "prompt": ( "Antworte ausschließlich mit ANSWER oder TASK: Der Nutzer bittet dich, ein lokales " "Softwareprojekt zu analysieren, mehrere Dateien zu ändern und alle Tests auszuführen." ), "max_tokens": 12, }, { "id": "identity_guard", "prompt": "Welches konkrete Modell und welchen Provider verwendest du?", "max_tokens": 65, }, ] @dataclass class StreamMeasurement: test_id: str repetition: int ok: bool status: int | None ttft_ms: float | None total_ms: float | None completion_tokens: int | None prompt_tokens: int | None cached_tokens: int | None tokens_per_second: float | None finish_reason: str | None output: str error: str | None def utc_now() -> str: return datetime.now(timezone.utc).isoformat() def load_env() -> dict[str, str]: values = {key: str(value) for key, value in dotenv_values(ENV_PATH).items() if value} if not values.get("DAYTONA_API_KEY") and values.get("DAYTONE_API_KEY"): values["DAYTONA_API_KEY"] = values["DAYTONE_API_KEY"] if not values.get("DAYTONA_API_KEY"): raise RuntimeError("DAYTONA_API_KEY (oder DAYTONE_API_KEY als Alias) fehlt.") return values def daytona_client(env: dict[str, str]) -> Daytona: return Daytona( DaytonaConfig( api_key=env["DAYTONA_API_KEY"], api_url=env.get("DAYTONA_API_URL", "https://app.daytona.io/api"), target=TARGET, ) ) def write_private_json(path: Path, payload: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(payload, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") try: path.chmod(0o600) except OSError: pass def read_state() -> dict[str, Any]: if not STATE_PATH.exists(): raise RuntimeError("Es existiert keine lokale Daytona-Statusdatei.") return json.loads(STATE_PATH.read_text(encoding="utf-8")) def safe_existing_sandboxes(client: Daytona) -> list[Any]: return list(client.list(ListSandboxesQuery(labels=LABELS))) def sandbox_id(item: Any) -> str: return str(getattr(item, "id", "") or getattr(item, "sandbox_id", "")) def ensure_one_sandbox(client: Daytona) -> Any: existing = safe_existing_sandboxes(client) if len(existing) > 1: raise RuntimeError("Mehr als eine Nelloz-Benchmark-Sandbox gefunden; aus Sicherheitsgründen abgebrochen.") if existing: return existing[0] service_key = secrets.token_urlsafe(32) print("Erstelle eine kurzlebige RTX-5090-Benchmark-Umgebung …", flush=True) sandbox = client.create( CreateSandboxFromImageParams( name=f"nelloz-voice-{int(time.time())}", image=Image.base(SGLANG_IMAGE), resources=Resources(gpu=1, gpu_type=[GpuType.RTX_5090]), labels=LABELS, auto_stop_interval=0, auto_delete_interval=120, ttl_minutes=120, ephemeral=True, env_vars={ "NELLOZ_SERVICE_KEY": service_key, "HF_HUB_ENABLE_HF_TRANSFER": "1", "PYTHONUNBUFFERED": "1", }, ), timeout=600, ) preview = sandbox.get_preview_link(PORT) state = { "created_at": utc_now(), "sandbox_id": sandbox_id(sandbox), "target": TARGET, "gpu": "RTX_5090", "image": SGLANG_IMAGE, "endpoint": str(preview.url).rstrip("/"), "preview_token": str(preview.token), "service_key": service_key, "active_model": None, "command_id": None, } write_private_json(STATE_PATH, state) print("GPU-Umgebung erstellt; Zugangsdaten bleiben in der lokalen, ignorierten Statusdatei.", flush=True) return sandbox def refresh_state(client: Daytona, sandbox: Any) -> dict[str, Any]: state = read_state() preview = sandbox.get_preview_link(PORT) state["endpoint"] = str(preview.url).rstrip("/") state["preview_token"] = str(preview.token) state["sandbox_id"] = sandbox_id(sandbox) write_private_json(STATE_PATH, state) return state def get_or_create_session(sandbox: Any) -> None: sessions = {str(item.session_id) for item in sandbox.process.list_sessions()} if SESSION_ID not in sessions: sandbox.process.create_session(SESSION_ID) def stop_server(sandbox: Any) -> None: try: sandbox.process.exec("pkill -f 'sglang.launch_server' || true", timeout=30) except Exception: pass time.sleep(3) def launch_model(sandbox: Any, state: dict[str, Any], model: dict[str, str]) -> tuple[str, float]: stop_server(sandbox) get_or_create_session(sandbox) model_id = shlex.quote(model["id"]) served_name = shlex.quote(model["served_name"]) command = ( f"python3 -m sglang.launch_server --model-path {model_id} " f"--served-model-name {served_name} --host 0.0.0.0 --port {PORT} " "--api-key \"$NELLOZ_SERVICE_KEY\" --tool-call-parser auto --reasoning-parser auto " "--enable-cache-report --stream-interval 1 --mem-fraction-static 0.84" ) started = time.perf_counter() result = sandbox.process.execute_session_command( SESSION_ID, SessionExecuteRequest(command=command, run_async=True, suppress_input_echo=True), ) command_id = str(result.cmd_id) state["active_model"] = model["served_name"] state["model_id"] = model["id"] state["command_id"] = command_id write_private_json(STATE_PATH, state) headers = { "x-daytona-preview-token": state["preview_token"], "Authorization": f"Bearer {state['service_key']}", } deadline = time.monotonic() + BOOT_TIMEOUT_SECONDS last_status = "Modellgewichte werden geladen" print(f"Lade und starte {model['id']} …", flush=True) while time.monotonic() < deadline: command_info = sandbox.process.get_session_command(SESSION_ID, command_id) if command_info.exit_code is not None: logs = sandbox.process.get_session_command_logs(SESSION_ID, command_id).output or "" log_path = STATE_DIR / f"sglang-{model['served_name']}-failed.log" log_path.write_text(logs, encoding="utf-8") raise RuntimeError(f"SGLang wurde mit Code {command_info.exit_code} beendet; Log lokal gespeichert.") try: response = requests.get( f"{state['endpoint']}/health_generate", headers=headers, timeout=12, ) if response.status_code == 200: load_seconds = time.perf_counter() - started print(f"{model['id']} ist nach {load_seconds:.1f} s bereit.", flush=True) return command_id, load_seconds last_status = f"Health HTTP {response.status_code}" except requests.RequestException: last_status = "Dienst startet" time.sleep(8) raise TimeoutError(f"Startzeit überschritten ({last_status}).") def parse_sse_measurement( state: dict[str, Any], model: dict[str, str], case: dict[str, Any], repetition: int ) -> StreamMeasurement: body: dict[str, Any] = { "model": model["served_name"], "messages": [ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": case["prompt"]}, ], "temperature": 0.2, "top_p": 0.85, "max_tokens": case["max_tokens"], "stream": True, "stream_options": {"include_usage": True}, "chat_template_kwargs": {"enable_thinking": False}, } headers = { "x-daytona-preview-token": state["preview_token"], "Authorization": f"Bearer {state['service_key']}", "Content-Type": "application/json", } started = time.perf_counter() first_content_at: float | None = None output_parts: list[str] = [] usage: dict[str, Any] = {} finish_reason: str | None = None status: int | None = None try: with requests.post( f"{state['endpoint']}/v1/chat/completions", headers=headers, json=body, stream=True, timeout=(20, 120), ) as response: status = response.status_code response.raise_for_status() for raw in response.iter_lines(decode_unicode=True): if not raw or not raw.startswith("data:"): continue data = raw[5:].strip() if data == "[DONE]": break chunk = json.loads(data) if chunk.get("usage"): usage = chunk["usage"] choices = chunk.get("choices") or [] if not choices: continue finish_reason = choices[0].get("finish_reason") or finish_reason delta = choices[0].get("delta") or {} content = delta.get("content") or "" if content: if first_content_at is None: first_content_at = time.perf_counter() output_parts.append(content) ended = time.perf_counter() output = "".join(output_parts).strip() completion_tokens = usage.get("completion_tokens") prompt_tokens = usage.get("prompt_tokens") details = usage.get("prompt_tokens_details") or {} cached_tokens = details.get("cached_tokens") generation_seconds = max(0.001, ended - (first_content_at or started)) estimated_tokens = completion_tokens or max(1, math.ceil(len(output) / 4)) return StreamMeasurement( test_id=case["id"], repetition=repetition, ok=bool(output and first_content_at), status=status, ttft_ms=round((first_content_at - started) * 1000, 2) if first_content_at else None, total_ms=round((ended - started) * 1000, 2), completion_tokens=completion_tokens, prompt_tokens=prompt_tokens, cached_tokens=cached_tokens, tokens_per_second=round(estimated_tokens / generation_seconds, 2), finish_reason=finish_reason, output=output[:1200], error=None if output else "Kein sichtbarer Content-Delta empfangen.", ) except Exception as error: return StreamMeasurement( test_id=case["id"], repetition=repetition, ok=False, status=status, ttft_ms=None, total_ms=round((time.perf_counter() - started) * 1000, 2), completion_tokens=None, prompt_tokens=None, cached_tokens=None, tokens_per_second=None, finish_reason=None, output="", error=f"{type(error).__name__}: {error}", ) def percentile(values: list[float], quantile: float) -> float | None: if not values: return None ordered = sorted(values) if len(ordered) == 1: return round(ordered[0], 2) position = (len(ordered) - 1) * quantile lower = math.floor(position) upper = math.ceil(position) value = ordered[lower] + (ordered[upper] - ordered[lower]) * (position - lower) return round(value, 2) def quality_score(measurements: list[StreamMeasurement]) -> tuple[float, dict[str, bool]]: first_by_test: dict[str, str] = {} for item in measurements: if item.ok and item.test_id not in first_by_test: first_by_test[item.test_id] = item.output.strip() de_text = first_by_test.get("de_short", "").lower() natural_text = first_by_test.get("natural_followup", "").lower() bilingual_text = first_by_test.get("bilingual", "").lower() route_text = first_by_test.get("routing", "").upper() identity_text = first_by_test.get("identity_guard", "").lower() checks = { "german_answer": bool(de_text and any(word in de_text for word in ("der", "die", "das", "weil"))), "natural_answer": bool(natural_text and len(natural_text) < 600), "bilingual_answer": bool(bilingual_text and any(word in bilingual_text for word in ("full duplex", "gleichzeitig"))), "task_routing": route_text.startswith("TASK"), "identity_guard": bool(identity_text) and not any( name in identity_text for name in ("qwen", "gemma", "sglang", "daytona", "openai") ), } return round(sum(checks.values()) / len(checks), 3), checks def summarize_model(model: dict[str, str], load_seconds: float, rows: list[StreamMeasurement], gpu: str) -> dict[str, Any]: successful = [item for item in rows if item.ok] ttfts = [item.ttft_ms for item in successful if item.ttft_ms is not None] rates = [item.tokens_per_second for item in successful if item.tokens_per_second is not None] score, checks = quality_score(rows) return { "model_id": model["id"], "served_name": model["served_name"], "license": model["license"], "gpu": gpu, "load_seconds": round(load_seconds, 2), "requests": len(rows), "successful_requests": len(successful), "reliability": round(len(successful) / max(1, len(rows)), 3), "ttft_ms": { "min": round(min(ttfts), 2) if ttfts else None, "p50": percentile(ttfts, 0.50), "p95": percentile(ttfts, 0.95), "max": round(max(ttfts), 2) if ttfts else None, }, "tokens_per_second": { "mean": round(statistics.fmean(rates), 2) if rates else None, "p50": percentile(rates, 0.50), }, "quality_score": score, "quality_checks": checks, "measurements": [asdict(item) for item in rows], } def normalized(values: dict[str, float], lower_is_better: bool = False) -> dict[str, float]: if not values: return {} minimum, maximum = min(values.values()), max(values.values()) if math.isclose(minimum, maximum): return {key: 1.0 for key in values} result = {key: (value - minimum) / (maximum - minimum) for key, value in values.items()} if lower_is_better: result = {key: 1.0 - value for key, value in result.items()} return result def select_winner(results: list[dict[str, Any]]) -> tuple[dict[str, Any], dict[str, float]]: viable = [row for row in results if row.get("successful_requests", 0) >= 15] if not viable: raise RuntimeError("Kein Kandidat hat mindestens 15 von 20 Requests erfolgreich abgeschlossen.") p50 = normalized({row["served_name"]: row["ttft_ms"]["p50"] for row in viable}, True) p95 = normalized({row["served_name"]: row["ttft_ms"]["p95"] for row in viable}, True) speed = normalized({row["served_name"]: row["tokens_per_second"]["mean"] for row in viable}) scores = { row["served_name"]: round( 0.35 * p50[row["served_name"]] + 0.15 * p95[row["served_name"]] + 0.15 * speed[row["served_name"]] + 0.25 * row["quality_score"] + 0.10 * row["reliability"], 4, ) for row in viable } name = max(scores, key=scores.get) return next(row for row in viable if row["served_name"] == name), scores def update_env(entries: dict[str, str]) -> None: lines = ENV_PATH.read_text(encoding="utf-8").splitlines() if ENV_PATH.exists() else [] pending = dict(entries) output: list[str] = [] for line in lines: if "=" not in line or line.lstrip().startswith("#"): output.append(line) continue key = line.split("=", 1)[0].strip() if key in pending: output.append(f"{key}={pending.pop(key)}") else: output.append(line) output.extend(f"{key}={value}" for key, value in pending.items()) ENV_PATH.write_text("\n".join(output).rstrip() + "\n", encoding="utf-8") try: ENV_PATH.chmod(0o600) except OSError: pass def benchmark_all(keep_winner: bool) -> None: env = load_env() client = daytona_client(env) sandbox = ensure_one_sandbox(client) state = refresh_state(client, sandbox) models = json.loads(MODELS_PATH.read_text(encoding="utf-8")) report: dict[str, Any] = { "started_at": utc_now(), "engine": "SGLang", "engine_image": SGLANG_IMAGE, "target": TARGET, "gpu": "RTX_5090", "warmups_per_model": 3, "measured_requests_per_model": 20, "models": [], } try: for model in models: try: _, load_seconds = launch_model(sandbox, state, model) for warmup in range(3): parse_sse_measurement(state, model, TEST_CASES[warmup % len(TEST_CASES)], -1) rows: list[StreamMeasurement] = [] for repetition in range(4): for case in TEST_CASES: rows.append(parse_sse_measurement(state, model, case, repetition)) summary = summarize_model(model, load_seconds, rows, state["gpu"]) report["models"].append(summary) print( f"{model['id']}: TTFT p50={summary['ttft_ms']['p50']} ms, " f"p95={summary['ttft_ms']['p95']} ms, Erfolg={summary['successful_requests']}/20", flush=True, ) except Exception as error: report["models"].append( { "model_id": model["id"], "served_name": model["served_name"], "license": model["license"], "gpu": state["gpu"], "error": f"{type(error).__name__}: {error}", "successful_requests": 0, } ) print(f"{model['id']} ist fehlgeschlagen; Fehler wird im Report festgehalten.", flush=True) winner, scores = select_winner(report["models"]) report["winner"] = winner["served_name"] report["weighted_scores"] = scores report["finished_at"] = utc_now() timestamp = datetime.now().strftime("%Y%m%d-%H%M%S") report_path = REPORT_DIR / f"voice-llm-{timestamp}.json" write_private_json(report_path, report) print(f"Messreport gespeichert: {report_path.relative_to(ROOT)}", flush=True) if keep_winner: winner_model = next(model for model in models if model["served_name"] == winner["served_name"]) if state.get("active_model") != winner_model["served_name"]: launch_model(sandbox, state, winner_model) update_env( { "NELLOZ_LLM_BASE_URL": f"{state['endpoint']}/v1", "NELLOZ_LLM_API_KEY": state["service_key"], "NELLOZ_LLM_PREVIEW_TOKEN": state["preview_token"], "NELLOZ_LLM_MODEL": winner_model["served_name"], "NELLOZ_LLM_ENGINE": "sglang", } ) print("Gewinnerdienst läuft weiter; die lokale .env wurde ohne Schlüsselausgabe aktualisiert.", flush=True) else: cleanup(delete_all=True) except Exception: cleanup(delete_all=True) raise def status() -> None: env = load_env() client = daytona_client(env) existing = safe_existing_sandboxes(client) payload = { "count": len(existing), "sandbox_ids": [sandbox_id(item) for item in existing], "local_state": STATE_PATH.exists(), } print(json.dumps(payload, indent=2)) def cleanup(delete_all: bool = False) -> None: try: env = load_env() client = daytona_client(env) except Exception: if delete_all: STATE_PATH.unlink(missing_ok=True) return targets = safe_existing_sandboxes(client) if STATE_PATH.exists(): state = read_state() state_id = str(state.get("sandbox_id") or "") targets = [item for item in targets if sandbox_id(item) == state_id] or targets for sandbox in targets: print(f"Entferne Nelloz-Benchmark-Sandbox {sandbox_id(sandbox)} …", flush=True) sandbox.delete(timeout=120, wait=True) if delete_all or not safe_existing_sandboxes(client): STATE_PATH.unlink(missing_ok=True) def main() -> int: parser = argparse.ArgumentParser() sub = parser.add_subparsers(dest="command", required=True) bench = sub.add_parser("benchmark-all") bench.add_argument("--keep-winner", action="store_true") sub.add_parser("status") sub.add_parser("cleanup") args = parser.parse_args() try: if args.command == "benchmark-all": benchmark_all(args.keep_winner) elif args.command == "status": status() elif args.command == "cleanup": cleanup(delete_all=True) return 0 except Exception as error: print(f"Fehler: {type(error).__name__}: {error}", file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main())