Files
cluster-knaster/core/src/knaster/bench.py
T
2026-10-12 02:12:30 +02:00

39 lines
1.6 KiB
Python

"""Benchmarks, deren Ergebnisse in Katalog/Registry zurückfließen (Kalibrierung am Deployment-Tag)."""
from __future__ import annotations
import time
from statistics import median
from typing import Any
from .audio import rtf, wav_duration
from .llm import OpenAIChatClient
DEFAULT_TEXT = (
"Guten Tag. Dies ist ein Test der Sprachausgabe im Cluster. Wir messen, wie lange die Berechnung "
"im Verhältnis zur Länge des erzeugten Audios dauert. Zahlen wie 1.250,75 Euro und Daten wie der "
"3. Oktober 2026 sollen korrekt gesprochen werden."
)
def bench_tts(url: str, model: str, *, api_key: str = "", runs: int = 3, text: str | None = None,
client: OpenAIChatClient | None = None) -> dict[str, Any]:
client = client or OpenAIChatClient(url, api_key, timeout=600)
text = text or DEFAULT_TEXT
results = []
errors = []
for _ in range(runs):
start = time.monotonic()
try:
audio = client.speech(model, text) # type: ignore[attr-defined]
except Exception as exc: # noqa: BLE001
errors.append(str(exc))
continue
elapsed = time.monotonic() - start
duration = wav_duration(audio)
results.append({"seconds": round(elapsed, 2), "audio_seconds": round(duration, 2),
"rtf": round(rtf(elapsed, duration), 3) if duration else None})
rtfs = [r["rtf"] for r in results if r["rtf"] is not None]
return {"model": model, "chars": len(text), "runs": results, "errors": errors,
"rtf_median": round(median(rtfs), 3) if rtfs else None}