548 lines
28 KiB
Python
548 lines
28 KiB
Python
"""Ebene 1: Inventar. Atome aus dem Korpus extrahieren — mit Verbatim-Anker als
|
||
Identitätskern. Dedup dreistufig (Lektionen 18/25/31):
|
||
1. Anker-Überlappung → Auto-Merge (deterministisch, gleiche Quellstelle = gleiches Atom)
|
||
2. Embedding-Kandidaten auf DEFINITIONEN (nie Titel) + Negations-Guard
|
||
3. 2er-Judge-Panel, Merge nur EINSTIMMIG; sonst „verwandt"-Kante (Falsch-Merge >> Dublette)
|
||
Danach Soll-Zuordnung je Atom; Soll-Punkte ohne Atom lösen gezielte Nachextraktion aus."""
|
||
|
||
import asyncio
|
||
import logging
|
||
import re
|
||
|
||
import db
|
||
import embedding
|
||
import korpus
|
||
import llm
|
||
import textkit
|
||
from config import (ANKER_OVERLAP_MERGE, LUECKEN_RUNDEN_MAX,
|
||
MERGE_KANDIDAT_COS, MERGE_KANDIDAT_JACCARD,
|
||
MERGE_PANEL, READER_JE_ABSCHNITT, ZIELE_CHUNK_ATOME)
|
||
|
||
log = logging.getLogger("creator2.inventar")
|
||
|
||
EBENE = "inventar"
|
||
TYPEN = ("begriff", "aussage", "verfahren")
|
||
LEVELS = ("E", "M", "S")
|
||
PAAR_CHUNK = 10 # Merge-Paare pro Judge-Call (40 riss das Output-Cap: 58–71 % Parse-Fehler)
|
||
|
||
|
||
def aktive_atome(topic: str) -> list[dict]:
|
||
return db.query("SELECT * FROM atome WHERE topic=? AND status NOT IN"
|
||
" ('gemerged','verworfen') ORDER BY id", (topic,))
|
||
|
||
|
||
# ── Extraktion ────────────────────────────────────────────────────────────────
|
||
|
||
async def _extrahiere_quelle(ctx: llm.Kontext, quelle: dict, force: bool = False) -> None:
|
||
# Idempotenz-Guard über `atome_stand` (nicht „hat irgendein Anker-Atom"): dieser
|
||
# Merker wird erst NACH einem vollständigen Extraktions-Pass gesetzt. Bricht der
|
||
# Lauf mitten in der Extraktion ab (einige Reader haben schon eingefügt, Merker
|
||
# noch leer), liest der Resume die Quelle komplett neu — sonst gingen die Atome
|
||
# der ungelesenen Abschnitte still verloren. Doppelte fängt der Anker-Overlap-Merge.
|
||
# force=True (Repair quelle_unvollstaendig) liest ohnehin neu.
|
||
stand = db.one("SELECT atome_stand FROM quellen WHERE id=?", (quelle["id"],))
|
||
if stand and stand["atome_stand"] and not force:
|
||
db.update("quellen", "id", quelle["id"], status=stand["atome_stand"])
|
||
return
|
||
text = korpus.quelltext(quelle)
|
||
# aufgaben-Quellen: das GEÜBTE Konzept destillieren, nie die Aufgabeninstanz
|
||
template = ("Atom-Extraktion-Aufgaben" if quelle.get("rolle") == "aufgaben"
|
||
else "Atom-Extraktion")
|
||
fehl: list[int] = []
|
||
|
||
async def reader(offset: int, chunk: str, r: int, tiefe: int = 0) -> None:
|
||
res = await llm.call(ctx, stage="extraktion", template=template,
|
||
werte={"topic": ctx.topic, "quelle": quelle["titel"], "text": chunk},
|
||
role="quick",
|
||
item=f"q{quelle['id']}-o{offset}-r{r}" + "-h" * tiefe,
|
||
erwartet=list)
|
||
if res is None:
|
||
# Leerantwort/Parse-Fehler ist meist das Output-Cap (stop=max_tokens,
|
||
# deterministisch — stumpfes Wiederholen hilft nie): Chunk halbieren.
|
||
if tiefe < 2 and len(chunk) > 2000:
|
||
mitte = len(chunk) // 2
|
||
await llm.alle((reader(offset, chunk[:mitte], r, tiefe + 1),
|
||
reader(offset + mitte, chunk[mitte:], r, tiefe + 1)))
|
||
else:
|
||
fehl.append(offset) # nicht still: Quelle bleibt unvollständig
|
||
return
|
||
for a in res or []:
|
||
titel = str(a.get("titel", "")).strip()
|
||
definition = str(a.get("definition", "")).strip()
|
||
zitat = str(a.get("zitat", "")).strip()
|
||
typ = str(a.get("typ", "")).strip().lower()
|
||
level = str(a.get("level", "M")).strip().upper()
|
||
if not titel or not definition or typ not in TYPEN:
|
||
continue
|
||
span = textkit.finde_zitat(chunk, zitat)
|
||
atom_id = db.insert("atome", topic=ctx.topic, titel=titel, typ=typ,
|
||
definition=definition,
|
||
level=level if level in LEVELS else "M",
|
||
status="neu" if span else "ohne_anker")
|
||
if span:
|
||
db.insert("anker", atom_id=atom_id, quelle_id=quelle["id"],
|
||
start=offset + span[0], ende=offset + span[1], zitat=zitat)
|
||
elif zitat:
|
||
# Fehlgeschlagenes Zitat NICHT wegwerfen (start=-1 = unverankert):
|
||
# der Repair matcht es erst deterministisch neu, LLM nur als Fallback.
|
||
db.insert("anker", atom_id=atom_id, quelle_id=quelle["id"],
|
||
start=-1, ende=-1, zitat=zitat)
|
||
|
||
stuecke = textkit.abschnitte(text)
|
||
await llm.alle(reader(off, chunk, r)
|
||
for off, chunk in stuecke
|
||
for r in range(READER_JE_ABSCHNITT))
|
||
if fehl: # stiller Chunk-Verlust hieß bisher trotzdem „atome" (Lernen: 41 Calls leer)
|
||
log.warning("Quelle %s: %d Abschnitt(e) ohne Extraktion", quelle["id"], len(fehl))
|
||
endstand = "teilweise" if fehl else "atome"
|
||
# Merker erst JETZT setzen (nach vollständigem Pass) — er ist das Resume-Gate
|
||
db.update("quellen", "id", quelle["id"], status=endstand, atome_stand=endstand)
|
||
|
||
|
||
# ── Dedup ─────────────────────────────────────────────────────────────────────
|
||
|
||
def _wurzel(atom_id: int) -> int:
|
||
"""merged_into-Kette bis zum lebenden Atom folgen (Merge-Ketten im selben Lauf)."""
|
||
for _ in range(20):
|
||
row = db.one("SELECT merged_into FROM atome WHERE id=?", (atom_id,))
|
||
if not row or not row["merged_into"]:
|
||
return atom_id
|
||
atom_id = row["merged_into"]
|
||
return atom_id
|
||
|
||
|
||
def _merge(topic: str, gewinner: int, verlierer: int) -> None:
|
||
"""Anker wandern zum Gewinner; der Verlierer bleibt als gemerged sichtbar."""
|
||
gewinner, verlierer = _wurzel(gewinner), _wurzel(verlierer)
|
||
if gewinner == verlierer:
|
||
return
|
||
g = db.one("SELECT * FROM atome WHERE id=?", (gewinner,))
|
||
v = db.one("SELECT * FROM atome WHERE id=?", (verlierer,))
|
||
if not g or not v or g["status"] == "verworfen" or v["status"] == "verworfen":
|
||
return
|
||
db.execute("UPDATE anker SET atom_id=? WHERE atom_id=?", (gewinner, verlierer))
|
||
db.update("atome", "id", verlierer, status="gemerged", merged_into=gewinner)
|
||
# Kanten einzeln umhängen: ein blindes UPDATE kollidiert mit dem UNIQUE-Index,
|
||
# wenn der Gewinner dieselbe Kante schon hat; Selbstkanten entstehen beim
|
||
# Merge zweier verbundener Atome — beide Fälle werden gelöscht statt verschoben.
|
||
for k in db.query("SELECT * FROM kanten WHERE von_atom=? OR zu_atom=?",
|
||
(verlierer, verlierer)):
|
||
von = gewinner if k["von_atom"] == verlierer else k["von_atom"]
|
||
zu = gewinner if k["zu_atom"] == verlierer else k["zu_atom"]
|
||
doppel = db.one("SELECT id FROM kanten WHERE von_atom=? AND zu_atom=? AND art=?"
|
||
" AND id!=?", (von, zu, k["art"], k["id"]))
|
||
if von == zu or doppel:
|
||
db.execute("DELETE FROM kanten WHERE id=?", (k["id"],))
|
||
else:
|
||
db.execute("UPDATE kanten SET von_atom=?, zu_atom=? WHERE id=?",
|
||
(von, zu, k["id"]))
|
||
|
||
|
||
def _anker_dedup(topic: str) -> None:
|
||
"""Gleiche Quellstelle = gleiches Atom. Deterministisch, kein Judge (Lektion: Anker
|
||
ist die Identität). Gewinner = längere Definition."""
|
||
atome = aktive_atome(topic)
|
||
anker = {a["id"]: db.query("SELECT * FROM anker WHERE atom_id=? AND start>=0", (a["id"],))
|
||
for a in atome}
|
||
je_def = {a["id"]: a["definition"] for a in atome}
|
||
ids = [a["id"] for a in atome if anker[a["id"]]]
|
||
for i, a_id in enumerate(ids):
|
||
for b_id in ids[i + 1:]:
|
||
if db.one("SELECT status FROM atome WHERE id=?", (a_id,))["status"] == "gemerged":
|
||
break
|
||
if db.one("SELECT status FROM atome WHERE id=?", (b_id,))["status"] == "gemerged":
|
||
continue
|
||
passt = any(ka["quelle_id"] == kb["quelle_id"]
|
||
and textkit.ueberlappung((ka["start"], ka["ende"]),
|
||
(kb["start"], kb["ende"])) >= ANKER_OVERLAP_MERGE
|
||
for ka in anker[a_id] for kb in anker[b_id])
|
||
if passt:
|
||
gew, ver = (a_id, b_id) if len(je_def[a_id]) >= len(je_def[b_id]) else (b_id, a_id)
|
||
_merge(topic, gew, ver)
|
||
# Verlierer-Anker in-memory erben, sonst übersteht eine transitive
|
||
# Dublette (C überlappt nur mit B, B ging in A) diesen Pass.
|
||
anker[gew].extend(anker[ver])
|
||
|
||
|
||
def _geprueft(a: int, b: int) -> bool:
|
||
lo, hi = min(a, b), max(a, b)
|
||
return db.one("SELECT id FROM kanten WHERE von_atom=? AND zu_atom=? AND art='verwandt'",
|
||
(lo, hi)) is not None
|
||
|
||
|
||
async def _judge_dedup(ctx: llm.Kontext) -> None:
|
||
"""Kandidaten (Definitions-Ähnlichkeit ≥ Floor ODER gleicher Titel-Kern,
|
||
gleiche Negationsmenge) ans 2er-Panel. Einstimmig gleich → Merge. Alles
|
||
andere → „verwandt" (persistiert, damit dasselbe Paar nie zweimal Tokens
|
||
kostet). Titel-Kern-Paare sind deterministisch — Dubletten aus
|
||
verschiedenen Quellen haben disjunkte Anker und paraphrasierte
|
||
Definitionen, das Embedding allein übersah sie (aak Lauf 8: 26 Gruppen)."""
|
||
atome = aktive_atome(topic := ctx.topic)
|
||
if len(atome) < 2:
|
||
return
|
||
je_kern: dict[str, list[int]] = {}
|
||
for idx, a in enumerate(atome):
|
||
if kern := textkit.titel_kern(a["titel"]):
|
||
je_kern.setdefault(kern, []).append(idx)
|
||
titel_paare = [(g[x], g[y], 1.0) for g in je_kern.values() if len(g) > 1
|
||
for x in range(len(g)) for y in range(x + 1, len(g))]
|
||
paare = embedding.kandidaten_paare([a["definition"] for a in atome],
|
||
MERGE_KANDIDAT_COS, MERGE_KANDIDAT_JACCARD)
|
||
offen = []
|
||
gesehen: set[tuple[int, int]] = set()
|
||
for i, j, cos in titel_paare + paare:
|
||
a, b = atome[i], atome[j]
|
||
if (a["id"], b["id"]) in gesehen:
|
||
continue
|
||
gesehen.add((a["id"], b["id"]))
|
||
if textkit.negations_menge(a["definition"]) != textkit.negations_menge(b["definition"]):
|
||
continue # Antonyme messen 0.91–0.95 (Lektion 31)
|
||
if _geprueft(a["id"], b["id"]):
|
||
continue
|
||
offen.append((a, b))
|
||
if not offen:
|
||
return
|
||
|
||
async def chunk_urteil(chunk: list) -> None:
|
||
liste = "\n".join(
|
||
f"PAAR {n}:\nA(id {a['id']}): {a['titel']} — {a['definition']}\n"
|
||
f" Beleg A: {_erster_anker(a['id'])}\n"
|
||
f"B(id {b['id']}): {b['titel']} — {b['definition']}\n"
|
||
f" Beleg B: {_erster_anker(b['id'])}"
|
||
for n, (a, b) in enumerate(chunk, 1))
|
||
stimmen = await llm.panel(ctx, MERGE_PANEL, stage="merge", template="Atom-Merge",
|
||
werte={"paare": liste}, role="judge",
|
||
n=len(chunk), item=f"c{chunk[0][0]['id']}", erwartet=list)
|
||
for n, (a, b) in enumerate(chunk, 1):
|
||
urteile = []
|
||
for stimme in stimmen:
|
||
for e in stimme:
|
||
if isinstance(e, dict) and e.get("paar") == n:
|
||
urteile.append(bool(e.get("gleich")))
|
||
break
|
||
if len(urteile) < MERGE_PANEL:
|
||
continue # Panel unvollständig (Parse-/Infra-Ausfall): Paar bleibt
|
||
# offen für die nächste Runde — ein dauerhaftes „nein" ließ
|
||
# Dubletten systematisch überleben (aak: 42/73 Calls unparsbar)
|
||
if all(urteile):
|
||
gew, ver = ((a["id"], b["id"]) if len(a["definition"]) >= len(b["definition"])
|
||
else (b["id"], a["id"]))
|
||
_merge(topic, gew, ver)
|
||
else:
|
||
lo, hi = min(a["id"], b["id"]), max(a["id"], b["id"])
|
||
db.execute("INSERT OR IGNORE INTO kanten(topic, von_atom, zu_atom, art, status)"
|
||
" VALUES(?,?,?,'verwandt','nein')", (topic, lo, hi))
|
||
|
||
chunks = [offen[i:i + PAAR_CHUNK] for i in range(0, len(offen), PAAR_CHUNK)]
|
||
await llm.alle(chunk_urteil(c) for c in chunks)
|
||
|
||
|
||
def _erster_anker(atom_id: int) -> str:
|
||
row = db.one("SELECT zitat FROM anker WHERE atom_id=? AND start>=0 LIMIT 1", (atom_id,))
|
||
return (row["zitat"][:300] if row else "(kein Anker)")
|
||
|
||
|
||
# ── Soll-Zuordnung ────────────────────────────────────────────────────────────
|
||
|
||
async def _soll_zuordnen(ctx: llm.Kontext, nur_offene: bool = True) -> None:
|
||
punkte = db.query("SELECT * FROM soll WHERE topic=? AND status='bestaetigt'", (ctx.topic,))
|
||
if not punkte:
|
||
return
|
||
atome = [a for a in aktive_atome(ctx.topic) if not (nur_offene and a["soll_id"])]
|
||
soll_liste = "\n".join(f"{p['id']}: {p['punkt']}" for p in punkte)
|
||
gueltig = {p["id"] for p in punkte}
|
||
|
||
async def chunk_zuordnen(chunk: list) -> None:
|
||
liste = "\n".join(f"{a['id']}: {a['titel']} — {a['definition']}" for a in chunk)
|
||
res = await llm.call(ctx, stage="soll_zuordnung", template="Atom-Soll-Zuordnung",
|
||
werte={"soll": soll_liste, "atome": liste}, role="judge",
|
||
n=len(chunk), item=f"z{chunk[0]['id']}", erwartet=list)
|
||
je_atom = {e.get("atom"): e.get("soll") for e in res or [] if isinstance(e, dict)}
|
||
for a in chunk:
|
||
sid = je_atom.get(a["id"])
|
||
if sid in gueltig:
|
||
db.update("atome", "id", a["id"], soll_id=sid)
|
||
|
||
chunks = [atome[i:i + ZIELE_CHUNK_ATOME] for i in range(0, len(atome), ZIELE_CHUNK_ATOME)]
|
||
await llm.alle(chunk_zuordnen(c) for c in chunks)
|
||
|
||
|
||
async def bauen(ctx: llm.Kontext) -> None:
|
||
ctx.ebene = EBENE
|
||
offene = db.query("SELECT * FROM quellen WHERE topic=? AND status='extrahiert'", (ctx.topic,))
|
||
await llm.alle(_extrahiere_quelle(ctx, q) for q in offene)
|
||
_anker_dedup(ctx.topic)
|
||
await _judge_dedup(ctx)
|
||
await _soll_zuordnen(ctx)
|
||
|
||
|
||
# ── QA + Repair ───────────────────────────────────────────────────────────────
|
||
|
||
_TITEL_KATALOG = re.compile(r"^(Satz|Lemma|Korollar|Bemerkung|Definition|Aufgabe"
|
||
r"|Übung|Transformation|Theorem)\s+[IVX]*\d", re.IGNORECASE)
|
||
|
||
|
||
def _titel_kaputt(titel: str) -> bool:
|
||
"""Katalognummern-Titel (Satz 6.26 — der Leser hat kein nummeriertes Skript),
|
||
abgerissene Fragmente (aak: „Teste für alle V 0 ⊆ V mit |") und LaTeX-Markup
|
||
(Titel sind Identifikatoren — Board/Dedup/Verweise rendern kein KaTeX)."""
|
||
if _TITEL_KATALOG.match(titel):
|
||
return True
|
||
if "$" in titel or "\\" in titel:
|
||
return True
|
||
if titel.rstrip().endswith(("|", ",", ";", ":", "-", "–", "(", "{", "⊆", "=")):
|
||
return True
|
||
return titel.count("(") != titel.count(")")
|
||
|
||
|
||
def messen(ctx: llm.Kontext) -> list[dict]:
|
||
befunde = []
|
||
atome = aktive_atome(ctx.topic)
|
||
belegte_soll = set()
|
||
for a in atome:
|
||
if a["status"] == "ohne_anker" or not db.one(
|
||
"SELECT id FROM anker WHERE atom_id=? AND start>=0 LIMIT 1", (a["id"],)):
|
||
befunde.append({"art": "atom_ohne_anker", "item": str(a["id"]), "detail": a["titel"]})
|
||
if not a["soll_id"]:
|
||
befunde.append({"art": "atom_ohne_soll", "item": str(a["id"]), "detail": a["titel"]})
|
||
else:
|
||
belegte_soll.add(a["soll_id"])
|
||
if _titel_kaputt(a["titel"]):
|
||
befunde.append({"art": "titel_katalog", "item": str(a["id"]), "detail": a["titel"]})
|
||
for p in db.query("SELECT * FROM soll WHERE topic=? AND status='bestaetigt'", (ctx.topic,)):
|
||
if p["id"] not in belegte_soll:
|
||
befunde.append({"art": "soll_ohne_atom", "item": str(p["id"]), "detail": p["punkt"]})
|
||
for q in db.query("SELECT * FROM quellen WHERE topic=? AND status='teilweise'",
|
||
(ctx.topic,)):
|
||
befunde.append({"art": "quelle_unvollstaendig", "item": str(q["id"]),
|
||
"detail": q["titel"]})
|
||
return befunde
|
||
|
||
|
||
async def reparieren(ctx: llm.Kontext, befunde: list[dict]) -> bool:
|
||
ctx.ebene = EBENE
|
||
|
||
async def einer(b: dict) -> bool:
|
||
try:
|
||
item = int(b["item"])
|
||
except (TypeError, ValueError):
|
||
return False
|
||
if b["art"] == "atom_ohne_soll":
|
||
return await _soll_stichentscheid(ctx, item)
|
||
if b["art"] == "soll_ohne_atom":
|
||
return await _luecke_schliessen(ctx, item)
|
||
if b["art"] == "quelle_unvollstaendig":
|
||
q = db.one("SELECT * FROM quellen WHERE id=?", (item,))
|
||
if q:
|
||
vorher = db.one(
|
||
"SELECT COUNT(*) AS n FROM anker k JOIN atome a ON a.id=k.atom_id"
|
||
" WHERE a.topic=? AND k.quelle_id=? AND a.status NOT IN"
|
||
" ('gemerged','verworfen')", (ctx.topic, q["id"]))["n"]
|
||
await _extrahiere_quelle(ctx, q, force=True)
|
||
nachher = db.one(
|
||
"SELECT COUNT(*) AS n FROM anker k JOIN atome a ON a.id=k.atom_id"
|
||
" WHERE a.topic=? AND k.quelle_id=? AND a.status NOT IN"
|
||
" ('gemerged','verworfen')", (ctx.topic, q["id"]))["n"]
|
||
# ehrliches bewegt: blieb die Quelle 'teilweise' ohne neue Anker,
|
||
# erkennt auto_loop Stillstand statt 10 Token-Runden zu drehen
|
||
nq = db.one("SELECT status FROM quellen WHERE id=?", (q["id"],))
|
||
return nq["status"] == "atome" or nachher > vorher
|
||
return False
|
||
|
||
# Anker-Fixes gebündelt (Batch je Quelle statt Atom×Quelle — war 83 % der
|
||
# Lauf-Tokens); Rest PARALLEL (Semaphoren deckeln die echte Last).
|
||
anker_ids = [int(b["item"]) for b in befunde
|
||
if b["art"] == "atom_ohne_anker" and str(b["item"]).isdigit()]
|
||
bewegt = await _anker_fixen_batch(ctx, anker_ids) if anker_ids else False
|
||
titel_ids = [int(b["item"]) for b in befunde
|
||
if b["art"] == "titel_katalog" and str(b["item"]).isdigit()]
|
||
if titel_ids:
|
||
bewegt |= await _titel_fixen(ctx, titel_ids)
|
||
rest = [b for b in befunde if b["art"] not in ("atom_ohne_anker", "titel_katalog")]
|
||
bewegt |= any(await llm.alle(einer(b) for b in rest))
|
||
if bewegt:
|
||
_anker_dedup(ctx.topic)
|
||
await _judge_dedup(ctx)
|
||
await _soll_zuordnen(ctx)
|
||
return bewegt
|
||
|
||
|
||
async def _titel_fixen(ctx: llm.Kontext, atom_ids: list[int]) -> bool:
|
||
"""Kaputte Titel (Katalognummern, Fragmente) aus der Definition neu benennen."""
|
||
atome = [a for i in atom_ids if (a := db.one("SELECT * FROM atome WHERE id=?", (i,)))
|
||
and a["status"] not in ("gemerged", "verworfen")]
|
||
if not atome:
|
||
return False
|
||
liste = "\n".join(f"{a['id']}: {a['titel']} — {a['definition']}" for a in atome)
|
||
res = await llm.call(ctx, stage="titel_fix", template="Atom-Titel-Fix",
|
||
schritt="fix", werte={"atome": liste}, role="judge",
|
||
n=len(atome), item=f"t{atome[0]['id']}", erwartet=list)
|
||
gueltig = {a["id"] for a in atome}
|
||
bewegt = False
|
||
for e in res or []:
|
||
if isinstance(e, dict) and e.get("atom") in gueltig:
|
||
titel = str(e.get("titel", "")).strip()
|
||
if titel and not _titel_kaputt(titel):
|
||
db.update("atome", "id", e["atom"], titel=titel)
|
||
bewegt = True
|
||
return bewegt
|
||
|
||
|
||
ANKER_FIX_CHUNK = 20 # Atome pro Batch-Call (ein Quelltext-Abschnitt trägt viele Fixes)
|
||
|
||
|
||
async def _anker_fixen_batch(ctx: llm.Kontext, atom_ids: list[int]) -> bool:
|
||
"""Stufe 1 deterministisch (kostenlos): gespeicherte, unverankerte Zitate
|
||
(start=-1) mit der toleranten Suche gegen alle Quellen neu matchen.
|
||
Stufe 2 gebündelt: EIN Call je (Quelltext-Abschnitt × Atom-Chunk) statt
|
||
Atom×Quelle einzeln (war 83 % der Lauf-Tokens). Ohne Fund → verworfen —
|
||
ein Atom ohne Quellstelle verletzt das Kernprinzip."""
|
||
offen: dict[int, dict] = {}
|
||
for atom_id in atom_ids:
|
||
a = db.one("SELECT * FROM atome WHERE id=?", (atom_id,))
|
||
if a and a["status"] not in ("gemerged", "verworfen"):
|
||
offen[atom_id] = a
|
||
quellen = db.query("SELECT * FROM quellen WHERE topic=?", (ctx.topic,))
|
||
texte = {q["id"]: korpus.quelltext(q) for q in quellen}
|
||
start_ids = set(offen)
|
||
ausgefallen: set[int] = set() # Atome, deren zuständiger Call ausfiel (None)
|
||
|
||
for atom_id in list(offen):
|
||
for roh in db.query("SELECT * FROM anker WHERE atom_id=? AND start<0", (atom_id,)):
|
||
treffer = next(((qid, span) for qid, text in texte.items()
|
||
if (span := textkit.finde_zitat(text, roh["zitat"]))), None)
|
||
if treffer:
|
||
db.update("anker", "id", roh["id"], quelle_id=treffer[0],
|
||
start=treffer[1][0], ende=treffer[1][1])
|
||
db.update("atome", "id", atom_id, status="neu")
|
||
offen.pop(atom_id, None)
|
||
break
|
||
|
||
async def abschnitt_call(q: dict, offset: int, chunk_text: str, ids: list[int]) -> None:
|
||
liste = "\n".join(f"{i}: {offen[i]['titel']} — {offen[i]['definition']}"
|
||
for i in ids if i in offen)
|
||
if not liste:
|
||
return
|
||
res = await llm.call(ctx, stage="anker_fix", template="Atom-Anker-Fix-Batch",
|
||
schritt="fix", role="judge", n=len(ids),
|
||
item=f"q{q['id']}-o{offset}",
|
||
werte={"atome": liste, "text": chunk_text}, erwartet=list)
|
||
if res is None: # Call-Ausfall ≠ „keine Quellstelle" → nicht verwerfen
|
||
ausgefallen.update(i for i in ids if i in offen)
|
||
return
|
||
for e in res or []:
|
||
atom_id = e.get("atom")
|
||
zitat = str(e.get("zitat", "")).strip()
|
||
if atom_id not in offen or not zitat:
|
||
continue
|
||
span = textkit.finde_zitat(texte[q["id"]], zitat)
|
||
if span:
|
||
db.insert("anker", atom_id=atom_id, quelle_id=q["id"],
|
||
start=span[0], ende=span[1], zitat=zitat)
|
||
db.update("atome", "id", atom_id, status="neu")
|
||
offen.pop(atom_id, None)
|
||
|
||
for q in quellen:
|
||
if not offen:
|
||
break
|
||
for offset, chunk_text in textkit.abschnitte(texte[q["id"]], 40_000):
|
||
ids = list(offen)
|
||
if not ids:
|
||
break
|
||
await llm.alle(abschnitt_call(q, offset, chunk_text, ids[i:i + ANKER_FIX_CHUNK])
|
||
for i in range(0, len(ids), ANKER_FIX_CHUNK))
|
||
# Verwerfen nur, wo ALLE zuständigen Calls liefen (kein Ausfall) und keiner
|
||
# eine Quellstelle fand. Ausgefallene bleiben ohne_anker → nächste Runde erneut.
|
||
verworfen = [i for i in offen if i not in ausgefallen]
|
||
for atom_id in verworfen:
|
||
db.update("atome", "id", atom_id, status="verworfen")
|
||
verankert = start_ids - set(offen)
|
||
return bool(verankert or verworfen) # ehrliches „bewegt" (kein Ausfall-Dauerschleifen)
|
||
|
||
|
||
async def _soll_stichentscheid(ctx: llm.Kontext, atom_id: int) -> bool:
|
||
"""2er-Panel entscheidet: Punkt zuordnen oder fremd. Verwerfen (destruktiv) nur
|
||
einstimmig — im Zweifel behalten (Lektion 67), dann bekommt es den Mehrheits-Punkt."""
|
||
a = db.one("SELECT * FROM atome WHERE id=?", (atom_id,))
|
||
if not a or a["status"] in ("gemerged", "verworfen") or a["soll_id"]:
|
||
return False
|
||
punkte = db.query("SELECT * FROM soll WHERE topic=? AND status='bestaetigt'", (ctx.topic,))
|
||
soll_liste = "\n".join(f"{p['id']}: {p['punkt']}" for p in punkte)
|
||
stimmen = await llm.panel(ctx, 2, stage="soll_stich", template="Atom-Soll-Stich",
|
||
schritt="soll_zuordnung", role="judge", item=f"a{atom_id}",
|
||
werte={"titel": a["titel"], "definition": a["definition"],
|
||
"beleg": _erster_anker(atom_id), "soll": soll_liste},
|
||
erwartet=dict)
|
||
# Verwerfen ist destruktiv → nur bei vollzähligem „fremd"-Panel. Eine Stimme
|
||
# ohne soll-Feld ist KEIN fremd-Votum (zählte bisher fälschlich mit).
|
||
if llm.einstimmig(stimmen, 2, lambda s: (s.get("soll") == "fremd")
|
||
if "soll" in s else None):
|
||
db.update("atome", "id", atom_id, status="verworfen")
|
||
return True
|
||
gueltig = {p["id"] for p in punkte}
|
||
for s in stimmen: # Zuordnung ist nicht destruktiv → Mehrheit/erste gültige reicht
|
||
if s.get("soll") in gueltig:
|
||
db.update("atome", "id", atom_id, soll_id=s["soll"])
|
||
return True
|
||
return False
|
||
|
||
|
||
async def _luecke_schliessen(ctx: llm.Kontext, soll_id: int) -> bool:
|
||
"""Gezielte Nachextraktion an der Belegstelle; nach LUECKEN_RUNDEN_MAX entscheidet
|
||
ein 2er-Panel, ob der Punkt durch bestehende Atome abgedeckt ist (Freispruch,
|
||
persistiert — Lektion 69)."""
|
||
p = db.one("SELECT * FROM soll WHERE id=?", (soll_id,))
|
||
if not p or p["status"] != "bestaetigt":
|
||
return False
|
||
# bisherige ECHTE Nachextraktions-Runden: nur erfolgreiche Calls (status='ok';
|
||
# Parse-/Infra-Retries schreiben eigene Zeilen) und nur seit dem letzten
|
||
# Inventar-Reset (Alt-Läufe zählten sonst mit → zu früher Freispruch).
|
||
resets = db.uj(db.one("SELECT resets FROM topics WHERE name=?", (ctx.topic,))["resets"], {})
|
||
seit = resets.get("inventar", 0)
|
||
runden = db.one(
|
||
"SELECT COUNT(*) AS n FROM events e JOIN runs r ON e.run_id=r.id"
|
||
" WHERE r.topic=? AND e.stage='luecke' AND e.item=? AND e.status='ok' AND e.id>?",
|
||
(ctx.topic, f"s{soll_id}", seit))["n"]
|
||
if runden >= LUECKEN_RUNDEN_MAX:
|
||
atome = aktive_atome(ctx.topic)
|
||
liste = "\n".join(f"- {a['titel']}: {a['definition']}" for a in atome[:200])
|
||
stimmen = await llm.panel(ctx, 2, stage="luecke_stich", template="Soll-Abgedeckt",
|
||
schritt="soll_zuordnung", role="judge", item=f"s{soll_id}",
|
||
werte={"punkt": p["punkt"], "atome": liste}, erwartet=dict)
|
||
# Freispruch ist destruktiv fürs Soll → nur bei vollzähligem Panel
|
||
if llm.einstimmig(stimmen, 2, lambda s: bool(s.get("abgedeckt"))
|
||
if "abgedeckt" in s else None):
|
||
db.update("soll", "id", soll_id, status="abgedeckt")
|
||
return True
|
||
return False
|
||
belege = db.uj(p["belege"])
|
||
if not belege:
|
||
return False
|
||
q = db.one("SELECT * FROM quellen WHERE id=?", (belege[0]["quelle"],))
|
||
if not q:
|
||
return False
|
||
text = korpus.quelltext(q)
|
||
pos = textkit.finde_zitat(text, belege[0]["zitat"])
|
||
fenster = text[max(0, (pos[0] if pos else 0) - 6000):(pos[1] if pos else 0) + 6000]
|
||
res = await llm.call(ctx, stage="luecke", template="Atom-Luecke",
|
||
role="quick", item=f"s{soll_id}",
|
||
werte={"punkt": p["punkt"], "quelle": q["titel"], "text": fenster},
|
||
erwartet=list)
|
||
neu = False
|
||
for a in res or []:
|
||
titel = str(a.get("titel", "")).strip()
|
||
definition = str(a.get("definition", "")).strip()
|
||
zitat = str(a.get("zitat", "")).strip()
|
||
typ = str(a.get("typ", "")).strip().lower()
|
||
span = textkit.finde_zitat(fenster, zitat)
|
||
if not titel or not definition or typ not in TYPEN or span is None:
|
||
continue
|
||
offset = text.find(fenster)
|
||
atom_id = db.insert("atome", topic=ctx.topic, titel=titel, typ=typ,
|
||
definition=definition, level="M", status="neu",
|
||
soll_id=soll_id)
|
||
db.insert("anker", atom_id=atom_id, quelle_id=q["id"],
|
||
start=offset + span[0], ende=offset + span[1], zitat=zitat)
|
||
neu = True
|
||
return neu
|