379 lines
17 KiB
Python
379 lines
17 KiB
Python
"""Ebene 0: Korpus & Soll. Selbst-verifizierend, null Eingriffspunkte.
|
|
|
|
Themen: je Runde ein Recherche-Agent pro Lens (Curriculum/Lehrbuch/Syllabus/Übungen)
|
|
→ Quellen-Snapshots lokal. Uni: Dateien aus topics/<name>/ werden die Quellen.
|
|
Danach Soll-Extraktion je Quelle, Konsens deterministisch im Code (thema: Punkt
|
|
bestätigt ab Belegen aus ≥2 unabhängigen Quellen — Lektion 24; uni: 1 Beleg,
|
|
das Zitat-Gate reicht), Runden bis Sättigung."""
|
|
|
|
import asyncio
|
|
import hashlib
|
|
import logging
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import unicodedata
|
|
from pathlib import Path
|
|
|
|
import db
|
|
import llm
|
|
import textkit
|
|
from config import (KORPUS_DIR, RECHERCHE_LENSES, RECHERCHE_RUNDEN_MAX,
|
|
SOLL_CHUNK_CHARS, SOLL_MIN_BELEGE, SOLL_PUNKTE_MIN,
|
|
SOLL_PUNKTE_PER_SQRT, TOPICS_DIR)
|
|
|
|
log = logging.getLogger("creator2.korpus")
|
|
|
|
EBENE = "korpus"
|
|
|
|
|
|
# Quellen-Rolle: aufgaben-Quellen liefern keine eigenen Konzept-Instanzen als Atome
|
|
# (Lektion aak: 76 % der Atome kamen aus Aufgabensammlungen — der Guide dozierte
|
|
# Aufgabentexte). Heuristik am Import; generische Muster, keine Domänen-Regeln.
|
|
_AUFGABEN_MUSTER = re.compile(
|
|
r"aufgab|klausur|serie|übung|uebung|präsenz|praesenz|blatt|exercise|exam|sheet|homework",
|
|
re.IGNORECASE)
|
|
|
|
|
|
def _rolle(titel: str, lens: str | None = None) -> str:
|
|
if lens == "uebungen":
|
|
return "aufgaben"
|
|
return "aufgaben" if _AUFGABEN_MUSTER.search(titel) else "stoff"
|
|
|
|
|
|
def quelltext(quelle: dict) -> str:
|
|
try:
|
|
return Path(quelle["snapshot"]).read_text(encoding="utf-8")
|
|
except OSError:
|
|
return ""
|
|
|
|
|
|
# C0/C1-Kontrollzeichen außer \n und \t: PDF-Schriften mappen Sonderglyphen (ε) auf
|
|
# Steuerbytes — der Reader echot sie als Müll („ž") und das Zitat wird unmatchbar.
|
|
# Ersetzung durch LEERZEICHEN muss VOR ftfy laufen: ftfy löscht Steuerzeichen ersatzlos
|
|
# und verklebt sonst die Nachbarwörter.
|
|
_KONTROLLZEICHEN = re.compile(r"[\x00-\x08\x0b-\x1f\x7f-\x9f]")
|
|
|
|
|
|
def _text_reparieren(inhalt: str) -> str:
|
|
"""Byte-Schicht am Import heilen (Lektionen 82/83): Steuerzeichen→Leerzeichen,
|
|
dann ftfy (Mojibake, Ligaturen fi→fi, C1, NFC). Ohne ftfy → NFC-Fallback —
|
|
gleiches Muster wie der sentence-transformers-Fallback."""
|
|
inhalt = _KONTROLLZEICHEN.sub(" ", inhalt)
|
|
try:
|
|
import ftfy
|
|
return ftfy.fix_text(inhalt, normalization="NFC")
|
|
except ImportError:
|
|
return unicodedata.normalize("NFC", inhalt)
|
|
|
|
|
|
def _snapshot_schreiben(topic: str, inhalt: str) -> tuple[str, str] | None:
|
|
"""→ (pfad, hash) oder None bei Duplikat (gleicher Inhalt schon registriert).
|
|
Text-Reparatur am Import (Lektion 82/83): pdftotext liefert dekomponierte
|
|
Umlaute, Steuerbytes und Ligaturen — ohne Bereinigung scheitert die
|
|
Zitat-Suche an unsichtbar anderen Bytes."""
|
|
inhalt = _text_reparieren(inhalt)
|
|
h = hashlib.sha256(inhalt.encode()).hexdigest()[:16]
|
|
if db.one("SELECT id FROM quellen WHERE topic=? AND hash=?", (topic, h)):
|
|
return None
|
|
ordner = KORPUS_DIR / topic
|
|
ordner.mkdir(parents=True, exist_ok=True)
|
|
pfad = ordner / f"q-{h}.md"
|
|
pfad.write_text(inhalt, encoding="utf-8")
|
|
return str(pfad), h
|
|
|
|
|
|
def _datei_lesen(pfad: Path) -> str:
|
|
"""txt/md direkt; PDF via pdftotext (treu, strukturarm — Lektion 38)."""
|
|
if pfad.suffix.lower() == ".pdf":
|
|
if shutil.which("pdftotext") is None:
|
|
log.warning("pdftotext fehlt — %s übersprungen", pfad.name)
|
|
return ""
|
|
res = subprocess.run(["pdftotext", "-layout", str(pfad), "-"],
|
|
capture_output=True, text=True, timeout=120)
|
|
return res.stdout if res.returncode == 0 else ""
|
|
try:
|
|
return pfad.read_text(encoding="utf-8", errors="replace")
|
|
except OSError:
|
|
return ""
|
|
|
|
|
|
async def _uni_quellen(ctx: llm.Kontext) -> int:
|
|
"""Dateien aus topics/<name>/ als Quellen registrieren. → Anzahl neu."""
|
|
ordner = TOPICS_DIR / ctx.topic
|
|
neu = 0
|
|
for pfad in sorted(ordner.glob("*")) if ordner.is_dir() else []:
|
|
if pfad.suffix.lower() not in (".txt", ".md", ".pdf"):
|
|
continue
|
|
# .pdf überspringen, wenn eine gleichnamige .txt daneben liegt (vorkonvertiert)
|
|
if pfad.suffix.lower() == ".pdf" and pfad.with_suffix(".txt").exists():
|
|
continue
|
|
inhalt = _datei_lesen(pfad)
|
|
if not inhalt.strip():
|
|
continue
|
|
snap = _snapshot_schreiben(ctx.topic, inhalt)
|
|
if snap is None:
|
|
continue
|
|
db.insert("quellen", topic=ctx.topic, art="datei", titel=pfad.name,
|
|
snapshot=snap[0], hash=snap[1], status="neu", rolle=_rolle(pfad.name))
|
|
neu += 1
|
|
return neu
|
|
|
|
|
|
async def _recherche_runde(ctx: llm.Kontext, runde: int) -> int:
|
|
"""Ein full-Agent je Lens. → Anzahl neuer Quellen."""
|
|
async def eine(lens: str) -> list[dict]:
|
|
res = await llm.call(ctx, stage="recherche", template="Korpus-Recherche",
|
|
werte={"topic": ctx.topic, "lens": lens},
|
|
role="quick", caps="full", item=f"r{runde}-{lens}",
|
|
erwartet=list)
|
|
return res or []
|
|
|
|
ergebnisse = await asyncio.gather(*(eine(lens) for lens in RECHERCHE_LENSES))
|
|
neu = 0
|
|
for lens, quellen in zip(RECHERCHE_LENSES, ergebnisse):
|
|
for q in quellen:
|
|
inhalt = str(q.get("inhalt", "")).strip()
|
|
if len(inhalt) < 500: # leere/dünne Funde sind keine Quelle
|
|
continue
|
|
snap = _snapshot_schreiben(ctx.topic, inhalt)
|
|
if snap is None:
|
|
continue
|
|
titel = str(q.get("titel", ""))[:200]
|
|
db.insert("quellen", topic=ctx.topic, art="web", titel=titel,
|
|
url=str(q.get("url", ""))[:500], snapshot=snap[0], hash=snap[1],
|
|
runde=runde, status="neu", rolle=_rolle(titel, lens))
|
|
neu += 1
|
|
return neu
|
|
|
|
|
|
async def _soll_extrahieren(ctx: llm.Kontext) -> None:
|
|
"""Je neue Quelle Soll-Kandidaten ziehen (Quelltext INLINE — Lektion 48)."""
|
|
offene = db.query("SELECT * FROM quellen WHERE topic=? AND status='neu'", (ctx.topic,))
|
|
|
|
async def eine(q: dict) -> None:
|
|
text = quelltext(q)
|
|
for i, (_, chunk) in enumerate(textkit.abschnitte(text, SOLL_CHUNK_CHARS)):
|
|
res = await llm.call(ctx, stage="soll", template="Korpus-Soll",
|
|
werte={"topic": ctx.topic, "quelle": q["titel"], "text": chunk},
|
|
role="judge", item=f"q{q['id']}-{i}", erwartet=list)
|
|
for kand in res or []:
|
|
punkt = str(kand.get("punkt", "")).strip()
|
|
zitat = str(kand.get("zitat", "")).strip()
|
|
if not punkt or textkit.finde_zitat(chunk, zitat) is None:
|
|
continue # Beleg muss wörtlich in der Quelle stehen
|
|
db.insert("soll", topic=ctx.topic, punkt=punkt, status="kandidat",
|
|
belege=db.j([{"quelle": q["id"], "zitat": zitat}]))
|
|
db.update("quellen", "id", q["id"], status="extrahiert")
|
|
|
|
await asyncio.gather(*(eine(q) for q in offene))
|
|
|
|
|
|
def _min_belege(topic: str) -> int:
|
|
"""uni: Korpus ist vertrauenswürdig und das Zitat-Gate fängt Halluzinationen —
|
|
1 Beleg genügt (die ≥2-Regel warf Skript-only-Stoff weg, aak: Definition P,
|
|
FPTAS, Cook-Levin). thema: ≥2 unabhängige Quellen, weil eine einzelne
|
|
Web-Quelle selbst falsch sein kann."""
|
|
if db.one("SELECT art FROM topics WHERE name=?", (topic,))["art"] == "uni":
|
|
return 1
|
|
n = db.one("SELECT COUNT(*) AS n FROM quellen WHERE topic=?", (topic,))["n"]
|
|
return min(SOLL_MIN_BELEGE, max(1, n))
|
|
|
|
|
|
async def _konsens(ctx: llm.Kontext) -> int:
|
|
"""Kandidaten zu Punkten falten. Der Judge GRUPPIERT nur — die Zählung
|
|
(≥ min unabhängige Quellen) macht der Code. Untergrenze k = c·√(Kandidaten):
|
|
ohne sie faltete der Konsens 254 Atome auf 7 Punkte (aak) — zu grob als
|
|
Kapitel-Ebene. Nach oben offen — eine Obergrenze drückte den Judge dazu,
|
|
Kandidaten still wegzulassen (aak: LPT-Scheduling fehlte im Soll).
|
|
Vollständigkeits-Invariante: jede id wird zugeordnet oder explizit
|
|
abgelehnt; Übrige bekommen Nachrunden. → Anzahl bestätigt."""
|
|
kandidaten = db.query("SELECT * FROM soll WHERE topic=? AND"
|
|
" status IN ('kandidat','gefaltet')", (ctx.topic,))
|
|
if not kandidaten:
|
|
return 0
|
|
je_id = {k["id"]: k for k in kandidaten}
|
|
k_soll = max(SOLL_PUNKTE_MIN, round(SOLL_PUNKTE_PER_SQRT * len(kandidaten) ** 0.5))
|
|
band = str(max(3, int(0.7 * k_soll)))
|
|
min_belege = _min_belege(ctx.topic)
|
|
|
|
async def falten(kand: list[dict], hinweis: str, tag: str) -> list:
|
|
liste = "\n".join(f"{k['id']}: {k['punkt']}" for k in kand)
|
|
res = await llm.call(ctx, stage="soll_konsens", template="Korpus-Soll-Konsens",
|
|
werte={"topic": ctx.topic, "kandidaten": liste,
|
|
"band": band, "hinweis": hinweis},
|
|
role="judge", n=len(kand), item=tag, erwartet=list)
|
|
return res or []
|
|
|
|
punkte: list[dict] = [] # {"punkt": str, "ids": [int]}
|
|
je_punkt: dict[str, dict] = {}
|
|
zugeordnet: set[int] = set()
|
|
abgelehnt: set[int] = set()
|
|
|
|
def einsortieren(res: list) -> None:
|
|
for gruppe in res:
|
|
if not isinstance(gruppe, dict):
|
|
continue
|
|
ids = [i for i in gruppe.get("kandidaten", []) if isinstance(i, int)
|
|
and i in je_id and i not in zugeordnet and i not in abgelehnt]
|
|
if not ids:
|
|
continue
|
|
if gruppe.get("abgelehnt"):
|
|
abgelehnt.update(ids)
|
|
continue
|
|
zugeordnet.update(ids)
|
|
punkt = str(gruppe.get("punkt", "")).strip() or je_id[ids[0]]["punkt"]
|
|
kern = punkt.casefold()
|
|
if kern in je_punkt:
|
|
je_punkt[kern]["ids"].extend(ids)
|
|
else:
|
|
je_punkt[kern] = {"punkt": punkt, "ids": ids}
|
|
punkte.append(je_punkt[kern])
|
|
|
|
res = await falten(kandidaten, "", "k1")
|
|
gruppen = sum(1 for g in res if isinstance(g, dict) and not g.get("abgelehnt"))
|
|
if gruppen < 0.5 * k_soll and len(kandidaten) >= 2 * SOLL_PUNKTE_MIN:
|
|
nochmal = await falten(
|
|
kandidaten,
|
|
f"ACHTUNG: Der letzte Versuch hat zu grob gruppiert ({gruppen} Punkte)."
|
|
f" Trenne feiner — verschiedene Teilthemen NICHT zusammenfassen.", "k2")
|
|
if len(nochmal) > len(res):
|
|
res = nochmal
|
|
einsortieren(res)
|
|
|
|
for runde in (1, 2): # Vollständigkeit: fehlende ids gezielt nachfassen
|
|
fehlend = [k for i, k in je_id.items()
|
|
if i not in zugeordnet and i not in abgelehnt]
|
|
if not fehlend:
|
|
break
|
|
vorhanden = "\n".join(f"- {p['punkt']}" for p in punkte)
|
|
einsortieren(await falten(
|
|
fehlend,
|
|
"NACHRUNDE — diese Kandidaten sind noch keinem Punkt zugeordnet."
|
|
" Ordne JEDEN zu: an einen BESTEHENDEN Punkt (punkt wörtlich"
|
|
" wiederverwenden) oder als neue Gruppe. Ablehnen nur, wenn es kein"
|
|
" lernbares Teilthema ist.\nBESTEHENDE PUNKTE:\n" + vorhanden,
|
|
f"n{runde}"))
|
|
|
|
bestaetigt = []
|
|
for p in punkte:
|
|
belege = [b for i in p["ids"] for b in db.uj(je_id[i]["belege"])]
|
|
if len({b["quelle"] for b in belege}) >= min_belege:
|
|
bestaetigt.append((p, belege))
|
|
for r in db.query("SELECT id FROM soll WHERE topic=? AND status='bestaetigt'", (ctx.topic,)):
|
|
db.execute("DELETE FROM soll WHERE id=?", (r["id"],)) # idempotent: neu aufbauen
|
|
for p, belege in bestaetigt:
|
|
db.insert("soll", topic=ctx.topic, punkt=p["punkt"], status="bestaetigt",
|
|
belege=db.j(belege))
|
|
ok_ids = {i for p, _ in bestaetigt for i in p["ids"]}
|
|
for i, k in je_id.items(): # Buchführung: kein Kandidat verschwindet still
|
|
status = ("abgelehnt" if i in abgelehnt else
|
|
"gefaltet" if i in ok_ids else "kandidat")
|
|
if status != k["status"]:
|
|
db.update("soll", "id", i, status=status)
|
|
return len(bestaetigt)
|
|
|
|
|
|
def _rollen_auffrischen(topic: str) -> None:
|
|
"""Rolle für Bestands-Quellen nachziehen (Reset behält die Zeilen; die Heuristik
|
|
kam ggf. erst später dazu). Nur Hochstufung stoff→aufgaben — eine Lens-basierte
|
|
aufgaben-Rolle wird nie zurückgestuft."""
|
|
for q in db.query("SELECT * FROM quellen WHERE topic=? AND rolle='stoff'", (topic,)):
|
|
if _rolle(q["titel"]) == "aufgaben":
|
|
db.update("quellen", "id", q["id"], rolle="aufgaben")
|
|
|
|
|
|
async def bauen(ctx: llm.Kontext) -> None:
|
|
ctx.ebene = EBENE
|
|
_rollen_auffrischen(ctx.topic)
|
|
topic = db.one("SELECT * FROM topics WHERE name=?", (ctx.topic,))
|
|
if topic["status"] not in ("neu", "korpus"):
|
|
return # Ebene fertig — Resume überspringt
|
|
db.update("topics", "name", ctx.topic, status="korpus")
|
|
if topic["art"] == "uni":
|
|
await _uni_quellen(ctx)
|
|
await _soll_extrahieren(ctx)
|
|
await _konsens(ctx)
|
|
else:
|
|
vorher = -1
|
|
for runde in range(1, RECHERCHE_RUNDEN_MAX + 1):
|
|
await _recherche_runde(ctx, runde)
|
|
await _soll_extrahieren(ctx)
|
|
jetzt = await _konsens(ctx)
|
|
if jetzt <= vorher >= 0: # Sättigung: Runde ohne neuen bestätigten Punkt
|
|
break
|
|
vorher = jetzt
|
|
db.update("topics", "name", ctx.topic, status="korpus_fertig") # Auto-Freeze
|
|
|
|
|
|
# ── QA + Repair ───────────────────────────────────────────────────────────────
|
|
|
|
def messen(ctx: llm.Kontext) -> list[dict]:
|
|
"""Invarianten, deterministisch. → Befunde [{art, item, detail}]."""
|
|
befunde = []
|
|
n_quellen = db.one("SELECT COUNT(*) AS n FROM quellen WHERE topic=?", (ctx.topic,))["n"]
|
|
if n_quellen == 0:
|
|
befunde.append({"art": "korpus_leer", "item": "", "detail": "keine Quellen"})
|
|
punkte = db.query("SELECT * FROM soll WHERE topic=? AND status='bestaetigt'", (ctx.topic,))
|
|
if not punkte and n_quellen:
|
|
befunde.append({"art": "soll_leer", "item": "", "detail": "kein bestätigter Soll-Punkt"})
|
|
min_belege = _min_belege(ctx.topic)
|
|
for p in punkte:
|
|
quellen = {b["quelle"] for b in db.uj(p["belege"])}
|
|
if len(quellen) < min_belege:
|
|
befunde.append({"art": "soll_wenig_belege", "item": str(p["id"]),
|
|
"detail": f"{p['punkt']}: {len(quellen)} < {min_belege}"})
|
|
if punkte: # Konsens lief: übrige Kandidaten sind unerledigt, kein stiller Verlust
|
|
for k in db.query("SELECT * FROM soll WHERE topic=? AND status='kandidat'",
|
|
(ctx.topic,)):
|
|
befunde.append({"art": "soll_kandidat_offen", "item": str(k["id"]),
|
|
"detail": k["punkt"]})
|
|
return befunde
|
|
|
|
|
|
async def _beleg_nachsuchen(ctx: llm.Kontext, soll_id: int) -> bool:
|
|
"""Gezielter Beleg-Judge in noch nicht zitierten Quellen; ein bestätigter
|
|
Punkt ohne Fund wird zum Kandidaten zurückgestuft (Konsens-Regel bleibt
|
|
hart). → True nur, wenn sich etwas geändert hat."""
|
|
p = db.one("SELECT * FROM soll WHERE id=?", (soll_id,))
|
|
if not p:
|
|
return False
|
|
belegte = {bl["quelle"] for bl in db.uj(p["belege"])}
|
|
andere = [q for q in db.query("SELECT * FROM quellen WHERE topic=?", (ctx.topic,))
|
|
if q["id"] not in belegte]
|
|
for q in andere:
|
|
res = await llm.call(ctx, stage="soll_beleg", template="Korpus-Soll-Beleg",
|
|
schritt="soll", role="judge", item=f"s{p['id']}-q{q['id']}",
|
|
werte={"punkt": p["punkt"],
|
|
"text": quelltext(q)[:SOLL_CHUNK_CHARS]},
|
|
erwartet=dict)
|
|
zitat = str((res or {}).get("zitat", "")).strip()
|
|
if zitat and textkit.finde_zitat(quelltext(q), zitat) is not None:
|
|
belege = db.uj(p["belege"]) + [{"quelle": q["id"], "zitat": zitat}]
|
|
db.update("soll", "id", p["id"], belege=db.j(belege))
|
|
return True
|
|
if p["status"] == "bestaetigt":
|
|
db.update("soll", "id", p["id"], status="kandidat")
|
|
return True
|
|
return False
|
|
|
|
|
|
async def reparieren(ctx: llm.Kontext, befunde: list[dict]) -> bool:
|
|
ctx.ebene = EBENE
|
|
punkte = [int(b["item"]) for b in befunde if b["art"] == "soll_wenig_belege"]
|
|
offene = [int(b["item"]) for b in befunde if b["art"] == "soll_kandidat_offen"]
|
|
bewegt = any(await llm.alle(_beleg_nachsuchen(ctx, s) for s in punkte + offene))
|
|
if offene: # neu falten: die Nachrunden im Konsens erzwingen die Zuordnung
|
|
await _konsens(ctx)
|
|
noch = db.query("SELECT id FROM soll WHERE topic=? AND status='kandidat'",
|
|
(ctx.topic,))
|
|
bewegt = bewegt or len(noch) < len(offene)
|
|
if any(b["art"] in ("korpus_leer", "soll_leer") for b in befunde):
|
|
topic = db.one("SELECT art FROM topics WHERE name=?", (ctx.topic,))
|
|
if topic["art"] == "thema":
|
|
await _recherche_runde(ctx, runde=99)
|
|
await _soll_extrahieren(ctx)
|
|
await _konsens(ctx)
|
|
bewegt = True
|
|
return bewegt
|