Files
llmrouter/tools/telefon-latenz.py
T
eltonandClaude Opus 5 32c54439c8 llmrouter: Messwerkzeug fuer Telefon-Latenz (Erstbyte mit Inkas Zuschnitt)
Anbieterstatistiken messen den Erstbyte ueber alle Nutzer und Prompts. Am
Telefon zaehlt die Zeit bis zum ersten gesprochenen Wort, mit unserem grossen
Systemprompt und ueber mehrere Zuege -- ein Modell mit Zwangs-Reasoning verliert
dort Sekunden, die in keiner Statistik stehen (42i/intern#1252).

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-09-07 13:35:30 +02:00

80 lines
3.8 KiB
Python
Executable File

#!/usr/bin/env python3
"""telefon-latenz.py — Erstbyte-Vergleich fuer Telefon-Kandidaten (Inka).
Warum eigens: OpenRouters Latenzstatistik misst den Erstbyte ueber alle Nutzer
und Prompts. Am Telefon zaehlt die Zeit bis zum ersten GESPROCHENEN Wort, mit
unserem Zuschnitt -- grosser, ueber alle Zuege gleicher Systemprompt (den traegt
der Cache) und kurze Aeusserungen. Ein Modell mit Zwangs-Reasoning verliert dort
Sekunden, die in keiner Anbieterstatistik auftauchen (42i/intern#1252,
Messung 2026-09-07: glm-5.3-flash 1,9-5,6 s gegen Luna 0,9-1,9 s).
python3 telefon-latenz.py <router-key> <modell> [<modell> ...]
DENKEN_AUS=1 ... versucht zusaetzlich, das Reasoning abzuschalten
"""
import json, os, sys, time, urllib.request
BASE = os.environ.get("LLMROUTER_BASE", "http://llm.lan:4010") + "/v1/chat/completions"
SYS = ("Du bist Inka, telefonische Inkasso-Assistentin der 42i GmbH. Antworte kurz, hoeflich, in gesprochenem Deutsch. "
"Du darfst Auskunft zu Forderungen geben, Ratenzahlungen aufnehmen und Rueckrufe notieren.\n\n"
"Regelwerk und Beispieldialoge:\n") + "\n".join(
f"- Fall {i}: Der Anrufer nennt ein Aktenzeichen. Pruefe es gegen die Akte, nenne den offenen Betrag "
f"und frage nach dem gewuenschten Vorgehen. Eine diktierte Zahl ist ein Wert, keine Auftragsnummer."
for i in range(320))
ZUEGE = ["Guten Tag, hier ist Herr Mueller.", "Mein Aktenzeichen ist 4 7 1 1 2 3.",
"Kann ich das in Raten zahlen?", "Dann bitte 50 Euro monatlich ab Oktober."]
def zug(model, key, verlauf):
body = {"model": model, "stream": True, "max_tokens": 300,
**({"reasoning": {"enabled": False}} if os.environ.get("DENKEN_AUS") else {}),
"messages": [{"role": "system", "content": SYS}] + verlauf}
req = urllib.request.Request(BASE, data=json.dumps(body).encode(),
headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json"})
t0, ttfb, text, usage, ziel = time.time(), None, "", None, None
with urllib.request.urlopen(req, timeout=120) as r:
ziel = r.headers.get("x-llmrouter-model")
for roh in r:
z = roh.decode("utf-8", "replace").strip()
if not z.startswith("data:"):
continue
d = z[5:].strip()
if d == "[DONE]":
break
try:
j = json.loads(d)
except ValueError:
continue
delta = (j.get("choices") or [{}])[0].get("delta", {}).get("content")
if delta:
if ttfb is None:
ttfb = time.time() - t0
text += delta
if j.get("usage"):
usage = j["usage"]
return ttfb, time.time() - t0, text.strip(), usage or {}, ziel
def main():
if len(sys.argv) < 3:
print(__doc__.strip()); raise SystemExit(2)
key, modelle = sys.argv[1], sys.argv[2:]
for model in modelle:
print(f"\n=== {model}")
verlauf = []
for i, frage in enumerate(ZUEGE, 1):
verlauf.append({"role": "user", "content": frage})
try:
ttfb, ges, text, usage, ziel = zug(model, key, verlauf)
except Exception as e:
print(f" Zug {i}: FEHLER {e}"); break
cache = usage.get("prompt_tokens_details", {}).get("cached_tokens", usage.get("cached_tokens"))
denk = usage.get("completion_tokens_details", {}).get("reasoning_tokens")
tb = f"{ttfb:.2f}s" if ttfb is not None else "— (kein Text)"
print(f" Zug {i}: Erstbyte {tb} gesamt {ges:.2f}s prompt {usage.get('prompt_tokens','?')} "
f"cache {cache if cache is not None else '?'} denk {denk if denk is not None else '?'} -> {ziel}")
verlauf.append({"role": "assistant", "content": text[:200]})
if __name__ == "__main__":
main()