Spaces:
Running
Running
| """Eval Kurations-Qualität (Kurator-Pipeline-v3-Konzept, Welle 4 – «Beweis statt Gefühl»). | |
| Testset: 3 Räume (bad/wohnen/kueche) × 3 Stilprofile (zwei Extreme + mittig- | |
| unentschieden) × 5 Seeds = 45 Läufe je Port. | |
| Metriken je Port (Baseline immer; `llm-api` nur wenn `FP_KURATOR_URL` gesetzt): | |
| - Validität (nach Repair), Stil-Treue (cos Profil↔Auswahl), Diversität | |
| (distinkte Auswahl-Sets über das Testset) – wie in Welle 1–3. | |
| - Solver-Überlebensrate (NEU): `kuratiere()`-Auswahl durch `solve()` gejagt – | |
| platzierte / gewählte Items. Nutzt dieselbe generische `solve()`-Signatur wie | |
| `/solve` in `api.py` (Zweig ohne `req.auswahl`, s. dort) – auch für Küche: | |
| die Eval prüft die generische Auswahl+Solver-Pipeline, nicht die | |
| spezialisierte Baugruppen-Lösung `solve_kueche`. | |
| - Platz-Auslastung (NEU): Footprint-Summe der Boden-Items (w×d×2.5, wandmontiert | |
| ausgenommen – Daumenregel aus `kurator._footprint`) / Bodenfläche. | |
| - Palette-Treffer (NEU): gewählte Farbe je Item (`farben`-Feld, sonst erste | |
| `farbVariante`) als Hex → minimale RGB-Distanz zur Profil-Palette, normiert | |
| auf 0..1 (1 = Treffer). Nur Profile mit Palette fliessen ein (mittig hat | |
| bewusst eine schwache, aber nicht-leere Palette – sonst nicht messbar). | |
| - Norm-Korrektur-Rate (NEU): Anteil Läufe mit Reparatur-Marker | |
| (`CURATOR_FLAECHEN_NORMREPAIR`/`_NORMKORREKTUR`/`CURATOR_FARBEN_BEREINIGT`) | |
| in der `begruendung` – misst Prompt-Qualität. Baseline erzeugt diese Marker | |
| nie (kein Norm-Repair-Pfad) → `n/a`. | |
| - 0-❌-Invariante: jeder von `solve()` gelieferte Plan hat 0 harte Verstösse | |
| (`constraintReport.hard.ok`) – hart assert, kein Soft-Fail. | |
| Gate: Schlägt der Kurator die Baseline NICHT (Stil-Treue + Diversität, wie | |
| bisher), bleibt die Baseline aktiv – Architektur identisch (Port), kein Umbau. | |
| Ohne FP_KURATOR_URL wird nur die Baseline gemessen. | |
| Determinismus: gleiche Seeds ⇒ gleicher Report (Baseline) – `solve()` und | |
| `BaselineKurator` sind seed-deterministisch (Engineering-Grundlagen §1), die | |
| Eval selbst zieht keine Zufallszahlen ausserhalb der Seeds. | |
| Aufruf: uv run python scripts/kurator_eval.py (aus services/engines/) | |
| """ | |
| import json | |
| import math | |
| import os | |
| import time | |
| from pathlib import Path | |
| from statistics import mean | |
| from typing import Any | |
| from fp_engines.kurator import ( | |
| BaselineKurator, | |
| KuratorPort, | |
| _footprint, | |
| mengen_aus_antwort, | |
| stil_score, | |
| waehle_port, | |
| ) | |
| from fp_engines.solver import NoFeasiblePlacement, solve | |
| REPO_ROOT = Path(__file__).resolve().parents[3] | |
| FIXTURES = REPO_ROOT / "packages" / "shared" / "fixtures" / "artefakte" | |
| DATA_CATALOG = REPO_ROOT / "data" / "catalog" | |
| DATA_RULES = REPO_ROOT / "data" / "rules" | |
| REPORT_DATEI = Path(__file__).resolve().parents[1] / "reports" / "kurator_eval.json" | |
| # Slug → Hex, lokale Kopie der 16 kuratierten Interieur-Farben aus dem | |
| # Katalog-Schema-Enum (`katalog-item.schema.json#/$defs/farbSlug`). Die | |
| # Anzeige-Quelle bleibt `apps/web/src/farben.ts` (Client-Optik); hier nur für | |
| # die RGB-Distanzmessung der Palette-Treffer-Metrik – kein Import über die | |
| # Sprachgrenze TS→Python, daher lokal dupliziert (Werte müssen bei einer | |
| # Palette-Änderung dort UND hier gepflegt werden). | |
| FARBSLUG_HEX: dict[str, str] = { | |
| "weiss": "#f4f4f2", | |
| "creme": "#ece5d8", | |
| "sand": "#d9c9a8", | |
| "beige": "#cbb99a", | |
| "hellgrau": "#c9c9c6", | |
| "anthrazit": "#3f4145", | |
| "schwarz": "#232323", | |
| "eiche-hell": "#c9a76c", | |
| "nussbaum": "#6f4e33", | |
| "salbei": "#9aa88e", | |
| "olive": "#6f7250", | |
| "terracotta": "#b0603f", | |
| "bordeaux": "#6e2b35", | |
| "blaugrau": "#7d8b96", | |
| "dunkelblau": "#2d3a55", | |
| "messing": "#ad8a4c", | |
| } | |
| _MAX_RGB_DIST = math.sqrt(3 * 255**2) | |
| RAEUME: dict[str, str] = { | |
| "bad": "raummodell.bad-sample.json", | |
| "wohnen": "raummodell.wohnen-sample.json", | |
| "kueche": "raummodell.kueche-sample.json", | |
| } | |
| NORM_MARKER = ( | |
| "CURATOR_FLAECHEN_NORMREPAIR", | |
| "CURATOR_FLAECHEN_NORMKORREKTUR", | |
| "CURATOR_FARBEN_BEREINIGT", | |
| ) | |
| PROFILE = { | |
| "warm-natuerlich (extrem)": { | |
| "styleVector": {"temperatur": 0.9, "materialitaet": 0.9, "farbigkeit": 0.5}, | |
| "derivedRequirements": [], | |
| "palette": ["#3a5a40", "#6f4e33", "#b0603f"], | |
| }, | |
| "kuehl-minimal (extrem)": { | |
| "styleVector": {"temperatur": -0.9, "opulenz": -0.9, "epoche": 0.9}, | |
| "derivedRequirements": [], | |
| "palette": ["#f5f5f5", "#c9c9c6", "#3f4145"], | |
| }, | |
| "mittig-unentschieden": { | |
| "styleVector": {"temperatur": 0.1, "opulenz": -0.1}, | |
| "derivedRequirements": [], | |
| "palette": ["#cbb99a", "#7d8b96"], | |
| }, | |
| } | |
| SEEDS = range(1, 6) | |
| def _lade_raum(name: str) -> dict[str, Any]: | |
| return json.loads((FIXTURES / RAEUME[name]).read_text(encoding="utf-8")) | |
| def _lade_katalog(room_type: str) -> list[dict[str, Any]]: | |
| return json.loads((DATA_CATALOG / f"{room_type}.json").read_text(encoding="utf-8")) | |
| def _lade_regeln(room_type: str) -> list[dict[str, Any]]: | |
| """Basis- + Raumtyp-Regelset – identisches Ladeschema wie `api._load_rulesets`.""" | |
| regeln: list[dict[str, Any]] = [] | |
| for name in ("basis", room_type): | |
| pfad = DATA_RULES / f"{name}.json" | |
| regeln.extend(json.loads(pfad.read_text(encoding="utf-8"))) | |
| return regeln | |
| _RaumEintrag = tuple[dict[str, Any], list[dict[str, Any]], list[dict[str, Any]]] | |
| RAUM_KATALOG_REGELN: dict[str, _RaumEintrag] = { | |
| name: (_lade_raum(name), _lade_katalog(name), _lade_regeln(name)) for name in RAEUME | |
| } | |
| def _hex_zu_rgb(hex_farbe: str) -> tuple[int, int, int]: | |
| h = hex_farbe.lstrip("#") | |
| return (int(h[0:2], 16), int(h[2:4], 16), int(h[4:6], 16)) | |
| def _rgb_distanz(a: str, b: str) -> float: | |
| ar, ag, ab = _hex_zu_rgb(a) | |
| br, bg, bb = _hex_zu_rgb(b) | |
| return math.sqrt((ar - br) ** 2 + (ag - bg) ** 2 + (ab - bb) ** 2) | |
| def _palette_treffer( | |
| auswahl: list[str], | |
| by_id: dict[str, dict[str, Any]], | |
| farben: dict[str, str] | None, | |
| palette: list[str], | |
| ) -> float | None: | |
| """Mittel über Items der normierten Palette-Nähe (1 = Treffer, 0 = maximal fern). | |
| Farbwahl je Item nach derselben Rangfolge wie `farben.ts::loeseFarbSlug` | |
| (ohne Manuell-Stufe, die gibt es im Eval nicht): KI-Farbe > erste | |
| `farbVariante`. `None`, wenn kein Item eine auflösbare Farbe hat. | |
| """ | |
| treffer: list[float] = [] | |
| for iid in auswahl: | |
| item = by_id.get(iid) | |
| if item is None: | |
| continue | |
| slug = (farben or {}).get(iid) or next(iter(item.get("farbVarianten") or []), None) | |
| hex_ = FARBSLUG_HEX.get(slug) if slug else None | |
| if hex_ is None: | |
| continue | |
| distanz = min(_rgb_distanz(hex_, p) for p in palette) | |
| treffer.append(1 - distanz / _MAX_RGB_DIST) | |
| return mean(treffer) if treffer else None | |
| def _port_seeds(port: KuratorPort) -> range: | |
| """Seeds je Port: FP_EVAL_LLM_SEEDS drosselt NUR den LLM-Port (Free-Tier- | |
| Rate-Limits, Tokens/Minute); die Baseline misst immer die volle Matrix. | |
| Vergleich dann entsprechend gröber – bewusster Trade-off, im Report sichtbar | |
| über die Läufe-Zeile.""" | |
| if port.name != "baseline": | |
| n = os.environ.get("FP_EVAL_LLM_SEEDS") | |
| if n: | |
| return range(1, max(1, int(n)) + 1) | |
| return SEEDS | |
| def _miss(port: KuratorPort) -> dict[str, Any]: | |
| seeds = _port_seeds(port) | |
| laeufe = 0 | |
| gueltig = 0 | |
| treue: list[float] = [] | |
| sets: set[str] = set() | |
| ueberleben: list[float] = [] | |
| auslastung: list[float] = [] | |
| palette_treffer: list[float] = [] | |
| norm_korrektur: list[bool] = [] | |
| invarianten: list[bool] = [] | |
| anker_erfuellt: list[float] = [] | |
| for raum_name in RAEUME: | |
| room, catalog, regeln = RAUM_KATALOG_REGELN[raum_name] | |
| by_id = {c["id"]: c for c in catalog} | |
| bodenflaeche = room["shell"]["floor"].get("area") or 0.0 | |
| for profil in PROFILE.values(): | |
| for seed in seeds: | |
| if port.name != "baseline": | |
| # Free-Tier-TPM: Pause zwischen LLM-Läufen (Serien-Drosselung). | |
| time.sleep(float(os.environ.get("FP_EVAL_PAUSE_S", "0"))) | |
| antwort = port.kuratiere(profil, room, catalog, None, seed) | |
| laeufe += 1 | |
| auswahl = antwort.get("auswahl") or [] | |
| if not auswahl: | |
| continue | |
| gueltig += 1 | |
| # Mengen (Objekt-Ebenen-Modell): Instanz-Anzahl je Item (Default 1). | |
| mengen = mengen_aus_antwort(antwort) | |
| gesamt_instanzen = sum(mengen.get(i, 1) for i in auswahl) | |
| scores = [stil_score(profil, by_id[i]) for i in auswahl if i in by_id] | |
| if scores: | |
| treue.append(mean(scores)) | |
| sets.add(json.dumps(sorted(auswahl))) | |
| # Anker-Erfüllung (NEU): Anteil verankerter Ergänzungen, deren | |
| # ankerTyp durch ein gewähltes Haupt-Objekt gedeckt ist. Die harte | |
| # Validierung erzwingt das bereits (Erwartung ~1.0) – die Spalte | |
| # macht es im Report sichtbar (Diagnose). | |
| haupt_ids = antwort.get("hauptObjekte") or auswahl | |
| haupt_typen = {by_id[i]["funktionsTyp"] for i in haupt_ids if i in by_id} | |
| verankert = [ | |
| e | |
| for e in (antwort.get("ergaenzungen") or []) | |
| if e["itemId"] in by_id and by_id[e["itemId"]].get("ankerTyp") | |
| ] | |
| if verankert: | |
| erfuellt = sum( | |
| 1 for e in verankert if by_id[e["itemId"]]["ankerTyp"] in haupt_typen | |
| ) | |
| anker_erfuellt.append(erfuellt / len(verankert)) | |
| if bodenflaeche > 0: | |
| footprint_summe = sum( | |
| _footprint(by_id[i]) * mengen.get(i, 1) for i in auswahl if i in by_id | |
| ) | |
| auslastung.append(footprint_summe / bodenflaeche) | |
| if profil["palette"]: | |
| treffer = _palette_treffer( | |
| auswahl, by_id, antwort.get("farben"), profil["palette"] | |
| ) | |
| if treffer is not None: | |
| palette_treffer.append(treffer) | |
| begruendung = antwort.get("begruendung") or "" | |
| norm_korrektur.append(any(m in begruendung for m in NORM_MARKER)) | |
| try: | |
| plan = solve( | |
| room, | |
| auswahl, | |
| antwort.get("relationaleAbsichten") or [], | |
| catalog, | |
| regeln, | |
| seed=seed, | |
| norm_profile="ch", | |
| stilprofil_ref=None, | |
| style_profile=profil, | |
| anordnung=antwort.get("anordnung"), | |
| farben=antwort.get("farben"), | |
| mengen=mengen, | |
| ) | |
| except NoFeasiblePlacement: | |
| # Ehrliches Solver-Ergebnis (Engineering-Grundlagen §1): die | |
| # KI-Auswahl war für diesen Raum nicht platzierbar → 0 | |
| # Überlebensrate für diesen Lauf, kein Abbruch der Eval. | |
| ueberleben.append(0.0) | |
| continue | |
| report = plan["constraintReport"] | |
| assert report["hard"]["summary"]["verletzt"] == 0, ( | |
| f"Solver-Invariante verletzt ({raum_name}/{seed}): {report['hard']}" | |
| ) | |
| invarianten.append(bool(report["hard"]["ok"])) | |
| # Überlebensrate zählt jetzt INSTANZEN (Haupt + Ergänzungen×anzahl): | |
| # ein 4er-Stuhlsatz sind 4 platzierbare Instanzen, nicht ein Item. | |
| ueberleben.append(len(plan["placements"]) / gesamt_instanzen) | |
| norm_korrektur_pct: float | str | |
| if port.name == "baseline": | |
| norm_korrektur_pct = "n/a" # Baseline hat keinen Norm-Repair-Pfad (kein LLM-Call C). | |
| else: | |
| norm_korrektur_pct = round(100 * mean(norm_korrektur), 1) if norm_korrektur else 0.0 | |
| return { | |
| "validitaet_pct": round(100 * gueltig / laeufe, 1) if laeufe else 0.0, | |
| "stil_treue_mittel": round(mean(treue), 3) if treue else 0.0, | |
| "diversitaet_sets": len(sets), | |
| "ueberlebensrate_mittel": round(mean(ueberleben), 3) if ueberleben else 0.0, | |
| "auslastung_mittel": round(mean(auslastung), 3) if auslastung else 0.0, | |
| "palette_treffer_mittel": round(mean(palette_treffer), 3) if palette_treffer else 0.0, | |
| "anker_erfuellt_pct": round(100 * mean(anker_erfuellt), 1) if anker_erfuellt else "n/a", | |
| "norm_korrektur_pct": norm_korrektur_pct, | |
| "invariante_gehalten": all(invarianten) if invarianten else True, | |
| "laeufe": laeufe, | |
| } | |
| _SPALTEN: list[tuple[str, str]] = [ | |
| ("Validität %", "validitaet_pct"), | |
| ("Stil-Treue (cos)", "stil_treue_mittel"), | |
| ("Diversität (Sets)", "diversitaet_sets"), | |
| ("Solver-Überlebensrate", "ueberlebensrate_mittel"), | |
| ("Platz-Auslastung", "auslastung_mittel"), | |
| ("Palette-Treffer", "palette_treffer_mittel"), | |
| ("Anker-Erfüllung %", "anker_erfuellt_pct"), | |
| ("Norm-Korrektur %", "norm_korrektur_pct"), | |
| ("0-❌-Invariante", "invariante_gehalten"), | |
| ("Läufe", "laeufe"), | |
| ] | |
| def _zelle(wert: Any) -> str: | |
| if isinstance(wert, bool): | |
| return "✅" if wert else "❌" | |
| return str(wert) | |
| def _drucke_tabelle(zeilen: list[tuple[str, dict[str, Any]]]) -> None: | |
| kopf = "| Metrik | " + " | ".join(name for name, _ in zeilen) + " |" | |
| trenner = "|" + "---|" * (len(zeilen) + 1) | |
| print(kopf) | |
| print(trenner) | |
| for label, key in _SPALTEN: | |
| werte = " | ".join(_zelle(m[key]) for _, m in zeilen) | |
| print(f"| {label} | {werte} |") | |
| def main() -> None: | |
| baseline = _miss(BaselineKurator()) | |
| zeilen = [("baseline", baseline)] | |
| if os.environ.get("FP_KURATOR_URL"): | |
| zeilen.append(("llm-api", _miss(waehle_port()))) | |
| else: | |
| print("Hinweis: FP_KURATOR_URL nicht gesetzt → nur Baseline gemessen.\n") | |
| _drucke_tabelle(zeilen) | |
| if len(zeilen) == 2: | |
| base_m, llm_m = baseline, zeilen[1][1] | |
| schlaegt = ( | |
| llm_m["stil_treue_mittel"] >= base_m["stil_treue_mittel"] | |
| and llm_m["diversitaet_sets"] > base_m["diversitaet_sets"] | |
| ) | |
| print(f"\nGate: {'✅ LLM schlägt Baseline' if schlaegt else '❌ Baseline bleibt aktiv'}") | |
| else: | |
| print("\nGate-Stand: Baseline aktiv (kein LLM konfiguriert).") | |
| REPORT_DATEI.parent.mkdir(parents=True, exist_ok=True) | |
| REPORT_DATEI.write_text( | |
| json.dumps( | |
| { | |
| "testset": { | |
| "raeume": sorted(RAEUME), | |
| "profile": sorted(PROFILE), | |
| "seeds": list(SEEDS), | |
| }, | |
| "ports": dict(zeilen), | |
| }, | |
| indent=2, | |
| ensure_ascii=False, | |
| sort_keys=True, | |
| ) | |
| + "\n", | |
| encoding="utf-8", | |
| ) | |
| print(f"\nReport geschrieben: {REPORT_DATEI.relative_to(REPO_ROOT)}") | |
| if __name__ == "__main__": | |
| main() | |