boxwart: Modell-Radar sucht, testet nachts selbst und empfiehlt (Hirn und Coder)
Ampel / ampel (push) Failing after 21s
Ampel / ampel (push) Failing after 21s
Suche aus Merkliste (deploy/radar-watchlist.json) und Hugging-Face-Entdeckung; nur Kandidaten mit Bild-Projektor, die neben das Warm-Set passen (<= 115 GB inkl. KV-Cache). Nachtlauf mc2-radar.timer 00:30, Tests nur bis 02:30 (um 03:00 kommt NerdQuiz), hoechstens ein neuer Kandidat pro Woche, Notbremse 02:35, RuntimeMaxSec als letzte Sicherung. Pruefstand als Modul (deploy/bench/pruefstand.py): Tempo, Werkzeuge, Deutsch/JSON bzw. Programmieraufgaben und Bild-Probe gegen das heutige Modell der Rolle. Durchgefallene werden geloescht, Bestandene gemeldet und erst nach "Uebernehmen" getauscht. Beim Uebernehmen wandern die Zweitrollen mit (fast beim Hirn, heavy beim Coder), und das Warm-Set des Stewards wird selbst umgestellt statt als Handgriff zu bleiben. Modell-Code im Pruefstand darf keine Prozesse starten (RLIMIT_NPROC=0). Radar steht im Flugplan und unter Waechter-Aufsicht; deploy.sh spielt seine Units ein. Oberflaeche: Vergleichswerte je Kandidat, Rueckfrage vor Uebernehmen und Verwerfen, Status auf Deutsch. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Hitonabi
co-authored by
Claude Opus 5.5
Claude Opus 5.5
parent
4b226d5d22
commit
177c9a311c
@@ -32,6 +32,7 @@ from routers import (
|
||||
hermes_ui,
|
||||
maintenance,
|
||||
models,
|
||||
radar,
|
||||
routing,
|
||||
system,
|
||||
voice,
|
||||
@@ -107,6 +108,7 @@ async def no_cache(
|
||||
|
||||
app.include_router(health.router)
|
||||
app.include_router(boxwart.router) # Cockpit: Start, Hinweise, Modell-Nutzung, Zeitplan
|
||||
app.include_router(radar.router) # Modell-Radar: Kandidaten, Nachttests, Übernehmen/Verwerfen
|
||||
app.include_router(models.router)
|
||||
app.include_router(routing.router)
|
||||
app.include_router(system.router)
|
||||
|
||||
@@ -0,0 +1,112 @@
|
||||
"""
|
||||
Nächtlicher Lauf des Modell-Radars (Box-Wart, 09/2026) — gestartet von mc2-radar.timer um 00:30.
|
||||
|
||||
Warum ein eigener Einstieg: Der Test lädt ein bis zu ~85 GB großes Modell neben Lucys Hirn. Das
|
||||
darf nur nachts passieren und muss um 02:30 sicher vorbei sein, weil um 03:00 der NerdQuiz-
|
||||
Nachtlauf das Hirn braucht. Ablauf:
|
||||
|
||||
1. Suchen — höchstens einmal am Tag (Merkliste + Entdeckung).
|
||||
2. Offene Urteile nachholen („getestet“: damals fehlte die Vergleichsmessung).
|
||||
3. Testen — nur im Fenster 00:30–02:30 und höchstens ein neuer Kandidat pro Woche.
|
||||
|
||||
Die Messungen prüfen die Frist selbst. Hängt trotzdem etwas, zieht fünf Minuten nach der Frist die
|
||||
Notbremse: Kandidaten-Server und Download werden gestoppt, der Prozess endet hart. systemd
|
||||
(RuntimeMaxSec) ist die letzte Sicherung. Ist nichts fällig, endet der Lauf sofort mit Code 0.
|
||||
"""
|
||||
|
||||
import logging
|
||||
import os
|
||||
import signal
|
||||
import subprocess
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
from services import radar
|
||||
|
||||
logging.basicConfig(
|
||||
level=os.environ.get("MC_LOG_LEVEL", "INFO").upper(),
|
||||
format="%(asctime)s %(levelname)-7s %(name)s: %(message)s",
|
||||
)
|
||||
log = logging.getLogger("radar_lauf")
|
||||
|
||||
NOTBREMSE_S = int(os.environ.get("MC_RADAR_NOTBREMSE_S", "300"))
|
||||
|
||||
|
||||
def _beim_beenden(signum, _frame) -> None:
|
||||
"""SIGTERM (systemd stop, RuntimeMaxSec) → normal auslaufen, damit der Kandidaten-Server gestoppt wird."""
|
||||
raise SystemExit(f"Signal {signum}")
|
||||
|
||||
|
||||
def _notbremse(frist: float) -> threading.Timer:
|
||||
def ziehen() -> None:
|
||||
log.error("Radar: Notbremse – %s s nach der Frist läuft noch etwas, der Lauf wird hart beendet.", NOTBREMSE_S)
|
||||
try:
|
||||
radar.notstopp()
|
||||
finally:
|
||||
os._exit(3)
|
||||
|
||||
bremse = threading.Timer(max(frist - time.time(), 0) + NOTBREMSE_S, ziehen)
|
||||
bremse.daemon = True
|
||||
bremse.start()
|
||||
return bremse
|
||||
|
||||
|
||||
NOTIFY = Path(__file__).resolve().parents[1] / "deploy" / "notify.sh"
|
||||
|
||||
|
||||
def melde_bestanden(test: dict) -> None:
|
||||
"""Ein bestandener Kandidat wartet auf eine Entscheidung — einmal Bescheid geben, sonst liegt er
|
||||
unbemerkt herum (Lehre der Pins vom September). notify.sh sammelt Nachtmeldungen für den Morgen."""
|
||||
k = radar.lade_stand()["kandidaten"].get(test.get("kandidat")) or {}
|
||||
rolle = "das Hirn" if test.get("rolle") == "hirn" else "den Coder"
|
||||
text = (f"{k.get('name') or test.get('kandidat')} hat den Nachttest für {rolle} bestanden. "
|
||||
f"{str(test.get('zusammenfassung') or '')[:300]} "
|
||||
"In MC2 unter Modelle kannst du ihn übernehmen oder verwerfen.")
|
||||
try:
|
||||
subprocess.run(["bash", str(NOTIFY), "-s", "[Modell-Radar]", text], timeout=120, check=False,
|
||||
stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
||||
except (OSError, subprocess.SubprocessError):
|
||||
log.warning("Radar: Meldung über den bestandenen Kandidaten ging nicht raus.", exc_info=True)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
beginn = radar.jetzt()
|
||||
if radar.suche_faellig(beginn):
|
||||
try:
|
||||
ergebnis = radar.suche()
|
||||
log.info("Radar: Suche – %s neu, %s warten", len(ergebnis.get("neu") or []), ergebnis.get("wartend"))
|
||||
except Exception:
|
||||
log.warning("Radar: Suche fehlgeschlagen", exc_info=True)
|
||||
else:
|
||||
log.info("Radar: heute schon gesucht.")
|
||||
|
||||
if not radar.im_fenster(radar.jetzt()):
|
||||
log.info("Radar: außerhalb des Test-Fensters %s–%s, nichts zu testen.", radar.FENSTER_START,
|
||||
radar.FENSTER_ENDE)
|
||||
return 0
|
||||
|
||||
frist = radar.fenster_ende(radar.jetzt()).timestamp()
|
||||
signal.signal(signal.SIGTERM, _beim_beenden)
|
||||
bremse = _notbremse(frist)
|
||||
try:
|
||||
if (nachgeholt := radar.nachurteilen(frist)):
|
||||
log.info("Radar: %s offene Urteile nachgeholt.", nachgeholt)
|
||||
plan = radar.plane_test(radar.lade_stand(), radar.jetzt())
|
||||
if not plan.get("kandidat"):
|
||||
log.info("Radar: nichts fällig (%s).", plan.get("grund"))
|
||||
return 0
|
||||
log.info("Radar: teste %s bis spätestens %s.", plan.get("grund"), radar.FENSTER_ENDE)
|
||||
test = radar.teste(plan["kandidat"], frist)
|
||||
if test:
|
||||
log.info("Radar: %s – %s", test.get("ergebnis"), test.get("zusammenfassung"))
|
||||
if test.get("ergebnis") == "bestanden":
|
||||
melde_bestanden(test)
|
||||
return 0
|
||||
finally:
|
||||
bremse.cancel()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,44 @@
|
||||
"""
|
||||
Modell-Radar-Schnittstellen (Box-Wart, Umbau 09/2026).
|
||||
|
||||
GET /api/radar Nächster und letzter Test, Kandidaten, Test-Verlauf
|
||||
POST /api/radar/suche Jetzt suchen (Merkliste + Hugging-Face-Entdeckung)
|
||||
POST /api/radar/{id}/uebernehmen Bestandenen Kandidaten in Betrieb nehmen (Modell-Tausch)
|
||||
POST /api/radar/{id}/verwerfen Kandidaten verwerfen (Dateien weg, nie wieder testen)
|
||||
|
||||
Dünn: die Logik liegt in services/radar.py. Getestet wird nur nachts (backend/radar_lauf.py).
|
||||
"""
|
||||
|
||||
from fastapi import APIRouter, HTTPException
|
||||
from services import radar
|
||||
|
||||
router = APIRouter(prefix="/api", tags=["radar"])
|
||||
|
||||
|
||||
def _antwort(ergebnis: dict) -> dict:
|
||||
if not ergebnis.get("ok"):
|
||||
raise HTTPException(status_code=int(ergebnis.get("status") or 409),
|
||||
detail=ergebnis.get("detail") or "Das ging nicht.")
|
||||
return ergebnis
|
||||
|
||||
|
||||
@router.get("/radar")
|
||||
def radar_stand() -> dict:
|
||||
return radar.uebersicht()
|
||||
|
||||
|
||||
@router.post("/radar/suche")
|
||||
def radar_suche() -> dict:
|
||||
"""Sucht sofort (Netz: Hugging Face) und liefert danach den neuen Stand."""
|
||||
ergebnis = radar.suche()
|
||||
return {**radar.uebersicht(), "suche": ergebnis}
|
||||
|
||||
|
||||
@router.post("/radar/{kandidat_id}/uebernehmen")
|
||||
def radar_uebernehmen(kandidat_id: str) -> dict:
|
||||
return _antwort(radar.uebernehmen(kandidat_id))
|
||||
|
||||
|
||||
@router.post("/radar/{kandidat_id}/verwerfen")
|
||||
def radar_verwerfen(kandidat_id: str) -> dict:
|
||||
return _antwort(radar.verwerfen(kandidat_id))
|
||||
@@ -474,6 +474,21 @@ def set_role(model_id: str, role: str | None) -> bool:
|
||||
return True
|
||||
|
||||
|
||||
def add_role(model_id: str, role: str) -> bool:
|
||||
"""Wie set_role, aber die übrigen Aliase des Ziel-Modells bleiben — für Modelle mit zwei Rollen
|
||||
(Coder: coder UND heavy). set_role behielte nur die geschützten Aliase und würfe coder wieder weg."""
|
||||
cfg = read_config()
|
||||
models = cfg.get("models") or {}
|
||||
rolle = (role or "").strip().lower()
|
||||
if model_id not in models or not rolle or not isinstance(models[model_id], dict):
|
||||
return False
|
||||
bisher = [a for a in (models[model_id].get("aliases") or []) if str(a).lower() != rolle]
|
||||
set_role_alias(cfg, model_id, rolle) # nimmt die Rolle allen anderen Modellen weg
|
||||
models[model_id]["aliases"] = bisher + [rolle]
|
||||
write_config(cfg)
|
||||
return True
|
||||
|
||||
|
||||
def set_ctx(model_id: str, ctx: int) -> bool:
|
||||
"""Kontextlänge (-c) eines bestehenden Modells ändern."""
|
||||
cfg = read_config()
|
||||
|
||||
@@ -24,7 +24,7 @@ SYSTEM_SERVICES = {"llama-swap"}
|
||||
# erneut laufen lassen“ — der Wächter bietet das als Knopf an (services/waechter.py).
|
||||
USER_SERVICES = {"mission-control-2", "mc2-gateway", "mc2-steward", "box-console",
|
||||
"hermes-gateway", "hermes-builtin-ui", "voice-service", "lucy-stimme",
|
||||
"projekte-sync", "mc2-backup"}
|
||||
"projekte-sync", "mc2-backup", "mc2-radar"}
|
||||
|
||||
# Engine-Update: lädt den neuesten Vulkan-Build (deploy/update-engine.sh, läuft als root).
|
||||
_REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -83,6 +83,7 @@ DIENSTE: dict[str, tuple[bool, bool, str]] = {
|
||||
TIMER_DIENSTE: dict[str, tuple[str, bool]] = {
|
||||
"projekte-sync": ("Projekte-Abgleich", False),
|
||||
"mc2-backup": ("Sicherung", True),
|
||||
"mc2-radar": ("Modell-Radar", False),
|
||||
}
|
||||
|
||||
HERMES_JOBS_PATH = HERMES_HOME / "cron" / "jobs.json"
|
||||
|
||||
@@ -14,7 +14,7 @@ from zoneinfo import ZoneInfo
|
||||
|
||||
from services import modell_nutzung, waechter
|
||||
|
||||
TIMER = {"mc2-backup.timer": "Sicherung"}
|
||||
TIMER = {"mc2-backup.timer": "Sicherung", "mc2-radar.timer": "Modell-Radar"}
|
||||
LOCAL_TZ = ZoneInfo(os.environ.get("MC_LOCAL_TZ", "Europe/Berlin"))
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,617 @@
|
||||
"""Tests für das Modell-Radar: reine Logik (Zeitfenster, Wochengrenze, Filter, Speicher-Rechnung,
|
||||
Zustandsübergänge) und die Urteile und Programmieraufgaben des Prüfstands — ohne Netz und ohne Box.
|
||||
Zahlen wie auf der Box am 23.09.2026 (Ornith 21,7 GB + mmproj, Flash-Next UD-IQ3_XXS 82 GB)."""
|
||||
|
||||
import json
|
||||
import os
|
||||
import struct
|
||||
import time
|
||||
import zlib
|
||||
from datetime import date, datetime
|
||||
|
||||
import pytest
|
||||
from services import radar
|
||||
|
||||
BERLIN = radar.LOCAL_TZ
|
||||
|
||||
|
||||
def _zeit(text: str) -> datetime:
|
||||
return datetime.fromisoformat(text).replace(tzinfo=BERLIN)
|
||||
|
||||
|
||||
def _kandidat(kid: str, rolle: str = "hirn", status: str = "neu", quelle: str = "watchlist", rang: int = 0,
|
||||
**extra) -> dict:
|
||||
return {"id": kid, "name": kid, "rolle": rolle, "repo": f"org/{kid}-GGUF", "quant": "Q4_K_M", "quelle": quelle,
|
||||
"rang": rang, "datei": f"{kid}-Q4_K_M.gguf", "dateien": [[f"{kid}-Q4_K_M.gguf", 100]],
|
||||
"mmproj": ["mmproj-F16.gguf", 10], "draft": None, "flags": [], "groesse_gb": 21.7, "passt": True,
|
||||
"eng": False, "ctx": 131072, "status": status, "begruendung": "", **extra}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def zustand(monkeypatch, tmp_path):
|
||||
"""Radar-Zustand und Radar-Ordner in einem Temp-Ordner statt unter /srv/models."""
|
||||
monkeypatch.setattr(radar, "STORE_PATH", tmp_path / "mc2-radar.json")
|
||||
monkeypatch.setattr(radar, "RADAR_DIR", tmp_path / "radar")
|
||||
(tmp_path / "radar").mkdir()
|
||||
return tmp_path
|
||||
|
||||
|
||||
# --- Zeitfenster und Wochengrenze ------------------------------------------------------------
|
||||
|
||||
def test_zeitfenster_nur_von_0030_bis_0230():
|
||||
assert not radar.im_fenster(_zeit("2026-09-29T00:29:59"))
|
||||
assert radar.im_fenster(_zeit("2026-09-29T00:30:00"))
|
||||
assert radar.im_fenster(_zeit("2026-09-29T02:29:59"))
|
||||
assert not radar.im_fenster(_zeit("2026-09-29T02:30:00"))
|
||||
assert not radar.im_fenster(_zeit("2026-09-29T14:00:00"))
|
||||
assert radar.fenster_ende(_zeit("2026-09-29T00:45:00")) == _zeit("2026-09-29T02:30:00")
|
||||
|
||||
|
||||
def test_fensterende_am_umstellungstag_nie_nach_0300():
|
||||
# 28.03.2027: 02:00 springt auf 03:00, 02:30 gibt es nicht. Lieber früher enden als in den NerdQuiz-Lauf.
|
||||
ende = radar.fenster_ende(_zeit("2027-03-28T00:45:00"))
|
||||
assert ende.timestamp() <= datetime(2027, 3, 28, 3, 0, tzinfo=BERLIN).timestamp()
|
||||
|
||||
|
||||
def test_naechster_start_ist_die_naechste_nacht():
|
||||
assert radar.naechster_start(_zeit("2026-09-29T00:10:00")) == _zeit("2026-09-29T00:30:00")
|
||||
assert radar.naechster_start(_zeit("2026-09-29T01:00:00")) == _zeit("2026-09-30T00:30:00")
|
||||
assert radar.naechster_start(_zeit("2026-09-29T14:00:00")) == _zeit("2026-09-30T00:30:00")
|
||||
|
||||
|
||||
def test_wochengrenze_ein_neuer_kandidat_pro_woche():
|
||||
stand = radar._leer()
|
||||
stand["kandidaten"] = {"a": _kandidat("a", status="bestanden",
|
||||
erster_start=_zeit("2026-09-29T00:31:00").timestamp()),
|
||||
"b": _kandidat("b")}
|
||||
assert radar.wochen_frei_ab(stand, date(2026, 10, 3)) == date(2026, 10, 6)
|
||||
assert radar.wochen_frei_ab(stand, date(2026, 10, 6)) is None
|
||||
|
||||
|
||||
def test_plan_testet_nur_im_fenster_mit_freier_woche_und_setzt_angefangene_fort():
|
||||
stand = radar._leer()
|
||||
stand["kandidaten"] = {"a": _kandidat("a", status="bestanden",
|
||||
erster_start=_zeit("2026-09-29T00:31:00").timestamp()),
|
||||
"b": _kandidat("b", rang=1)}
|
||||
radar._ordne(stand)
|
||||
assert stand["kandidaten"]["b"]["status"] == "wartet"
|
||||
|
||||
plan = radar.plane_test(stand, _zeit("2026-10-03T00:31:00"))
|
||||
assert plan["kandidat"] is None and "06.10." in plan["grund"]
|
||||
assert radar.plane_test(stand, _zeit("2026-10-06T00:31:00"))["kandidat"] == "b"
|
||||
assert "außerhalb" in radar.plane_test(stand, _zeit("2026-10-06T14:00:00"))["grund"]
|
||||
assert "Minuten" in radar.plane_test(stand, _zeit("2026-10-06T02:15:00"))["grund"]
|
||||
|
||||
# angefangen (z. B. Download unterbrochen): darf in der nächsten Nacht weiter, die Woche gilt ab dem 1. Start
|
||||
stand["kandidaten"]["b"]["erster_start"] = _zeit("2026-10-06T00:31:00").timestamp()
|
||||
assert radar.plane_test(stand, _zeit("2026-10-07T00:31:00"))["kandidat"] == "b"
|
||||
|
||||
|
||||
def test_laufender_test_blockiert_zweiten_lauf_und_zeigt_sich(zustand):
|
||||
stand = radar._leer()
|
||||
stand["kandidaten"]["b"] = _kandidat("b", status="wartet", erster_start=time.time())
|
||||
stand["lauf"] = {"pid": os.getpid(), "kandidat": "b", "schritt": "test", "seit": time.time(),
|
||||
"frist": time.time() + 3600}
|
||||
assert radar.plane_test(stand, _zeit("2026-10-06T00:40:00"))["grund"] == "es läuft schon ein Test"
|
||||
radar._speichere(stand)
|
||||
assert radar.uebersicht()["kandidaten"][0]["begruendung"].startswith("Läuft gerade seit")
|
||||
stand["lauf"]["pid"] = 999_999_999 # Prozess tot → Lauf gilt als beendet
|
||||
assert radar._aktiver_lauf(stand) is None
|
||||
|
||||
|
||||
# --- Namen, Filter, Dateien, Speicher -----------------------------------------------------------
|
||||
|
||||
def test_namen_und_staemme():
|
||||
assert radar.anzeigename("bartowski/Qwen_Qwen3.6-35B-A3B-GGUF") == "Qwen3.6-35B-A3B"
|
||||
assert radar.anzeigename("unsloth/Qwen3.8-Flash-Next-GGUF") == "Qwen3.8-Flash-Next"
|
||||
assert radar.stamm("Qwen3.8-Flash-Next-UD-IQ3_XXS-00001-of-00003.gguf") == "qwen3.8-flash-next"
|
||||
assert radar.stamm("Qwen3.6-35B-A3B") == radar.stamm("unsloth/Qwen3.6-35B-A3B-GGUF")
|
||||
assert radar.generation("Qwen3.8-27B") == ("qwen", 3.8)
|
||||
assert radar.kandidat_id("hirn", "ornith-ai/Ornith-1.5-35B-A3B-GGUF", "Q4_K_M") == \
|
||||
"hirn-ornith-ai-ornith-1-5-35b-a3b-gguf-q4-k-m"
|
||||
|
||||
|
||||
def test_entdeckung_filtert_alte_kleine_bekannte_und_dichte_hirne():
|
||||
heute = {"hirn": "Qwen3.6-35B-A3B", "coder": "Qwen3.8-27B"}
|
||||
bekannt = {"qwen3.6-35b-a3b", "qwen3.8-27b"}
|
||||
|
||||
def fund(repo, params, aktiv=None, kuratiert=False):
|
||||
return {"repo": repo, "params_b": params, "caps": {"active_b": aktiv}, "curated": kuratiert}
|
||||
|
||||
def grund(m, rolle):
|
||||
return radar.entdeckung_ungeeignet(m, rolle, bekannt, heute)
|
||||
|
||||
assert "Katalog" in grund(fund("unsloth/Qwen3-Coder-30B-A3B-Instruct-GGUF", 30, 3, kuratiert=True), "coder")
|
||||
assert "auf der Box" in grund(fund("bartowski/Qwen_Qwen3.6-35B-A3B-GGUF", 35, 3), "hirn")
|
||||
assert "zu klein" in grund(fund("lmstudio-community/Qwen2.5-Coder-14B-Instruct-GGUF", 14), "coder")
|
||||
assert "dichte" in grund(fund("bartowski/Hermes-3-Llama-3.1-70B-GGUF", 70), "hirn")
|
||||
assert "ältere Generation" in grund(fund("bartowski/Qwen_Qwen3.5-35B-A3B-GGUF", 35, 3), "hirn")
|
||||
assert grund(fund("unsloth/Qwen-AgentWorld-35B-A3B-GGUF", 35, 3), "hirn") is None
|
||||
|
||||
|
||||
BAUM_FLASH = [
|
||||
{"type": "directory", "path": "UD-IQ3_XXS"},
|
||||
{"path": "UD-IQ3_XXS/Qwen3.8-Flash-Next-UD-IQ3_XXS-00001-of-00003.gguf", "size": 10_000_000},
|
||||
{"path": "UD-IQ3_XXS/Qwen3.8-Flash-Next-UD-IQ3_XXS-00002-of-00003.gguf", "size": 49_570_000_000},
|
||||
{"path": "UD-IQ3_XXS/Qwen3.8-Flash-Next-UD-IQ3_XXS-00003-of-00003.gguf", "lfs": {"size": 32_380_000_000}},
|
||||
{"path": "UD-IQ4_XS/Qwen3.8-Flash-Next-UD-IQ4_XS-00001-of-00003.gguf", "size": 10_000_000},
|
||||
{"path": "MTP/mtp-Qwen3.8-Flash-Next-Q8_0.gguf", "size": 4_140_000_000},
|
||||
{"path": "mmproj-BF16.gguf", "size": 910_000_000},
|
||||
{"path": "mmproj-F16.gguf", "size": 900_000_000},
|
||||
]
|
||||
KOPF_FLASH = {"general.architecture": "qwen4exp", "qwen4exp.block_count": 48, "qwen4exp.attention.head_count": 24,
|
||||
"qwen4exp.attention.head_count_kv": 2, "qwen4exp.attention.key_length": 256,
|
||||
"qwen4exp.attention.value_length": 256, "qwen4exp.full_attention_interval": 4}
|
||||
|
||||
|
||||
def test_dateiauswahl_nimmt_alle_teile_des_quants_und_das_f16_mmproj():
|
||||
wahl = radar.waehle_dateien(BAUM_FLASH, "UD-IQ3_XXS")
|
||||
assert [p.rsplit("-", 3)[-3] for p, _ in wahl["dateien"]] == ["00001", "00002", "00003"]
|
||||
assert all(p.startswith("UD-IQ3_XXS/") for p, _ in wahl["dateien"])
|
||||
assert wahl["mmproj"] == ["mmproj-F16.gguf", 900_000_000]
|
||||
assert radar.waehle_dateien(BAUM_FLASH, "Q4_K_M") is None
|
||||
assert radar.waehle_dateien(BAUM_FLASH, "Q8_0") is None # der MTP-Kopf ist kein Modell
|
||||
ornith = [{"path": "Ornith-1.5-35B-Q4_K_M.gguf", "size": 21_710_000_000},
|
||||
{"path": "Ornith-1.5-35B-Q8_0.gguf", "size": 37_800_000_000},
|
||||
{"path": "mmproj-Ornith-1.5-35B-BF16.gguf", "size": 900_000_000}]
|
||||
wahl = radar.waehle_dateien(ornith, "Q4_K_M")
|
||||
assert wahl["dateien"] == [["Ornith-1.5-35B-Q4_K_M.gguf", 21_710_000_000]]
|
||||
assert wahl["mmproj"][0] == "mmproj-Ornith-1.5-35B-BF16.gguf"
|
||||
|
||||
|
||||
def _gguf_kopf_bytes(werte: dict) -> bytes:
|
||||
"""Minimaler GGUF-Kopf wie von llama.cpp: Architektur-Schlüssel, danach der Tokenizer (abgeschnitten)."""
|
||||
def s(text: str) -> bytes:
|
||||
return struct.pack("<Q", len(text.encode())) + text.encode()
|
||||
|
||||
teile = [b"GGUF", struct.pack("<I", 3), struct.pack("<Q", 0), struct.pack("<Q", len(werte) + 1)]
|
||||
for k, v in werte.items():
|
||||
if isinstance(v, str):
|
||||
teile += [s(k), struct.pack("<I", 8), s(v)]
|
||||
elif isinstance(v, list):
|
||||
teile += [s(k), struct.pack("<I", 9), struct.pack("<I", 5), struct.pack("<Q", len(v)),
|
||||
struct.pack(f"<{len(v)}i", *v)]
|
||||
else:
|
||||
teile += [s(k), struct.pack("<I", 4), struct.pack("<I", v)]
|
||||
teile += [s("tokenizer.ggml.tokens"), struct.pack("<I", 9)]
|
||||
return b"".join(teile)
|
||||
|
||||
|
||||
def test_kv_cache_aus_dem_gguf_kopf_kennt_hybrid_modelle():
|
||||
daten = _gguf_kopf_bytes({"general.architecture": "qwen35moe", "qwen35moe.block_count": 41,
|
||||
"qwen35moe.attention.head_count": 16, "qwen35moe.attention.head_count_kv": 2,
|
||||
"qwen35moe.rope.dimension_sections": [11, 11, 10, 0],
|
||||
"qwen35moe.attention.key_length": 256, "qwen35moe.attention.value_length": 256,
|
||||
"qwen35moe.full_attention_interval": 4})
|
||||
kopf = radar.gguf_kopf_aus_bytes(daten)
|
||||
assert kopf["general.architecture"] == "qwen35moe" and kopf["qwen35moe.block_count"] == 41
|
||||
# 11 von 41 Schichten mit KV-Cache × 2 KV-Köpfe × (256 + 256) × 1,0625 Byte (q8_0)
|
||||
assert radar.kv_je_token_gb(kopf) == pytest.approx(11 * 2 * 512 * 1.0625 / 1e9)
|
||||
assert radar.kv_je_token_gb(kopf) * 131072 == pytest.approx(1.57, abs=0.01)
|
||||
ohne_hybrid = {k: v for k, v in kopf.items() if not k.endswith("full_attention_interval")}
|
||||
assert radar.kv_je_token_gb(ohne_hybrid) == pytest.approx(41 * 2 * 512 * 1.0625 / 1e9)
|
||||
assert radar.gguf_kopf_aus_bytes(b"kein gguf") is None
|
||||
assert radar.gguf_kopf_aus_bytes(daten[:20]) is None
|
||||
|
||||
|
||||
def test_speicher_passung_nach_der_115_gb_regel():
|
||||
budget = radar.SPEICHER_GRENZE_GB - radar.WARMSET_GB
|
||||
p = radar.speicher_passung("hirn", 22.6, 11968 / 1e9) # Ornith: bequem, voller Kontext
|
||||
assert p["passt"] and not p["eng"] and p["ctx"] == 131072
|
||||
p = radar.speicher_passung("coder", 82.9, 13056 / 1e9, eng_markiert=True) # Flash-Next: knapp
|
||||
assert p["passt"] and p["eng"] and p["bedarf_gb"] <= budget
|
||||
p = radar.speicher_passung("coder", 82.9, 52224 / 1e9) # großer KV-Cache → halber Kontext
|
||||
assert p["passt"] and p["eng"] and p["ctx"] == 65536
|
||||
p = radar.speicher_passung("coder", 94.0, 13056 / 1e9) # UD-IQ4_XS passt nicht
|
||||
assert not p["passt"] and p["text"].startswith("Passt nicht")
|
||||
p = radar.speicher_passung("hirn", 85.0, 1e-6) # Test ja, im Betrieb neben dem Coder knapp
|
||||
assert p["passt"] and p["eng"] and "Im Betrieb" in p["text"]
|
||||
assert radar.platte_nach_download(265, 1900, 82) == pytest.approx(18.26, abs=0.01)
|
||||
|
||||
|
||||
def test_bewertung_ohne_netz(monkeypatch):
|
||||
monkeypatch.setattr(radar, "_hf_baum", lambda repo: BAUM_FLASH)
|
||||
monkeypatch.setattr(radar, "_gguf_kopf", lambda repo, datei: KOPF_FLASH)
|
||||
monkeypatch.setattr(radar, "engine_kennt", lambda arch: True)
|
||||
fund = {"name": "Qwen3.8-Flash-Next", "rolle": "coder", "repo": "unsloth/Qwen3.8-Flash-Next-GGUF",
|
||||
"quant": "UD-IQ3_XXS", "eng": True, "quelle": "watchlist", "rang": 1}
|
||||
k = radar._bewerte(fund)
|
||||
assert k["tauglich"] and k["passt"] and k["eng"] and k["ctx"] == 131072
|
||||
assert k["datei"] == "Qwen3.8-Flash-Next-UD-IQ3_XXS-00001-of-00003.gguf"
|
||||
assert k["groesse_gb"] == pytest.approx(82.9, abs=0.05)
|
||||
assert k["begruendung"].startswith("Aus der Merkliste. Passt knapp")
|
||||
|
||||
monkeypatch.setattr(radar, "engine_kennt", lambda arch: False)
|
||||
k = radar._bewerte(fund)
|
||||
assert not k["tauglich"] and "qwen4exp" in k["begruendung"]
|
||||
|
||||
# Kopf nicht lesbar: die grobe Schätzung sagt „passt nicht“ — das darf nicht zum Verwerfen führen
|
||||
monkeypatch.setattr(radar, "_gguf_kopf", lambda repo, datei: None)
|
||||
k = radar._bewerte(fund)
|
||||
assert k["netzfehler"] and not k["tauglich"] and k["begruendung"].startswith("Speicherbedarf unklar")
|
||||
|
||||
|
||||
def test_merkliste_im_repo():
|
||||
funde, ok = radar._merkliste()
|
||||
assert ok
|
||||
nach_name = {f["name"]: f for f in funde}
|
||||
assert nach_name["Ornith-1.5-35B-A3B"]["rolle"] == "hirn"
|
||||
assert nach_name["Qwen3.8-Flash-Next"]["quant"] == "UD-IQ3_XXS" and nach_name["Qwen3.8-Flash-Next"]["eng"] is True
|
||||
k2 = radar._bewerte(nach_name["K2-Horizon-MoVA-36B-A4B"]) # übersprungen, ganz ohne Netz
|
||||
assert not k2["tauglich"] and k2["begruendung"].startswith("Übersprungen") and "llama.cpp" in k2["begruendung"]
|
||||
|
||||
|
||||
# --- Zustandsübergänge ---------------------------------------------------------------------
|
||||
|
||||
def test_suche_uebernimmt_merkliste_und_laesst_geschichte_in_ruhe():
|
||||
stand = radar._leer()
|
||||
stand["kandidaten"] = {"alt": _kandidat("alt", status="durchgefallen", quelle="discover"),
|
||||
"weg": _kandidat("weg", quelle="discover"),
|
||||
"nein": _kandidat("nein", status="verworfen", verworfen_von="nutzer")}
|
||||
bewertet = [
|
||||
{**_kandidat("ornith", rang=0), "tauglich": True, "begruendung": "Aus der Merkliste. Passt."},
|
||||
{**_kandidat("k2", rang=2), "tauglich": False, "passt": False,
|
||||
"begruendung": "Übersprungen: läuft noch nicht im offiziellen llama.cpp."},
|
||||
{**_kandidat("alt", quelle="discover"), "tauglich": True},
|
||||
{**_kandidat("nein"), "tauglich": True},
|
||||
{**_kandidat("klein", quelle="discover"), "tauglich": False, "begruendung": "Kein GGUF mit Q4_K_M im Repo."},
|
||||
]
|
||||
ergebnis = radar._suche_anwenden(stand, bewertet, {"watchlist": True, "discover": True}, 1000.0)
|
||||
k = stand["kandidaten"]
|
||||
assert k["ornith"]["status"] == "wartet" # Kopf der Warteschlange
|
||||
assert k["k2"]["status"] == "verworfen" and k["k2"]["verworfen_von"] == "radar"
|
||||
assert k["alt"]["status"] == "durchgefallen" # Geschichte bleibt
|
||||
assert k["nein"]["status"] == "verworfen" and k["nein"]["verworfen_von"] == "nutzer"
|
||||
assert "weg" not in k and "klein" not in k
|
||||
assert "klein" in stand["abgelehnt"]
|
||||
assert ergebnis["neu"] == ["ornith"] and stand["letzte_suche"] == 1000.0
|
||||
|
||||
stand["kandidaten"]["weg"] = _kandidat("weg", quelle="discover")
|
||||
radar._suche_anwenden(stand, [], {"watchlist": True, "discover": False}, 2000.0) # Entdeckung offline
|
||||
assert "weg" in stand["kandidaten"]
|
||||
|
||||
|
||||
def test_bestanden_bleibt_liegen_durchgefallen_wird_geloescht(zustand):
|
||||
stand = radar._leer()
|
||||
for kid in ("gut", "schlecht"):
|
||||
ordner = radar.RADAR_DIR / kid
|
||||
ordner.mkdir()
|
||||
(ordner / "m.gguf").write_bytes(b"x")
|
||||
stand["kandidaten"][kid] = _kandidat(kid, status="wartet", ordner=str(ordner), erster_start=1.0)
|
||||
test = radar._abschluss(stand, "gut", "bestanden", "Bestanden: versteht Bilder (heute nicht).",
|
||||
werte={"tiefe": {"tg_tps": 95.0}}, vergleich={"Bilder": {"urteil": "besser"}})
|
||||
radar._abschluss(stand, "schlecht", "durchgefallen", "Durchgefallen: Werkzeuge 3/6 (heute 6/6).")
|
||||
assert stand["kandidaten"]["gut"]["status"] == "bestanden" and (radar.RADAR_DIR / "gut").exists()
|
||||
assert "wartet auf deine Entscheidung" in stand["kandidaten"]["gut"]["begruendung"]
|
||||
assert stand["kandidaten"]["schlecht"]["status"] == "durchgefallen" and not (radar.RADAR_DIR / "schlecht").exists()
|
||||
assert [t["ergebnis"] for t in stand["tests"]] == ["bestanden", "durchgefallen"]
|
||||
assert set(test) == set(radar.TEST_FELDER)
|
||||
|
||||
|
||||
def test_abbrueche_zaehlen_nur_mit_schuld_des_kandidaten(zustand):
|
||||
ordner = radar.RADAR_DIR / "zaeh"
|
||||
ordner.mkdir()
|
||||
stand = radar._leer()
|
||||
stand["kandidaten"]["zaeh"] = _kandidat("zaeh", status="wartet", ordner=str(ordner), erster_start=1.0)
|
||||
radar._abschluss(stand, "zaeh", "abgebrochen", "Port 5899 ist belegt.", zaehlt=False)
|
||||
assert stand["kandidaten"]["zaeh"]["status"] == "wartet" and not stand["kandidaten"]["zaeh"].get("versuche")
|
||||
for _ in range(radar.VERSUCHE_MAX):
|
||||
radar._abschluss(stand, "zaeh", "abgebrochen", "Das Nachtfenster endete um 02:30, bevor der Test fertig war.")
|
||||
k = stand["kandidaten"]["zaeh"]
|
||||
assert k["status"] == "durchgefallen" and k["begruendung"].startswith(f"{radar.VERSUCHE_MAX}-mal abgebrochen")
|
||||
assert not ordner.exists()
|
||||
assert {t["ergebnis"] for t in stand["tests"]} == {"abgebrochen"}
|
||||
|
||||
|
||||
def test_ohne_vergleichswerte_bleibt_der_kandidat_getestet(zustand):
|
||||
stand = radar._leer()
|
||||
stand["kandidaten"]["g"] = _kandidat("g", status="wartet", erster_start=1.0)
|
||||
assert radar._abschluss(stand, "g", None, "", werte={"tiefe": {"tg_tps": 80.0}}) is None
|
||||
assert stand["kandidaten"]["g"]["status"] == "getestet" and stand["kandidaten"]["g"]["messung"]
|
||||
assert stand["tests"] == []
|
||||
|
||||
|
||||
def test_verwerfen_und_uebernehmen_pruefen_den_status(zustand):
|
||||
ordner = radar.RADAR_DIR / "b"
|
||||
ordner.mkdir()
|
||||
stand = radar._leer()
|
||||
stand["kandidaten"] = {"b": _kandidat("b", status="bestanden", ordner=str(ordner)),
|
||||
"u": _kandidat("u", status="uebernommen"), "n": _kandidat("n")}
|
||||
radar._speichere(stand)
|
||||
assert radar.uebernehmen("n")["status"] == 409 # nur bestandene
|
||||
assert radar.uebernehmen("gibt-es-nicht")["status"] == 404
|
||||
assert radar.verwerfen("u")["status"] == 409 # schon übernommen
|
||||
antwort = radar.verwerfen("b")
|
||||
assert antwort["ok"] and antwort["kandidat"]["status"] == "verworfen"
|
||||
assert not ordner.exists()
|
||||
assert radar._lade()["kandidaten"]["b"]["verworfen_von"] == "nutzer"
|
||||
|
||||
|
||||
def test_uebersicht_hat_die_felder_der_oberflaeche(zustand):
|
||||
stand = radar._leer()
|
||||
stand["kandidaten"] = {"a": _kandidat("a", status="wartet", erster_start=_zeit("2026-09-29T00:31:00").timestamp()),
|
||||
"b": _kandidat("b", rang=1)}
|
||||
radar._abschluss(stand, "a", "bestanden", "Bestanden: Prefill bei 13k 18 % schneller.",
|
||||
zeit=_zeit("2026-09-29T01:40:00").timestamp())
|
||||
radar._speichere(stand)
|
||||
u = radar.uebersicht(_zeit("2026-10-01T12:00:00"))
|
||||
assert set(u) == {"naechster_test", "letzter_test", "kandidaten", "tests"}
|
||||
assert set(u["kandidaten"][0]) == set(radar.KANDIDAT_FELDER)
|
||||
assert [k["status"] for k in u["kandidaten"]] == ["wartet", "bestanden"]
|
||||
assert u["naechster_test"] == "2026-10-06T00:30:00+02:00" # die Woche ab dem 29.09. ist belegt
|
||||
assert u["letzter_test"].startswith("2026-09-29T01:40")
|
||||
assert set(u["tests"][0]) == set(radar.TEST_FELDER)
|
||||
|
||||
|
||||
def _uebernahme_vorbereiten(zustand, monkeypatch, rolle: str, alt: dict) -> tuple[list, dict]:
|
||||
"""Bestandenen Kandidaten „o“ anlegen und llama-swap/Hirn-Wechsel durch Protokoll-Attrappen ersetzen."""
|
||||
modelle = zustand / "models"
|
||||
modelle.mkdir()
|
||||
monkeypatch.setattr(radar, "MODELS_DIR", modelle)
|
||||
ordner = radar.RADAR_DIR / "o"
|
||||
ordner.mkdir()
|
||||
(ordner / "o-Q4_K_M.gguf").write_bytes(b"x")
|
||||
(ordner / "mmproj-F16.gguf").write_bytes(b"x")
|
||||
stand = radar._leer()
|
||||
stand["kandidaten"]["o"] = _kandidat("o", rolle=rolle, status="bestanden", ordner=str(ordner))
|
||||
radar._speichere(stand)
|
||||
aufrufe, cfg = [], {"models": {}}
|
||||
|
||||
def eintragen(pfad, **kw):
|
||||
aufrufe.append(("eintragen", pfad, kw["role"], kw["set_alias"]))
|
||||
cfg["models"]["o"] = {"cmd": "vorlage"}
|
||||
return "o"
|
||||
|
||||
from services import agent
|
||||
monkeypatch.setattr(radar, "_heutige_modelle", lambda: {rolle: alt})
|
||||
monkeypatch.setattr(radar.llamaswap, "register_model", eintragen)
|
||||
monkeypatch.setattr(radar.llamaswap, "read_config", lambda: cfg)
|
||||
monkeypatch.setattr(radar.llamaswap, "write_config", lambda c: aufrufe.append(("schreiben",)))
|
||||
monkeypatch.setattr(radar.llamaswap, "set_role", lambda mid, r: aufrufe.append(("rolle", mid, r)) or True)
|
||||
monkeypatch.setattr(radar.llamaswap, "add_role", lambda mid, r: aufrufe.append(("dazu", mid, r)) or True)
|
||||
monkeypatch.setattr(radar, "_steward_neu_starten", lambda: aufrufe.append(("steward",)) or None)
|
||||
monkeypatch.setattr(radar.llamaswap, "set_ttl", lambda mid, ttl: aufrufe.append(("ttl", mid, ttl)) or True)
|
||||
monkeypatch.setattr(agent, "set_agent_brain", lambda mid: aufrufe.append(("hirn", mid)) or {"ok": True})
|
||||
monkeypatch.setattr(radar, "STEWARD_WARMSET", zustand / "warmset.conf")
|
||||
return aufrufe, cfg
|
||||
|
||||
|
||||
def test_uebernehmen_hirn_nutzt_den_hirn_wechsel_und_nimmt_fast_mit(zustand, monkeypatch):
|
||||
aufrufe, cfg = _uebernahme_vorbereiten(zustand, monkeypatch, "hirn",
|
||||
{"name": "Qwen3.6-35B-A3B", "aliases": ["hermes", "fast"], "ttl": 0})
|
||||
(zustand / "warmset.conf").write_text('[Service]\nEnvironment="MC_WARMSET=Qwen3-Embedding-0.6B Qwen3.6-35B-A3B"\n',
|
||||
encoding="utf-8")
|
||||
antwort = radar.uebernehmen("o")
|
||||
assert antwort["ok"] and antwort["modell_id"] == "o"
|
||||
assert [a[0] for a in aufrufe] == ["eintragen", "schreiben", "hirn", "dazu", "steward"]
|
||||
assert aufrufe[0][2:] == ("hermes", False) and aufrufe[3] == ("dazu", "o", "fast")
|
||||
befehl = str(cfg["models"]["o"]["cmd"])
|
||||
assert "--parallel 2" in befehl and "-ctk q8_0" in befehl and "--mmproj" in befehl
|
||||
assert (zustand / "models" / "o-GGUF" / "o-Q4_K_M.gguf").exists() and not (radar.RADAR_DIR / "o").exists()
|
||||
k = radar._lade()["kandidaten"]["o"]
|
||||
assert k["status"] == "uebernommen" and "vorher Qwen3.6-35B-A3B" in k["begruendung"]
|
||||
# Der Steward hält danach das neue Hirn warm, nicht mehr das alte — ohne Handgriff.
|
||||
assert 'MC_WARMSET=Qwen3-Embedding-0.6B o"' in (zustand / "warmset.conf").read_text(encoding="utf-8")
|
||||
assert antwort["text"] is None and "Noch zu tun" not in k["begruendung"]
|
||||
|
||||
|
||||
def test_uebernehmen_coder_nimmt_heavy_mit(zustand, monkeypatch):
|
||||
aufrufe, cfg = _uebernahme_vorbereiten(zustand, monkeypatch, "coder",
|
||||
{"name": "Qwen3.8-27B", "aliases": ["coder", "heavy"], "ttl": 5400})
|
||||
antwort = radar.uebernehmen("o")
|
||||
assert antwort["ok"] and antwort["text"] is None
|
||||
assert aufrufe[2:] == [("rolle", "o", "coder"), ("dazu", "o", "heavy"), ("ttl", "o", 5400)]
|
||||
assert "--parallel 1" in str(cfg["models"]["o"]["cmd"])
|
||||
|
||||
|
||||
def test_betriebsbefehl_uebernimmt_die_getesteten_flags():
|
||||
k = _kandidat("f", rolle="coder", ctx=65536, draft={"typ": "draft-mtp", "n_max": 2, "datei": None}, flags=["-ub", "512"])
|
||||
befehl = radar.betriebs_befehl("coder", {"gguf": "/srv/models/F-GGUF/f.gguf", "mmproj": "/srv/models/F-GGUF/mm.gguf",
|
||||
"draft": None}, k)
|
||||
assert "-c 65536" in befehl and "--parallel 1" in befehl and "-ctk q8_0 -ctv q8_0" in befehl
|
||||
assert "--mmproj /srv/models/F-GGUF/mm.gguf" in befehl and "--load-mode none" in befehl
|
||||
assert befehl.endswith("--spec-type draft-mtp --spec-draft-n-max 2 -ub 512")
|
||||
assert "--parallel 2" in radar.betriebs_befehl("hirn", {"gguf": "x.gguf", "mmproj": None, "draft": None},
|
||||
_kandidat("h"))
|
||||
|
||||
|
||||
# --- Prüfstand: Urteil, Programmieraufgaben, Bild ----------------------------------------------------
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def ps():
|
||||
return radar._pruefstand()
|
||||
|
||||
|
||||
BILD_GUT = {"direkt": 1.0, "gefunden": ["Wort ZEBRA", "Zahl 4711"], "agentisch": True}
|
||||
|
||||
|
||||
def _werte(rolle: str, decode: float = 100.0, prefill: float = 13.0, werkzeuge: int = 6, code: int = 8,
|
||||
bild: dict | None = None, deutsch: bool = True) -> dict:
|
||||
werte = {"rolle": rolle, "tiefe": {"tg_tps": decode, "pp_s": prefill}, "werkzeuge": {"ok": werkzeuge, "von": 6},
|
||||
"bild": bild}
|
||||
if rolle == "hirn":
|
||||
werte.update(deutsch={"ok": deutsch}, json={"ok": True})
|
||||
else:
|
||||
werte["code"] = {"ok": code, "von": 10}
|
||||
return werte
|
||||
|
||||
|
||||
def test_urteil_hirn_mit_bildern_besteht_ohne_verschlechterung(ps):
|
||||
u = ps.urteil("hirn", _werte("hirn", decode=95.0, bild=BILD_GUT), _werte("hirn", decode=100.8))
|
||||
assert u["ergebnis"] == "bestanden"
|
||||
assert u["zusammenfassung"] == "Bestanden: versteht Bilder (heute nicht)."
|
||||
assert u["vergleich"]["Decode bei 13k"]["urteil"] == "gleich" # −6 % liegt in der Toleranz
|
||||
assert list(u["vergleich"]) == ["Decode bei 13k", "Prefill bei 13k", "Werkzeug-Aufrufe", "Deutsch", "JSON",
|
||||
"Bilder"]
|
||||
|
||||
|
||||
def test_urteil_hirn_faellt_bei_schwaeche_oder_ohne_vorteil_durch(ps):
|
||||
basis = _werte("hirn", decode=100.8)
|
||||
u = ps.urteil("hirn", _werte("hirn", werkzeuge=5, bild=BILD_GUT), basis)
|
||||
assert u["ergebnis"] == "durchgefallen" and "Werkzeuge 5/6 (heute 6/6)" in u["zusammenfassung"]
|
||||
u = ps.urteil("hirn", _werte("hirn", decode=70.0, bild=BILD_GUT), basis)
|
||||
assert u["ergebnis"] == "durchgefallen" and "Decode bei 13k 70 statt 100,8 t/s" in u["zusammenfassung"]
|
||||
u = ps.urteil("hirn", _werte("hirn", bild=BILD_GUT), _werte("hirn", bild=BILD_GUT))
|
||||
assert u["zusammenfassung"] == "Durchgefallen: kein Vorteil gegenüber dem heutigen Modell."
|
||||
u = ps.urteil("hirn", _werte("hirn", decode=130.0), basis) # schneller, aber blind
|
||||
assert u["ergebnis"] == "durchgefallen" and "versteht keine Bilder" in u["zusammenfassung"]
|
||||
|
||||
|
||||
def test_urteil_coder_zaehlt_programmieraufgaben_und_bilder(ps):
|
||||
basis = _werte("coder", decode=30.4, prefill=41.2, code=8, bild=BILD_GUT)
|
||||
u = ps.urteil("coder", _werte("coder", decode=45.0, prefill=12.0, code=9, bild=BILD_GUT), basis)
|
||||
assert u["ergebnis"] == "bestanden" and "Programmieraufgaben 9/10 statt 8/10" in u["zusammenfassung"]
|
||||
u = ps.urteil("coder", _werte("coder", decode=45.0, prefill=12.0, code=7, bild=BILD_GUT), basis)
|
||||
assert u["ergebnis"] == "durchgefallen" and "Programmieraufgaben 7/10 (heute 8/10)" in u["zusammenfassung"]
|
||||
u = ps.urteil("coder", _werte("coder", code=9), basis) # der heutige Coder kann Bilder
|
||||
assert u["ergebnis"] == "durchgefallen" and "versteht keine Bilder" in u["zusammenfassung"]
|
||||
|
||||
|
||||
REFERENZ = {
|
||||
"palindrom": "def ist_palindrom(text):\n z = [c.lower() for c in text if c.isalnum()]\n return z == z[::-1]\n",
|
||||
"roemisch": (
|
||||
"def roemisch(zahl):\n"
|
||||
" if not isinstance(zahl, int) or not 1 <= zahl <= 3999:\n"
|
||||
" raise ValueError(zahl)\n"
|
||||
" out = ''\n"
|
||||
" for wert, zeichen in ((1000, 'M'), (900, 'CM'), (500, 'D'), (400, 'CD'), (100, 'C'), (90, 'XC'),\n"
|
||||
" (50, 'L'), (40, 'XL'), (10, 'X'), (9, 'IX'), (5, 'V'), (4, 'IV'), (1, 'I')):\n"
|
||||
" while zahl >= wert:\n"
|
||||
" out += zeichen\n"
|
||||
" zahl -= wert\n"
|
||||
" return out\n"),
|
||||
"intervalle": (
|
||||
"def zusammenfassen(intervalle):\n"
|
||||
" out = []\n"
|
||||
" for a, b in sorted(intervalle):\n"
|
||||
" if out and a <= out[-1][1]:\n"
|
||||
" out[-1][1] = max(out[-1][1], b)\n"
|
||||
" else:\n"
|
||||
" out.append([a, b])\n"
|
||||
" return out\n"),
|
||||
"flach": (
|
||||
"def flach(liste):\n"
|
||||
" out = []\n"
|
||||
" for x in liste:\n"
|
||||
" out.extend(flach(x) if isinstance(x, list) else [x])\n"
|
||||
" return out\n"),
|
||||
"iban": (
|
||||
"def iban_gueltig(iban):\n"
|
||||
" s = iban.replace(' ', '').upper()\n"
|
||||
" if not 15 <= len(s) <= 34 or not s.isalnum() or not s.isascii():\n"
|
||||
" return False\n"
|
||||
" return int(''.join(str(int(c, 36)) for c in s[4:] + s[:4])) % 97 == 1\n"),
|
||||
"lru": (
|
||||
"from collections import OrderedDict\n"
|
||||
"class LRUCache:\n"
|
||||
" def __init__(self, kapazitaet):\n"
|
||||
" self.k, self.d = kapazitaet, OrderedDict()\n"
|
||||
" def get(self, schluessel):\n"
|
||||
" if schluessel not in self.d:\n"
|
||||
" return None\n"
|
||||
" self.d.move_to_end(schluessel)\n"
|
||||
" return self.d[schluessel]\n"
|
||||
" def put(self, schluessel, wert):\n"
|
||||
" self.d[schluessel] = wert\n"
|
||||
" self.d.move_to_end(schluessel)\n"
|
||||
" if len(self.d) > self.k:\n"
|
||||
" self.d.popitem(last=False)\n"),
|
||||
"klammern": (
|
||||
"def klammern_ok(text):\n"
|
||||
" paare, stapel = {')': '(', ']': '[', '}': '{'}, []\n"
|
||||
" for c in text:\n"
|
||||
" if c in '([{':\n"
|
||||
" stapel.append(c)\n"
|
||||
" elif c in paare and (not stapel or stapel.pop() != paare[c]):\n"
|
||||
" return False\n"
|
||||
" return not stapel\n"),
|
||||
"wege": (
|
||||
"import heapq\n"
|
||||
"def kuerzeste_wege(graph, start):\n"
|
||||
" dist, heap = {start: 0}, [(0, start)]\n"
|
||||
" while heap:\n"
|
||||
" d, u = heapq.heappop(heap)\n"
|
||||
" if d > dist[u]:\n"
|
||||
" continue\n"
|
||||
" for v, w in graph.get(u, {}).items():\n"
|
||||
" if v not in dist or d + w < dist[v]:\n"
|
||||
" dist[v] = d + w\n"
|
||||
" heapq.heappush(heap, (d + w, v))\n"
|
||||
" return dist\n"),
|
||||
"median": (
|
||||
"def median(zahlen):\n"
|
||||
" if not zahlen:\n"
|
||||
" raise ValueError('leer')\n"
|
||||
" s, n = sorted(zahlen), len(zahlen)\n"
|
||||
" return s[n // 2] if n % 2 else (s[n // 2 - 1] + s[n // 2]) / 2\n"),
|
||||
"csv": (
|
||||
"def spaltensumme(csv_text, spalte):\n"
|
||||
" zeilen = [z for z in csv_text.splitlines() if z.strip()]\n"
|
||||
" kopf = zeilen[0].split(';')\n"
|
||||
" if spalte not in kopf:\n"
|
||||
" raise KeyError(spalte)\n"
|
||||
" i, summe = kopf.index(spalte), 0.0\n"
|
||||
" for z in zeilen[1:]:\n"
|
||||
" zellen = z.split(';')\n"
|
||||
" wert = zellen[i].strip() if i < len(zellen) else ''\n"
|
||||
" if wert:\n"
|
||||
" summe += float(wert.replace('.', '').replace(',', '.'))\n"
|
||||
" return summe\n"),
|
||||
}
|
||||
|
||||
|
||||
def test_referenzloesungen_bestehen_alle_programmieraufgaben(ps):
|
||||
assert {a["id"] for a in ps.AUFGABEN_CODE} == set(REFERENZ)
|
||||
for aufgabe in ps.AUFGABEN_CODE:
|
||||
ok, grund = ps.fuehre_aufgabe_aus(REFERENZ[aufgabe["id"]], aufgabe["tests"])
|
||||
assert ok, f"{aufgabe['id']}: {grund}"
|
||||
|
||||
|
||||
def test_falsche_verbotene_und_endlose_loesungen_fallen_durch(ps):
|
||||
tests = next(a["tests"] for a in ps.AUFGABEN_CODE if a["id"] == "median")
|
||||
ok, _ = ps.fuehre_aufgabe_aus("def median(z):\n z.sort()\n return z[len(z) // 2]\n", tests)
|
||||
assert not ok
|
||||
ok, grund = ps.fuehre_aufgabe_aus("import os\ndef median(z):\n return os.getpid()\n", tests)
|
||||
assert not ok and grund.startswith("unerlaubter Aufruf")
|
||||
ok, grund = ps.fuehre_aufgabe_aus("def median(z):\n while True:\n pass\n", tests, zeitlimit=2)
|
||||
assert not ok and "Zeitlimit" in grund
|
||||
|
||||
|
||||
def test_code_aus_antwort_nimmt_den_loesungsblock(ps):
|
||||
antwort = "<think>kurz nachdenken</think>Hier:\n```python\ndef f():\n return 1\n```\nAufruf:\n```python\nf()\n```"
|
||||
assert ps.code_aus_antwort(antwort) == "def f():\n return 1\n"
|
||||
assert ps.code_aus_antwort("def g(): return 2") == "def g(): return 2"
|
||||
|
||||
|
||||
def test_bild_ist_ein_gueltiges_png_mit_rotem_kreis(ps):
|
||||
png = ps.bild_png()
|
||||
assert png.startswith(b"\x89PNG\r\n\x1a\n")
|
||||
breite, hoehe = struct.unpack(">II", png[16:24])
|
||||
assert (breite, hoehe) == (448, 300)
|
||||
laenge = struct.unpack(">I", png[33:37])[0]
|
||||
roh = zlib.decompress(png[41:41 + laenge])
|
||||
zeile = 1 + breite * 3
|
||||
assert roh[95 * zeile + 1 + 80 * 3:95 * zeile + 1 + 81 * 3] == b"\xdc\x14\x14" # Mitte des Kreises
|
||||
beschreibung = "Ein roter Kreis, ein blaues Quadrat, ein grünes Dreieck und darunter der Text ZEBRA 4711."
|
||||
assert ps.bild_merkmale(beschreibung) == list(ps.BILD_MERKMALE)
|
||||
assert ps.bild_data_url().startswith("data:image/png;base64,")
|
||||
|
||||
|
||||
def test_werkzeug_treffer_liest_json_argumente(ps):
|
||||
aufruf = [{"function": {"name": "send_telegram", "arguments": json.dumps({"text": "Prüfstand läuft."})}}]
|
||||
assert ps.trifft(aufruf, [("send_telegram", r"pr(ü|ue)fstand")])
|
||||
assert not ps.trifft(aufruf, [("terminal", r".*")])
|
||||
assert len(ps.WERKZEUGE_HIRN) >= 90 # Lucy hat ~100 Werkzeuge
|
||||
assert ps.tiefen_prompt("ab12").startswith("Messlauf ab12.")
|
||||
|
||||
|
||||
def test_add_role_behaelt_die_uebrigen_aliase(monkeypatch):
|
||||
"""add_role ohne Attrappe: coder bleibt, heavy wandert vom alten Coder herüber."""
|
||||
from services import llamaswap
|
||||
cfg = {"models": {"alt": {"cmd": "x", "aliases": ["coder", "heavy"]}, "neu": {"cmd": "y", "aliases": ["coder"]}}}
|
||||
monkeypatch.setattr(llamaswap, "read_config", lambda: cfg)
|
||||
monkeypatch.setattr(llamaswap, "write_config", lambda c: None)
|
||||
assert llamaswap.add_role("neu", "heavy") is True
|
||||
assert cfg["models"]["neu"]["aliases"] == ["coder", "heavy"]
|
||||
assert cfg["models"]["alt"]["aliases"] == ["coder"]
|
||||
assert llamaswap.add_role("gibt-es-nicht", "heavy") is False
|
||||
Reference in New Issue
Block a user