305 lines
14 KiB
Python
305 lines
14 KiB
Python
"""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 `<!-- diagramm: id -->`; 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_MIN_KANTEN, 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,))
|
|
|
|
|
|
def _interne_kanten(topic: str, ids: set[int]) -> list[dict]:
|
|
"""aktive braucht-Kanten, deren BEIDE Enden im Baustein liegen."""
|
|
if len(ids) < 2:
|
|
return []
|
|
return [k for k in db.query(
|
|
"SELECT * FROM kanten WHERE topic=? AND art='braucht' AND status='aktiv'", (topic,))
|
|
if k["von_atom"] in ids and k["zu_atom"] in ids]
|
|
|
|
|
|
# ── Spec (JSON-Zwischenrepräsentation) ────────────────────────────────────────
|
|
|
|
def _dag_spec(atome: list[dict], kanten: list[dict]) -> dict:
|
|
"""Abhängigkeitsdiagramm direkt aus dem braucht-Graph. `von braucht zu` ⇒ Pfeil
|
|
zu→von (Voraussetzung zeigt auf das, was sie ermöglicht). Kein LLM."""
|
|
knoten = [{"id": str(a["id"]), "label": a["titel"]} for a in atome]
|
|
kanten_spec = [{"von": str(k["zu_atom"]), "zu": str(k["von_atom"]), "label": ""}
|
|
for k in kanten]
|
|
return {"typ": "dependency", "knoten": knoten, "kanten": kanten_spec}
|
|
|
|
|
|
# ── 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 _planen(topic: str) -> None:
|
|
"""Deterministischer Zweig: Baustein mit ≥DIAGRAMM_MIN_KANTEN internen
|
|
braucht-Kanten → Abhängigkeitsdiagramm aus dem DAG. Idempotent: Bausteine mit
|
|
bestehender Diagramm-Zeile werden übersprungen. Kein LLM, null Halluzination."""
|
|
for b in _bausteine(topic):
|
|
if db.one("SELECT id FROM diagramme WHERE baustein_id=?", (b["id"],)):
|
|
continue
|
|
atome = _atome_von(b["id"])
|
|
ids = {a["id"] for a in atome}
|
|
kanten = _interne_kanten(topic, ids)
|
|
if len(kanten) >= DIAGRAMM_MIN_KANTEN:
|
|
db.insert("diagramme", topic=topic, baustein_id=b["id"], typ="dependency",
|
|
quelle="dag", spec=db.j(_dag_spec(atome, kanten)), status="geplant")
|
|
|
|
|
|
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 and typ != "dependency":
|
|
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
|
|
_planen(ctx.topic) # welche/Typ: deterministisch (DAG)
|
|
await _plan_judge(ctx) # welche/Typ: Judge (Automaten/Abläufe/Bäume)
|
|
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
|
|
# dag-Diagramme sind per Konstruktion gegroundet → nach bestandenem Gate verifiziert
|
|
for d in db.query("SELECT id FROM diagramme WHERE topic=? AND status='kandidat'"
|
|
" AND quelle='dag'", (ctx.topic,)):
|
|
db.update("diagramme", "id", d["id"], status="verifiziert")
|
|
await _verifizieren(ctx) # LLM-Diagramme: 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
|