"""Ebene 3.5: Diagramme. Pro Baustein ein Mermaid-Diagramm, wenn es Struktur oder Prozess sichtbar macht (Multimedia-Prinzip: instruktive Bilder heben Transfer, Deko schadet). Zuverlässigkeits-Prinzip: der LLM liefert NIE rohes Mermaid, nur Struktur-JSON; Python kompiliert deterministisch; ein Parse-Gate (mermaid_check.mjs) ist das Sicherheitsnetz. Abhängigkeitsdiagramme kommen direkt aus dem braucht-DAG, ganz ohne LLM — korrekt per Konstruktion, null Halluzination. Phasen von bauen(): planen (welche + Typ) → Spec (nur LLM-Zweig) → kompilieren → Gate → verifizieren (Grounding). Platziert werden Diagramme erst im Guide-Writer (E4) über ``; das Assembly tauscht den Marker gegen den Fence.""" import json import logging import os import re import subprocess from pathlib import Path import db import llm from config import DIAGRAMM_TYPEN, DURCHGANG, VERIFY_PANEL log = logging.getLogger("creator2.diagramme") EBENE = "diagramme" _MERMAID_SKRIPT = Path(__file__).resolve().parent / "mermaid_check.mjs" _MERMAID_BASE = Path(__file__).resolve().parent.parent / "frontend" _mermaid_gewarnt = False def _mermaid_fehler(diagramme: list[dict]) -> dict[int, str]: """Parse-Gate: Diagramme durch DENSELBEN Parser wie das Frontend jagen (Zwilling zu guide._katex_fehler). → {id: fehlermeldung} nur der fehlerhaften. Fake-Modus überspringt (deterministisch); fail-open MIT Warnung, wenn node/mermaid fehlt.""" global _mermaid_gewarnt if not diagramme or os.getenv("CREATOR_FAKE_AGENTS"): return {} try: res = subprocess.run( ["node", str(_MERMAID_SKRIPT)], input=json.dumps([{"id": d["id"], "code": d["mermaid"]} for d in diagramme]), capture_output=True, text=True, timeout=60, env={**os.environ, "MERMAID_BASE": str(_MERMAID_BASE)}) if res.returncode: raise RuntimeError(res.stderr.strip()[:200]) return {e["id"]: e["fehler"] for e in json.loads(res.stdout)} except Exception as e: # nie fail-open OHNE Signal (Befund-Prinzip) if not _mermaid_gewarnt: _mermaid_gewarnt = True log.warning("Mermaid-Gate nicht verfügbar (%s) — Diagramm-Validierung fällt aus", e) return {} def _bausteine(topic: str) -> list[dict]: return db.query("SELECT * FROM bausteine WHERE topic=? ORDER BY ord", (topic,)) def _atome_von(baustein_id: int) -> list[dict]: return db.query("SELECT * FROM atome WHERE baustein_id=? AND status NOT IN" " ('gemerged','verworfen') ORDER BY ord", (baustein_id,)) # ── Kompilieren (deterministisch, kein LLM) ─────────────────────────────────── def _sid(raw) -> str: """Mermaid-Knoten-ID: nur alnum/_ erlaubt, führender Buchstabe.""" return "n" + re.sub(r"[^A-Za-z0-9]", "_", str(raw)) def _lbl(text) -> str: """Label für Mermaid entschärfen: keine Anführungszeichen/Klammern/Zeilenumbrüche, die die Syntax brechen. Gekürzt (Lesbarkeit).""" t = str(text).replace('"', "'").replace("\n", " ") t = t.replace("[", "(").replace("]", ")").replace("{", "(").replace("}", ")") return t.strip()[:80] or "?" def _spec_zu_mermaid(spec: dict) -> str: """Struktur-JSON → Mermaid-Quelltext. state → stateDiagram-v2, sonst flowchart.""" knoten = spec.get("knoten") or [] kanten = spec.get("kanten") or [] if not knoten: return "" if spec.get("typ") == "state": zeilen = ["stateDiagram-v2"] for k in kanten: lab = _lbl(k.get("label", "")) pfeil = f": {lab}" if lab and lab != "?" else "" zeilen.append(f" {_sid(k['von'])} --> {_sid(k['zu'])}{pfeil}") return "\n".join(zeilen) zeilen = ["flowchart TD"] for n in knoten: zeilen.append(f' {_sid(n["id"])}["{_lbl(n["label"])}"]') for k in kanten: lab = _lbl(k.get("label", "")) pfeil = f"-- {lab} -->" if lab and lab != "?" else "-->" zeilen.append(f' {_sid(k["von"])} {pfeil} {_sid(k["zu"])}') return "\n".join(zeilen) def _kompilieren(topic: str) -> None: for d in db.query("SELECT * FROM diagramme WHERE topic=? AND status='geplant'" " AND spec!='{}'", (topic,)): mermaid = _spec_zu_mermaid(db.uj(d["spec"], {})) if mermaid: db.update("diagramme", "id", d["id"], mermaid=mermaid, status="kandidat") # ── Planung (welche + Typ) ──────────────────────────────────────────────────── def _atom_liste(atome: list[dict]) -> str: return "\n".join(f"{a['id']}: {a['titel']} — {a['definition']}" for a in atome) async def _plan_judge(ctx: llm.Kontext) -> None: """LLM-Zweig, Schritt „welche + Typ": je Baustein OHNE Diagramm-Zeile entscheidet ein Judge, ob ein Automaten-/Ablauf-/Baum-Diagramm nötig ist (Multimedia-Regel im Prompt). „noetig" → status='geplant'/quelle='llm'; sonst eine 'kein'-Zeile, damit der Judge nicht jede Repair-Iteration erneut fragt (idempotent).""" alle = _bausteine(ctx.topic) je_kap: dict = {} for b in alle: je_kap.setdefault(b["kapitel_id"], []).append(b["titel"]) async def einer(b: dict) -> None: if db.one("SELECT id FROM diagramme WHERE baustein_id=?", (b["id"],)): return atome = _atome_von(b["id"]) if not atome: return nachbarn = "\n".join(f"- {t}" for t in je_kap.get(b["kapitel_id"], []) if t != b["titel"]) or "(keine)" res = await llm.call(ctx, stage="diagramm_plan", template="Diagramm-Plan", role="judge", n=len(atome), item=f"dp{b['id']}", werte={"titel": b["titel"], "atome": _atom_liste(atome), "nachbarn": nachbarn, "durchgang": DURCHGANG.get(b["level"], DURCHGANG["M"])}, erwartet=dict) if res is None: return # Ausfall → nächste Runde erneut (kein 'kein'-Eintrag) typ = str(res.get("typ", "")).strip() if res.get("noetig") and typ in DIAGRAMM_TYPEN: db.insert("diagramme", topic=ctx.topic, baustein_id=b["id"], typ=typ, quelle="llm", status="geplant") else: db.insert("diagramme", topic=ctx.topic, baustein_id=b["id"], typ="", quelle="llm", status="kein") # kein Diagramm — nicht erneut fragen await llm.alle(einer(b) for b in alle) def _spec_normalisieren(res: dict, typ: str) -> dict: """LLM-Spec säubern: nur Knoten mit id+label, nur Kanten zwischen echten Knoten.""" knoten = [{"id": str(n.get("id", "")).strip(), "label": str(n.get("label", "")).strip()} for n in (res.get("knoten") or []) if isinstance(n, dict) and n.get("id")] gueltig = {n["id"] for n in knoten} kanten = [{"von": str(k.get("von", "")).strip(), "zu": str(k.get("zu", "")).strip(), "label": str(k.get("label", "")).strip()} for k in (res.get("kanten") or []) if isinstance(k, dict) and str(k.get("von")) in gueltig and str(k.get("zu")) in gueltig] return {"typ": typ, "knoten": knoten, "kanten": kanten} async def _spec_bauen(ctx: llm.Kontext) -> None: """LLM liefert NUR Struktur-JSON (kein Mermaid) — die Hauptfehlerquelle (roh geschriebenes Mermaid) fällt so weg. Kompiliert wird deterministisch.""" offen = db.query("SELECT * FROM diagramme WHERE topic=? AND status='geplant'" " AND quelle='llm' AND spec='{}'", (ctx.topic,)) async def einer(d: dict) -> None: b = db.one("SELECT * FROM bausteine WHERE id=?", (d["baustein_id"],)) atome = _atome_von(d["baustein_id"]) res = await llm.call(ctx, stage="diagramm_spec", template="Diagramm-Spec", role="guide", n=len(atome), item=f"ds{d['id']}", werte={"titel": b["titel"] if b else "", "typ": d["typ"], "atome": _atom_liste(atome)}, erwartet=dict) if res is None: return # Ausfall → bleibt 'geplant', nächste Runde erneut spec = _spec_normalisieren(res, d["typ"]) if spec["knoten"]: db.update("diagramme", "id", d["id"], spec=db.j(spec)) await llm.alle(einer(d) for d in offen) async def _verifizieren(ctx: llm.Kontext) -> None: """Grounding-Panel für LLM-Diagramme: jeder Knoten muss aus einem Atom ableitbar sein, keine erfundene Struktur. Verwerfen (destruktiv) nur einstimmig (llm.einstimmig) — Panel-Ausfall lässt den Kandidaten stehen.""" offen = db.query("SELECT * FROM diagramme WHERE topic=? AND status='kandidat'" " AND quelle='llm'", (ctx.topic,)) def urteil(stimme): if stimme and isinstance(stimme[0], dict) and "ok" in stimme[0]: return bool(stimme[0]["ok"]) return None async def einer(d: dict) -> None: atome = _atome_von(d["baustein_id"]) stimmen = await llm.panel(ctx, VERIFY_PANEL, stage="diagramm_verify", template="Diagramm-Verify", schritt="diagramm_verify", role="judge", n=len(atome), item=f"dv{d['id']}", werte={"atome": _atom_liste(atome), "mermaid": d["mermaid"]}, erwartet=list) if llm.einstimmig(stimmen, VERIFY_PANEL, urteil): db.update("diagramme", "id", d["id"], status="verifiziert") elif llm.einstimmig(stimmen, VERIFY_PANEL, lambda s: None if urteil(s) is None else not urteil(s)): db.update("diagramme", "id", d["id"], status="verworfen", befunde=db.j(["Grounding-Panel: nicht aus den Atomen ableitbar"])) # sonst: Kandidat bleibt (Panel-Ausfall / uneins) — nächste Runde erneut await llm.alle(einer(d) for d in offen) # ── Ebenen-Vertrag ──────────────────────────────────────────────────────────── async def bauen(ctx: llm.Kontext) -> None: ctx.ebene = EBENE await _plan_judge(ctx) # welche/Typ: Judge (Automaten/Abläufe/Bäume) — kein DAG mehr await _spec_bauen(ctx) # LLM → Struktur-JSON (kein rohes Mermaid) _kompilieren(ctx.topic) # Spec → Mermaid (deterministisch) _gate_kandidaten(ctx.topic) # Parse-Gate: ungültige Mermaid-Quelle → verworfen await _verifizieren(ctx) # Grounding-Panel gegen die Atome def _gate_kandidaten(topic: str) -> None: """Kandidaten durchs Parse-Gate: Syntaxfehler → verworfen (mit Befund). Ein kaputtes Diagramm darf nie in den Guide (fail-closed).""" kandidaten = db.query("SELECT * FROM diagramme WHERE topic=? AND status='kandidat'" " AND mermaid!=''", (topic,)) fehler = _mermaid_fehler(kandidaten) for d in kandidaten: if d["id"] in fehler: db.update("diagramme", "id", d["id"], status="verworfen", befunde=db.j([f"Parse-Fehler: {fehler[d['id']]}"])) def messen(ctx: llm.Kontext) -> list[dict]: befunde = [] for d in db.query("SELECT * FROM diagramme WHERE topic=? AND status='kandidat'", (ctx.topic,)): # hängengebliebener Kandidat (Spec/Verify offen) — sichtbar, kein stiller Verlust befunde.append({"art": "diagramm_unverifiziert", "item": str(d["id"]), "detail": f"Baustein {d['baustein_id']}"}) return befunde async def reparieren(ctx: llm.Kontext, befunde: list[dict]) -> bool: ctx.ebene = EBENE if not befunde: return False await bauen(ctx) return True def gate(ctx: llm.Kontext) -> str | None: """Diagramme sind optional — KEINE Diagramme blockiert den Lauf nicht. Backstop: ein verifiziertes Diagramm ohne Mermaid-Quelltext darf nie ausgeliefert werden.""" tot = db.one("SELECT id FROM diagramme WHERE topic=? AND status='verifiziert'" " AND mermaid='' LIMIT 1", (ctx.topic,)) return "verifiziertes Diagramm ohne Quelltext" if tot else None