"""Stufe 3: Lernziele je Soll-Punkt → Bausteine (Band 4-8) → Ordnung (Kapitel-Schnitt + Permutation, exakt validiert, deterministischer Fallback). Gate: exakte Partition + Coverage (jeder Soll-Punkt ≥1 Baustein).""" import math from . import config, db, engine, graph, llm, textkit def _atom_rang(atom_id: int) -> tuple: a = db.one("SELECT quelle_id, start FROM anker WHERE atom_id=? AND start>=0 " "ORDER BY quelle_id, start LIMIT 1", atom_id) return (a["quelle_id"], a["start"]) if a else (9999, 10**9) @engine.worker("struktur.lernziele", buendel=True) async def lernziele(task: dict) -> engine.Ergebnis: mitglieder = [task] + task.get("_buendel", []) teil, aktive, neue = {}, {}, [] # idx → (mitglied, soll_id, punkt, atome) idx = 0 for m in mitglieder: soll_id = db.uj(m["payload"], {}).get("soll_id") if soll_id: soll = db.one("SELECT * FROM soll WHERE id=?", soll_id) atome = db.query("SELECT * FROM atome WHERE soll_id=? AND " "status='aktiv'", soll_id) punkt = soll["punkt"] else: atome = db.query("SELECT * FROM atome WHERE topic=? AND " "status='aktiv' AND soll_id IS NULL", task["topic"]) punkt = config.TEXTE["sammel_punkt"] if not atome: teil[m["id"]] = "fertig" continue if db.one("SELECT id FROM lernziele WHERE topic=? AND " "COALESCE(soll_id,-1)=COALESCE(?,-1)", task["topic"], soll_id): teil[m["id"]] = "fertig" neue += _baustein_tasks(task["topic"], soll_id) continue idx += 1 aktive[idx] = (m, soll_id, punkt, atome) if not aktive: return engine.Ergebnis(daten={"skip": True}, teil_status=teil, neue_tasks=neue) antwort = await llm.call( run_id=task["run_id"], stufe="struktur", knoten="lernziele", item=task["item"], role="judge", n=sum(len(a) for (_, _, _, a) in aktive.values()), skill_namen=graph.skills_von(task["knoten"]), werte={"soll_punkte": "\n\n".join( f"=== SOLL {i} ===\nPUNKT: {punkt}\nATOME:\n" + "\n".join(f"[{a['id']}] {a['titel']}" for a in atome) for i, (_, _, punkt, atome) in aktive.items())}) if antwort is None: raise RuntimeError("lernziele ohne Ergebnis") je_soll: dict[int, list] = {} for block in textkit.bloecke(antwort, "ZIEL"): s = block.get("soll", "").strip() if s.isdigit(): je_soll.setdefault(int(s), []).append(block) for i, (m, soll_id, punkt, atome) in aktive.items(): if i not in je_soll: teil[m["id"]] = "neu" # Soll-Punkt fehlt in der Antwort → neuer Versuch continue ids_alle = {a["id"] for a in atome} zugeordnet: set[int] = set() for block in je_soll[i]: atom_ids = [int(x) for x in block.get("atome", "").replace(" ", "").split(",") if x.isdigit() and int(x) in ids_alle - zugeordnet] if not atom_ids: continue ziel_id = db.insert("lernziele", topic=task["topic"], titel=block.get("titel", "")[:80], text=block.get("text", "")[:300], soll_id=soll_id) for aid in atom_ids: db.update("atome", "id=?", (aid,), ziel_id=ziel_id) zugeordnet |= set(atom_ids) rest = ids_alle - zugeordnet # deterministisches Sammelziel — nichts verlieren if rest: ziel_id = db.insert( "lernziele", topic=task["topic"], titel=config.TEXTE["sammel_titel"], soll_id=soll_id, text=config.TEXTE["sammel_ziel"].format(punkt=punkt)) for aid in rest: db.update("atome", "id=?", (aid,), ziel_id=ziel_id) teil[m["id"]] = "fertig" neue += _baustein_tasks(task["topic"], soll_id) return engine.Ergebnis(daten={"ziele": True}, teil_status=teil, neue_tasks=neue) def _baustein_tasks(topic: str, soll_id) -> list: return [{"knoten": "bausteine", "item": f"ziel:{z['id']}", "payload": {"ziel_id": z["id"]}} for z in db.query("SELECT id FROM lernziele WHERE topic=? AND " "COALESCE(soll_id,-1)=COALESCE(?,-1)", topic, soll_id)] @engine.worker("struktur.bausteine") async def bausteine(task: dict) -> engine.Ergebnis: payload = db.uj(task["payload"], {}) ziel = db.one("SELECT * FROM lernziele WHERE id=?", payload["ziel_id"]) atome = db.query("SELECT * FROM atome WHERE ziel_id=? AND status='aktiv'", ziel["id"]) if not atome: return engine.Ergebnis(daten={"skip": True}) if db.one("SELECT id FROM bausteine WHERE ziel_id=?", ziel["id"]): return engine.Ergebnis(daten={"skip": True}, neue_tasks=[{"knoten": "ordnung", "item": "ordnung"}]) atome.sort(key=lambda a: _atom_rang(a["id"])) gruppen: list[list[dict]] if len(atome) <= config.BAUSTEIN_MAX_ATOME: gruppen = [atome] else: antwort = await llm.call( run_id=task["run_id"], stufe="struktur", knoten="bausteine", item=task["item"], role="judge", n=len(atome), skill_namen=graph.skills_von(task["knoten"]), werte={"ziel": ziel["text"], "band": f"{config.BAUSTEIN_MIN_ATOME}-{config.BAUSTEIN_MAX_ATOME}", "atome": "\n".join(f"[{a['id']}] {a['titel']}" for a in atome)}) if antwort is None: # Call-Ausfall: neuer Versuch statt stiller Fallback raise RuntimeError("bausteine ohne Ergebnis") gruppen = _min_erzwingen(_gruppen_validieren(antwort, atome)) for i, gruppe in enumerate(gruppen): titel = ziel["titel"] or gruppe[0]["titel"] if len(gruppen) > 1: titel = config.TEXTE["teil"].format(titel=titel, n=i + 1) bid = db.insert("bausteine", topic=task["topic"], ziel_id=ziel["id"], titel=titel[:120], ord=i) for j, a in enumerate(gruppe): db.update("atome", "id=?", (a["id"],), baustein_id=bid, ord=j) return engine.Ergebnis(daten={"bausteine": len(gruppen)}, neue_tasks=[{"knoten": "ordnung", "item": "ordnung"}]) def _min_erzwingen(gruppen: list[list[dict]]) -> list[list[dict]]: """Splitter-Bremse (Audit: 8 „Teil"-Sections mit 1-2 Atomen): Gruppen unter BAUSTEIN_MIN_ATOME mit dem kleineren Nachbarn mergen, solange das MAX hält. Quellreihenfolge bleibt erhalten.""" geaendert = True while geaendert and len(gruppen) > 1: geaendert = False for i, g in enumerate(gruppen): if len(g) >= config.BAUSTEIN_MIN_ATOME: continue nachbarn = sorted((j for j in (i - 1, i + 1) if 0 <= j < len(gruppen)), key=lambda j: len(gruppen[j])) for j in nachbarn: if len(g) + len(gruppen[j]) <= config.BAUSTEIN_MAX_ATOME: a, b = min(i, j), max(i, j) gruppen[a] = gruppen[a] + gruppen[b] del gruppen[b] geaendert = True break if geaendert: break return gruppen def _gruppen_validieren(antwort: str | None, atome: list[dict]) -> list[list[dict]]: """LLM-Split validieren (jede ID genau einmal); sonst deterministischer divmod-Split entlang der Quellreihenfolge.""" nach_id = {a["id"]: a for a in atome} if antwort: gruppen, gesehen = [], set() for block in textkit.bloecke(antwort, "BAUSTEIN"): ids = [int(i) for i in block.get("atome", "").replace(" ", "").split(",") if i.isdigit()] if ids and all(i in nach_id and i not in gesehen for i in ids): gruppen.append([nach_id[i] for i in ids]) gesehen |= set(ids) if gesehen == set(nach_id) and all( len(g) <= config.BAUSTEIN_MAX_ATOME for g in gruppen): return gruppen n_gruppen = math.ceil(len(atome) / config.BAUSTEIN_MAX_ATOME) basis, rest = divmod(len(atome), n_gruppen) gruppen, pos = [], 0 for i in range(n_gruppen): k = basis + (1 if i < rest else 0) gruppen.append(atome[pos:pos + k]) pos += k return gruppen def _ziel_kerne(ziele: list[dict]) -> list[set]: """Inhaltswort-Kerne aller Lernziele: Boilerplate wird GEMESSEN statt gelistet (Lektion 106 — die alte deutsche Floskel-Liste wäre in jeder anderen Sprache ein No-op). Stämme, die in der Mehrheit der Ziele vorkommen, sind Schablone („Der Lernende kann …") und fliegen raus.""" stamm_je_ziel = [textkit.stamm_tokens(z["text"]) for z in ziele] n = len(stamm_je_ziel) alle = set().union(*stamm_je_ziel) if stamm_je_ziel else set() boilerplate = {t for t in alle if sum(t in s for s in stamm_je_ziel) > max(2, n * 0.5)} return [{t for t in s if len(t) > 3} - boilerplate for s in stamm_je_ziel] def _ziele_dedup(topic: str) -> int: """Synonyme Lernziele deterministisch mergen (Audit: „einordnen" vs. „verorten" erzeugte Doppel-Bausteine → Writer-Echos). Nur Paare mit gleichem Soll-Punkt; Falsch-Merge ist harmlos: Atome bleiben erhalten, hängen nur am selben Ziel.""" ziele = db.query("SELECT * FROM lernziele WHERE topic=? ORDER BY id", topic) kerne = _ziel_kerne(ziele) merges = 0 for i, a in enumerate(ziele): if a.get("_weg"): continue for j, b in enumerate(ziele[i + 1:], i + 1): if b.get("_weg") or a["soll_id"] != b["soll_id"]: continue ka, kb = kerne[i], kerne[j] kern_jaccard = len(ka & kb) / len(ka | kb) if ka and kb else 0.0 if textkit.titel_kern(a["titel"] or a["text"]) == \ textkit.titel_kern(b["titel"] or b["text"]) or \ kern_jaccard >= config.ZIEL_DEDUP_JACCARD: db.update("atome", "ziel_id=?", (b["id"],), ziel_id=a["id"]) db.execute("DELETE FROM bausteine WHERE ziel_id=?", b["id"]) db.execute("DELETE FROM lernziele WHERE id=?", b["id"]) b["_weg"] = True merges += 1 if merges: # verwaiste Baustein-Zuordnungen lösen → Gate-Reparatur baut neu db.execute("UPDATE atome SET baustein_id=NULL WHERE topic=? AND " "baustein_id IS NOT NULL AND baustein_id NOT IN " "(SELECT id FROM bausteine)", topic) return merges @engine.worker("struktur.ordnung") async def ordnung(task: dict) -> engine.Ergebnis: """Barriere: Ziel-Dedup, dann Kapitel-Schnitt über Soll-Punkte + Baustein- Permutation je Kapitel. Prior = Median der Atom-Ränge; exakt validiert.""" topic = task["topic"] ziel_merges = _ziele_dedup(topic) soll = db.query("SELECT * FROM soll WHERE topic=? AND status='bestaetigt' " "ORDER BY id", topic) db.execute("DELETE FROM kapitel WHERE topic=?", topic) # idempotent neu kapitel_gruppen = await _kapitel_schnitt(task, soll) for i, (titel, punkt_ids) in enumerate(kapitel_gruppen): kid = db.insert("kapitel", topic=topic, titel=titel[:120], ord=i) for sid in punkt_ids: db.update("soll", "id=?", (sid,), kapitel_id=kid) # Bausteine → Kapitel (über Ziel→Soll); Sammel-Bausteine ans letzte Kapitel letztes = db.one("SELECT id FROM kapitel WHERE topic=? ORDER BY ord DESC " "LIMIT 1", topic) for b in db.query("SELECT b.*, z.soll_id sz FROM bausteine b JOIN lernziele z " "ON z.id=b.ziel_id WHERE b.topic=?", topic): kap = db.one("SELECT kapitel_id k FROM soll WHERE id=?", b["sz"]) \ if b["sz"] else None db.update("bausteine", "id=?", (b["id"],), kapitel_id=(kap or {}).get("k") or letztes["id"]) # Permutation je Kapitel (Prior: Median-Anker-Rang) for kap in db.query("SELECT * FROM kapitel WHERE topic=? ORDER BY ord", topic): bs = db.query("SELECT * FROM bausteine WHERE topic=? AND kapitel_id=?", topic, kap["id"]) prior = sorted(bs, key=lambda b: _baustein_rang(b["id"])) reihenfolge = prior if len(bs) > 1: ids = [b["id"] for b in prior] antwort = await llm.call( run_id=task["run_id"], stufe="struktur", knoten="ordnung", item=f"perm:kapitel:{kap['id']}", role="judge", n=len(bs), skill_namen=graph.skills_von(task["knoten"]), werte={"art": "Bausteine", "liste": "\n".join(f"[{b['id']}] {b['titel']}" for b in prior), "prior": ",".join(map(str, ids))}) perm = _permutation_validieren(antwort, ids) if perm: nach_id = {b["id"]: b for b in bs} reihenfolge = [nach_id[i] for i in perm] else: db.insert("befunde", run_id=task["run_id"], stufe="struktur", knoten="ordnung", art="ordnung_fallback", item=f"kapitel:{kap['id']}", status="freispruch", detail="Judge-Permutation ungültig, Prior verwendet") for i, b in enumerate(reihenfolge): db.update("bausteine", "id=?", (b["id"],), ord=i, status="geordnet") return engine.Ergebnis(daten={"kapitel": len(kapitel_gruppen), "ziel_merges": ziel_merges}) async def _kapitel_schnitt(task: dict, soll: list[dict]) -> list[tuple[str, list[int]]]: thema = db.one("SELECT COALESCE(NULLIF(titel,''), name) t FROM topics " "WHERE name=?", task["topic"])["t"] ids = [s["id"] for s in soll] if len(ids) <= config.KAPITEL_SOLL_PUNKTE: return [(thema, ids)] antwort = await llm.call( run_id=task["run_id"], stufe="struktur", knoten="ordnung", item="themen_schnitt", role="judge", n=len(ids), skill_namen=graph.skills_von(task["knoten"], extra="schnitt"), werte={"thema": thema, "richtwert": config.KAPITEL_SOLL_PUNKTE, "punkte": "\n".join(f"[{s['id']}] {s['punkt']}" for s in soll)}) if antwort: gruppen, gesehen = [], set() for block in textkit.bloecke(antwort, "KAPITEL"): pids = [int(i) for i in block.get("punkte", "").replace(" ", "").split(",") if i.isdigit() and int(i) in set(ids) - gesehen] if pids: gruppen.append((block.get("titel", "Kapitel"), pids)) gesehen |= set(pids) if gesehen == set(ids): # exakte Partition — sonst Fallback return gruppen db.insert("befunde", run_id=task["run_id"], stufe="struktur", knoten="ordnung", art="themen_fallback", item=task["topic"], status="freispruch", detail="Schnitt ungültig, √n-Fallback") n_kap = max(1, round(math.sqrt(len(ids)))) groesse = math.ceil(len(ids) / n_kap) return [(config.TEXTE["teil"].format(titel=thema, n=i + 1), ids[i * groesse:(i + 1) * groesse]) for i in range(n_kap) if ids[i * groesse:(i + 1) * groesse]] def _baustein_rang(baustein_id: int) -> tuple: raenge = sorted(_atom_rang(a["id"]) for a in db.query( "SELECT id FROM atome WHERE baustein_id=? AND status='aktiv'", baustein_id)) return raenge[len(raenge) // 2] if raenge else (9999, 10**9) def _permutation_validieren(antwort: str | None, ids: list[int]) -> list[int] | None: if not antwort: return None block = textkit.bloecke(antwort, "ORDNUNG") if not block: return None perm = [int(i) for i in block[0].get("ids", "").replace(" ", "").split(",") if i.isdigit()] return perm if sorted(perm) == sorted(ids) else None @engine.worker("struktur.gate") async def gate(task: dict) -> engine.Ergebnis: topic, runde = task["topic"], task["runde"] befunde, neue = [], [] ohne_baustein = db.query("SELECT id FROM atome WHERE topic=? AND " "status='aktiv' AND baustein_id IS NULL", topic) for a in ohne_baustein: befunde.append({"art": "partition", "item": f"atom:{a['id']}", "detail": "Atom ohne Baustein"}) zu_gross = db.query( "SELECT baustein_id b, COUNT(*) n FROM atome WHERE topic=? AND " "status='aktiv' GROUP BY baustein_id HAVING n>?", topic, config.BAUSTEIN_MAX_ATOME) for z in zu_gross: befunde.append({"art": "band", "item": f"baustein:{z['b']}", "detail": f"{z['n']} Atome"}) ziel = db.one("SELECT ziel_id z FROM bausteine WHERE id=?", z["b"]) db.execute("DELETE FROM bausteine WHERE id=?", z["b"]) neue.append({"knoten": "bausteine", "item": f"ziel:{ziel['z']}:r{runde + 1}", "art": "reparatur", "runde": runde + 1, "payload": {"ziel_id": ziel["z"]}}) # Coverage Stufe 3: jeder bestätigte Soll-Punkt (ohne Freispruch) ≥1 Baustein ungedeckt = db.query( "SELECT s.id FROM soll s WHERE s.topic=? AND s.status='bestaetigt' AND " "s.freispruch='' AND NOT EXISTS (SELECT 1 FROM atome a WHERE a.soll_id=s.id " "AND a.status='aktiv' AND a.baustein_id IS NOT NULL)", topic) for s in ungedeckt: befunde.append({"art": "soll_ohne_baustein", "item": f"soll:{s['id']}", "detail": ""}) # Kennzahl-Befund (blockiert nicht): Struktur aufgebläht? n_bausteine = db.one("SELECT COUNT(*) c FROM bausteine WHERE topic=?", topic)["c"] n_soll = db.one("SELECT COUNT(*) c FROM soll WHERE topic=? AND " "status='bestaetigt'", topic)["c"] if n_soll and n_bausteine > n_soll * 3: db.insert("befunde", run_id=task["run_id"], stufe="struktur", knoten="gate_struktur", art="struktur_aufgeblaeht", item=topic, status="freispruch", detail=f"{n_bausteine} Bausteine aus {n_soll} Soll-Punkten") if befunde: return engine.Ergebnis(gate_status="rot", befunde=befunde, neue_tasks=neue) db.update("topics", "name=?", (topic,), status="struktur_fertig") neue = [] for b in db.query("SELECT id FROM bausteine WHERE topic=?", topic): db.insert("sections", ignore=True, baustein_id=b["id"]) neue.append({"knoten": "writer", "item": f"r1:baustein:{b['id']}", "payload": {"baustein_id": b["id"]}}) for k in db.query("SELECT id FROM kapitel WHERE topic=?", topic): neue.append({"knoten": "kapitel_intro", "item": f"kapitel:{k['id']}", "payload": {"kapitel_id": k["id"]}}) return engine.Ergebnis(gate_status="gruen", neue_tasks=neue)