Trend-Radar + Pruefstand: Box beobachtet Modell-/Engine-Trends und testet Kandidaten selbst
- trend-radar-feed.sh (Cron Sa 05:15): Live-Puls ALLER Rollen mit Vorwochen-Vergleich, Engine-A/B gegen neuen llama.cpp-Build (Fruehwarnung vor So-Auto-Update), mechanische Watch-Liste (MMQ-Issue, ROCm-Releases, Community-Grid), HF-Kandidaten-Suche mit Zitat-Gate -> max. 1 Pruefstand-Kandidat/Woche - pruefstand.sh + harness.py (Cron So 01:00, no-agent): volles Programm je Kandidat - 6 Coding-Aufgaben mit echten Unit-Tests, 4 Agenten-Aufgaben (Tool-Loop + Endzustand), 4 Recherche-Fragen mit Zitat-Ehrlichkeits-Gate, Deutsch-Richter, Vision-Testbild, Tempo kurz+lang; Baseline der Amtsinhaber als Vergleichsmassstab - Leitplanken (User-Entscheid 17.07.): Download-Deckel 35 GB (drueber -> Karte), 1 Kandidat/Woche, Cache mit Auto-Aufraeumen, RAM-/Platz-Wache, Bench-Port :5899, Radar/Pruefstand EMPFEHLEN nur - Rollen-Wechsel bleibt Commander-Klick - E2E bewiesen (Mock-LLM): coding 6/6, recherche 4/4 inkl. Zitat-Gate, Agent-Roundtrip, Richter-Parsing; Suiten-Selbsttest mit Referenzloesungen 6/6 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Hitonabi
@@ -0,0 +1,571 @@
|
||||
#!/usr/bin/env python3
|
||||
# Prüfstand-Harness (17.07.2026): fährt ECHTE Prüfungen gegen ein OpenAI-kompatibles
|
||||
# Endpoint (llama-swap :8080 für Amtsinhaber, Bench-Port :5899 für Kandidaten) und
|
||||
# bewertet MECHANISCH — bestanden/durchgefallen statt Bauchgefühl.
|
||||
#
|
||||
# Suiten (deploy/pruefstand/suites/):
|
||||
# coding Aufgabe → Modell schreibt Code → echte Unit-Tests laufen im Sandkasten
|
||||
# agent Mehrschritt-Aufgabe mit Werkzeugen (Dateien lesen/schreiben/löschen im
|
||||
# Sandkasten, Tool-Calling-API) → bewertet am ENDZUSTAND + Tool-Fehlerquote
|
||||
# recherche Fragen gegen mitgelieferte Dokumente → Pflicht-Stichworte + wörtliches
|
||||
# Zitat, das mechanisch gegen die Dokumente geprüft wird (Ehrlichkeits-Gate)
|
||||
# deutsch Lucy-Alltags-Turns → Richter-Modell (Zwei-Kritiker-Muster) vergibt Noten
|
||||
# vision generiertes Testbild beschreiben (nur für Vision-Rollen)
|
||||
# speed kurze + lange Probe, pp/tg aus den llama.cpp-Timings
|
||||
#
|
||||
# Nutzung:
|
||||
# python3 harness.py --endpoint http://127.0.0.1:8080/v1 --model hermes \
|
||||
# --suites coding,agent --out ergebnis.json [--judge-model gpt-oss-120b] \
|
||||
# [--deadline-min 90] [--vision-image pfad.png]
|
||||
#
|
||||
# Lehren eingebaut: Reasoning-Modelle brauchen großzügiges max_tokens (Verdikt 14.07.:
|
||||
# knappes Budget = leerer content, nie "kaputt"); Timings stehen im "timings"-Feld der
|
||||
# llama.cpp-Antwort; jeder API-Fehler = Aufgabe durchgefallen mit Grund, nie Abbruch.
|
||||
import argparse
|
||||
import base64
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
import unicodedata
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
|
||||
HIER = os.path.dirname(os.path.abspath(__file__))
|
||||
SUITES_DIR = os.path.join(HIER, "suites")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- HTTP-Helfer
|
||||
def api_chat(endpoint, model, messages, max_tokens=1800, tools=None, timeout=420,
|
||||
temperature=0.1):
|
||||
"""Ein Chat-Call. Gibt (message-dict, timings-dict, fehler-string) zurück."""
|
||||
body = {"model": model, "messages": messages, "max_tokens": max_tokens,
|
||||
"temperature": temperature}
|
||||
if tools:
|
||||
body["tools"] = tools
|
||||
body["tool_choice"] = "auto"
|
||||
req = urllib.request.Request(
|
||||
endpoint.rstrip("/") + "/chat/completions",
|
||||
data=json.dumps(body).encode("utf-8"),
|
||||
headers={"Content-Type": "application/json"})
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=timeout) as r:
|
||||
d = json.loads(r.read().decode("utf-8", "replace"))
|
||||
msg = d["choices"][0]["message"]
|
||||
return msg, d.get("timings") or {}, ""
|
||||
except Exception as e: # Timeout, HTTP-Fehler, kaputtes JSON — alles ehrlich melden
|
||||
return None, {}, f"{type(e).__name__}: {e}"
|
||||
|
||||
|
||||
def inhalt_von(msg):
|
||||
"""content zuerst, reasoning_content als Notnagel (Reasoning-Modelle)."""
|
||||
if not msg:
|
||||
return ""
|
||||
return (msg.get("content") or msg.get("reasoning_content") or "").strip()
|
||||
|
||||
|
||||
def norm(s):
|
||||
"""Normalisierung fürs Zitat-Gate (identisch zur idle-radar-Mechanik)."""
|
||||
s = unicodedata.normalize("NFKC", s or "")
|
||||
s = "".join(ch for ch in s if ch.isalnum() or ch.isspace())
|
||||
return " ".join(s.lower().split())
|
||||
|
||||
|
||||
def code_block(text, sprache="python"):
|
||||
"""Letzten passenden Code-Block extrahieren; ohne Zaun: ganzen Text nehmen."""
|
||||
bloecke = re.findall(r"```(?:%s)?\s*\n(.*?)```" % re.escape(sprache),
|
||||
text or "", re.DOTALL | re.IGNORECASE)
|
||||
if bloecke:
|
||||
return bloecke[-1]
|
||||
bloecke = re.findall(r"```\s*\n(.*?)```", text or "", re.DOTALL)
|
||||
return bloecke[-1] if bloecke else (text or "")
|
||||
|
||||
|
||||
def lade_task(pfad):
|
||||
with open(pfad, encoding="utf-8") as f:
|
||||
return json.load(f)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- Suite: coding
|
||||
def suite_coding(cfg, ergebnisse):
|
||||
basis = os.path.join(SUITES_DIR, "coding", "aufgaben")
|
||||
for tid in sorted(os.listdir(basis)):
|
||||
tdir = os.path.join(basis, tid)
|
||||
if not os.path.isfile(os.path.join(tdir, "task.json")):
|
||||
continue
|
||||
if cfg.abgelaufen():
|
||||
ergebnisse.append({"suite": "coding", "id": tid, "pass": False,
|
||||
"grund": "Zeitbudget erschöpft — übersprungen"})
|
||||
continue
|
||||
task = lade_task(os.path.join(tdir, "task.json"))
|
||||
t0 = time.time()
|
||||
prompt = task["prompt"]
|
||||
for name, inhalt in (task.get("dateien") or {}).items():
|
||||
prompt += f"\n\n--- Datei `{name}` (Ist-Stand) ---\n```python\n{inhalt}\n```"
|
||||
prompt += ("\n\nAntworte mit GENAU EINEM ```python-Code-Block, der den "
|
||||
f"vollständigen Inhalt der Datei `{task['antwort_datei']}` enthält. "
|
||||
"Keine Erklärungen außerhalb des Blocks.")
|
||||
msg, tim, err = api_chat(cfg.endpoint, cfg.model,
|
||||
[{"role": "user", "content": prompt}],
|
||||
max_tokens=task.get("max_tokens", 2600))
|
||||
eintrag = {"suite": "coding", "id": tid, "dauer_s": round(time.time() - t0, 1),
|
||||
"tg_tps": tim.get("predicted_per_second"),
|
||||
"pp_tps": tim.get("prompt_per_second")}
|
||||
if err:
|
||||
eintrag.update({"pass": False, "grund": f"API-Fehler: {err}"})
|
||||
ergebnisse.append(eintrag)
|
||||
continue
|
||||
code = code_block(inhalt_von(msg))
|
||||
sbx = tempfile.mkdtemp(prefix="pruefstand-coding-")
|
||||
try:
|
||||
for name, inhalt in (task.get("dateien") or {}).items():
|
||||
with open(os.path.join(sbx, name), "w", encoding="utf-8") as f:
|
||||
f.write(inhalt)
|
||||
with open(os.path.join(sbx, task["antwort_datei"]), "w",
|
||||
encoding="utf-8") as f:
|
||||
f.write(code)
|
||||
shutil.copy(os.path.join(tdir, "verify.py"), os.path.join(sbx, "verify.py"))
|
||||
p = subprocess.run([sys.executable, "verify.py"], cwd=sbx,
|
||||
capture_output=True, text=True,
|
||||
timeout=task.get("timeout_s", 120))
|
||||
eintrag["pass"] = (p.returncode == 0)
|
||||
if p.returncode != 0:
|
||||
eintrag["grund"] = (p.stdout + p.stderr).strip()[-400:]
|
||||
except subprocess.TimeoutExpired:
|
||||
eintrag.update({"pass": False, "grund": "verify-Timeout (Endlosschleife?)"})
|
||||
except Exception as e:
|
||||
eintrag.update({"pass": False, "grund": f"Sandkasten-Fehler: {e}"})
|
||||
finally:
|
||||
shutil.rmtree(sbx, ignore_errors=True)
|
||||
ergebnisse.append(eintrag)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- Suite: agent
|
||||
WERKZEUGE = [
|
||||
{"type": "function", "function": {
|
||||
"name": "liste_dateien",
|
||||
"description": "Listet alle Dateien im Arbeitsverzeichnis (rekursiv).",
|
||||
"parameters": {"type": "object", "properties": {}}}},
|
||||
{"type": "function", "function": {
|
||||
"name": "lies_datei",
|
||||
"description": "Liest eine Datei aus dem Arbeitsverzeichnis.",
|
||||
"parameters": {"type": "object", "properties": {
|
||||
"pfad": {"type": "string"}}, "required": ["pfad"]}}},
|
||||
{"type": "function", "function": {
|
||||
"name": "schreibe_datei",
|
||||
"description": "Schreibt/überschreibt eine Datei im Arbeitsverzeichnis.",
|
||||
"parameters": {"type": "object", "properties": {
|
||||
"pfad": {"type": "string"}, "inhalt": {"type": "string"}},
|
||||
"required": ["pfad", "inhalt"]}}},
|
||||
{"type": "function", "function": {
|
||||
"name": "loesche_datei",
|
||||
"description": "Löscht eine Datei im Arbeitsverzeichnis.",
|
||||
"parameters": {"type": "object", "properties": {
|
||||
"pfad": {"type": "string"}}, "required": ["pfad"]}}},
|
||||
{"type": "function", "function": {
|
||||
"name": "fertig",
|
||||
"description": "Aufgabe abgeschlossen. Kurze Abschluss-Meldung übergeben.",
|
||||
"parameters": {"type": "object", "properties": {
|
||||
"meldung": {"type": "string"}}, "required": ["meldung"]}}},
|
||||
]
|
||||
|
||||
|
||||
def sicherer_pfad(sbx, pfad):
|
||||
"""Pfad-Ausbruch verhindern — Sandkasten bleibt Sandkasten."""
|
||||
ziel = os.path.realpath(os.path.join(sbx, pfad or ""))
|
||||
if not ziel.startswith(os.path.realpath(sbx) + os.sep):
|
||||
raise ValueError(f"Pfad verlässt den Sandkasten: {pfad}")
|
||||
return ziel
|
||||
|
||||
|
||||
def werkzeug_ausfuehren(sbx, name, args):
|
||||
if name == "liste_dateien":
|
||||
alle = []
|
||||
for wurzel, _, dateien in os.walk(sbx):
|
||||
for d in dateien:
|
||||
alle.append(os.path.relpath(os.path.join(wurzel, d), sbx))
|
||||
return "\n".join(sorted(alle)) or "(leer)"
|
||||
if name == "lies_datei":
|
||||
with open(sicherer_pfad(sbx, args["pfad"]), encoding="utf-8",
|
||||
errors="replace") as f:
|
||||
return f.read()[:20000]
|
||||
if name == "schreibe_datei":
|
||||
ziel = sicherer_pfad(sbx, args["pfad"])
|
||||
os.makedirs(os.path.dirname(ziel), exist_ok=True)
|
||||
with open(ziel, "w", encoding="utf-8") as f:
|
||||
f.write(args.get("inhalt", ""))
|
||||
return f"geschrieben: {args['pfad']}"
|
||||
if name == "loesche_datei":
|
||||
os.remove(sicherer_pfad(sbx, args["pfad"]))
|
||||
return f"gelöscht: {args['pfad']}"
|
||||
raise ValueError(f"unbekanntes Werkzeug: {name}")
|
||||
|
||||
|
||||
def suite_agent(cfg, ergebnisse):
|
||||
basis = os.path.join(SUITES_DIR, "agent", "aufgaben")
|
||||
for tid in sorted(os.listdir(basis)):
|
||||
tdir = os.path.join(basis, tid)
|
||||
if not os.path.isfile(os.path.join(tdir, "task.json")):
|
||||
continue
|
||||
if cfg.abgelaufen():
|
||||
ergebnisse.append({"suite": "agent", "id": tid, "pass": False,
|
||||
"grund": "Zeitbudget erschöpft — übersprungen"})
|
||||
continue
|
||||
task = lade_task(os.path.join(tdir, "task.json"))
|
||||
t0 = time.time()
|
||||
sbx = tempfile.mkdtemp(prefix="pruefstand-agent-")
|
||||
tool_fehler = 0
|
||||
schritte = 0
|
||||
try:
|
||||
for name, inhalt in (task.get("sandkasten") or {}).items():
|
||||
ziel = os.path.join(sbx, name)
|
||||
os.makedirs(os.path.dirname(ziel), exist_ok=True)
|
||||
with open(ziel, "w", encoding="utf-8") as f:
|
||||
f.write(inhalt)
|
||||
messages = [
|
||||
{"role": "system", "content":
|
||||
"Du bist ein Arbeits-Agent mit Datei-Werkzeugen in einem "
|
||||
"Sandkasten-Verzeichnis. Erledige die Aufgabe Schritt für Schritt "
|
||||
"mit den Werkzeugen und rufe am Ende `fertig` auf."},
|
||||
{"role": "user", "content": task["prompt"]},
|
||||
]
|
||||
fertig_gemeldet = False
|
||||
for _ in range(task.get("max_schritte", 8)):
|
||||
schritte += 1
|
||||
msg, _tim, err = api_chat(cfg.endpoint, cfg.model, messages,
|
||||
max_tokens=1800, tools=WERKZEUGE)
|
||||
if err:
|
||||
break
|
||||
calls = msg.get("tool_calls") or []
|
||||
if not calls:
|
||||
break # Agent redet statt zu handeln — Endzustand zählt trotzdem
|
||||
messages.append({"role": "assistant",
|
||||
"content": msg.get("content") or "",
|
||||
"tool_calls": calls})
|
||||
for call in calls:
|
||||
fn = (call.get("function") or {}).get("name") or "?"
|
||||
try:
|
||||
args = json.loads((call.get("function") or {})
|
||||
.get("arguments") or "{}")
|
||||
except Exception:
|
||||
args, fn_ok = {}, False
|
||||
tool_fehler += 1
|
||||
antwort = "FEHLER: Argumente waren kein gültiges JSON."
|
||||
else:
|
||||
fn_ok = True
|
||||
if fn_ok:
|
||||
if fn == "fertig":
|
||||
fertig_gemeldet = True
|
||||
antwort = "ok"
|
||||
else:
|
||||
try:
|
||||
antwort = werkzeug_ausfuehren(sbx, fn, args)
|
||||
except Exception as e:
|
||||
tool_fehler += 1
|
||||
antwort = f"FEHLER: {e}"
|
||||
messages.append({"role": "tool",
|
||||
"tool_call_id": call.get("id") or "0",
|
||||
"content": antwort})
|
||||
if fertig_gemeldet:
|
||||
break
|
||||
# -------- Endzustand prüfen (die einzige Wahrheit)
|
||||
fehlschlaege = []
|
||||
for check in task.get("checks", []):
|
||||
if "nicht_vorhanden" in check:
|
||||
if os.path.exists(os.path.join(sbx, check["nicht_vorhanden"])):
|
||||
fehlschlaege.append(
|
||||
f"{check['nicht_vorhanden']} existiert noch")
|
||||
continue
|
||||
pfad = os.path.join(sbx, check["datei"])
|
||||
if not os.path.isfile(pfad):
|
||||
fehlschlaege.append(f"{check['datei']} fehlt")
|
||||
continue
|
||||
with open(pfad, encoding="utf-8", errors="replace") as f:
|
||||
text = f.read()
|
||||
if check.get("regex") and not re.search(check["regex"], text,
|
||||
re.IGNORECASE | re.MULTILINE):
|
||||
fehlschlaege.append(
|
||||
f"{check['datei']}: Muster fehlt ({check['regex']})")
|
||||
ergebnisse.append({
|
||||
"suite": "agent", "id": tid, "pass": not fehlschlaege,
|
||||
"grund": "; ".join(fehlschlaege)[:400] if fehlschlaege else "",
|
||||
"schritte": schritte, "tool_fehler": tool_fehler,
|
||||
"fertig_gemeldet": fertig_gemeldet,
|
||||
"dauer_s": round(time.time() - t0, 1)})
|
||||
except Exception as e:
|
||||
ergebnisse.append({"suite": "agent", "id": tid, "pass": False,
|
||||
"grund": f"Harness-Fehler: {e}",
|
||||
"dauer_s": round(time.time() - t0, 1)})
|
||||
finally:
|
||||
shutil.rmtree(sbx, ignore_errors=True)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- Suite: recherche
|
||||
def suite_recherche(cfg, ergebnisse):
|
||||
basis = os.path.join(SUITES_DIR, "recherche")
|
||||
dok_dir = os.path.join(basis, "dokumente")
|
||||
dokumente = {}
|
||||
for d in sorted(os.listdir(dok_dir)):
|
||||
with open(os.path.join(dok_dir, d), encoding="utf-8") as f:
|
||||
dokumente[d] = f.read()
|
||||
alle_norm = norm("\n".join(dokumente.values()))
|
||||
material = "\n\n".join(f"===== DOKUMENT {n} =====\n{t}"
|
||||
for n, t in dokumente.items())
|
||||
for tid in sorted(os.listdir(os.path.join(basis, "aufgaben"))):
|
||||
tpfad = os.path.join(basis, "aufgaben", tid)
|
||||
if not tpfad.endswith(".json"):
|
||||
continue
|
||||
if cfg.abgelaufen():
|
||||
ergebnisse.append({"suite": "recherche", "id": tid, "pass": False,
|
||||
"grund": "Zeitbudget erschöpft — übersprungen"})
|
||||
continue
|
||||
task = lade_task(tpfad)
|
||||
t0 = time.time()
|
||||
prompt = (f"{material}\n\nFRAGE: {task['frage']}\n\n"
|
||||
"Beantworte die Frage NUR aus den Dokumenten oben, auf Deutsch, "
|
||||
"in höchstens 4 Sätzen. Letzte Zeile deiner Antwort: "
|
||||
'ZITAT: "<eine wörtlich aus einem Dokument übernommene Zeile, '
|
||||
'die deine Antwort belegt>"')
|
||||
msg, tim, err = api_chat(cfg.endpoint, cfg.model,
|
||||
[{"role": "user", "content": prompt}],
|
||||
max_tokens=1200)
|
||||
eintrag = {"suite": "recherche", "id": task.get("id", tid),
|
||||
"dauer_s": round(time.time() - t0, 1),
|
||||
"tg_tps": tim.get("predicted_per_second"),
|
||||
"pp_tps": tim.get("prompt_per_second")}
|
||||
if err:
|
||||
eintrag.update({"pass": False, "grund": f"API-Fehler: {err}"})
|
||||
ergebnisse.append(eintrag)
|
||||
continue
|
||||
antwort = inhalt_von(msg)
|
||||
fehlschlaege = []
|
||||
for gruppe in task.get("muss_gruppen", []):
|
||||
if not any(k.lower() in antwort.lower() for k in gruppe):
|
||||
fehlschlaege.append(f"Pflicht-Stichwort fehlt: {'/'.join(gruppe)}")
|
||||
m = re.search(r'ZITAT:\s*[„"]?(.+?)["“”]?\s*$', antwort,
|
||||
re.MULTILINE | re.DOTALL)
|
||||
zitat = (m.group(1).strip() if m else "")
|
||||
if task.get("zitat_pflicht", True):
|
||||
if not zitat:
|
||||
fehlschlaege.append("kein ZITAT geliefert")
|
||||
elif norm(zitat) not in alle_norm:
|
||||
fehlschlaege.append("ZITAT nicht wörtlich in den Dokumenten "
|
||||
"(Ehrlichkeits-Gate)")
|
||||
eintrag["pass"] = not fehlschlaege
|
||||
if fehlschlaege:
|
||||
eintrag["grund"] = "; ".join(fehlschlaege)[:400]
|
||||
ergebnisse.append(eintrag)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- Suite: deutsch
|
||||
RICHTER_RASTER = (
|
||||
"Du bist ein strenger Deutsch-Richter für eine Sprach-Assistentin (Lucy): "
|
||||
"locker, natürlich, Du-Form, kurze gesprochene Sätze, kein Beamten- oder "
|
||||
"Übersetzungsdeutsch, keine Anglizismen-Ketten, fachlich korrekt. Bewerte die "
|
||||
"ANTWORT unten auf einer Skala 1–10 (10 = klingt wie eine deutsche "
|
||||
"Muttersprachlerin im Alltag). Antworte EXAKT in diesem Format:\n"
|
||||
"NOTE: <Zahl 1-10>\nBEGRUENDUNG: <ein Satz>")
|
||||
|
||||
|
||||
def suite_deutsch(cfg, ergebnisse):
|
||||
aufgaben = lade_task(os.path.join(SUITES_DIR, "deutsch", "aufgaben.json"))
|
||||
for task in aufgaben:
|
||||
if cfg.abgelaufen():
|
||||
ergebnisse.append({"suite": "deutsch", "id": task["id"], "pass": False,
|
||||
"grund": "Zeitbudget erschöpft — übersprungen"})
|
||||
continue
|
||||
t0 = time.time()
|
||||
msg, tim, err = api_chat(cfg.endpoint, cfg.model,
|
||||
[{"role": "user", "content": task["prompt"]}],
|
||||
max_tokens=900)
|
||||
eintrag = {"suite": "deutsch", "id": task["id"],
|
||||
"dauer_s": round(time.time() - t0, 1),
|
||||
"tg_tps": tim.get("predicted_per_second"),
|
||||
"pp_tps": tim.get("prompt_per_second")}
|
||||
if err:
|
||||
eintrag.update({"pass": False, "grund": f"API-Fehler: {err}"})
|
||||
ergebnisse.append(eintrag)
|
||||
continue
|
||||
antwort = inhalt_von(msg)
|
||||
jmsg, _jt, jerr = api_chat(cfg.judge_endpoint, cfg.judge_model, [
|
||||
{"role": "system", "content": RICHTER_RASTER},
|
||||
{"role": "user", "content":
|
||||
f"AUFGABE AN DIE ASSISTENTIN: {task['prompt']}\n\n"
|
||||
f"ANTWORT:\n{antwort}"}],
|
||||
max_tokens=2200, temperature=0.2)
|
||||
note = None
|
||||
if not jerr:
|
||||
m = re.search(r"NOTE:\s*(\d{1,2})", inhalt_von(jmsg))
|
||||
if m:
|
||||
note = max(1, min(10, int(m.group(1))))
|
||||
eintrag["note"] = note
|
||||
eintrag["pass"] = (note is not None and note >= 6)
|
||||
if note is None:
|
||||
eintrag["grund"] = f"Richter ohne Note ({jerr or 'Format'})"
|
||||
ergebnisse.append(eintrag)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- Suite: vision
|
||||
def suite_vision(cfg, ergebnisse):
|
||||
task = lade_task(os.path.join(SUITES_DIR, "vision", "task.json"))
|
||||
bild = cfg.vision_image or os.path.join(SUITES_DIR, "vision", "testbild.png")
|
||||
if not os.path.isfile(bild):
|
||||
# Testbild deterministisch erzeugen (reines Python, kein PIL nötig)
|
||||
subprocess.run([sys.executable,
|
||||
os.path.join(SUITES_DIR, "vision", "gen_testbild.py"), bild],
|
||||
check=False, timeout=30)
|
||||
if not os.path.isfile(bild):
|
||||
ergebnisse.append({"suite": "vision", "id": "testbild", "pass": False,
|
||||
"grund": "Testbild fehlt und konnte nicht erzeugt werden"})
|
||||
return
|
||||
with open(bild, "rb") as f:
|
||||
b64 = base64.b64encode(f.read()).decode("ascii")
|
||||
t0 = time.time()
|
||||
msg, tim, err = api_chat(cfg.endpoint, cfg.model, [
|
||||
{"role": "user", "content": [
|
||||
{"type": "text", "text": task["prompt"]},
|
||||
{"type": "image_url",
|
||||
"image_url": {"url": f"data:image/png;base64,{b64}"}}]}],
|
||||
max_tokens=900, timeout=600)
|
||||
eintrag = {"suite": "vision", "id": "testbild",
|
||||
"dauer_s": round(time.time() - t0, 1),
|
||||
"tg_tps": tim.get("predicted_per_second")}
|
||||
if err:
|
||||
eintrag.update({"pass": False, "grund": f"API-Fehler: {err}"})
|
||||
ergebnisse.append(eintrag)
|
||||
return
|
||||
antwort = inhalt_von(msg)
|
||||
fehlschlaege = [f"nicht erkannt: {'/'.join(g)}"
|
||||
for g in task.get("muss_gruppen", [])
|
||||
if not any(k.lower() in antwort.lower() for k in g)]
|
||||
eintrag["pass"] = not fehlschlaege
|
||||
if fehlschlaege:
|
||||
eintrag["grund"] = "; ".join(fehlschlaege)[:300]
|
||||
eintrag["antwort_auszug"] = antwort[:200]
|
||||
ergebnisse.append(eintrag)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- Suite: speed
|
||||
LANG_BAUSTEIN = ("Die Box protokolliert jeden Dienststart mit Zeitstempel, "
|
||||
"Dienstname und Ergebnis, damit die Morgenlage Abweichungen "
|
||||
"gegen die Vorwoche erkennen und einordnen kann. ")
|
||||
|
||||
|
||||
def suite_speed(cfg, ergebnisse):
|
||||
# Kurz-Probe: misst tg (Alltags-Turn) — 2 Läufe, erster wärmt den Cache an.
|
||||
for lauf in ("warm", "kurz"):
|
||||
t0 = time.time()
|
||||
msg, tim, err = api_chat(cfg.endpoint, cfg.model, [
|
||||
{"role": "user", "content":
|
||||
"Erkläre in drei kurzen Sätzen, was ein Mixture-of-Experts-Modell "
|
||||
"ist."}], max_tokens=200, temperature=0)
|
||||
if lauf == "warm":
|
||||
continue
|
||||
ergebnisse.append({"suite": "speed", "id": "kurz", "pass": not err,
|
||||
"grund": err[:200] if err else "",
|
||||
"dauer_s": round(time.time() - t0, 1),
|
||||
"tg_tps": tim.get("predicted_per_second"),
|
||||
"pp_tps": tim.get("prompt_per_second")})
|
||||
# Lang-Probe: ~12k-Token-Prompt → misst pp bei echtem Kontext (ROCm-Watch-Metrik).
|
||||
if cfg.abgelaufen():
|
||||
ergebnisse.append({"suite": "speed", "id": "lang", "pass": False,
|
||||
"grund": "Zeitbudget erschöpft — übersprungen"})
|
||||
return
|
||||
lang = LANG_BAUSTEIN * 550
|
||||
t0 = time.time()
|
||||
msg, tim, err = api_chat(cfg.endpoint, cfg.model, [
|
||||
{"role": "user", "content":
|
||||
lang + "\n\nWie oft steht sinngemäß derselbe Satz oben? Antworte in "
|
||||
"einem Satz."}], max_tokens=80, temperature=0, timeout=900)
|
||||
ergebnisse.append({"suite": "speed", "id": "lang", "pass": not err,
|
||||
"grund": err[:200] if err else "",
|
||||
"dauer_s": round(time.time() - t0, 1),
|
||||
"tg_tps": tim.get("predicted_per_second"),
|
||||
"pp_tps": tim.get("prompt_per_second"),
|
||||
"prompt_tokens": tim.get("prompt_n")})
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- Hauptlauf
|
||||
SUITE_FUNKS = {"coding": suite_coding, "agent": suite_agent,
|
||||
"recherche": suite_recherche, "deutsch": suite_deutsch,
|
||||
"vision": suite_vision, "speed": suite_speed}
|
||||
|
||||
|
||||
class Konfig:
|
||||
def __init__(self, args):
|
||||
self.endpoint = args.endpoint
|
||||
self.model = args.model
|
||||
self.judge_endpoint = args.judge_endpoint or args.endpoint
|
||||
self.judge_model = args.judge_model
|
||||
self.vision_image = args.vision_image
|
||||
self.deadline = (time.time() + args.deadline_min * 60
|
||||
if args.deadline_min else None)
|
||||
|
||||
def abgelaufen(self):
|
||||
return self.deadline is not None and time.time() > self.deadline
|
||||
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser(description="Prüfstand-Harness")
|
||||
ap.add_argument("--endpoint", required=True)
|
||||
ap.add_argument("--model", required=True)
|
||||
ap.add_argument("--suites", required=True,
|
||||
help="kommagetrennt: coding,agent,recherche,deutsch,vision,speed")
|
||||
ap.add_argument("--out", required=True)
|
||||
ap.add_argument("--judge-endpoint", default="")
|
||||
ap.add_argument("--judge-model", default="gpt-oss-120b")
|
||||
ap.add_argument("--vision-image", default="")
|
||||
ap.add_argument("--deadline-min", type=float, default=0)
|
||||
args = ap.parse_args()
|
||||
|
||||
cfg = Konfig(args)
|
||||
ergebnisse = []
|
||||
for name in [s.strip() for s in args.suites.split(",") if s.strip()]:
|
||||
if name not in SUITE_FUNKS:
|
||||
ergebnisse.append({"suite": name, "id": "-", "pass": False,
|
||||
"grund": "unbekannte Suite"})
|
||||
continue
|
||||
try:
|
||||
SUITE_FUNKS[name](cfg, ergebnisse)
|
||||
except Exception as e: # kaputte Suite darf den Gesamtlauf nicht töten
|
||||
ergebnisse.append({"suite": name, "id": "-", "pass": False,
|
||||
"grund": f"Suite-Absturz: {type(e).__name__}: {e}"})
|
||||
|
||||
# -------- Zusammenfassung je Suite (bestanden/gesamt + Tempo-Mittel)
|
||||
zsm = {}
|
||||
for e in ergebnisse:
|
||||
s = zsm.setdefault(e["suite"], {"bestanden": 0, "gesamt": 0,
|
||||
"tg": [], "pp": [], "noten": []})
|
||||
s["gesamt"] += 1
|
||||
s["bestanden"] += 1 if e.get("pass") else 0
|
||||
if e.get("tg_tps"):
|
||||
s["tg"].append(e["tg_tps"])
|
||||
if e.get("pp_tps"):
|
||||
s["pp"].append(e["pp_tps"])
|
||||
if e.get("note") is not None:
|
||||
s["noten"].append(e["note"])
|
||||
for s in zsm.values():
|
||||
s["tg_mittel"] = round(sum(s["tg"]) / len(s["tg"]), 1) if s["tg"] else None
|
||||
s["pp_mittel"] = round(sum(s["pp"]) / len(s["pp"]), 1) if s["pp"] else None
|
||||
s["note_mittel"] = (round(sum(s["noten"]) / len(s["noten"]), 1)
|
||||
if s["noten"] else None)
|
||||
del s["tg"], s["pp"], s["noten"]
|
||||
|
||||
with open(args.out, "w", encoding="utf-8") as f:
|
||||
json.dump({"modell": args.model, "endpoint": args.endpoint,
|
||||
"erhoben": time.strftime("%Y-%m-%d %H:%M"),
|
||||
"zusammenfassung": zsm, "einzeln": ergebnisse},
|
||||
f, ensure_ascii=False, indent=1)
|
||||
# Kurzfassung auf stdout (für Runner-Logs)
|
||||
for name, s in zsm.items():
|
||||
extra = f" · Note {s['note_mittel']}" if s.get("note_mittel") else ""
|
||||
tempo = (f" · tg {s['tg_mittel']} t/s" if s.get("tg_mittel") else "")
|
||||
print(f"{name}: {s['bestanden']}/{s['gesamt']}{extra}{tempo}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user