"""Ebene 3: Struktur. Backward Design: Lernziele aus den Atomen je Soll-Punkt (Zuordnung im selben Call — Lektion 52), dann wird Struktur RECHENBAR: Baustein-Schnitt = Partition der Ziel-Gruppen ins Größenband, Reihenfolge = topologische Sortierung des braucht-Graphen. Zyklen bricht ein Judge an der schwächsten Kante (deterministischer Fallback).""" import asyncio import logging from collections import defaultdict import db import llm from config import (BAUSTEIN_MAX_ATOME, BAUSTEIN_MIN_ATOME, KAPITEL_BAUSTEINE, ZIELE_CHUNK_ATOME) log = logging.getLogger("creator2.struktur") EBENE = "struktur" def _atome(topic: str) -> list[dict]: return db.query("SELECT * FROM atome WHERE topic=? AND status NOT IN" " ('gemerged','verworfen') ORDER BY id", (topic,)) def _braucht_kanten(topic: str) -> list[dict]: return db.query("SELECT * FROM kanten WHERE topic=? AND art='braucht' AND status='aktiv'", (topic,)) # ── Lernziele ───────────────────────────────────────────────────────────────── async def _ziele_bilden(ctx: llm.Kontext) -> None: atome = [a for a in _atome(ctx.topic) if not a["ziel_id"]] gruppen: dict = defaultdict(list) for a in atome: gruppen[a["soll_id"]].append(a) async def eine_gruppe(soll_id, gruppe: list[dict]) -> None: punkt = db.one("SELECT punkt FROM soll WHERE id=?", (soll_id,)) or {"punkt": ctx.topic} rest = gruppe for runde in ("1", "2"): # Schlussrunde MUSS alles entscheiden (Lektion 23) if not rest: return chunk = rest[:ZIELE_CHUNK_ATOME] liste = "\n".join(f"{a['id']}: {a['titel']} — {a['definition']}" for a in chunk) res = await llm.call(ctx, stage="ziele", template="Lernziele", werte={"punkt": punkt["punkt"], "atome": liste, "pflicht": "Jedes Atom MUSS genau einem Ziel" " zugeordnet sein." if runde == "2" else ""}, role="judge", n=len(chunk), item=f"s{soll_id}-r{runde}", erwartet=list) gueltig = {a["id"] for a in chunk} zugeordnet: set[int] = set() for z in res or []: ids = [i for i in z.get("atome", []) if isinstance(i, int) and i in gueltig and i not in zugeordnet] text = str(z.get("ziel", "")).strip() if not ids or not text: continue ziel_id = db.insert("lernziele", topic=ctx.topic, text=text, titel=str(z.get("titel", "")).strip()[:80], soll_id=soll_id, status="aktiv") for i in ids: db.update("atome", "id", i, ziel_id=ziel_id) zugeordnet.update(ids) rest = [a for a in gruppe if a["id"] not in zugeordnet and not db.one("SELECT ziel_id FROM atome WHERE id=?", (a["id"],))["ziel_id"]] if rest: # nach der Schlussrunde: deterministisches Sammelziel statt Lücke ziel_id = db.insert("lernziele", topic=ctx.topic, text=f"Grundlagen: {punkt['punkt']}", soll_id=soll_id, status="aktiv") for a in rest: db.update("atome", "id", a["id"], ziel_id=ziel_id) await asyncio.gather(*(eine_gruppe(s, g) for s, g in gruppen.items())) # ── Zyklen brechen ──────────────────────────────────────────────────────────── def _finde_zyklus(knoten: list[int], kanten: list[tuple[int, int]]) -> list[tuple[int, int]] | None: """Ein Zyklus als Kantenliste, oder None (iterative DFS mit Farben).""" nach = defaultdict(list) for v, z in kanten: nach[v].append(z) farbe = {k: 0 for k in knoten} # 0 weiß, 1 grau, 2 schwarz eltern: dict[int, int] = {} for start in knoten: if farbe[start]: continue stapel = [(start, iter(nach[start]))] farbe[start] = 1 while stapel: v, it = stapel[-1] fortschritt = False for z in it: if z not in farbe: continue if farbe[z] == 0: farbe[z] = 1 eltern[z] = v stapel.append((z, iter(nach[z]))) fortschritt = True break if farbe[z] == 1: # Rückkante → Zyklus rekonstruieren pfad = [(v, z)] k = v while k != z: pfad.append((eltern[k], k)) k = eltern[k] return pfad if not fortschritt: farbe[v] = 2 stapel.pop() return None async def _zyklen_brechen(ctx: llm.Kontext) -> None: while True: atome = _atome(ctx.topic) ids = [a["id"] for a in atome] kanten = [(k["von_atom"], k["zu_atom"]) for k in _braucht_kanten(ctx.topic) if k["von_atom"] in ids and k["zu_atom"] in ids] zyklus = _finde_zyklus(ids, kanten) if not zyklus: return je_id = {a["id"]: a for a in atome} liste = "\n".join(f"{v}→{z}: {je_id[v]['titel']} braucht {je_id[z]['titel']}" for v, z in zyklus) res = await llm.call(ctx, stage="zyklus", template="Kanten-Zyklus", werte={"kanten": liste}, role="judge", item=f"z{zyklus[0][0]}", erwartet=dict) wahl = None if res: paar = (res.get("von"), res.get("zu")) if paar in zyklus: wahl = paar if wahl is None: # deterministischer Fallback: kleinste Kante wahl = min(zyklus) db.execute("UPDATE kanten SET status='gebrochen' WHERE topic=? AND von_atom=?" " AND zu_atom=? AND art='braucht'", (ctx.topic, wahl[0], wahl[1])) log.info("Zyklus gebrochen: %s→%s", wahl[0], wahl[1]) # ── Bausteine schneiden + ordnen ────────────────────────────────────────────── def _topo(knoten: list[int], kanten: list[tuple[int, int]], rang: dict[int, tuple]) -> list[int]: """Kahn; kanten (v, z) = v braucht z ⇒ z kommt vor v. Ties nach rang (stabil). Kanten STRIKT auf die Knotenmenge filtern — Kanten zu fremden Atomen zogen sonst fremde Knoten in die Ausgabe und verdrängten Gruppenmitglieder (gemessen aak: 22 Atome ohne Baustein).""" kn = set(knoten) vorher = defaultdict(set) nachher = defaultdict(set) for v, z in kanten: if v in kn and z in kn: vorher[v].add(z) nachher[z].add(v) offen = sorted([k for k in knoten if not vorher[k]], key=lambda k: rang[k]) out = [] erledigt: set[int] = set() while offen: k = offen.pop(0) out.append(k) erledigt.add(k) neu = [] for n in nachher[k]: if not (vorher[n] - erledigt) and n not in erledigt and n not in offen and n not in neu: neu.append(n) offen = sorted(offen + neu, key=lambda kk: rang[kk]) out += sorted([k for k in knoten if k not in erledigt], key=lambda k: rang[k]) return out def _anker_rang(topic: str) -> dict[int, tuple]: """Natürliche Quellreihenfolge als Tie-Break: (quelle_id, start) des ersten Ankers.""" out = {} for a in _atome(topic): row = db.one("SELECT quelle_id, start FROM anker WHERE atom_id=? AND start>=0" " ORDER BY quelle_id, start LIMIT 1", (a["id"],)) out[a["id"]] = (row["quelle_id"], row["start"]) if row else (999_999, a["id"]) return out def _bausteine_schneiden(ctx: llm.Kontext) -> None: """Deterministisch, idempotent: alte Zuordnung verwerfen, neu schneiden. Ziel-Gruppen ins Band bringen — kleine Ziele desselben Soll-Punkts mergen (das Atom wechselt sein Ziel, Baustein bleibt EIN Ziel), große entlang der Topo-Reihenfolge splitten ("Teil n").""" topic = ctx.topic db.execute("DELETE FROM bausteine WHERE topic=?", (topic,)) atome = _atome(topic) rang = _anker_rang(topic) kanten = [(k["von_atom"], k["zu_atom"]) for k in _braucht_kanten(topic)] gruppen: dict = defaultdict(list) for a in atome: gruppen[a["ziel_id"]].append(a) # kleine Ziele mergen (kleinste zuerst). Partnerwahl thematisch statt per # Soll-Punkt — das Soll ist seit der Entkopplung reine Checkliste (152 feine # Punkte hätten sonst keine Merge-Partner): meiste braucht-Kanten zwischen # den Gruppen, sonst Nachbar in der Quellreihenfolge. def kanten_score(g1: list[dict], g2: list[dict]) -> int: ids1 = {a["id"] for a in g1} ids2 = {a["id"] for a in g2} return sum(1 for v, z in kanten if (v in ids1 and z in ids2) or (v in ids2 and z in ids1)) def gruppen_rang(g: list[dict]) -> tuple: return min(rang[a["id"]] for a in g) geaendert = True while geaendert: geaendert = False kleine = sorted([z for z, g in gruppen.items() if len(g) < BAUSTEIN_MIN_ATOME], key=lambda z: len(gruppen[z])) for z in kleine: passende = [p for p in gruppen if p != z and len(gruppen[p]) + len(gruppen[z]) <= BAUSTEIN_MAX_ATOME] if not passende: continue eigener = gruppen_rang(gruppen[z]) def naehe(pp: int) -> tuple: r = gruppen_rang(gruppen[pp]) return (-kanten_score(gruppen[z], gruppen[pp]), (abs(r[0] - eigener[0]), abs(r[1] - eigener[1])), len(gruppen[pp])) p = min(passende, key=naehe) for a in gruppen[z]: db.update("atome", "id", a["id"], ziel_id=p) gruppen[p] += gruppen.pop(z) db.update("lernziele", "id", z, status="gemerged") geaendert = True break # Bausteine anlegen (große Ziele splitten), Atome zuordnen ziele = {z["id"]: z for z in db.query("SELECT * FROM lernziele WHERE topic=?", (topic,))} for ziel_id, gruppe in gruppen.items(): ordnung = _topo([a["id"] for a in gruppe], kanten, rang) z = ziele.get(ziel_id, {}) titel = z.get("titel") or z.get("text", "Baustein") # Kurztitel; Zieltext nur Fallback n_teile = -(-len(gruppe) // BAUSTEIN_MAX_ATOME) groesse = -(-len(gruppe) // n_teile) for teil in range(n_teile): teil_ids = ordnung[teil * groesse:(teil + 1) * groesse] if not teil_ids: continue b_titel = titel if n_teile == 1 else f"{titel} (Teil {teil + 1})" b_id = db.insert("bausteine", topic=topic, ziel_id=ziel_id, titel=b_titel, status="neu") for pos, atom_id in enumerate(teil_ids): db.update("atome", "id", atom_id, baustein_id=b_id, ord=pos) _bausteine_ordnen(topic, kanten, rang) def _bausteine_ordnen(topic: str, kanten: list[tuple[int, int]], rang: dict) -> None: """Baustein-DAG aus aggregierten Atom-Kanten; Zyklen deterministisch an der schwächsten Aggregat-Kante gebrochen (wenigste Atom-Kanten).""" bausteine = db.query("SELECT * FROM bausteine WHERE topic=?", (topic,)) je_atom = {a["id"]: a["baustein_id"] for a in _atome(topic)} gewicht: dict = defaultdict(int) for v, z in kanten: # v braucht z ⇒ Baustein(z) vor Baustein(v) bv, bz = je_atom.get(v), je_atom.get(z) if bv and bz and bv != bz: gewicht[(bv, bz)] += 1 b_kanten = set(gewicht) b_ids = [b["id"] for b in bausteine] # Tie-Break: Quellposition des Bausteins (früheste Belegstelle); Kanten gewinnen # weiter. Kapitel entstehen erst NACH der Ordnung (Entkopplung vom Soll). atome = _atome(topic) b_rang = {} for b in bausteine: b_rang[b["id"]] = min((rang.get(a["id"], (9, 9)) for a in atome if a["baustein_id"] == b["id"]), default=(9, 9)) while True: zyklus = _finde_zyklus(b_ids, list(b_kanten)) if not zyklus: break schwach = min(zyklus, key=lambda k: (gewicht[k], k)) b_kanten.discard(schwach) # Die Atom-Kanten hinter der geopferten Aggregat-Kante stilllegen — sonst # meldet die QA sie ewig als Vorwärts-Kante (Baustein-Zyklen kann keine # Ordnung vollständig rückwärts auflösen). bv, bz = schwach db.execute("UPDATE kanten SET status='geopfert' WHERE topic=? AND art='braucht'" " AND status='aktiv'" " AND von_atom IN (SELECT id FROM atome WHERE baustein_id=?)" " AND zu_atom IN (SELECT id FROM atome WHERE baustein_id=?)", (topic, bv, bz)) for pos, b_id in enumerate(_topo(b_ids, list(b_kanten), b_rang)): db.update("bausteine", "id", b_id, ord=pos) async def _kapitel_bilden(ctx: llm.Kontext) -> None: """Kapitel = kontiguierliche Segmente der geordneten Baustein-Folge. Das Soll ist seit der Entkopplung reine Checkliste — Kapitel sind Struktur. Der Judge liefert nur GRENZEN (titel + bis-id) — id-Listen schrieb er unvollständig (aak Lauf 17: 516 Tokens für 68 Bausteine → alles verworfen). Der Code erzwingt Reihenfolge + Lückenlosigkeit; ein ungültiges Ergebnis bekommt EINEN Retry, danach √n-Fallback mit Befund (kapitel_fallback). Idempotent.""" topic = ctx.topic bausteine = db.query("SELECT * FROM bausteine WHERE topic=? ORDER BY ord", (topic,)) if not bausteine: return ziele = {z["id"]: z for z in db.query("SELECT * FROM lernziele WHERE topic=?", (topic,))} liste = "\n".join(f"{b['id']}: {b['titel']} — {ziele.get(b['ziel_id'], {}).get('text', '')}" for b in bausteine) richtwert = max(3, round(len(bausteine) / KAPITEL_BAUSTEINE)) folge = [b["id"] for b in bausteine] pos_von = {bid: i for i, bid in enumerate(folge)} async def schneiden(hinweis: str, tag: str) -> list[tuple[str, list[int]]] | None: res = await llm.call(ctx, stage="kapitel_schnitt", template="Kapitel-Schnitt", schritt="kapitel", role="judge", n=len(bausteine), item=tag, erwartet=list, werte={"bausteine": liste, "richtwert": str(richtwert), "hinweis": hinweis}) segmente: list[tuple[str, list[int]]] = [] start = 0 for gruppe in res or []: if not isinstance(gruppe, dict): return None titel = str(gruppe.get("titel", "")).strip() ende = pos_von.get(gruppe.get("bis")) if not titel or ende is None or ende < start: return None segmente.append((titel, folge[start:ende + 1])) start = ende + 1 return segmente if segmente and start == len(folge) else None segmente = await schneiden("", "kapitel") if segmente is None: segmente = await schneiden( "ACHTUNG: Der letzte Versuch war ungültig. Jede Zeile braucht" " \"titel\" und \"bis\" (eine id aus der Liste, streng aufsteigend);" " das letzte \"bis\" MUSS die letzte id der Liste sein.", "kapitel-2") art = "judge" if segmente is None: # deterministischer Fallback: gleichmäßige √n-Segmente art = "fallback" je_id = {b["id"]: b for b in bausteine} n = max(3, round(len(folge) ** 0.5)) groesse = -(-len(folge) // n) segmente = [(je_id[teil[0]]["titel"], teil) for i in range(n) if (teil := folge[i * groesse:(i + 1) * groesse])] db.execute("DELETE FROM kapitel WHERE topic=?", (topic,)) for pos, (titel, ids) in enumerate(segmente): k_id = db.insert("kapitel", topic=topic, titel=titel[:120], ord=pos, art=art) for b_id in ids: db.update("bausteine", "id", b_id, kapitel_id=k_id) async def bauen(ctx: llm.Kontext) -> None: ctx.ebene = EBENE await _ziele_bilden(ctx) await _zyklen_brechen(ctx) _bausteine_schneiden(ctx) await _kapitel_bilden(ctx) def messen(ctx: llm.Kontext) -> list[dict]: befunde = [] atome = _atome(ctx.topic) for a in atome: if not a["ziel_id"] or not a["baustein_id"]: befunde.append({"art": "partition", "item": str(a["id"]), "detail": a["titel"]}) bausteine = db.query("SELECT * FROM bausteine WHERE topic=? ORDER BY ord", (ctx.topic,)) je_baustein: dict = defaultdict(list) for a in atome: je_baustein[a["baustein_id"]].append(a) # Band-Unterschreitung nur, wo ein Merge MÖGLICH gewesen wäre (kombiniert # ≤ Band) — dieselbe Bedingung wie im Schnitt; eine unerfüllbare Messlatte # pendelt nur die Note (Lektion 75). for b in bausteine: n = len(je_baustein[b["id"]]) partner = any(bb["id"] != b["id"] and len(je_baustein[bb["id"]]) + n <= BAUSTEIN_MAX_ATOME for bb in bausteine) if n > BAUSTEIN_MAX_ATOME or (n < BAUSTEIN_MIN_ATOME and partner): befunde.append({"art": "band", "item": str(b["id"]), "detail": f"{b['titel']}: {n} Atome"}) kaps = db.query("SELECT * FROM kapitel WHERE topic=?", (ctx.topic,)) kap_ids = {k["id"] for k in kaps} for b in bausteine: if b["kapitel_id"] not in kap_ids: befunde.append({"art": "baustein_ohne_kapitel", "item": str(b["id"]), "detail": b["titel"]}) belegt = {b["kapitel_id"] for b in bausteine} for k in kaps: if k["id"] not in belegt: befunde.append({"art": "kapitel_leer", "item": str(k["id"]), "detail": k["titel"]}) if k["art"] == "fallback": # stiller Fallback wäre unsichtbarer Qualitätsverlust befunde.append({"art": "kapitel_fallback", "item": str(k["id"]), "detail": k["titel"]}) ord_von = {b["id"]: b["ord"] for b in bausteine} for k in _braucht_kanten(ctx.topic): bv = next((a["baustein_id"] for a in atome if a["id"] == k["von_atom"]), None) bz = next((a["baustein_id"] for a in atome if a["id"] == k["zu_atom"]), None) if bv and bz and bv != bz and ord_von.get(bz, 0) > ord_von.get(bv, 0): befunde.append({"art": "vorwaerts_kante", "item": str(k["id"]), "detail": f"Atom {k['von_atom']} braucht {k['zu_atom']} (später)"}) ids = [a["id"] for a in atome] kanten = [(k["von_atom"], k["zu_atom"]) for k in _braucht_kanten(ctx.topic) if k["von_atom"] in ids and k["zu_atom"] in ids] if _finde_zyklus(ids, kanten): befunde.append({"art": "zyklus", "item": "", "detail": "braucht-Graph hat Zyklus"}) return befunde async def reparieren(ctx: llm.Kontext, befunde: list[dict]) -> bool: ctx.ebene = EBENE if not befunde: return False await _ziele_bilden(ctx) # fängt Partition-Lücken await _zyklen_brechen(ctx) # fängt Zyklen _bausteine_schneiden(ctx) # Neu-Schnitt fängt Band + Vorwärts-Kanten await _kapitel_bilden(ctx) # fängt Kapitel-Befunde return True