#!/usr/bin/env python3 """ Prüfstand als Modul (Modell-Radar, 09/2026) — herausgelöst aus pruefstand-hirn.py (17.09.). Warum: Der Prüfstand vom 17.09. lief von Hand, mit festen Pfaden und nur für das Hirn. Das Modell-Radar (backend/services/radar.py) misst Kandidaten nachts selbst — für beide Rollen, gegen das heutige Modell der Rolle und mit einem Urteil, das ohne Menschen trägt. Deshalb nimmt hier jede Messung einen Kandidaten-Pfad (GGUF, optional mmproj und Draft), und jede Anfrage kennt eine Frist (das Nachtfenster endet hart um 02:30). • Server Kandidat standalone auf 127.0.0.1:5899 (llama-server, Flags wie am 17.09.); der Live-Stack bleibt unberührt. • hirn Tempo kurz + 13k-Tiefe (Prefill, Decode, Draft-Akzeptanz), Werkzeug-Aufrufe zwischen ~100 Werkzeugen mit ~12k Kontext, Deutsch- und JSON-Probe. • coder Tempo, Werkzeug-Aufrufe eines Coding-Agenten, 10 kleine Programmieraufgaben, deren Lösung mit Unit-Tests in einem Temp-Ordner mit Zeitlimit läuft. • Bild Nur mit mmproj: ein im Code gezeichnetes PNG (Formen, Farben, Text) als image_url — einmal direkt, einmal mitten in einer Werkzeug-Aufgabe. • Baseline Das heutige Modell der Rolle über llama-swap (:8080). Das Ergebnis wird gecacht und höchstens monatlich neu gemessen. • Urteil bestanden = keine Verschlechterung bei den Pflichtwerten UND mindestens ein echter Vorteil (siehe urteil()). Lehren vom 17.09. (gelten weiter): Denk-Modelle brauchen für Werkzeug-Aufrufe >= 1500 Tokens Budget; /no_think wirkt bei Qwen 3.6 nicht (chat_template_kwargs.enable_thinking=false); der erste Lauf nach dem Laden ist ~40 % langsamer (Warmlauf verwerfen). Neu: Das warme Live-Hirn hatte den 13k-Text im Prompt-Cache (Prefill 0,1 s statt ~13 s) — jede Tiefenmessung beginnt deshalb mit einer Zufallsmarke. Nur Standardbibliothek: läuft mit dem System-Python der Box genauso wie im MC2-venv. """ import atexit import base64 import json import os import re import secrets import signal import socket import struct import subprocess import sys import tempfile import textwrap import time import urllib.error import urllib.request import zlib from pathlib import Path # --- Grundeinstellungen --------------------------------------------------------------- BIN = os.environ.get("MC_PRUEFSTAND_BIN", "/opt/llamacpp-vulkan/llama-server") LIB_DIR = os.environ.get("MC_PRUEFSTAND_LIB", os.path.dirname(BIN)) PORT = int(os.environ.get("MC_PRUEFSTAND_PORT", "5899")) KANDIDAT_URL = f"http://127.0.0.1:{PORT}" SWAP_URL = os.environ.get("MC_LLAMA_SWAP_URL", "http://127.0.0.1:8080").rstrip("/") SERVER_LOG = os.path.join(tempfile.gettempdir(), "pruefstand-server.log") HERMES = os.path.expanduser("~/.local/bin/hermes") CTX_STANDARD = 65536 # Stand der Proben. Ändern sich Aufgaben oder Werkzeuge, ist eine gecachte Baseline nicht mehr # vergleichbar — baseline() misst dann neu. Bei jeder Änderung an den Proben hochzählen. PRUEFSTAND_VERSION = 2 # Flags wie am 17.09. (der Kontext kommt je Kandidat dazu): voller GPU-Offload, Flash-Attention, # kein mmap (--load-mode none, das alte --no-mmap gibt es seit b10936 nicht mehr), ein Slot, # KV-Cache q8_0 wie im Betrieb. BASIS_FLAGS = ["-ngl", "999", "-fa", "on", "--load-mode", "none", "--jinja", "--parallel", "1", "-ctk", "q8_0", "-ctv", "q8_0"] LADEZEIT_MAX_S = int(os.environ.get("MC_PRUEFSTAND_LADEZEIT", "600")) ANFRAGE_MAX_S = 900 WERKZEUG_TOKENS = 1500 # Denk-Modelle brauchen so viel Luft vor dem Werkzeug-Aufruf CODE_TOKENS = 2500 CODE_ZEITLIMIT_S = 10 # Baseline = das heutige Modell der Rolle, angesprochen über seinen llama-swap-Alias. ALIAS = {"hirn": "hermes", "coder": "coder"} # Urteil: Faktor Kandidat/Baseline. *_min = Pflicht (darunter durchgefallen), *_plus = Vorteil. # Beim Hirn zählt Tempo mehr (Lucy antwortet live), beim Coder die Code-Qualität. REGELN = { "hirn": {"decode_min": 0.85, "decode_plus": 1.10, "prefill_min": 0.80, "prefill_plus": 1.10}, "coder": {"decode_min": 0.80, "decode_plus": 1.15, "prefill_min": 0.67, "prefill_plus": 1.25}, } WERKZEUG_BODEN = 0.5 # so viele Werkzeug-Aufgaben muss jeder schaffen, auch wenn heute alle scheitern BILD_MIN = 0.75 # Anteil erkannter Bild-Merkmale # Beide Rollen sollen Bilder verstehen (User-Entscheid 23.09.) — wer es nicht kann, fällt durch. BILD_PFLICHT = os.environ.get("MC_RADAR_BILD_PFLICHT", "1") != "0" class Zeitende(Exception): """Die Frist (Ende des Nachtfensters) ist erreicht — Messung abbrechen.""" class ServerFehler(Exception): """Der Kandidat startet nicht oder stürzt ab (Laden, Port belegt, Absturz im Test).""" def log(*teile) -> None: """Protokollzeile. Eine Konsole ohne UTF-8 darf nie eine Messung abbrechen.""" try: print(*teile, flush=True) except UnicodeEncodeError: print(*(str(t).encode("ascii", "replace").decode() for t in teile), flush=True) # --- HTTP --------------------------------------------------------------------------- def _restzeit(frist: float | None) -> float | None: if frist is None: return None rest = frist - time.time() if rest <= 5: raise Zeitende("Frist erreicht.") return rest def post(url: str, nutzlast: dict, timeout: float = ANFRAGE_MAX_S, frist: float | None = None) -> dict: """JSON-POST; der Timeout endet spätestens mit der Frist.""" rest = _restzeit(frist) if rest is not None: timeout = min(timeout, rest) anfrage = urllib.request.Request(url, data=json.dumps(nutzlast).encode(), headers={"Content-Type": "application/json"}) try: with urllib.request.urlopen(anfrage, timeout=timeout) as antwort: return json.loads(antwort.read()) except (TimeoutError, urllib.error.URLError) as e: if frist is not None and time.time() >= frist - 5: raise Zeitende("Frist während einer Anfrage erreicht.") from e raise def chat(basis: str, modell: str, messages: list, max_tokens: int = 200, tools: list | None = None, think: bool = True, frist: float | None = None) -> dict: p: dict = {"model": modell, "temperature": 0, "max_tokens": max_tokens, "messages": messages} if not think: p["chat_template_kwargs"] = {"enable_thinking": False} if tools: p["tools"] = tools p["tool_choice"] = "auto" t0 = time.time() j = post(f"{basis}/v1/chat/completions", p, frist=frist) j["_wall"] = time.time() - t0 return j def _nachricht(j: dict) -> dict: return ((j.get("choices") or [{}])[0].get("message")) or {} def _text(j: dict) -> str: """Antworttext ohne Denkblock (manche Templates lassen im content stehen).""" text = _nachricht(j).get("content") or "" return re.sub(r".*?", "", text, flags=re.DOTALL).strip() def timings(j: dict) -> dict: t = j.get("timings") or {} acc = (t.get("draft_n_accepted", 0) / t["draft_n"]) if t.get("draft_n") else None return {"pp_n": t.get("prompt_n"), "pp_s": round((t.get("prompt_ms") or 0) / 1000, 1), "pp_tps": round(t.get("prompt_per_second") or 0), "tg_n": t.get("predicted_n"), "tg_tps": round(t.get("predicted_per_second") or 0, 1), "acc": None if acc is None else round(acc, 2)} # --- Tempo --------------------------------------------------------------------------- PARA = ("Notiz {i}: Die Box läuft mit llama-swap auf Port 8080, das Hirn ist ein Mixture-of-Experts-Modell mit drei " "Milliarden aktiven Parametern, der Coder ein dichtes 27B-Modell mit Draft-Beschleunigung. Lucy hört über " "Parakeet, spricht über pocket-tts und meldet sich per Telegram. Der Sonntags-Job aktualisiert Router, Engine " "und Agent und rollt bei Rot zurück. ") def tiefen_prompt(marke: str) -> str: """~13k Tokens. Die Zufallsmarke vorn verhindert, dass ein warmes Modell den Text aus dem Prompt-Cache holt (am 17.09. kam das Live-Hirn so auf 0,1 s Prefill).""" return (f"Messlauf {marke}.\n" + "".join(PARA.format(i=i) for i in range(130)) + "\n\nFrage: Fasse in drei Sätzen zusammen, was in den Notizen steht, und nenne die Nummer der letzten Notiz.") def messe_tempo(basis: str, modell: str, frist: float | None = None) -> dict: chat(basis, modell, [{"role": "user", "content": "Sag kurz hallo."}], 32, frist=frist) # Warmlauf kurz = chat(basis, modell, [{"role": "user", "content": "Erkläre in fünf Sätzen, was ein Mixture-of-Experts-Modell ist."}], 200, frist=frist) tiefe = chat(basis, modell, [{"role": "user", "content": tiefen_prompt(secrets.token_hex(4))}], 160, frist=frist) return {"kurz": timings(kurz), "tiefe": timings(tiefe)} # --- Werkzeuge ----------------------------------------------------------------------- def _werkzeug(name: str, beschreibung: str, **parameter: str) -> dict: return {"type": "function", "function": { "name": name, "description": beschreibung, "parameters": {"type": "object", "properties": {k: {"type": "string", "description": v} for k, v in parameter.items()}, "required": list(parameter)}}} SCREENSHOT = _werkzeug("screenshot", "Nimmt ein Bildschirmfoto vom PC des Nutzers auf und zeigt es dir.") WERKZEUGE_HIRN_ECHT = [ _werkzeug("terminal", "Führt einen Shell-Befehl auf der Box aus und gibt stdout zurück.", command="Der Befehl"), _werkzeug("read_file", "Liest eine Datei und gibt den Inhalt zurück.", path="Absoluter Pfad"), _werkzeug("send_telegram", "Schickt dem Nutzer eine Telegram-Nachricht.", text="Nachrichtentext"), _werkzeug("web_fetch", "Ruft eine Webseite ab und liefert ihren Text.", url="Adresse der Seite"), _werkzeug("erinnerung_anlegen", "Legt eine Erinnerung für den Nutzer an.", text="Worum es geht", zeit="Wann (ISO-Zeit oder natürlich)"), SCREENSHOT, ] # Lucy hat ~100 Werkzeuge. Die Ablenker machen die Auswahl so schwer wie im Alltag. _ABLENKER = { "kalender": ("Kalender", ("lesen", "eintragen", "verschieben", "loeschen")), "notiz": ("Notizen", ("lesen", "anlegen", "suchen", "loeschen")), "mail": ("E-Mails", ("lesen", "suchen", "entwerfen", "archivieren")), "kontakt": ("Kontakte", ("suchen", "anlegen", "aendern")), "einkauf": ("Einkaufsliste", ("lesen", "ergaenzen", "leeren")), "licht": ("Licht", ("einschalten", "ausschalten", "dimmen")), "heizung": ("Heizung", ("lesen", "stellen")), "rollladen": ("Rollläden", ("oeffnen", "schliessen")), "musik": ("Musik", ("abspielen", "pausieren", "suchen")), "jellyfin": ("Jellyfin-Mediathek", ("suchen", "abspielen")), "docker": ("Docker-Container", ("auflisten", "neustarten", "logs")), "gitea": ("Gitea-Repos", ("auflisten", "issue_anlegen", "pr_lesen")), "backup": ("Sicherungen", ("auflisten", "starten", "pruefen")), "wetter": ("Wetter", ("heute", "vorhersage")), "bahn": ("Bahnverbindungen", ("suchen", "verspaetung")), "paket": ("Paketsendungen", ("verfolgen",)), "rezept": ("Rezepte", ("suchen", "speichern")), "wissen": ("Wissensartikel", ("suchen", "zusammenfassen")), "uebersetzung": ("Übersetzungen", ("deutsch_englisch", "englisch_deutsch")), "rechner": ("Taschenrechner", ("rechnen",)), "timer": ("Küchentimer", ("stellen", "stoppen")), "wecker": ("Wecker", ("stellen", "loeschen")), "drucker": ("Drucker", ("drucken", "status")), "sauger": ("Saugroboter", ("starten", "stoppen", "status")), "tuer": ("Haustür", ("status",)), "kamera": ("Kamera an der Haustür", ("bild_holen",)), "strom": ("Stromverbrauch", ("heute", "monat")), "nerdquiz": ("NerdQuiz", ("frage_holen", "punkte")), "projekt": ("Projekte", ("status", "anlegen")), "modell": ("KI-Modelle", ("auflisten", "laden")), "system": ("Box-System", ("temperatur", "arbeitsspeicher", "prozesse")), # nicht „speicher“: klang nach Platte "datei": ("Dateien", ("suchen", "kopieren", "verschieben")), "aufgabe": ("Aufgaben", ("anlegen", "erledigen", "auflisten")), "browser": ("Browser-Tabs", ("auflisten", "schliessen")), "ablage": ("Zwischenablage", ("lesen", "setzen")), "sprache": ("Sprachausgabe", ("sprechen",)), "tagebuch": ("Tagebuch", ("eintragen", "lesen")), "nachrichten": ("Nachrichten", ("schlagzeilen", "suchen")), } WERKZEUGE_HIRN = WERKZEUGE_HIRN_ECHT + [ _werkzeug(f"{objekt}_{verb}", f"{anzeige}: {verb.replace('_', ' ')}.", eingabe="Freitext") for objekt, (anzeige, verben) in _ABLENKER.items() for verb in verben] SYSTEM_HIRN = ("Du bist Lucy, die persönliche Assistentin auf der Box. Du antwortest auf Deutsch, knapp und freundlich. " "Wenn eine Aufgabe ein Werkzeug braucht, rufst du genau das passende auf, statt zu raten.\n\n" "Hintergrundwissen aus deinem Gedächtnis:\n" + "".join(PARA.format(i=i) for i in range(80))) # (Frage, [(Werkzeug, Muster in den Argumenten), …]) — eine Alternative reicht. AUFGABEN_HIRN = [ ("Wie viel Platz ist auf /srv/models noch frei? Nutze das passende Werkzeug.", [("terminal", r"\bd[fu]\b")]), ("Lies die Datei /etc/hostname und sag mir, wie die Box heißt.", [("read_file", r"/etc/hostname")]), ("Schick mir per Telegram die Nachricht: Prüfstand läuft.", [("send_telegram", r"pr(ü|ue)fstand")]), ("Was steht gerade auf https://example.org? Ruf die Seite ab.", [("web_fetch", r"example\.org")]), ("Erinnere mich morgen um 8 Uhr an den Zahnarzttermin.", [("erinnerung_anlegen", r"zahnarzt")]), ("Schau auf meinen Bildschirm: Was ist da gerade offen?", [("screenshot", r"")]), ] WERKZEUGE_CODER = [ _werkzeug("bash", "Führt einen Shell-Befehl im Projektordner aus.", command="Befehl", description="Wozu, kurz"), _werkzeug("read", "Liest eine Datei.", filePath="Pfad"), _werkzeug("write", "Schreibt eine Datei (legt sie an oder überschreibt sie).", filePath="Pfad", content="Inhalt"), _werkzeug("edit", "Ersetzt in einer Datei einen Textabschnitt.", filePath="Pfad", oldString="Alter Text", newString="Neuer Text"), _werkzeug("glob", "Findet Dateien per Muster.", pattern="z. B. src/**/*.py"), _werkzeug("grep", "Durchsucht Dateien nach einem regulären Ausdruck.", pattern="Suchmuster", include="Dateimuster"), _werkzeug("list", "Listet einen Ordner auf.", path="Ordner"), _werkzeug("webfetch", "Lädt eine Webseite.", url="Adresse"), _werkzeug("todowrite", "Schreibt die Aufgabenliste des Agenten.", todos="Aufgaben als Text"), _werkzeug("task", "Startet einen Unter-Agenten für eine Teilaufgabe.", description="Kurzbeschreibung", prompt="Auftrag"), ] SYSTEM_CODER = ("Du bist ein Coding-Agent in OpenCode. Das Projekt liegt in /home/nutzer/projekt (Python, Tests mit " "pytest, Quellcode unter src/). Erledige Aufträge mit den Werkzeugen, ohne nachzufragen. " "In src/config.py steht die Zeile `PORT = 8080`.") AUFGABEN_CODER = [ ("Führe die Tests des Projekts mit pytest aus.", [("bash", r"pytest")]), ("Zeig mir den Inhalt von src/app.py.", [("read", r"src/app\.py")]), ("Ersetze in src/config.py den Port 8080 durch 9090.", [("edit", r"8080[\s\S]*9090"), ("write", r"9090"), ("bash", r"sed[\s\S]*9090")]), ("Finde alle Stellen im Code, an denen TODO steht.", [("grep", r"TODO"), ("bash", r"grep[\s\S]*TODO")]), ("Lege die Datei hallo.py an, mit einer Funktion hallo(), die 'Hallo' zurückgibt.", [("write", r"hallo\.py[\s\S]*def hallo")]), ] def _werkzeug_satz(rolle: str) -> tuple[str, list, list]: if rolle == "hirn": return SYSTEM_HIRN, WERKZEUGE_HIRN, AUFGABEN_HIRN return SYSTEM_CODER, WERKZEUGE_CODER, AUFGABEN_CODER def argumente_text(roh) -> str: """Argumente eines Werkzeug-Aufrufs als durchsuchbarer Text (JSON-Escapes wie \\u00fc aufgelöst).""" werte = roh if not isinstance(roh, dict): try: werte = json.loads(roh or "{}") except (TypeError, ValueError): return str(roh or "") if isinstance(werte, dict): return " ".join(str(v) for v in werte.values()) return str(werte) def trifft(aufrufe: list, erwartet: list) -> bool: for aufruf in aufrufe or []: fn = (aufruf or {}).get("function") or {} text = argumente_text(fn.get("arguments")) if any(fn.get("name") == name and re.search(muster, text, re.IGNORECASE) for name, muster in erwartet): return True return False def _aufrufe_kurz(aufrufe: list) -> list: return [[(a.get("function") or {}).get("name"), argumente_text((a.get("function") or {}).get("arguments"))[:80]] for a in aufrufe or []] def messe_werkzeuge(basis: str, modell: str, rolle: str, frist: float | None = None) -> dict: system, werkzeuge, aufgaben = _werkzeug_satz(rolle) details, ok, kontext = [], 0, None for frage, erwartet in aufgaben: eintrag: dict = {"aufgabe": frage[:60], "ok": False} try: j = chat(basis, modell, [{"role": "system", "content": system}, {"role": "user", "content": frage}], WERKZEUG_TOKENS, werkzeuge, frist=frist) t = j.get("timings") or {} kontext = kontext or ((t.get("prompt_n") or 0) + (t.get("cache_n") or 0)) or None aufrufe = _nachricht(j).get("tool_calls") or [] eintrag["ok"] = trifft(aufrufe, erwartet) eintrag["aufrufe"] = _aufrufe_kurz(aufrufe) if not aufrufe: eintrag["antwort"] = _text(j)[:120] except Zeitende: raise except Exception as e: eintrag["fehler"] = str(e)[:120] ok += eintrag["ok"] details.append(eintrag) return {"ok": ok, "von": len(aufgaben), "kontext_tokens": kontext, "details": details} # --- Deutsch und JSON (Hirn) -------------------------------------------------------------- _DEUTSCHE_WOERTER = re.compile(r"\b(und|die|der|das|ist|nicht|ein|eine|für|mit|wird|sind|auch|oder)\b", re.IGNORECASE) def ist_deutsch(text: str) -> bool: return len(text) >= 60 and len({w.lower() for w in _DEUTSCHE_WOERTER.findall(text)}) >= 3 def ist_json(text: str) -> bool: try: json.loads(text[text.find("{"): text.rfind("}") + 1]) return True except ValueError: return False def messe_sprache(basis: str, modell: str, frist: float | None = None) -> dict: j = chat(basis, modell, [{"role": "user", "content": "Antworte auf Deutsch in genau drei Sätzen: Warum braucht ein Heimserver ein Backup?"}], 300, think=False, frist=frist) deutsch = _text(j) j = chat(basis, modell, [{"role": "user", "content": 'Antworte NUR mit JSON, ohne Erklärung: {"modell": "", "ok": true}'}], 120, think=False, frist=frist) roh = _text(j) return {"deutsch": {"ok": ist_deutsch(deutsch), "text": deutsch[:300]}, "json": {"ok": ist_json(roh), "text": roh[:120]}} # --- Programmieraufgaben (Coder) ------------------------------------------------------------- def _aufgabe(kennung: str, auftrag: str, tests: str) -> dict: return {"id": kennung, "auftrag": auftrag, "tests": textwrap.dedent(tests).strip() + "\n"} AUFGABEN_CODE = [ _aufgabe("palindrom", "Schreibe eine Python-Funktion `ist_palindrom(text: str) -> bool`. Sie gibt True zurück, wenn der Text " "vorwärts wie rückwärts gleich lautet. Groß- und Kleinschreibung, Leerzeichen und Satzzeichen zählen " "nicht; verglichen werden nur Buchstaben (auch Umlaute) und Ziffern. Ein leerer Text ist ein Palindrom.", """ assert ist_palindrom("Ein Esel lese nie") is True assert ist_palindrom("Reliefpfeiler") is True assert ist_palindrom("Otto!") is True assert ist_palindrom("Ölö") is True assert ist_palindrom("12a21") is True assert ist_palindrom("") is True assert ist_palindrom("Hallo Welt") is False """), _aufgabe("roemisch", "Schreibe `roemisch(zahl: int) -> str`, die eine ganze Zahl von 1 bis 3999 als römische Zahl schreibt " "(mit Subtraktion: 4 = IV, 9 = IX, 40 = XL, 90 = XC, 400 = CD, 900 = CM). Zahlen außerhalb des Bereichs " "lösen ValueError aus.", """ assert roemisch(1) == "I" assert roemisch(4) == "IV" assert roemisch(9) == "IX" assert roemisch(58) == "LVIII" assert roemisch(1994) == "MCMXCIV" assert roemisch(3999) == "MMMCMXCIX" for falsch in (0, 4000, -3): try: roemisch(falsch) except ValueError: pass else: raise AssertionError(f"kein ValueError für {falsch}") """), _aufgabe("intervalle", "Schreibe `zusammenfassen(intervalle: list[list[int]]) -> list[list[int]]`. Sie fasst sich überlappende " "oder berührende Intervalle [start, ende] zusammen und gibt sie nach Start sortiert zurück. Die Eingabe " "ist unsortiert und darf nicht verändert werden.", """ assert zusammenfassen([[1, 3], [2, 6], [8, 10], [15, 18]]) == [[1, 6], [8, 10], [15, 18]] assert zusammenfassen([[1, 4], [4, 5]]) == [[1, 5]] assert zusammenfassen([]) == [] assert zusammenfassen([[5, 7], [1, 2]]) == [[1, 2], [5, 7]] assert zusammenfassen([[1, 10], [2, 3], [4, 8]]) == [[1, 10]] eingabe = [[3, 4], [1, 2]] zusammenfassen(eingabe) assert eingabe == [[3, 4], [1, 2]] """), _aufgabe("flach", "Schreibe `flach(liste: list) -> list`, die beliebig tief verschachtelte Listen zu einer flachen Liste " "macht. Die Reihenfolge bleibt erhalten. Nur Listen werden aufgelöst; Tupel und Strings bleiben einzelne " "Elemente.", """ assert flach([1, [2, [3, [4]], 5]]) == [1, 2, 3, 4, 5] assert flach([]) == [] assert flach([[[]]]) == [] assert flach(["ab", ("c", "d"), ["e"]]) == ["ab", ("c", "d"), "e"] tief = [1] for _ in range(50): tief = [tief] assert flach(tief) == [1] """), _aufgabe("iban", "Schreibe `iban_gueltig(iban: str) -> bool`. Leerzeichen werden ignoriert, Groß-/Kleinschreibung ist " "egal. Prüfe die Länge (15 bis 34 Zeichen), dass nur Buchstaben und Ziffern vorkommen, und die Prüfsumme " "nach ISO 13616: die ersten vier Zeichen ans Ende stellen, Buchstaben durch Zahlen ersetzen (A=10 … Z=35), " "die entstehende Zahl modulo 97 muss 1 ergeben.", """ assert iban_gueltig("DE89 3704 0044 0532 0130 00") is True assert iban_gueltig("de89370400440532013000") is True assert iban_gueltig("GB82 WEST 1234 5698 7654 32") is True assert iban_gueltig("DE89 3704 0044 0532 0130 01") is False assert iban_gueltig("DE89-3704-0044") is False assert iban_gueltig("") is False """), _aufgabe("lru", "Schreibe eine Klasse `LRUCache` mit `__init__(self, kapazitaet: int)`, `get(self, schluessel)` (liefert " "den Wert oder None und markiert den Eintrag als zuletzt benutzt) und `put(self, schluessel, wert)` (legt " "an oder überschreibt, ebenfalls als zuletzt benutzt; ist die Kapazität überschritten, fliegt der am " "längsten nicht benutzte Eintrag raus).", """ c = LRUCache(2) c.put("a", 1) c.put("b", 2) assert c.get("a") == 1 c.put("c", 3) assert c.get("b") is None assert c.get("a") == 1 assert c.get("c") == 3 c.put("a", 10) c.put("d", 4) assert c.get("c") is None assert c.get("a") == 10 assert c.get("d") == 4 """), _aufgabe("klammern", "Schreibe `klammern_ok(text: str) -> bool`. Sie prüft, ob die Klammern (), [] und {} im Text korrekt " "geschachtelt und geschlossen sind. Alle anderen Zeichen werden ignoriert.", """ assert klammern_ok("") is True assert klammern_ok("f(a[1], {b: 2})") is True assert klammern_ok("x = {'a': [1, (2, 3)]}") is True assert klammern_ok("([)]") is False assert klammern_ok("((") is False assert klammern_ok("())") is False """), _aufgabe("wege", "Schreibe `kuerzeste_wege(graph: dict[str, dict[str, int]], start: str) -> dict[str, int]`. graph ordnet " "jedem Knoten seine Nachbarn mit nicht negativen Kantengewichten zu. Liefere die kürzeste Entfernung vom " "Start zu jedem erreichbaren Knoten (der Start selbst mit 0). Nicht erreichbare Knoten fehlen im " "Ergebnis. Nachbarn, die nicht als Schlüssel in graph stehen, sind trotzdem gültige Knoten.", """ g = {"A": {"B": 1, "C": 4}, "B": {"C": 2, "D": 5}, "C": {"D": 1}, "D": {}, "X": {"A": 1}} assert kuerzeste_wege(g, "A") == {"A": 0, "B": 1, "C": 3, "D": 4} assert kuerzeste_wege(g, "D") == {"D": 0} assert kuerzeste_wege(g, "X") == {"X": 0, "A": 1, "B": 2, "C": 4, "D": 5} assert kuerzeste_wege({"A": {"Z": 7}}, "A") == {"A": 0, "Z": 7} """), _aufgabe("median", "Diese Funktion soll den Median einer Liste von Zahlen liefern, ist aber fehlerhaft. Korrigiere sie. Die " "Eingabeliste darf nicht verändert werden; bei einer leeren Liste soll ValueError kommen.\n\n" "```python\ndef median(zahlen):\n zahlen.sort()\n mitte = len(zahlen) // 2\n" " return zahlen[mitte]\n```", """ assert median([3, 1, 2]) == 2 assert median([4, 1, 3, 2]) == 2.5 assert median([5]) == 5 daten = [9, 7, 8] median(daten) assert daten == [9, 7, 8] try: median([]) except ValueError: pass else: raise AssertionError("kein ValueError bei leerer Liste") """), _aufgabe("csv", "Schreibe `spaltensumme(csv_text: str, spalte: str) -> float`. Der Text ist eine CSV-Datei mit Semikolon " "als Trenner und einer Kopfzeile. Zahlen haben ein Komma als Dezimaltrenner und können einen Punkt als " "Tausendertrenner haben (z. B. 1.234,5). Leere Zellen zählen als 0. Gibt es die Spalte nicht, löse " "KeyError aus.", """ text = "name;betrag;menge\\nA;1,5;2\\nB;1.234,5;\\nC;;3\\n" assert abs(spaltensumme(text, "betrag") - 1236.0) < 1e-9 assert abs(spaltensumme(text, "menge") - 5.0) < 1e-9 try: spaltensumme(text, "preis") except KeyError: pass else: raise AssertionError("kein KeyError") """), ] SYSTEM_CODE = "Du bist ein erfahrener Python-Entwickler. Antworte nur mit einem Python-Codeblock, ohne Erklärung." # Modell-Code läuft auf der Box — nur reine Funktionen sind erlaubt, kein Zugriff auf Dateien, # Prozesse oder Netz. Die Sperrliste ist grob, aber für diese Aufgaben braucht es nichts davon. _VERBOTEN = re.compile( r"^\s*(?:import|from)\s+(?:os|subprocess|shutil|socket|pathlib|ctypes|multiprocessing|threading|urllib|http|" r"requests|signal|importlib|builtins|pty|resource|glob|tempfile|io|codecs)\b" r"|__import__|(? str: """Längster Python-Codeblock der Antwort (Lösung statt Beispielaufruf); ohne Block die ganze Antwort.""" text = re.sub(r".*?", "", text or "", flags=re.DOTALL) bloecke = re.findall(r"```[ \t]*(?:python|py|python3)?[^\n]*\n(.*?)```", text, re.DOTALL | re.IGNORECASE) return max(bloecke, key=len) if bloecke else text.strip() def fuehre_aufgabe_aus(code: str, tests: str, zeitlimit: float = CODE_ZEITLIMIT_S) -> tuple[bool, str]: """Lösung + Tests in einem frischen Temp-Ordner mit Zeit- und Speichergrenze ausführen.""" if not code.strip(): return False, "keine Lösung in der Antwort" if (treffer := _VERBOTEN.search(code)): return False, f"unerlaubter Aufruf: {treffer.group(0).strip()[:40]}" with tempfile.TemporaryDirectory(prefix="pruefstand-code-") as ordner: Path(ordner, "loesung.py").write_text(code, encoding="utf-8") Path(ordner, "pruefung.py").write_text(_RUNNER_KOPF + tests + "print('OK')\n", encoding="utf-8") umgebung = {"PATH": os.environ.get("PATH", ""), "HOME": ordner, "TMPDIR": ordner, "LANG": "C.UTF-8"} umgebung.update({k: os.environ[k] for k in ("SYSTEMROOT", "WINDIR") if k in os.environ}) try: p = subprocess.run([sys.executable, "-I", "pruefung.py"], cwd=ordner, capture_output=True, encoding="utf-8", errors="replace", timeout=zeitlimit, env=umgebung, stdin=subprocess.DEVNULL) except subprocess.TimeoutExpired: return False, f"Zeitlimit {zeitlimit:.0f} s überschritten" if p.returncode == 0 and "OK" in p.stdout: return True, "" zeilen = (p.stderr or p.stdout or "").strip().splitlines() return False, (zeilen[-1] if zeilen else f"Exit-Code {p.returncode}")[:200] def messe_code(basis: str, modell: str, frist: float | None = None) -> dict: """Denken aus: bei so kleinen Aufgaben misst das die Code-Qualität und hält die Nacht kurz.""" details, ok, t0 = [], 0, time.time() for aufgabe in AUFGABEN_CODE: try: j = chat(basis, modell, [{"role": "system", "content": SYSTEM_CODE}, {"role": "user", "content": aufgabe["auftrag"]}], CODE_TOKENS, think=False, frist=frist) geschafft, grund = fuehre_aufgabe_aus(code_aus_antwort(_text(j)), aufgabe["tests"]) except Zeitende: raise except Exception as e: geschafft, grund = False, f"Anfrage scheiterte: {e}"[:160] ok += geschafft details.append({"aufgabe": aufgabe["id"], "ok": geschafft, "grund": grund}) return {"ok": ok, "von": len(AUFGABEN_CODE), "sekunden": round(time.time() - t0), "details": details} # --- Bild-Probe ------------------------------------------------------------------------ BILD_TEXT = "ZEBRA 4711" BILD_MERKMALE = { "Wort ZEBRA": r"zebra", "Zahl 4711": r"4[\s.]?711", "rot": r"\brot(e|er|es|en)?\b|\bred\b", "blau": r"\bblau|\bblue\b", "grün": r"\bgr(ü|ue)n|\bgreen\b", "Kreis": r"kreis|circle", "Quadrat": r"quadrat|rechteck|viereck|square|rectangle", "Dreieck": r"dreieck|triangle", } _GLYPHEN = { "Z": ("#####", "....#", "...#.", "..#..", ".#...", "#....", "#####"), "E": ("#####", "#....", "#....", "####.", "#....", "#....", "#####"), "B": ("####.", "#...#", "#...#", "####.", "#...#", "#...#", "####."), "R": ("####.", "#...#", "#...#", "####.", "#.#..", "#..#.", "#...#"), "A": (".###.", "#...#", "#...#", "#####", "#...#", "#...#", "#...#"), "4": ("...#.", "..##.", ".#.#.", "#..#.", "#####", "...#.", "...#."), "7": ("#####", "....#", "...#.", "..#..", ".#...", ".#...", ".#..."), "1": ("..#..", ".##..", "..#..", "..#..", "..#..", "..#..", ".###."), " ": (".....",) * 7, } def bild_png(breite: int = 448, hoehe: int = 300) -> bytes: """Testbild ohne Pillow: roter Kreis, blaues Quadrat, grünes Dreieck, darunter „ZEBRA 4711“.""" px = bytearray(b"\xff" * (breite * hoehe * 3)) def setze(x: int, y: int, farbe: bytes) -> None: if 0 <= x < breite and 0 <= y < hoehe: i = (y * breite + x) * 3 px[i:i + 3] = farbe rot, blau, gruen, schwarz = b"\xdc\x14\x14", b"\x14\x3c\xdc", b"\x14\xa0\x28", b"\x10\x10\x10" cx, cy, r = 80, 95, 55 for y in range(cy - r, cy + r + 1): for x in range(cx - r, cx + r + 1): if (x - cx) ** 2 + (y - cy) ** 2 <= r * r: setze(x, y, rot) for y in range(40, 150): for x in range(170, 280): setze(x, y, blau) spitze_x, oben, unten, links, rechts = 370, 40, 150, 315, 425 for y in range(oben, unten + 1): anteil = (y - oben) / (unten - oben) for x in range(round(spitze_x + (links - spitze_x) * anteil), round(spitze_x + (rechts - spitze_x) * anteil) + 1): setze(x, y, gruen) skala, zelle = 6, 6 x0, y0 = (breite - len(BILD_TEXT) * zelle * skala) // 2, 205 for n, zeichen in enumerate(BILD_TEXT): for zy, reihe in enumerate(_GLYPHEN[zeichen]): for zx, punkt in enumerate(reihe): if punkt == "#": for dy in range(skala): for dx in range(skala): setze(x0 + (n * zelle + zx) * skala + dx, y0 + zy * skala + dy, schwarz) roh = b"".join(b"\x00" + bytes(px[y * breite * 3:(y + 1) * breite * 3]) for y in range(hoehe)) def block(typ: bytes, daten: bytes) -> bytes: return struct.pack(">I", len(daten)) + typ + daten + struct.pack(">I", zlib.crc32(typ + daten) & 0xFFFFFFFF) return (b"\x89PNG\r\n\x1a\n" + block(b"IHDR", struct.pack(">IIBBBBB", breite, hoehe, 8, 2, 0, 0, 0)) + block(b"IDAT", zlib.compress(roh, 9)) + block(b"IEND", b"")) def bild_data_url() -> str: return "data:image/png;base64," + base64.b64encode(bild_png()).decode() def bild_merkmale(text: str) -> list[str]: return [name for name, muster in BILD_MERKMALE.items() if re.search(muster, text or "", re.IGNORECASE)] def _bild_im_ablauf(basis: str, modell: str, rolle: str, url: str, frist: float | None) -> dict: """Das Bild kommt mitten in einer Werkzeug-Aufgabe (als Ergebnis von „screenshot“), wie bei Lucy am Bildschirm oder beim Coder mit einem Screenshot der App. Erwartet wird der nächste richtige Schritt.""" if rolle == "hirn": system, werkzeuge = SYSTEM_HIRN, WERKZEUGE_HIRN auftrag = "Schau auf meinen Bildschirm und schick mir per Telegram, welches Wort und welche Zahl dort stehen." erwartet = [("send_telegram", r"zebra[\s\S]*4[\s.]?711|4[\s.]?711[\s\S]*zebra")] else: system, werkzeuge = SYSTEM_CODER, WERKZEUGE_CODER + [SCREENSHOT] auftrag = ("Mach einen Screenshot der laufenden App. Die Zahl, die dort angezeigt wird, soll in src/config.py " "als PORT stehen. Pass die Datei an.") erwartet = [("edit", r"4711"), ("write", r"4711"), ("bash", r"4711")] verlauf = [ {"role": "system", "content": system}, {"role": "user", "content": auftrag}, {"role": "assistant", "content": "", "tool_calls": [ {"id": "call_bild_1", "type": "function", "function": {"name": "screenshot", "arguments": "{}"}}]}, {"role": "tool", "tool_call_id": "call_bild_1", "content": "Bildschirmfoto aufgenommen, es folgt als Bild."}, {"role": "user", "content": [{"type": "image_url", "image_url": {"url": url}}, {"type": "text", "text": "Bildschirmfoto aus dem Werkzeug screenshot."}]}, ] j = chat(basis, modell, verlauf, WERKZEUG_TOKENS, werkzeuge, frist=frist) aufrufe = _nachricht(j).get("tool_calls") or [] return {"ok": trifft(aufrufe, erwartet), "aufrufe": _aufrufe_kurz(aufrufe), "antwort": "" if aufrufe else _text(j)[:160]} def messe_bild(basis: str, modell: str, rolle: str, frist: float | None = None) -> dict: url = bild_data_url() j = chat(basis, modell, [{"role": "user", "content": [ {"type": "image_url", "image_url": {"url": url}}, {"type": "text", "text": "Beschreibe das Bild: Welche Formen in welchen Farben siehst du, und welcher Text " "steht darin?"}]}], 600, think=False, frist=frist) antwort = _text(j) gefunden = bild_merkmale(antwort) try: ablauf = _bild_im_ablauf(basis, modell, rolle, url, frist) except Zeitende: raise except Exception as e: ablauf = {"ok": False, "fehler": str(e)[:160]} return {"direkt": round(len(gefunden) / len(BILD_MERKMALE), 2), "gefunden": gefunden, "agentisch": bool(ablauf.get("ok")), "ablauf": ablauf, "antwort": antwort[:300]} # --- Gesamtprüfung --------------------------------------------------------------------- def pruefe(rolle: str, basis: str, modell: str, *, bild: bool = False, frist: float | None = None, protokoll=log) -> dict: """Alle Proben der Rolle gegen ein laufendes Modell (Kandidat auf :5899 oder Baseline über llama-swap).""" if rolle not in ALIAS: raise ValueError(f"Unbekannte Rolle {rolle!r}") t0 = time.time() werte: dict = {"rolle": rolle, "modell": modell, "zeitpunkt": time.strftime("%Y-%m-%dT%H:%M:%S%z")} werte.update(messe_tempo(basis, modell, frist)) protokoll(f" kurz : {werte['kurz']}") protokoll(f" 13k : {werte['tiefe']}") werte["werkzeuge"] = messe_werkzeuge(basis, modell, rolle, frist) protokoll(f" werkzeuge: {werte['werkzeuge']['ok']}/{werte['werkzeuge']['von']}") if rolle == "hirn": werte.update(messe_sprache(basis, modell, frist)) protokoll(f" deutsch: {'ja' if werte['deutsch']['ok'] else 'nein'} json: {'ja' if werte['json']['ok'] else 'nein'}") else: werte["code"] = messe_code(basis, modell, frist) protokoll(f" code : {werte['code']['ok']}/{werte['code']['von']} in {werte['code']['sekunden']} s") werte["bild"] = None if bild: try: werte["bild"] = messe_bild(basis, modell, rolle, frist) except Zeitende: raise except Exception as e: werte["bild"] = {"direkt": 0.0, "gefunden": [], "agentisch": False, "fehler": str(e)[:160]} protokoll(f" bild : {werte['bild'].get('direkt')} direkt, im Ablauf " f"{'ja' if werte['bild'].get('agentisch') else 'nein'}") werte["dauer_s"] = round(time.time() - t0) return werte # --- Kandidaten-Server ------------------------------------------------------------------- class Server: """Ein laufender Kandidaten-Server (llama-server auf PORT).""" def __init__(self, prozess: subprocess.Popen, befehl: list[str], log_pfad: str): self.prozess, self.befehl, self.log_pfad = prozess, befehl, log_pfad self.ladezeit = 0 _LAUFENDE: list[Server] = [] def draft_flags(pfad: str | None = None, typ: str | None = None, n_max: int | None = None) -> list[str]: """Spekulatives Dekodieren: Typ (draft-dflash, draft-mtp, draft-simple …) plus Entwurfsmodell. Ein MTP-Kopf im Modell selbst braucht nur den Typ.""" if not typ: return [] flags = ["--spec-type", typ] if pfad: flags += ["--spec-draft-model", pfad] if n_max: flags += ["--spec-draft-n-max", str(n_max)] return flags def server_befehl(gguf: str, *, mmproj: str | None = None, ctx: int = CTX_STANDARD, flags=(), binary: str | None = None) -> list[str]: befehl = [binary or BIN, "-m", gguf, "--host", "127.0.0.1", "--port", str(PORT), "-c", str(ctx), *BASIS_FLAGS] if mmproj: befehl += ["--mmproj", mmproj] return befehl + [str(f) for f in flags] def port_belegt(port: int = PORT) -> bool: try: with socket.create_connection(("127.0.0.1", port), timeout=1): return True except OSError: return False def _gesund() -> bool: try: with urllib.request.urlopen(f"{KANDIDAT_URL}/health", timeout=3) as antwort: return antwort.status == 200 and b"ok" in antwort.read() except Exception: return False def log_ende(pfad: str, zeichen: int = 400) -> str: try: return Path(pfad).read_text(encoding="utf-8", errors="replace")[-zeichen:].strip() except OSError: return "" def _signal(prozess: subprocess.Popen, hart: bool) -> None: try: if os.name == "posix": # ganze Prozessgruppe (start_new_session), falls llama-server Kinder hat os.killpg(prozess.pid, signal.SIGKILL if hart else signal.SIGTERM) elif hart: prozess.kill() else: prozess.terminate() except OSError: pass def stoppe_server(server: Server, warte_s: float = 60) -> None: p = server.prozess if p.poll() is None: _signal(p, hart=False) try: p.wait(timeout=warte_s) except subprocess.TimeoutExpired: _signal(p, hart=True) try: p.wait(timeout=15) except subprocess.TimeoutExpired: pass if server in _LAUFENDE: _LAUFENDE.remove(server) def stoppe_alle(warte_s: float = 20) -> None: """Alle Kandidaten-Server dieses Prozesses stoppen (Notbremse, Programmende).""" for server in _LAUFENDE.copy(): # Kopie: stoppe_server nimmt den Server aus der Liste stoppe_server(server, warte_s) atexit.register(stoppe_alle) def starte_server(gguf: str, *, mmproj: str | None = None, ctx: int = CTX_STANDARD, flags=(), frist: float | None = None, ladezeit_max: float = LADEZEIT_MAX_S, protokoll=log) -> Server: """Kandidat standalone starten und warten, bis /health „ok“ meldet.""" if not os.path.exists(gguf): raise ServerFehler(f"Modelldatei fehlt: {gguf}") if mmproj and not os.path.exists(mmproj): raise ServerFehler(f"mmproj fehlt: {mmproj}") if port_belegt(): raise ServerFehler(f"Port {PORT} ist belegt – läuft noch ein anderer Prüfstand?") befehl = server_befehl(gguf, mmproj=mmproj, ctx=ctx, flags=flags) umgebung = dict(os.environ) umgebung["LD_LIBRARY_PATH"] = os.pathsep.join(p for p in (LIB_DIR, umgebung.get("LD_LIBRARY_PATH")) if p) protokoll(" starte: " + " ".join(befehl)) with open(SERVER_LOG, "w", encoding="utf-8") as logdatei: prozess = subprocess.Popen(befehl, stdout=logdatei, stderr=subprocess.STDOUT, stdin=subprocess.DEVNULL, env=umgebung, start_new_session=(os.name == "posix")) server = Server(prozess, befehl, SERVER_LOG) _LAUFENDE.append(server) t0 = time.time() try: while True: if prozess.poll() is not None: raise ServerFehler(f"Absturz beim Laden (Code {prozess.returncode}): {log_ende(SERVER_LOG, 300)}") if _gesund(): server.ladezeit = round(time.time() - t0) protokoll(f" geladen in {server.ladezeit} s") return server if time.time() - t0 > ladezeit_max: raise ServerFehler(f"Nach {ladezeit_max:.0f} s noch nicht geladen: {log_ende(SERVER_LOG, 200)}") if frist is not None and time.time() > frist - 5: raise Zeitende("Frist erreicht, während der Kandidat noch lud.") time.sleep(2) except BaseException: stoppe_server(server) raise def pruefe_kandidat(rolle: str, gguf: str, *, mmproj: str | None = None, flags=(), ctx: int = CTX_STANDARD, basis: dict | None = None, frist: float | None = None, protokoll=log) -> dict: """Ganzer Durchlauf eines Kandidaten: Server starten → messen → stoppen → urteilen. basis = Werte der Baseline (ohne sie gibt es kein Urteil, ergebnis None). Rückgabe: {"werte", "ergebnis", "vergleich", "zusammenfassung"}.""" server = starte_server(gguf, mmproj=mmproj, ctx=ctx, flags=flags, frist=frist, protokoll=protokoll) def absturz() -> str: return f"Während des Tests abgestürzt (Code {server.prozess.returncode}): {log_ende(SERVER_LOG, 300)}" try: werte = pruefe(rolle, KANDIDAT_URL, "kandidat", bild=bool(mmproj), frist=frist, protokoll=protokoll) except Exception as e: if not isinstance(e, Zeitende) and server.prozess.poll() is not None: raise ServerFehler(absturz()) from e raise else: # Einzelproben fangen Fehler ab — ein toter Server darf nicht wie ein schlechtes Modell aussehen. if server.prozess.poll() is not None: raise ServerFehler(absturz()) finally: stoppe_server(server) time.sleep(3) # Speicher freigeben lassen, bevor jemand anderes lädt werte.update(ladezeit_s=server.ladezeit, ctx=ctx, befehl=" ".join(server.befehl)) if not basis: return {"werte": werte, "ergebnis": None, "vergleich": {}, "zusammenfassung": "Gemessen, aber ohne Vergleichswerte des heutigen Modells."} return {"werte": werte, **urteil(rolle, werte, basis)} # --- Baseline ------------------------------------------------------------------------ def _lies_json(pfad: str) -> dict: try: daten = json.loads(Path(pfad).read_text(encoding="utf-8")) return daten if isinstance(daten, dict) else {} except (OSError, ValueError): return {} def _schreibe_json(pfad: str, daten: dict) -> None: try: tmp = Path(pfad).with_suffix(".tmp") tmp.write_text(json.dumps(daten, ensure_ascii=False, indent=1), encoding="utf-8") tmp.replace(pfad) except OSError as e: log(f" Baseline-Cache nicht schreibbar ({pfad}): {e}") def baseline(rolle: str, cache_pfad: str, *, kennung: str, bild: bool = False, frist: float | None = None, max_alter_tage: int = 30, basis_url: str = SWAP_URL, modell: str | None = None, protokoll=log) -> dict | None: """Werte des heutigen Modells der Rolle, höchstens alle max_alter_tage neu gemessen. kennung = welches Modell mit welchen Flags (ändert sie sich, wird neu gemessen). Rückgabe {"kennung", "zeit", "werte"} oder None, wenn es weder Cache noch Messung gibt.""" cache = _lies_json(cache_pfad) eintrag = cache.get(rolle) if isinstance(cache.get(rolle), dict) else None passend = (eintrag if eintrag and eintrag.get("kennung") == kennung and eintrag.get("version") == PRUEFSTAND_VERSION else None) if passend and time.time() - float(passend.get("zeit") or 0) < max_alter_tage * 86400: return passend protokoll(f" Baseline {rolle} wird neu gemessen ({modell or ALIAS[rolle]} über llama-swap)") try: werte = pruefe(rolle, basis_url, modell or ALIAS[rolle], bild=bild, frist=frist, protokoll=protokoll) except Zeitende: raise except Exception as e: protokoll(f" Baseline {rolle} nicht messbar: {e}") return passend # lieber einen Monat alt als gar nicht vergleichen neu = {"kennung": kennung, "version": PRUEFSTAND_VERSION, "zeit": time.time(), "werte": werte} cache[rolle] = neu _schreibe_json(cache_pfad, cache) return neu # --- Urteil -------------------------------------------------------------------------- def _zahl(wert) -> str: if wert is None: return "–" if isinstance(wert, float) and wert % 1: return f"{wert:.1f}".replace(".", ",") return f"{wert:.0f}" def _prozent(faktor: float) -> str: return f"{(faktor - 1) * 100:+.0f} %".replace("-", "−") def _anteil(wert: dict | None) -> float: return (wert.get("ok", 0) / wert["von"]) if wert and wert.get("von") else 0.0 def bild_kann(bild: dict | None) -> bool: return bool(bild) and float(bild.get("direkt") or 0) >= BILD_MIN and bool(bild.get("agentisch")) def _bild_text(bild: dict | None) -> str: if not bild: return "kein mmproj" return (f"{len(bild.get('gefunden') or [])}/{len(BILD_MERKMALE)} Merkmale, im Ablauf " f"{'ja' if bild.get('agentisch') else 'nein'}") def urteil(rolle: str, k: dict, b: dict) -> dict: """Kandidat k gegen Baseline b (jeweils die Werte aus pruefe()). bestanden = keine Pflicht verfehlt UND mindestens ein Vorteil. Pflicht: Tempo nicht deutlich schlechter, Werkzeuge (und beim Coder der Code) nicht schlechter, Deutsch/JSON (Hirn), Bilder.""" b = b.get("werte", b) # auch ein ganzer Baseline-Eintrag ist recht r = REGELN[rolle] zeilen: dict[str, dict] = {} # Messwert → Kandidat/Baseline/Einheit/Urteil, in Anzeige-Reihenfolge maengel, vorteile = [], [] def zeile(name: str, kw, bw, einheit: str, bewertung: str) -> None: zeilen[name] = {"kandidat": kw, "baseline": bw, "einheit": einheit, "urteil": bewertung} kt, bt = k.get("tiefe") or {}, b.get("tiefe") or {} # Decode in t/s (mehr ist besser), Prefill als Zeit in s (weniger ist besser, Faktor = Tempo). kd, bd = kt.get("tg_tps"), bt.get("tg_tps") kp, bp = kt.get("pp_s"), bt.get("pp_s") for name, kw, bw, einheit, faktor, pmin, pplus in ( ("Decode bei 13k", kd, bd, "t/s", (kd / bd) if kd and bd else None, r["decode_min"], r["decode_plus"]), ("Prefill bei 13k", kp, bp, "s", (bp / kp) if kp and bp else None, r["prefill_min"], r["prefill_plus"])): if not kw: maengel.append(f"{name} nicht messbar") zeile(name, kw, bw, einheit, "schlechter") elif faktor is None: zeile(name, kw, bw, einheit, "unbekannt") elif faktor < pmin: maengel.append(f"{name} {_zahl(kw)} statt {_zahl(bw)} {einheit} (Tempo {_prozent(faktor)})") zeile(name, kw, bw, einheit, "schlechter") elif faktor >= pplus: vorteile.append(f"{name} {_prozent(faktor)} schneller".replace("+", "")) zeile(name, kw, bw, einheit, "besser") else: zeile(name, kw, bw, einheit, "gleich") kw_, bw_ = k.get("werkzeuge") or {}, b.get("werkzeuge") or {} ka, ba = _anteil(kw_), _anteil(bw_) ktext, btext = f"{kw_.get('ok', 0)}/{kw_.get('von', 0)}", f"{bw_.get('ok', 0)}/{bw_.get('von', 0)}" if ka + 1e-9 < ba or ka < WERKZEUG_BODEN: maengel.append(f"Werkzeuge {ktext} (heute {btext})") zeile("Werkzeug-Aufrufe", ktext, btext, "", "schlechter") elif ka > ba + 1e-9: vorteile.append(f"Werkzeuge {ktext} statt {btext}") zeile("Werkzeug-Aufrufe", ktext, btext, "", "besser") else: zeile("Werkzeug-Aufrufe", ktext, btext, "", "gleich") if rolle == "hirn": for schluessel, name in (("deutsch", "Deutsch"), ("json", "JSON")): ko, bo = bool((k.get(schluessel) or {}).get("ok")), bool((b.get(schluessel) or {}).get("ok")) if not ko: maengel.append(f"{name}-Probe nicht bestanden") zeile(name, "ja" if ko else "nein", "ja" if bo else "nein", "", "gleich" if ko == bo else ("besser" if ko else "schlechter")) else: kc, bc = k.get("code") or {}, b.get("code") or {} ktext, btext = f"{kc.get('ok', 0)}/{kc.get('von', 0)}", f"{bc.get('ok', 0)}/{bc.get('von', 0)}" if kc.get("ok", 0) < bc.get("ok", 0): maengel.append(f"Programmieraufgaben {ktext} (heute {btext})") zeile("Programmieraufgaben", ktext, btext, "", "schlechter") elif kc.get("ok", 0) > bc.get("ok", 0): vorteile.append(f"Programmieraufgaben {ktext} statt {btext}") zeile("Programmieraufgaben", ktext, btext, "", "besser") else: zeile("Programmieraufgaben", ktext, btext, "", "gleich") kb, bb = k.get("bild"), b.get("bild") kann_k, kann_b = bild_kann(kb), bild_kann(bb) if not kann_k and (BILD_PFLICHT or kann_b): maengel.append("versteht keine Bilder" if not kb else f"Bild-Probe nicht bestanden ({_bild_text(kb)})") zeile("Bilder", _bild_text(kb), _bild_text(bb), "", "schlechter") elif kann_k and not kann_b: vorteile.append("versteht Bilder (heute nicht)") zeile("Bilder", _bild_text(kb), _bild_text(bb), "", "besser") else: zeile("Bilder", _bild_text(kb), _bild_text(bb), "", "gleich") if maengel: ergebnis, satz = "durchgefallen", "Durchgefallen: " + "; ".join(maengel) + "." elif not vorteile: ergebnis, satz = "durchgefallen", "Durchgefallen: kein Vorteil gegenüber dem heutigen Modell." else: ergebnis, satz = "bestanden", "Bestanden: " + "; ".join(vorteile) + "." return {"ergebnis": ergebnis, "vergleich": zeilen, "zusammenfassung": satz, "maengel": maengel, "vorteile": vorteile} # --- Hermes-Smoke (nur für den Handbetrieb, pruefstand-hirn.py) --------------------------------- def hermes_smoke(hermes_args: list[str], protokoll=log) -> list[dict]: """Zwei Aufgaben durch den ECHTEN Hermes-Agenten (Provider `pruefstand` -> :5899 muss in `providers:` der Hermes-Config stehen).""" ergebnisse = [] for frage, erwartet in ((("Führe im Terminal exakt den Befehl echo pruefstand-ok aus und gib mir nur dessen " "Ausgabe zurück."), "pruefstand-ok"), ("Welche Verzeichnisse liegen direkt unter /srv/models? Nenne mir drei davon.", "GGUF")): t0 = time.time() try: p = subprocess.run([HERMES, "chat", *hermes_args, "-Q", "--oneshot", "--max-turns", "6", "-q", frage], capture_output=True, text=True, timeout=900) ausgabe = (p.stdout + p.stderr)[-600:] except subprocess.TimeoutExpired: ausgabe = "TIMEOUT" except OSError as e: ausgabe = f"FEHLER {e}" ok = erwartet.lower() in ausgabe.lower() ergebnisse.append({"ok": ok, "s": round(time.time() - t0), "tail": ausgabe[-160:].replace("\n", " ")}) protokoll(f" hermes: {'ok' if ok else 'FEHLT'} {round(time.time() - t0)} s {ausgabe[-120:].replace(chr(10), ' ')!r}") return ergebnisse