HitonabiandClaude Opus 5.5
4a05bc2a05
Ampel / ampel (push) Failing after 22s
Das heutige Modell misst mit seinem Draft (Hirn 106 t/s, ohne 68), Kandidaten liefen ohne - Ornith fiel am 24.09. deshalb durch, obwohl es dieselbe Architektur hat. Jetzt probiert das Radar kurz: ohne Draft, eingebauten MTP-Kopf und den Draft des heutigen Modells (gleiche Architektur, Speicher reicht) und testet mit der schnellsten. Spekulatives Dekodieren aendert die Antworten nicht, nur das Tempo. Mit Draft laeuft die Bild-Probe auf einem Zwilling ohne Draft (llama.cpp: Draft und Bild = HTTP 500). Uebernehmen baut wie der Betrieb seit 24.09.: Hauptmodell mit dem gewaehlten Draft ohne Projektor, dazu ein Bild-Zwilling; der alte Zwilling fliegt raus. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1187 lines
58 KiB
Python
1187 lines
58 KiB
Python
#!/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 = 3 # 24.09.: Coder darf vor dem Ändern lesen (Bild im Ablauf)
|
||
# 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 <think> im content stehen)."""
|
||
text = _nachricht(j).get("content") or ""
|
||
return re.sub(r"<think>.*?</think>", "", 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": "<dein Name>", "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__|(?<![\w.])(?:open|eval|exec|compile)\s*\(", re.MULTILINE)
|
||
# Die Sperrliste lässt sich mit Tricks umgehen. Der Box-Nutzer darf sudo ohne Passwort — darum
|
||
# verbietet RLIMIT_NPROC=0 dem Prüfling zusätzlich jeden neuen Prozess (kein os.system, kein sudo).
|
||
_RUNNER_KOPF = (
|
||
"import os, sys\n"
|
||
"try:\n"
|
||
" import resource\n"
|
||
" resource.setrlimit(resource.RLIMIT_AS, (2 << 30, 2 << 30))\n"
|
||
" resource.setrlimit(resource.RLIMIT_CPU, (10, 10))\n"
|
||
" resource.setrlimit(resource.RLIMIT_FSIZE, (10 << 20, 10 << 20))\n"
|
||
" resource.setrlimit(resource.RLIMIT_NPROC, (0, 0))\n"
|
||
"except (ImportError, ValueError, OSError):\n"
|
||
" pass\n"
|
||
"sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))\n"
|
||
"from loesung import * # noqa: F403\n"
|
||
)
|
||
|
||
|
||
def code_aus_antwort(text: str) -> str:
|
||
"""Längster Python-Codeblock der Antwort (Lösung statt Beispielaufruf); ohne Block die ganze Antwort."""
|
||
text = re.sub(r"<think>.*?</think>", "", 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 []
|
||
lesen = next((a for a in aufrufe if (a.get("function") or {}).get("name") == "read"), None)
|
||
if rolle != "hirn" and lesen and not trifft(aufrufe, erwartet):
|
||
# Ein guter Coding-Agent liest die Datei, bevor er sie ändert (24.09.: der Coder tat genau das und
|
||
# galt als gescheitert). Dann den Inhalt liefern und nach dem nächsten Schritt fragen.
|
||
verlauf += [{"role": "assistant", "content": "", "tool_calls": [lesen]},
|
||
{"role": "tool", "tool_call_id": lesen.get("id") or "call_lesen",
|
||
"content": 'PORT = 8080\nHOST = "0.0.0.0"\n'}]
|
||
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, bild_modell: str | None = None) -> dict:
|
||
"""Alle Proben der Rolle gegen ein laufendes Modell (Kandidat auf :5899 oder Baseline über llama-swap).
|
||
bild_modell: wer die Bild-Probe macht, wenn es nicht das Modell selbst ist — seit 24.09. sehen Hirn und
|
||
Coder über ihren Bild-Zwilling (vision, coder-bild), weil Draft und Bild in llama.cpp nicht zusammengehen."""
|
||
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"] = bild_probe(basis, bild_modell or modell, rolle, frist, protokoll) if bild else None
|
||
werte["dauer_s"] = round(time.time() - t0)
|
||
return werte
|
||
|
||
|
||
def bild_probe(basis: str, modell: str, rolle: str, frist: float | None = None, protokoll=log) -> dict:
|
||
"""Bild-Probe mit Fehlerfang: ein Absturz der Probe ist ein Messwert, kein Programmfehler."""
|
||
try:
|
||
ergebnis = messe_bild(basis, modell, rolle, frist)
|
||
except Zeitende:
|
||
raise
|
||
except Exception as e:
|
||
ergebnis = {"direkt": 0.0, "gefunden": [], "agentisch": False, "fehler": str(e)[:160]}
|
||
protokoll(f" bild : {ergebnis.get('direkt')} direkt, im Ablauf {'ja' if ergebnis.get('agentisch') else 'nein'}")
|
||
return ergebnis
|
||
|
||
|
||
# --- 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 mit_server(gguf: str, arbeit, *, mmproj: str | None = None, ctx: int = CTX_STANDARD, flags=(),
|
||
frist: float | None = None, protokoll=log):
|
||
"""Kandidat starten, arbeit() laufen lassen, stoppen. Gibt (Ergebnis, Server) zurück; ein Absturz
|
||
während der Arbeit wird zu ServerFehler — ein toter Server darf nicht wie ein schlechtes Modell aussehen."""
|
||
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:
|
||
ergebnis = arbeit()
|
||
except Exception as e:
|
||
if not isinstance(e, Zeitende) and server.prozess.poll() is not None:
|
||
raise ServerFehler(absturz()) from e
|
||
raise
|
||
else:
|
||
if server.prozess.poll() is not None: # Einzelproben fangen Fehler ab — darum hier nachsehen
|
||
raise ServerFehler(absturz())
|
||
finally:
|
||
stoppe_server(server)
|
||
time.sleep(3) # Speicher freigeben lassen, bevor jemand anderes lädt
|
||
return ergebnis, server
|
||
|
||
|
||
_SPEC_FLAGS = ("--spec-type", "--spec-draft-model", "--spec-draft-n-max")
|
||
|
||
|
||
def ohne_draft(flags) -> list[str]:
|
||
"""Flags ohne spekulatives Dekodieren (jeweils Schalter plus Wert)."""
|
||
rest, flags, i = [], list(flags), 0
|
||
while i < len(flags):
|
||
if flags[i] in _SPEC_FLAGS:
|
||
i += 2
|
||
continue
|
||
rest.append(flags[i])
|
||
i += 1
|
||
return rest
|
||
|
||
|
||
def waehle_draft(gguf: str, varianten: list[dict], *, frist: float | None = None, protokoll=log) -> dict:
|
||
"""Schnellste Art des spekulativen Dekodierens für einen Kandidaten. Jede Variante {"name", "flags", …}
|
||
wird gestartet und bei 13k Tiefe gemessen; startet eine nicht (z. B. passt das Vokabular des Drafts
|
||
nicht), fällt sie raus. Spekulatives Dekodieren ändert die Antworten nicht, nur das Tempo — darum
|
||
entscheidet allein das Tempo. Rückgabe: die gewählte Variante plus "messungen" (Name → t/s bzw. Grund)."""
|
||
messungen: dict[str, object] = {}
|
||
beste: tuple[dict, float] | None = None
|
||
for v in varianten:
|
||
try:
|
||
tempo, _ = mit_server(gguf, lambda: messe_tempo(KANDIDAT_URL, "kandidat", frist), ctx=32768,
|
||
flags=v["flags"], frist=frist, protokoll=protokoll)
|
||
except Zeitende:
|
||
raise
|
||
except Exception as e:
|
||
messungen[v["name"]] = f"geht nicht: {str(e)[:120]}"
|
||
protokoll(f" Draft-Variante {v['name']}: geht nicht ({str(e)[:120]})")
|
||
continue
|
||
tiefe = tempo.get("tiefe") or {}
|
||
tps = float(tiefe.get("tg_tps") or 0)
|
||
messungen[v["name"]] = tps
|
||
protokoll(f" Draft-Variante {v['name']}: {tps} t/s bei 13k (Akzeptanz {tiefe.get('acc')})")
|
||
if beste is None or tps > beste[1]:
|
||
beste = (v, tps)
|
||
gewaehlt = beste[0] if beste else next((v for v in varianten if not v["flags"]), varianten[0])
|
||
return {**gewaehlt, "messungen": messungen}
|
||
|
||
|
||
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).
|
||
Mit Draft wie im Betrieb seit 24.09.: Text auf einem Server ohne Projektor, die Bild-Probe auf einem
|
||
Zwilling ohne Draft — llama.cpp kann Draft und Bild nicht zusammen (HTTP 500).
|
||
Rückgabe: {"werte", "ergebnis", "vergleich", "zusammenfassung"}."""
|
||
mit_draft = "--spec-type" in list(flags)
|
||
text_mmproj = None if mit_draft else mmproj
|
||
werte, server = mit_server(
|
||
gguf, lambda: pruefe(rolle, KANDIDAT_URL, "kandidat", bild=bool(text_mmproj), frist=frist, protokoll=protokoll),
|
||
mmproj=text_mmproj, ctx=ctx, flags=flags, frist=frist, protokoll=protokoll)
|
||
befehle = [" ".join(server.befehl)]
|
||
if mit_draft and mmproj:
|
||
werte["bild"], zwilling = mit_server(
|
||
gguf, lambda: bild_probe(KANDIDAT_URL, "kandidat", rolle, frist, protokoll),
|
||
mmproj=mmproj, ctx=min(ctx, 32768), flags=ohne_draft(flags), frist=frist, protokoll=protokoll)
|
||
befehle.append(" ".join(zwilling.befehl))
|
||
werte.update(ladezeit_s=server.ladezeit, ctx=ctx, befehl=" | ".join(befehle))
|
||
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, bild_modell: str | None = None) -> 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,
|
||
bild_modell=bild_modell)
|
||
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
|