425 lines
20 KiB
Python
425 lines
20 KiB
Python
"""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
|