From 28d0b494cd2002ba735baec531318dc70dba96ad Mon Sep 17 00:00:00 2001 From: team3 Date: Mon, 22 Jun 2026 22:21:23 +0200 Subject: [PATCH] update --- backend/bausteine.py | 669 +++++++++++++++------- backend/database.py | 369 ++++++++++++ backend/guide.py | 31 +- backend/routes.py | 18 +- templates/Prompt/Bausteine-Klaerung.md | 17 + templates/Prompt/Bausteine-Recherche.md | 2 + templates/Prompt/Subbaustein-Recherche.md | 1 + 7 files changed, 875 insertions(+), 232 deletions(-) create mode 100644 templates/Prompt/Bausteine-Klaerung.md diff --git a/backend/bausteine.py b/backend/bausteine.py index 1bf1035..053148a 100644 --- a/backend/bausteine.py +++ b/backend/bausteine.py @@ -15,8 +15,10 @@ import logging import math import shutil import subprocess +import time from pathlib import Path +import database as db from agents import kill_process, cancel_scope, clear_scope from config import KONSENS_GRACE, RECHERCHE_GRACE, KONSENS_MAX_RUNDEN, DEFAULT_PROVIDER from fsutil import atomic_write_text, atomic_write_json @@ -39,6 +41,13 @@ SUBBAUSTEIN_MAX = 40 # Einstufen ist billig (kurzes Urteil, keine Websuche) → größere Pakete, weniger Dateien/Agenten. STUFE_CHUNK = 100 +# Recherche-Loop: 5 Agenten je Runde, Runden bis volle Crawl-Abdeckung / 0 neue / Zeit-Kappe. +RECHERCHE_AGENTEN = 5 +RECHERCHE_QUORUM = 3 # je Runde mind. so viele gültige Slot-Ergebnisse +RECHERCHE_KAPPE = 1800 # Loop-Gesamt-Sekunden (Notbremse, 30 min) +SUBBAUSTEIN_KAPPE = 900 # Subbaustein-Finde-Loop je Chunk (15 min) +KONSOLIDIERUNG_CHUNK = 150 # Konsolidierung ab so vielen Kandidaten chunken + log = logging.getLogger("creator.bausteine") _bausteine_progress: dict[str, str] = {} @@ -83,8 +92,12 @@ def _crawl_fertig(topic: str) -> bool: _STUFEN = ("einfach", "mittel", "schwer") -def subbausteine_titel(topic: str, baustein: str) -> list[str]: - """Subbaustein-Titel eines Bausteins aus der Sidecar (leer, wenn keine).""" +async def subbausteine_titel(topic: str, baustein: str) -> list[str]: + """Subbaustein-Titel eines Bausteins — DB-first (Konsens), Fallback Sidecar-Datei.""" + rows = [s["sub_titel"] for s in await db.list_subbausteine(topic, _norm_titel(baustein)) + if s["status"] == "konsens" and s["sub_titel"]] + if rows: + return rows sc = _json_datei(subbausteine_path(topic)) if not isinstance(sc, dict): return [] @@ -94,8 +107,11 @@ def subbausteine_titel(topic: str, baustein: str) -> list[str]: ] -def lade_frage_muster(topic: str, baustein: str) -> list[dict]: - """Vordefinierte Frage-Muster eines Bausteins aus dem Sidecar (leer = Fallback auf Live).""" +async def lade_frage_muster(topic: str, baustein: str) -> list[dict]: + """Vordefinierte Frage-Muster eines Bausteins — DB-first, Fallback Sidecar (leer = Live).""" + rows = await db.list_frage_muster(topic, _norm_titel(baustein)) + if rows: + return [{"subbaustein": r["sub_titel"], "typ": r["typ"], "frage": r["frage"]} for r in rows if r["frage"]] fm = _json_datei(frage_muster_path(topic)) if not isinstance(fm, dict): return [] @@ -108,12 +124,24 @@ def lade_frage_muster(topic: str, baustein: str) -> list[dict]: ] -def lade_uebersicht(topic: str) -> list[dict]: - """Strukturierte Baustein-Liste für die Übersicht: Titel + Beschreibung + Subbausteine/Stufen. - - Verbindet bausteine.md (Nummer/Titel/Beschreibung) mit der Sidecar subbausteine.json - (Key = Titel). Fehlt die Sidecar, sind die Subbaustein-Listen leer. - """ +async def lade_uebersicht(topic: str) -> list[dict]: + """Strukturierte Baustein-Liste für die Übersicht — DB-first (Konsens + Subs/Stufen/Relevanz), + Fallback bausteine.md + Sidecar (Alt-Themen).""" + bs = await db.list_bausteine(topic, status="konsens") + if bs: + out = [] + for num, b in enumerate(bs, 1): + subs = [s for s in await db.list_subbausteine(topic, b["titel_norm"]) if s["status"] == "konsens"] + out.append({ + "num": num, "titel": b["titel"], "beschreibung": b["beschreibung"], + "subbausteine": [ + {"titel": s["sub_titel"], + "stufe": s["stufe"] if s["stufe"] in _STUFEN else "mittel", + "relevanz": s["relevanz"] if s["relevanz"] in ("relevant", "rand") else None} + for s in subs if s["sub_titel"] + ], + }) + return out entries = _lade_bausteine(_read(bausteine_path(topic))) sidecar = _json_datei(subbausteine_path(topic)) sidecar = sidecar if isinstance(sidecar, dict) else {} @@ -238,22 +266,14 @@ def cancel_bausteine(topic: str) -> bool: def _resume_step(topic: str) -> int: - """Erster noch offener Schritt anhand der persistierten Zwischendateien.""" + """Erster noch offener Schritt anhand der persistierten Artefakte. + Inventar (Recherche→Konsolidierung→Klärung) gilt als fertig, sobald bausteine.md vorliegt.""" files = _bausteine_files(topic) q = lade_quelle(topic) if q["type"] == "link" and not _crawl_fertig(topic): return _step_idx(topic, "Quelle laden") - if sum(p.exists() for p in files["recherche"]) < 3: + if not files["final"].exists(): # Inventar (DB-Loop) noch offen return _step_idx(topic, "Recherche") - if not files["recherche_mapping"].exists(): - return _step_idx(topic, "Konsolidierung") - mapping = _mapping_schema(_json_datei(files["recherche_mapping"])) - geklaert = mapping is not None and ( - not mapping[1] # kein strittiger Rest - or any((r := _runde_schema(_json_datei(p))) is not None and not r[1] for p in files["mapping"].values()) - ) - if not geklaert: - return _step_idx(topic, "Klärung") if q["type"] == "projekt" and not files["ergaenzung"].exists(): return _step_idx(topic, "Ergänzung") sidecar = _json_datei(files["sidecar"]) @@ -339,9 +359,10 @@ def _reset_ab_phase(topic: str, phase: int) -> None: if phase <= 2: # Subbausteine files["sub_roh"].unlink(missing_ok=True) glob_del("subbaustein-*") - if phase <= 1: # Inventar = kompletter Frischstart (alle Zwischendateien) + if phase <= 1: # Inventar = kompletter Frischstart (alle Zwischendateien + bausteine.md) for p_alt in _alle_slot_dateien(files): p_alt.unlink(missing_ok=True) + files["final"].unlink(missing_ok=True) def _ergaenzung_schema(data): @@ -386,14 +407,14 @@ def _pdfs_konvertieren(project: Path) -> None: _QUELLE_TEMPLATE = {"projekt": "Bausteine-Quelle-Projekt", "uni": "Bausteine-Quelle-Uni", "link": "Bausteine-Quelle-Link"} -def _build_recherche_prompt(topic: str, out_path: Path, instructions: str, typ: str, ordner: Path | None) -> str: +def _build_recherche_prompt(topic: str, out_path: Path, instructions: str, typ: str, ordner: Path | None, fokus: str = "") -> str: if typ in _QUELLE_TEMPLATE: source = _prompt(_QUELLE_TEMPLATE[typ], project=ordner) else: source = _prompt("Bausteine-Quelle-Thema", topic=topic) return _prompt( "Bausteine-Recherche", - topic=topic, source=source, bausteine_path=out_path, extra=_extra(instructions), + topic=topic, source=source, bausteine_path=out_path, fokus=fokus, extra=_extra(instructions), ) @@ -566,88 +587,143 @@ def _judge_block(chunk: list[int], entries: dict, daten: dict) -> str: async def _subbausteine_block(ctx: GenContext, set_p, files: dict, entries: dict, instructions: str) -> dict | None: - """Block B: drei Phasen mit Barriere — Finden, Wählen (Code-Merge), Klären. - Pro Phase laufen alle Pakete parallel; der Schritt bleibt, bis das letzte fertig ist. - → {Baustein-Titel: [Subbaustein, …]} oder None bei Abbruch/Fehler.""" + """Block B (DB + Loop): je Paket Subbausteine in Runden finden (3 Finder, bis 0 neue/Kappe), + in der DB sammeln (≥2 Nennungen = Konsens, 1× verworfen), Judge bereinigt je Paket. + → {Baustein-Titel: [Subbaustein, …]} (Konsens) oder None. Befüllt DB-Tabelle `subbausteine`.""" topic, provider, is_cancelled = ctx.topic, ctx.provider, ctx.is_cancelled arbeit = files["arbeit"] - idx = _titel_index(entries) caps = "files" if quelle_ordner(topic) else "full" nums = list(entries) chunks = _chunk_nums(nums, _n_chunks(len(nums))) n = len(chunks) + titel_by_num = {num: _titel(entries[num]) for num in nums} + norm_by_num = {num: _norm_titel(titel_by_num[num]) for num in nums} + await db.delete_subbausteine(topic) # Frischstart des Blocks (idempotenter Zähler) - def finder_paths(c): - return [arbeit / f"subbaustein-c{c}-{i}.md" for i in (1, 2, 3)] + async def _bekannt_block(chunk): + bl = [] + for num in chunk: + subs = [s["sub_titel"] for s in await db.list_subbausteine(topic, norm_by_num[num])] + if subs: + bl.append(f"\n" + "\n".join(f"- {s}" for s in subs)) + if not bl: + return "" + return "\n\nBEREITS GEFUNDEN — bestätige diese Subbausteine erneut UND ergänze fehlende:\n" + "\n".join(bl) - def final_path(c): - return arbeit / f"subbaustein-final-c{c}.md" - - # Phase „Subbausteine finden": pro Paket 3 Finder (min. 2), alle Pakete parallel. + # Phase „Subbausteine finden": je Paket Loop bis 0 neue Subs / Zeit-Kappe. async def _finde(c, chunk): - paths = finder_paths(c) - vorhanden = sum(1 for p in paths if _parse_subbausteine(_read(p))) - if vorhanden >= 2: - return True zuteilung = "\n".join(f"- {entries[num]}" for num in chunk) - offen = [(i, p) for i, p in enumerate(paths, 1) if not _parse_subbausteine(_read(p))] - slots = [{ - "key": f"bausteine-{topic}-subbaustein-c{c}-{i}", - "prompt": _prompt("Subbaustein-Recherche", topic=topic, zuteilung=zuteilung, out_path=p, extra=_extra(instructions)), - "role": "quick", "capabilities": caps, - "payload": (lambda result, p=p: _parse_subbausteine(_read(p)) or None), - } for i, p in offen] - neu = await _race(topic, f"Subbausteine Paket {c}", slots, 2 - vorhanden, _timeout("subbaustein", len(chunk)), provider, cancelled=is_cancelled, grace=KONSENS_GRACE) - return not is_cancelled() and neu is not None + chunk_idx = _titel_index({num: titel_by_num[num] for num in chunk}) + start = time.monotonic() + runde = 0 + while not is_cancelled(): + runde += 1 + bekannt = await _bekannt_block(chunk) if runde > 1 else "" + paths = [arbeit / f"subbaustein-c{c}-r{runde}-{i}.md" for i in (1, 2, 3)] + for p in paths: + p.unlink(missing_ok=True) + slots = [{ + "key": f"bausteine-{topic}-subbaustein-c{c}-r{runde}-{i}", + "prompt": _prompt("Subbaustein-Recherche", topic=topic, zuteilung=zuteilung, bekannt=bekannt, out_path=p, extra=_extra(instructions)), + "role": "quick", "capabilities": caps, + "payload": (lambda result, p=p: _parse_subbausteine(_read(p)) or None), + } for i, p in enumerate(paths, 1)] + texte = await _race(topic, f"Subbausteine Paket {c} R{runde}", slots, 2, _timeout("subbaustein", len(chunk)), provider, cancelled=is_cancelled, grace=KONSENS_GRACE) + if is_cancelled(): + return False + if not texte: + return runde > 1 # Runde 1 ohne Ergebnis = Fehler; spätere = einfach Ende + vorhanden = {num: {s["sub_norm"] for s in await db.list_subbausteine(topic, norm_by_num[num])} for num in chunk} + neu = 0 + for d in texte: + for marker, subs in d.items(): + num = _titel_aufloesen(chunk_idx, marker) + if num is None: + continue + gesehen = set() + for sub in subs: + sn = _norm_titel(sub) + if not sn or sn in gesehen: + continue + gesehen.add(sn) + if sn not in vorhanden[num]: + neu += 1 + vorhanden[num].add(sn) + await db.upsert_subbaustein(topic, norm_by_num[num], sn, titel_by_num[num], sub) + if neu == 0: + break + if time.monotonic() - start > SUBBAUSTEIN_KAPPE: + _log(topic, f"Subbausteine Paket {c}: Zeit-Kappe erreicht (Runde {runde})") + break + return True - oks = await _gather_fortschritt([_finde(c, chunk) for c, chunk in enumerate(chunks, 1)], len(chunks), _melde_p(set_p, topic, "Subbausteine finden")) + oks = await _gather_fortschritt([_finde(c, chunk) for c, chunk in enumerate(chunks, 1)], n, _melde_p(set_p, topic, "Subbausteine finden")) if is_cancelled(): return None if not all(ok is True for ok in oks): _bausteine_errors[topic] = "Subbausteine fehlgeschlagen (Recherche)" return None - # Phase „Subbausteine wählen": Code-Merge je Paket (instant, kein Agent). + # Phase „Subbausteine wählen": ≥2 Nennungen = Konsens, 1× verworfen (Code). set_p(f"Subbausteine wählen ({n} Pakete)…", step=_step_idx(topic, "Subbausteine wählen")) - daten_by_c = {} - for c, chunk in enumerate(chunks, 1): - finder = [d for p in finder_paths(c) if (d := _parse_subbausteine(_read(p)))] - daten_by_c[c] = {num: _merge_finder(num, idx, finder) for num in chunk} + for num in nums: + for s in await db.list_subbausteine(topic, norm_by_num[num]): + await db.set_subbaustein_felder(topic, norm_by_num[num], s["sub_norm"], + status=("konsens" if s["nennungen"] >= 2 else "verworfen")) - # Phase „Subbausteine klären": Judge je Paket mit Strittigem, alle parallel. + # Phase „Subbausteine klären": Judge je Paket bereinigt die Konsens-Liste. async def _klaere(c, chunk): - daten = daten_by_c[c] - fp = final_path(c) + fp = arbeit / f"subbaustein-final-c{c}.md" if _parse_subbausteine(_read(fp)): return - if not any(daten[num][1] for num in chunk): - atomic_write_text(fp, _final_text(chunk, entries, daten)) + bloecke, hat = [], False + for num in chunk: + subs = [s["sub_titel"] for s in await db.list_subbausteine(topic, norm_by_num[num]) if s["status"] == "konsens"] + zeilen = "\n".join(f"- {s}" for s in subs) if subs else "- (keiner)" + bloecke.append(f"BAUSTEIN: {titel_by_num[num]}\nKonsens:\n{zeilen}") + if subs: + hat = True + if not hat: return status, _ = await run_single_slot( ctx, f"Subbaustein-Klärung {c}", key=f"bausteine-{topic}-subbaustein-final-c{c}", - prompt=_prompt("Subbaustein-Mapping", topic=topic, bausteine=_judge_block(chunk, entries, daten), out_path=fp, extra=_extra(instructions)), + prompt=_prompt("Subbaustein-Mapping", topic=topic, bausteine="\n\n".join(bloecke), out_path=fp, extra=_extra(instructions)), role="judge", capabilities="files", payload=lambda result, p=fp: _parse_subbausteine(_read(p)) or None, timeout=_timeout("subbaustein_check", len(chunk)), ) if status == FAILED: - _log(topic, f"Subbaustein-Klärung Paket {c} fehlgeschlagen — nur Konsens übernommen") + _log(topic, f"Subbaustein-Klärung Paket {c} fehlgeschlagen — Konsens übernommen") - await _gather_fortschritt([_klaere(c, chunk) for c, chunk in enumerate(chunks, 1)], len(chunks), _melde_p(set_p, topic, "Subbausteine klären")) + await _gather_fortschritt([_klaere(c, chunk) for c, chunk in enumerate(chunks, 1)], n, _melde_p(set_p, topic, "Subbausteine klären")) if is_cancelled(): return None - # Fehlende finale Dateien → Konsens-Fallback; dann alle parsen. + # Finale Liste je Baustein: Judge-Ausgabe, sonst Konsens-Fallback. DB reconcilen + roh bauen. roh: dict[str, list[str]] = {} for c, chunk in enumerate(chunks, 1): - fp = final_path(c) - if not _parse_subbausteine(_read(fp)): - atomic_write_text(fp, _final_text(chunk, entries, daten_by_c[c])) - for marker, subs in (_parse_subbausteine(_read(fp)) or {}).items(): - num = _titel_aufloesen(idx, marker) - if num is not None: - roh[_titel(entries[num])] = subs + final = _parse_subbausteine(_read(arbeit / f"subbaustein-final-c{c}.md")) or {} + chunk_idx = _titel_index({num: titel_by_num[num] for num in chunk}) + final_by_num = {_titel_aufloesen(chunk_idx, m): subs for m, subs in final.items() if _titel_aufloesen(chunk_idx, m) is not None} + for num in chunk: + titel = titel_by_num[num] + konsens = [s["sub_titel"] for s in await db.list_subbausteine(topic, norm_by_num[num]) if s["status"] == "konsens"] + subs = final_by_num.get(num) or konsens + if not subs: + continue + roh[titel] = subs + # DB an die finale Liste angleichen: finale = konsens, Rest verworfen, Neues ergänzen. + final_norms = {_norm_titel(s) for s in subs} + have = {s["sub_norm"] for s in await db.list_subbausteine(topic, norm_by_num[num])} + for s in await db.list_subbausteine(topic, norm_by_num[num]): + await db.set_subbaustein_felder(topic, norm_by_num[num], s["sub_norm"], + status=("konsens" if s["sub_norm"] in final_norms else "verworfen")) + for s in subs: + sn = _norm_titel(s) + if sn and sn not in have: + await db.upsert_subbaustein(topic, norm_by_num[num], sn, titel, s) + await db.set_subbaustein_felder(topic, norm_by_num[num], sn, status="konsens") if not roh: _bausteine_errors[topic] = "Keine Subbausteine ermittelt" return None @@ -1064,6 +1140,282 @@ async def _frage_muster_block(ctx: GenContext, set_p, files: dict, sidecar: dict return ergebnis +# ── Inventar in der DB: Recherche-Loop · Konsolidierung · Klärung ──────────── + +def _cited_sources(text: str) -> set[str]: + """Zitierte Quellen je Eintrag = 3. ' — '-Segment (URL bzw. Dateiname), kleingeschrieben.""" + out = set() + for eintrag in _parse_auswahl(text).values(): + teile = [t.strip() for t in eintrag.split(" — ")] + if len(teile) >= 3 and teile[-1]: + out.add(teile[-1].lower()) + return out + + +def _crawl_index(ordner) -> dict[str, str]: + """Alias (Dateiname ODER QUELLE:-URL, klein) → kanonischer Seiten-Key (Dateiname).""" + idx: dict[str, str] = {} + if not ordner or not Path(ordner).is_dir(): + return idx + for p in sorted(Path(ordner).glob("*.txt")): + key = p.name + idx[key.lower()] = key + try: + erste = p.read_text(encoding="utf-8").splitlines()[0] + except (OSError, IndexError): + erste = "" + if erste.startswith("QUELLE:"): + url = erste[len("QUELLE:"):].strip() + if url: + idx[url.lower()] = key + idx[url.rstrip("/").lower()] = key + return idx + + +def _abgedeckt(zitiert: set[str], crawl_idx: dict[str, str]) -> set[str]: + """Zitierte Quellen → Menge kanonischer Crawl-Seiten-Keys (was nicht passt, fällt weg).""" + out = set() + for z in zitiert: + key = crawl_idx.get(z) or crawl_idx.get(z.rstrip("/")) + if key: + out.add(key) + return out + + +def _fokus_text(bereits: list[str], offen: list[str]) -> str: + """Re-Prompt-Block: noch nicht abgedeckte Crawl-Seiten + bereits gefundene Titel.""" + teile = [] + if offen: + liste = "\n".join(f"- {n}" for n in offen[:150]) + teile.append( + "NOCH NICHT ABGEDECKTE QUELLEN-DATEIEN — lies ZUERST genau diese im Quell-Ordner " + f"und ergänze daraus die noch fehlenden Bausteine:\n{liste}" + ) + if bereits: + liste = "\n".join(f"- {t}" for t in bereits) + teile.append("BEREITS GEFUNDEN (NICHT wiederholen — liefere nur NEUE Bausteine):\n" + liste) + return ("\n\n" + "\n\n".join(teile)) if teile else "" + + +async def _set_inventar(topic: str, eintrag: str, status: str) -> None: + """Einen Inventar-Eintrag ('Titel — Beschreibung') mit Status in die DB schreiben.""" + titel = _titel(eintrag) + norm = _norm_titel(titel) + if not norm: + return + teile = [t.strip() for t in eintrag.split(" — ")] + besch = teile[1] if len(teile) >= 2 else "" + await db.upsert_baustein(topic, norm, titel, besch) + await db.set_baustein_status(topic, norm, status) + + +async def _recherche_loop(ctx: GenContext, set_p, files: dict, q: dict, ordner, instructions: str) -> bool: + """Füllt DB-Tabelle `bausteine` mit Kandidaten (+ Nennungszähler). Loop: je Runde + RECHERCHE_AGENTEN Agenten; re-promptet mit noch offenen Crawl-Seiten, bis alle gelesen / + 0 neue / Zeit-Kappe. Resume: fertiger Schritt wird übersprungen. → True (ok) / False (Abbruch).""" + topic, provider, is_cancelled = ctx.topic, ctx.provider, ctx.is_cancelled + if await db.get_step_status(topic, "Recherche") == "fertig": + return True + arbeit = files["arbeit"] + caps = "files" if ordner else "full" + crawl_idx = _crawl_index(ordner) + alle_seiten = set(crawl_idx.values()) + + await db.delete_bausteine(topic) + await db.delete_coverage(topic) + await db.set_step_status(topic, "Recherche", "laufend") + set_p("Recherche läuft (Runde 1)…", step=_step_idx(topic, "Recherche")) + + start = time.monotonic() + runde = 0 + while not is_cancelled(): + runde += 1 + vorhandene = {b["titel_norm"] for b in await db.list_bausteine(topic)} + gelesen = set(await db.list_coverage(topic)) + offen = sorted(alle_seiten - gelesen) + if runde > 1 and alle_seiten and not offen: + break # volle Abdeckung erreicht + fokus = _fokus_text([b["titel"] for b in await db.list_bausteine(topic)], offen) + paths = [arbeit / f"recherche-r{runde}-{i}.md" for i in range(1, RECHERCHE_AGENTEN + 1)] + for p in paths: + p.unlink(missing_ok=True) + slots = [{ + "key": f"bausteine-{topic}-recherche-r{runde}-{i}", + "prompt": _build_recherche_prompt(topic, p, instructions, q["type"], ordner, fokus=fokus), + "role": "quick", "capabilities": caps, + "payload": (lambda result, p=p: _file_payload(p)), + } for i, p in enumerate(paths, 1)] + set_p(f"Recherche läuft (Runde {runde})…") + texte = await _race(topic, f"Recherche R{runde}", slots, RECHERCHE_QUORUM, + _timeout("recherche"), provider, cancelled=is_cancelled, grace=RECHERCHE_GRACE) + if is_cancelled(): + return False + if not texte: + if runde == 1: + _bausteine_errors[topic] = "Recherche fehlgeschlagen (Minimum nicht erreicht)" + return False + break # keine neuen Ergebnisse mehr + neu, zitiert = 0, set() + for text in texte: + zitiert |= _cited_sources(text) + gesehen = set() + for eintrag in _parse_auswahl(text).values(): + titel = _titel(eintrag) + norm = _norm_titel(titel) + if not norm or norm in gesehen: + continue + gesehen.add(norm) + teile = [t.strip() for t in eintrag.split(" — ")] + besch = teile[1] if len(teile) >= 2 else "" + quelle = [teile[2]] if len(teile) >= 3 and teile[2] else [] + if norm not in vorhandene: + neu += 1 + vorhandene.add(norm) + await db.upsert_baustein(topic, norm, titel, besch, quelle) + await db.mark_quellen_gelesen(topic, sorted(_abgedeckt(zitiert, crawl_idx))) + gesamt = len(await db.list_bausteine(topic)) + deckung = len(await db.list_coverage(topic)) + _log(topic, f"Recherche R{runde}: +{neu} neu (gesamt {gesamt}), Abdeckung {deckung}/{len(alle_seiten) or '?'}") + if neu == 0: + break + if time.monotonic() - start > RECHERCHE_KAPPE: + _log(topic, f"Recherche: Zeit-Kappe {RECHERCHE_KAPPE}s erreicht (Runde {runde})") + break + await db.set_step_status(topic, "Recherche", "fertig") + return True + + +async def _konsolidiere(ctx: GenContext, set_p, files: dict) -> bool: + """Judge mergt Kandidaten semantisch + teilt in Konsens (≥2)/Rest (1×); Status in DB.""" + topic, is_cancelled = ctx.topic, ctx.is_cancelled + if await db.get_step_status(topic, "Konsolidierung") == "fertig": + return True + set_p("Konsolidiere Recherche…", step=_step_idx(topic, "Konsolidierung")) + kandidaten = await db.list_bausteine(topic) + if not kandidaten: + _bausteine_errors[topic] = "Konsolidierung: keine Kandidaten" + return False + arbeit = files["arbeit"] + chunks = _chunk_nums(kandidaten, max(1, math.ceil(len(kandidaten) / KONSOLIDIERUNG_CHUNK))) + konsens, rest = [], [] + for c, chunk in enumerate(chunks, 1): + fp = arbeit / f"konsolidierung-c{c}.json" + fp.unlink(missing_ok=True) + eintraege = "\n".join( + f"{i}. {b['titel']} — {b['beschreibung']} ({b['nennungen']}× genannt)" for i, b in enumerate(chunk, 1) + ) + status, mapping = await run_single_slot( + ctx, f"Konsolidierung {c}", + key=f"bausteine-{topic}-konsolidierung-c{c}", + prompt=_prompt("Bausteine-Recherche-Mapping", topic=topic, n=RECHERCHE_AGENTEN, eintraege=eintraege, out_path=fp), + role="judge", capabilities="files", + payload=lambda result, p=fp: _mapping_schema(_json_datei(p)), + timeout=_timeout("recherche_mapping", len(chunk)), + ) + if status == CANCELLED: + return False + if status == FAILED: + _bausteine_errors[topic] = "Recherche-Mapping fehlgeschlagen" + return False + k, r = mapping + konsens += k + rest += r + # Judge-Ausgabe ist maßgeblich → Inventar in der DB neu setzen. + await db.delete_bausteine(topic) + for t in konsens: + await _set_inventar(topic, t, "konsens") + for t in rest: + await _set_inventar(topic, t, "rest") + await db.set_step_status(topic, "Konsolidierung", "fertig") + return True + + +async def _klaere_inventar(ctx: GenContext, set_p, files: dict) -> bool: + """1 Judge entscheidet über den Rest (1×-Genannte): aufnehmen → Konsens, sonst verworfen.""" + topic, is_cancelled = ctx.topic, ctx.is_cancelled + if await db.get_step_status(topic, "Klärung") == "fertig": + return True + set_p("Klärung läuft…", step=_step_idx(topic, "Klärung")) + rest_rows = await db.list_bausteine(topic, status="rest") + if rest_rows: + konsens = [b["titel"] for b in await db.list_bausteine(topic, status="konsens")] + fp = files["arbeit"] / "klaerung.json" + fp.unlink(missing_ok=True) + status, ergebnis = await run_single_slot( + ctx, "Klärung", + key=f"bausteine-{topic}-klaerung", + prompt=_prompt( + "Bausteine-Klaerung", topic=topic, + konsens="\n".join(f"- {t}" for t in konsens) or "(noch leer)", + rest="\n".join(f"- {b['titel']}" for b in rest_rows), + final="\n- Entscheide JEDEN Eintrag. `rest` MUSS leer sein.", + out_path=fp, + ), + role="judge", capabilities="files", + payload=lambda result, p=fp: _runde_schema(_json_datei(p), final=True), + timeout=_timeout("auswahl_mapping", len(rest_rows)), + ) + if status == CANCELLED: + return False + if status == FAILED: + _bausteine_errors[topic] = "Klärung fehlgeschlagen" + return False + aufnehmen, _ = ergebnis + auf_norm = {_norm_titel(_titel(t)) for t in aufnehmen} + for b in rest_rows: + await db.set_baustein_status(topic, b["titel_norm"], "konsens" if b["titel_norm"] in auf_norm else "verworfen") + await db.set_step_status(topic, "Klärung", "fertig") + return True + + +async def _mirror_sidecar_db(topic: str, sidecar: dict) -> None: + """Sidecar {Baustein-Titel: [{titel, stufe, relevanz}]} in die DB-Tabelle subbausteine spiegeln.""" + for btitel, subs in sidecar.items(): + bnorm = _norm_titel(btitel) + if not bnorm or not isinstance(subs, list): + continue + for s in subs: + if not isinstance(s, dict): + continue + st = str(s.get("titel", "")).strip() + sn = _norm_titel(st) + if not sn: + continue + await db.put_subbaustein(topic, bnorm, sn, btitel, st, + stufe=s.get("stufe"), relevanz=s.get("relevanz"), status="konsens") + + +async def _mirror_frage_muster_db(topic: str, muster: dict) -> None: + """Frage-Muster {Baustein-Titel: [{subbaustein, typ, frage}]} in die DB-Tabelle frage_muster spiegeln.""" + await db.delete_frage_muster(topic) + for btitel, eintraege in muster.items(): + bnorm = _norm_titel(btitel) + if not bnorm or not isinstance(eintraege, list): + continue + for e in eintraege: + if not isinstance(e, dict): + continue + sub = str(e.get("subbaustein", "")).strip() + sn = _norm_titel(sub) + typ = str(e.get("typ", "")).strip() + frage = str(e.get("frage", "")).strip() + if not (sn and typ and frage): + continue + await db.upsert_frage_muster(topic, bnorm, sn, btitel, sub, typ, frage) + + +async def _reset_db_ab_phase(topic: str, ab_phase: int) -> None: + """DB-Inhalt der Phasen ≥ ab_phase verwerfen (Inventar=1 … Fragen=5).""" + if ab_phase <= 1: + await db.delete_bausteine(topic) + await db.delete_coverage(topic) + await db.delete_pipeline_state(topic, ["Recherche", "Konsolidierung", "Klärung"]) + if ab_phase <= 2: + await db.delete_subbausteine(topic) + if ab_phase <= 5: + await db.delete_frage_muster(topic) + + async def generate_bausteine(topic: str, instructions: str = "", provider: str = DEFAULT_PROVIDER, ab_phase: int | None = None) -> None: if topic in _bausteine_progress: return @@ -1096,6 +1448,7 @@ async def generate_bausteine(topic: str, instructions: str = "", provider: str = # unten wird übersprungen (er würde bei erhaltener Sidecar sonst alles wischen). if ab_phase is not None: _reset_ab_phase(topic, ab_phase) + await _reset_db_ab_phase(topic, ab_phase) # Link-Quelle: erst crawlen (gleiche Domain, begrenzt) → wird zur Ordner-Quelle. if q["type"] == "link" and not _crawl_fertig(topic): set_p("Quelle laden (Crawl)…", step=_step_idx(topic, "Quelle laden")) @@ -1116,160 +1469,30 @@ async def generate_bausteine(topic: str, instructions: str = "", provider: str = if fertig: for p_alt in _alle_slot_dateien(files): p_alt.unlink(missing_ok=True) + await db.delete_pipeline_state(topic) + await db.delete_bausteine(topic) + await db.delete_subbausteine(topic) + await db.delete_frage_muster(topic) + await db.delete_coverage(topic) - # Schritt 1: 5 Recherche-Agenten, min. 3 mit Grace-Fenster — alle gültigen - # Slot-Dateien fließen ins Mapping (kein Kappen mehr bei 3) - recherchen: list[str] = [] - offen = [] - for i, path in enumerate(files["recherche"], 1): - text = _file_payload(path) - if text is not None: - recherchen.append(text) - else: - offen.append((i, path)) - vorhanden = len(recherchen) - set_p(f"Recherche läuft ({vorhanden} gültig, min. 3)…", step=_step_idx(topic, "Recherche")) - if vorhanden < 3: - caps = "files" if ordner else "full" - slots = [ - { - "key": f"bausteine-{topic}-recherche-{i}", - "prompt": _build_recherche_prompt(topic, path, instructions, q["type"], ordner), - "role": "quick", "capabilities": caps, - "payload": (lambda result, p=path: _file_payload(p)), - } - for i, path in offen - ] - neue = await _race( - topic, "Recherche", slots, 3 - vorhanden, _timeout("recherche"), provider, - on_update=lambda c: set_p(f"Recherche läuft ({vorhanden + c} gültig, min. 3)…"), - cancelled=is_cancelled, grace=RECHERCHE_GRACE, - ) + # Inventar (DB): Recherche-Loop → Konsolidierung → Klärung. + if not await _recherche_loop(ctx, set_p, files, q, ordner, instructions): if is_cancelled(): abgebrochen() - return - if neue is None: - _bausteine_errors[topic] = "Recherche fehlgeschlagen (Minimum nicht erreicht)" - return - recherchen += neue - - # Schritt 2: Recherche-Mapping — Code-Vormerge (exakte Titel) + 1 Agent - # für semantische Dubletten und Konsens/Rest-Teilung (fatal) - mapping = _mapping_schema(_json_datei(files["recherche_mapping"])) - if mapping is None: - set_p("Konsolidiere Recherche…", step=_step_idx(topic, "Konsolidierung")) - files["recherche_mapping"].unlink(missing_ok=True) - gemergt = _vormerge([_parse_auswahl(t) for t in recherchen]) - eintraege = "\n".join(f"{i}. {text} ({n}× genannt)" for i, (text, n) in enumerate(gemergt, 1)) - status, mapping = await run_single_slot( - ctx, "Recherche-Mapping", - key=f"bausteine-{topic}-recherche-mapping", - prompt=_prompt( - "Bausteine-Recherche-Mapping", - topic=topic, n=len(recherchen), eintraege=eintraege, - out_path=files["recherche_mapping"], - ), - role="judge", capabilities="files", - payload=lambda result: _mapping_schema(_json_datei(files["recherche_mapping"])), - timeout=_timeout("recherche_mapping", len(gemergt)), - ) - if status == CANCELLED: + return + if not await _konsolidiere(ctx, set_p, files): + if is_cancelled(): abgebrochen() - return - if status == FAILED: - _bausteine_errors[topic] = "Recherche-Mapping fehlgeschlagen" - return - konsens, rest = mapping - - # Klärungs-Loop: 3 Auswahl-Agenten entscheiden über den Rest, ein - # Mapping-Agent sortiert in aufnehmen/verwerfen/weiter strittig. - # Leerer Rest beendet den Loop; Runde KONSENS_MAX_RUNDEN muss - # alles entscheiden. Der Konsens wächst nur hier im Code. - runde = 0 - while rest and runde < KONSENS_MAX_RUNDEN: - runde += 1 - final_runde = runde == KONSENS_MAX_RUNDEN - set_p(f"Klärung läuft (Runde {runde}/{KONSENS_MAX_RUNDEN})…", step=_step_idx(topic, "Klärung")) - mapping_path = files["mapping"][runde] - - # Resume: fertiges Runden-Mapping wird direkt übernommen - ergebnis = _runde_schema(_json_datei(mapping_path), final=final_runde) - if ergebnis is None: - mapping_path.unlink(missing_ok=True) - konsens_block = "\n".join(f"- {t}" for t in konsens) - rest_block = "\n".join(f"- {t}" for t in rest) - - # 3 Auswahl-Agenten, min. 2 mit Grace-Fenster - entscheidungen = [] - offen = [] - for i, path in enumerate(files["auswahl"][runde], 1): - res = _rest_schema(_json_datei(path)) - if res is not None: - entscheidungen.append(res) - else: - offen.append((i, path)) - if len(entscheidungen) < 2: - slots = [ - { - "key": f"bausteine-{topic}-auswahl-r{runde}-{i}", - "prompt": _prompt( - "Bausteine-Auswahl", - topic=topic, konsens=konsens_block, rest=rest_block, out_path=path, - ), - "role": "fast", "capabilities": "files", - "payload": (lambda result, p=path: _rest_schema(_json_datei(p))), - } - for i, path in offen - ] - neue = await _race( - topic, f"Auswahl r{runde}", slots, 2 - len(entscheidungen), - _timeout("auswahl", len(rest)), provider, - cancelled=is_cancelled, grace=KONSENS_GRACE, - ) - if is_cancelled(): - abgebrochen() - return - if neue is None: - _bausteine_errors[topic] = f"Auswahl fehlgeschlagen (Runde {runde}, Minimum nicht erreicht)" - return - entscheidungen += neue - - # Votum pro Rest-Eintrag deterministisch zählen - indizes = [_titel_index(dict(enumerate(e, 1))) for e in entscheidungen] - voten = "\n".join( - f"{i}. {text} (von {sum(1 for idx in indizes if _titel_aufloesen(idx, text) is not None)}" - f"/{len(entscheidungen)} Agenten übernommen)" - for i, text in enumerate(rest, 1) - ) - final_zusatz = ( - "\n- LETZTE RUNDE: Es gibt keine weitere Runde. `rest` MUSS leer sein" - " — entscheide JEDEN Eintrag selbst: aufnehmen oder verwerfen." - if final_runde else "" - ) - status, ergebnis = await run_single_slot( - ctx, f"Auswahl-Mapping r{runde}", - key=f"bausteine-{topic}-auswahl-mapping-r{runde}", - prompt=_prompt( - "Bausteine-Auswahl-Mapping", - topic=topic, n=len(entscheidungen), konsens=konsens_block, - rest=voten, final=final_zusatz, out_path=mapping_path, - ), - role="judge", capabilities="files", - payload=lambda result, p=mapping_path, f=final_runde: _runde_schema(_json_datei(p), final=f), - timeout=_timeout("auswahl_mapping", len(rest)), - ) - if status == CANCELLED: - abgebrochen() - return - if status == FAILED: - _bausteine_errors[topic] = f"Auswahl-Mapping fehlgeschlagen (Runde {runde})" - return - - aufnehmen, rest = ergebnis - _log(topic, f"Klärung Runde {runde}: {len(aufnehmen)} aufgenommen, {len(rest)} weiter strittig") - konsens = konsens + aufnehmen - - entries = {i: t for i, t in enumerate(konsens, 1)} + return + if not await _klaere_inventar(ctx, set_p, files): + if is_cancelled(): + abgebrochen() + return + konsens_rows = await db.list_bausteine(topic, status="konsens") + entries = { + i: (f"{b['titel']} — {b['beschreibung']}" if b["beschreibung"] else b["titel"]) + for i, b in enumerate(konsens_rows, 1) + } # Nur Projekte: Themenfeld-Ergänzung — Skript/Projekt ist ein Ausschnitt, # ein Web-Agent ergänzt kanonisch fehlende Bausteine, markiert mit [Ergänzung]. @@ -1360,6 +1583,14 @@ async def generate_bausteine(topic: str, instructions: str = "", provider: str = if muster is None: return # Abbruch atomic_write_json(files["frage_muster"], muster, indent=1) + + # DB-Spiegel (Brücke): finalen Sidecar- + Frage-Muster-Stand in die DB schreiben. + sidecar = _json_datei(files["sidecar"]) + if _sidecar_schema(sidecar) is not None: + await _mirror_sidecar_db(topic, sidecar) + muster = _json_datei(files["frage_muster"]) + if isinstance(muster, dict) and muster: + await _mirror_frage_muster_db(topic, muster) except Exception as e: log.exception("[%s] Bausteine-Generierung fehlgeschlagen", topic) _bausteine_errors[topic] = str(e)[:2000] diff --git a/backend/database.py b/backend/database.py index 061fbdb..aafa93a 100644 --- a/backend/database.py +++ b/backend/database.py @@ -73,6 +73,102 @@ CREATE TABLE IF NOT EXISTS baustein_progress ( ) """ +# --- Bausteine-Pipeline-Inhalt (ersetzt Datei-Sidecars) --- + +# Inventar: ein Baustein je (topic, titel_norm). nennungen = Anzahl Agenten/Runden, +# die ihn nannten (≥2 = Konsens). status: kandidat/konsens/rest/verworfen. +CREATE_BAUSTEINE = """ +CREATE TABLE IF NOT EXISTS bausteine ( + topic TEXT NOT NULL, + titel_norm TEXT NOT NULL, + titel TEXT NOT NULL, + beschreibung TEXT NOT NULL DEFAULT '', + nennungen INTEGER NOT NULL DEFAULT 1, + status TEXT NOT NULL DEFAULT 'kandidat', + quellen TEXT NOT NULL DEFAULT '[]', + updated_at TEXT NOT NULL, + PRIMARY KEY (topic, titel_norm) +) +""" + +# Subbausteine je Baustein. stufe (einfach/mittel/schwer) + relevanz (relevant/rand) +# werden später gesetzt. nennungen analog zum Inventar. +CREATE_SUBBAUSTEINE = """ +CREATE TABLE IF NOT EXISTS subbausteine ( + topic TEXT NOT NULL, + baustein_norm TEXT NOT NULL, + sub_norm TEXT NOT NULL, + baustein TEXT NOT NULL, + sub_titel TEXT NOT NULL, + nennungen INTEGER NOT NULL DEFAULT 1, + stufe TEXT, + relevanz TEXT, + status TEXT NOT NULL DEFAULT 'kandidat', + updated_at TEXT NOT NULL, + PRIMARY KEY (topic, baustein_norm, sub_norm) +) +""" + +# Ein Frage-Muster je (Baustein, Subbaustein, Typ). +CREATE_FRAGE_MUSTER = """ +CREATE TABLE IF NOT EXISTS frage_muster ( + topic TEXT NOT NULL, + baustein_norm TEXT NOT NULL, + sub_norm TEXT NOT NULL, + baustein TEXT NOT NULL, + sub_titel TEXT NOT NULL, + typ TEXT NOT NULL, + frage TEXT NOT NULL, + updated_at TEXT NOT NULL, + PRIMARY KEY (topic, baustein_norm, sub_norm, typ) +) +""" + +# Crawl-Seiten-Abdeckung: treibt den Recherche-Loop (welche Quelle wurde zitiert?). +CREATE_RECHERCHE_COVERAGE = """ +CREATE TABLE IF NOT EXISTS recherche_coverage ( + topic TEXT NOT NULL, + quelle TEXT NOT NULL, + gelesen INTEGER NOT NULL DEFAULT 0, + updated_at TEXT NOT NULL, + PRIMARY KEY (topic, quelle) +) +""" + +# Schritt-Status der Pipeline (ersetzt Datei-Existenz-Resume + Reset-Globs). +CREATE_PIPELINE_STATE = """ +CREATE TABLE IF NOT EXISTS pipeline_state ( + topic TEXT NOT NULL, + schritt TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'offen', + updated_at TEXT NOT NULL, + PRIMARY KEY (topic, schritt) +) +""" + +# Fertiger Guide-Inhalt je (Thema, Format) als JSON-Blob (ersetzt die Guide-JSON-Datei). +# Geteilt über alle Guide-Läufe desselben Thema+Formats (wie zuvor die Content-Datei). +CREATE_GUIDE_CONTENT = """ +CREATE TABLE IF NOT EXISTS guide_content ( + topic TEXT NOT NULL, + format TEXT NOT NULL, + json TEXT NOT NULL, + updated_at TEXT NOT NULL, + PRIMARY KEY (topic, format) +) +""" + +# Quellen-Wahl je Thema (ersetzt quelle.json). +CREATE_QUELLE = """ +CREATE TABLE IF NOT EXISTS quelle ( + topic TEXT PRIMARY KEY, + type TEXT NOT NULL, + ort TEXT NOT NULL DEFAULT '', + spec TEXT NOT NULL DEFAULT '', + updated_at TEXT NOT NULL +) +""" + _db: aiosqlite.Connection | None = None @@ -95,6 +191,13 @@ async def init_db(): await db.execute(CREATE_ELEMENTS) await db.execute(CREATE_BAUSTEIN_TEXTE) await db.execute(CREATE_BAUSTEIN_PROGRESS) + await db.execute(CREATE_BAUSTEINE) + await db.execute(CREATE_SUBBAUSTEINE) + await db.execute(CREATE_FRAGE_MUSTER) + await db.execute(CREATE_RECHERCHE_COVERAGE) + await db.execute(CREATE_PIPELINE_STATE) + await db.execute(CREATE_GUIDE_CONTENT) + await db.execute(CREATE_QUELLE) try: # Migration für Bestands-DBs ohne step-Spalte await db.execute("ALTER TABLE guides ADD COLUMN step INTEGER") except aiosqlite.OperationalError: @@ -471,3 +574,269 @@ async def delete_baustein_daten(topic: str) -> None: await db.execute("DELETE FROM baustein_texte WHERE topic = ?", (topic,)) await db.execute("DELETE FROM baustein_progress WHERE topic = ?", (topic,)) await db.commit() + + +# --- Bausteine-Pipeline-Inhalt: Inventar / Subbausteine / Frage-Muster / Coverage / State / Quelle --- + +async def upsert_baustein(topic: str, titel_norm: str, titel: str, beschreibung: str = "", quellen: list | None = None) -> None: + """Kandidat einfügen oder Nennungszähler erhöhen. Erst-Beschreibung bleibt erhalten.""" + db = await get_db() + await db.execute( + """INSERT INTO bausteine (topic, titel_norm, titel, beschreibung, nennungen, status, quellen, updated_at) + VALUES (?, ?, ?, ?, 1, 'kandidat', ?, ?) + ON CONFLICT(topic, titel_norm) DO UPDATE SET + nennungen = nennungen + 1, quellen = excluded.quellen, updated_at = excluded.updated_at""", + (topic, titel_norm, titel, beschreibung, json.dumps(quellen or [], ensure_ascii=False), _now()), + ) + await db.commit() + + +async def list_bausteine(topic: str, status: str | None = None) -> list[dict]: + db = await get_db() + if status is None: + cursor = await db.execute("SELECT * FROM bausteine WHERE topic = ? ORDER BY rowid", (topic,)) + else: + cursor = await db.execute("SELECT * FROM bausteine WHERE topic = ? AND status = ? ORDER BY rowid", (topic, status)) + rows = await cursor.fetchall() + out = [] + for row in rows: + d = _row_to_dict(row, cursor) + d["quellen"] = json.loads(d.get("quellen") or "[]") + out.append(d) + return out + + +async def set_baustein_status(topic: str, titel_norm: str, status: str, titel: str | None = None, beschreibung: str | None = None) -> None: + """Status setzen; optional Titel/Beschreibung aktualisieren (z.B. nach semantischem Merge).""" + db = await get_db() + fields = {"status": status, "updated_at": _now()} + if titel is not None: + fields["titel"] = titel + if beschreibung is not None: + fields["beschreibung"] = beschreibung + sets = ", ".join(f"{k} = :{k}" for k in fields) + await db.execute( + f"UPDATE bausteine SET {sets} WHERE topic = :topic AND titel_norm = :titel_norm", + {**fields, "topic": topic, "titel_norm": titel_norm}, + ) + await db.commit() + + +async def delete_bausteine(topic: str) -> None: + db = await get_db() + await db.execute("DELETE FROM bausteine WHERE topic = ?", (topic,)) + await db.commit() + + +async def upsert_subbaustein(topic: str, baustein_norm: str, sub_norm: str, baustein: str, sub_titel: str) -> None: + db = await get_db() + await db.execute( + """INSERT INTO subbausteine (topic, baustein_norm, sub_norm, baustein, sub_titel, nennungen, status, updated_at) + VALUES (?, ?, ?, ?, ?, 1, 'kandidat', ?) + ON CONFLICT(topic, baustein_norm, sub_norm) DO UPDATE SET + nennungen = nennungen + 1, updated_at = excluded.updated_at""", + (topic, baustein_norm, sub_norm, baustein, sub_titel, _now()), + ) + await db.commit() + + +async def put_subbaustein(topic: str, baustein_norm: str, sub_norm: str, baustein: str, sub_titel: str, + stufe: str | None = None, relevanz: str | None = None, status: str = "konsens") -> None: + """Insert/Update OHNE Nennungszähler (Spiegel aus dem Sidecar). stufe/relevanz nur überschreiben, + wenn ein neuer Wert übergeben wird (COALESCE schützt Bestehendes).""" + db = await get_db() + await db.execute( + """INSERT INTO subbausteine (topic, baustein_norm, sub_norm, baustein, sub_titel, nennungen, stufe, relevanz, status, updated_at) + VALUES (?, ?, ?, ?, ?, 1, ?, ?, ?, ?) + ON CONFLICT(topic, baustein_norm, sub_norm) DO UPDATE SET + baustein = excluded.baustein, sub_titel = excluded.sub_titel, + stufe = COALESCE(excluded.stufe, subbausteine.stufe), + relevanz = COALESCE(excluded.relevanz, subbausteine.relevanz), + status = excluded.status, updated_at = excluded.updated_at""", + (topic, baustein_norm, sub_norm, baustein, sub_titel, stufe, relevanz, status, _now()), + ) + await db.commit() + + +async def list_subbausteine(topic: str, baustein_norm: str | None = None) -> list[dict]: + db = await get_db() + if baustein_norm is None: + cursor = await db.execute("SELECT * FROM subbausteine WHERE topic = ? ORDER BY rowid", (topic,)) + else: + cursor = await db.execute( + "SELECT * FROM subbausteine WHERE topic = ? AND baustein_norm = ? ORDER BY rowid", (topic, baustein_norm) + ) + rows = await cursor.fetchall() + return [_row_to_dict(row, cursor) for row in rows] + + +async def set_subbaustein_felder(topic: str, baustein_norm: str, sub_norm: str, **fields) -> None: + """Setzt Felder (stufe/relevanz/status/sub_titel) einer Subbaustein-Zeile.""" + fields["updated_at"] = _now() + sets = ", ".join(f"{k} = :{k}" for k in fields) + db = await get_db() + await db.execute( + f"UPDATE subbausteine SET {sets} WHERE topic = :topic AND baustein_norm = :baustein_norm AND sub_norm = :sub_norm", + {**fields, "topic": topic, "baustein_norm": baustein_norm, "sub_norm": sub_norm}, + ) + await db.commit() + + +async def delete_subbausteine(topic: str) -> None: + db = await get_db() + await db.execute("DELETE FROM subbausteine WHERE topic = ?", (topic,)) + await db.commit() + + +async def upsert_frage_muster(topic: str, baustein_norm: str, sub_norm: str, baustein: str, sub_titel: str, typ: str, frage: str) -> None: + db = await get_db() + await db.execute( + """INSERT INTO frage_muster (topic, baustein_norm, sub_norm, baustein, sub_titel, typ, frage, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(topic, baustein_norm, sub_norm, typ) DO UPDATE SET + frage = excluded.frage, updated_at = excluded.updated_at""", + (topic, baustein_norm, sub_norm, baustein, sub_titel, typ, frage, _now()), + ) + await db.commit() + + +async def list_frage_muster(topic: str, baustein_norm: str | None = None) -> list[dict]: + db = await get_db() + if baustein_norm is None: + cursor = await db.execute("SELECT * FROM frage_muster WHERE topic = ? ORDER BY rowid", (topic,)) + else: + cursor = await db.execute( + "SELECT * FROM frage_muster WHERE topic = ? AND baustein_norm = ? ORDER BY rowid", (topic, baustein_norm) + ) + rows = await cursor.fetchall() + return [_row_to_dict(row, cursor) for row in rows] + + +async def delete_frage_muster(topic: str) -> None: + db = await get_db() + await db.execute("DELETE FROM frage_muster WHERE topic = ?", (topic,)) + await db.commit() + + +async def mark_quellen_gelesen(topic: str, quellen: list[str]) -> None: + """Markiert die zitierten Crawl-Seiten als gelesen (Recherche-Loop-Abdeckung).""" + if not quellen: + return + db = await get_db() + now = _now() + await db.executemany( + """INSERT INTO recherche_coverage (topic, quelle, gelesen, updated_at) VALUES (?, ?, 1, ?) + ON CONFLICT(topic, quelle) DO UPDATE SET gelesen = 1, updated_at = excluded.updated_at""", + [(topic, q, now) for q in quellen], + ) + await db.commit() + + +async def list_coverage(topic: str) -> dict[str, int]: + db = await get_db() + cursor = await db.execute("SELECT quelle, gelesen FROM recherche_coverage WHERE topic = ?", (topic,)) + rows = await cursor.fetchall() + return {q: g for q, g in rows} + + +async def delete_coverage(topic: str) -> None: + db = await get_db() + await db.execute("DELETE FROM recherche_coverage WHERE topic = ?", (topic,)) + await db.commit() + + +async def set_step_status(topic: str, schritt: str, status: str) -> None: + db = await get_db() + await db.execute( + """INSERT INTO pipeline_state (topic, schritt, status, updated_at) VALUES (?, ?, ?, ?) + ON CONFLICT(topic, schritt) DO UPDATE SET status = excluded.status, updated_at = excluded.updated_at""", + (topic, schritt, status, _now()), + ) + await db.commit() + + +async def get_step_status(topic: str, schritt: str) -> str: + db = await get_db() + cursor = await db.execute( + "SELECT status FROM pipeline_state WHERE topic = ? AND schritt = ?", (topic, schritt) + ) + row = await cursor.fetchone() + return row[0] if row else "offen" + + +async def list_pipeline_state(topic: str) -> dict[str, str]: + db = await get_db() + cursor = await db.execute("SELECT schritt, status FROM pipeline_state WHERE topic = ?", (topic,)) + rows = await cursor.fetchall() + return {s: st for s, st in rows} + + +async def delete_pipeline_state(topic: str, schritte: list[str] | None = None) -> None: + db = await get_db() + if schritte is None: + await db.execute("DELETE FROM pipeline_state WHERE topic = ?", (topic,)) + elif schritte: + marks = ",".join("?" for _ in schritte) + await db.execute(f"DELETE FROM pipeline_state WHERE topic = ? AND schritt IN ({marks})", (topic, *schritte)) + await db.commit() + + +async def set_quelle(topic: str, type: str, ort: str = "", spec: str = "") -> None: + db = await get_db() + await db.execute( + """INSERT INTO quelle (topic, type, ort, spec, updated_at) VALUES (?, ?, ?, ?, ?) + ON CONFLICT(topic) DO UPDATE SET + type = excluded.type, ort = excluded.ort, spec = excluded.spec, updated_at = excluded.updated_at""", + (topic, type, ort, spec, _now()), + ) + await db.commit() + + +async def get_quelle(topic: str) -> dict | None: + db = await get_db() + cursor = await db.execute("SELECT type, ort, spec FROM quelle WHERE topic = ?", (topic,)) + row = await cursor.fetchone() + if row is None: + return None + return {"type": row[0], "ort": row[1], "spec": row[2]} + + +async def delete_quelle(topic: str) -> None: + db = await get_db() + await db.execute("DELETE FROM quelle WHERE topic = ?", (topic,)) + await db.commit() + + +async def set_guide_content(topic: str, format: str, content_json: str) -> None: + """Fertigen Guide-Inhalt (JSON-Blob) je Thema+Format speichern.""" + db = await get_db() + await db.execute( + """INSERT INTO guide_content (topic, format, json, updated_at) VALUES (?, ?, ?, ?) + ON CONFLICT(topic, format) DO UPDATE SET json = excluded.json, updated_at = excluded.updated_at""", + (topic, format, content_json, _now()), + ) + await db.commit() + + +async def get_guide_content(topic: str, format: str) -> str | None: + db = await get_db() + cursor = await db.execute("SELECT json FROM guide_content WHERE topic = ? AND format = ?", (topic, format)) + row = await cursor.fetchone() + return row[0] if row else None + + +async def delete_guide_content(topic: str, format: str | None = None) -> None: + db = await get_db() + if format is None: + await db.execute("DELETE FROM guide_content WHERE topic = ?", (topic,)) + else: + await db.execute("DELETE FROM guide_content WHERE topic = ? AND format = ?", (topic, format)) + await db.commit() + + +async def delete_topic_pipeline(topic: str) -> None: + """Alle Pipeline-Inhalte eines Themas verwerfen (Inventar/Subs/Muster/Coverage/State/Quelle).""" + db = await get_db() + for tab in ("bausteine", "subbausteine", "frage_muster", "recherche_coverage", "pipeline_state", "quelle"): + await db.execute(f"DELETE FROM {tab} WHERE topic = ?", (topic,)) + await db.commit() diff --git a/backend/guide.py b/backend/guide.py index 1c8d557..e4799e2 100644 --- a/backend/guide.py +++ b/backend/guide.py @@ -9,6 +9,7 @@ Schritt-Dateien bleiben liegen → Abbruch erhält Fortschritt, ▶ setzt am off """ import asyncio +import json import logging import math from datetime import datetime, timezone @@ -21,7 +22,7 @@ from config import ( LESBARKEIT_AKTIV, TEMPLATES_DIR, ) import lesbarkeit -from database import list_guides, update_guide +from database import list_guides, update_guide, list_bausteine, list_subbausteine, set_guide_content from fsutil import atomic_write_json, atomic_write_text from jsonio import read_json_file as _json_datei from paths import bausteine_path, guide_content_path, project_dir, subbausteine_path @@ -52,12 +53,19 @@ GUIDE_CHUNK = 10 LESE_RUNDEN = 1 -def _load_subbausteine(topic: str) -> dict[str, list[dict]]: - """Sidecar laden: {Baustein-Titel: [{titel, stufe}, …]}. Fehlt sie → {} (Fallback).""" +async def _load_subbausteine(topic: str) -> dict[str, list[dict]]: + """Subbausteine je Baustein — DB-first ({titel, stufe, relevanz}), Fallback Sidecar-Datei. + Fehlt beides → {} (Guide nimmt alles).""" + out: dict[str, list[dict]] = {} + for r in await list_subbausteine(topic): + if r["status"] == "konsens" and r["sub_titel"] and r["stufe"] in ("einfach", "mittel", "schwer"): + out.setdefault(r["baustein"], []).append( + {"titel": r["sub_titel"], "stufe": r["stufe"], "relevanz": r["relevanz"]}) + if out: + return out data = _json_datei(subbausteine_path(topic)) if not isinstance(data, dict): return {} - out: dict[str, list[dict]] = {} for titel, subs in data.items(): if not isinstance(subs, list): continue @@ -478,9 +486,9 @@ async def _generate_sections( "Wähle, was diesem Zweck dient — lass weg, was dafür nicht nötig ist." ) - # Subbausteine je Baustein (Sidecar) — früh geladen: steuert Auswahl + Sub-Filter je Format. + # Subbausteine je Baustein (DB-first) — früh geladen: steuert Auswahl + Sub-Filter je Format. # Fehlt sie → {} (Fallback: Guide nimmt alles). - subs_raw = _load_subbausteine(topic) + subs_raw = await _load_subbausteine(topic) def _hat_relevanz(num, art): return any(isinstance(s, dict) and s.get("relevanz") == art for s in subs_raw.get(_titel(entries[num]), [])) @@ -932,7 +940,13 @@ async def generate_guide(guide_id: str, topic: str, format_name: str, instructio for p_alt in guide_slot_dateien(content_path): p_alt.unlink(missing_ok=True) - alle = _lade_bausteine(bausteine_path(topic).read_text(encoding="utf-8")) + bs = await list_bausteine(topic, status="konsens") + if bs: + alle = {i: (f"{b['titel']} — {b['beschreibung']}" if b["beschreibung"] else b["titel"]) + for i, b in enumerate(bs, 1)} + else: # Fallback: bausteine.md (Alt-Themen) + bp = bausteine_path(topic) + alle = _lade_bausteine(bp.read_text(encoding="utf-8")) if bp.exists() else {} if not alle: await _fail(guide_id, "Keine Bausteine gefunden") return @@ -946,7 +960,8 @@ async def generate_guide(guide_id: str, topic: str, format_name: str, instructio return content = {"topic": topic, "format": format_name, "chapters": chapters} - atomic_write_json(content_path, content, indent=1) + atomic_write_json(content_path, content, indent=1) # Brücke (Resume/Fallback) + await set_guide_content(topic, format_name, json.dumps(content, ensure_ascii=False)) now = datetime.now(timezone.utc).isoformat() await update_guide(guide_id, status="done", progress=None, step=None, updated_at=now) diff --git a/backend/routes.py b/backend/routes.py index 2daabec..420ad2b 100644 --- a/backend/routes.py +++ b/backend/routes.py @@ -4,7 +4,7 @@ import uuid from datetime import datetime, timezone from fastapi import APIRouter, HTTPException -from fastapi.responses import FileResponse +from fastapi.responses import FileResponse, Response from agents import provider_available from config import PROJECTS_DIR, UNI_DIR, PROVIDERS @@ -15,6 +15,7 @@ from database import ( create_element, list_elements, get_element, update_element, delete_element, list_baustein_progress, get_baustein_progress, set_offene_frage, set_baustein_score, set_baustein_absolviert, set_baustein_verstanden, set_baustein_gemeistert, delete_baustein_daten, + delete_topic_pipeline, get_guide_content, delete_guide_content, ) from bausteine import generate_bausteine, cancel_bausteine, bausteine_status, active_bausteine, reset_bausteine, lade_quelle, lade_uebersicht, subbausteine_titel, lade_frage_muster from elements import generate_element, chat_with_guide, chat_with_element, check_element, style_element, refine_suggestion @@ -86,6 +87,8 @@ async def add_topic(req: TopicCreateRequest): async def remove_topic(topic: str): await delete_topic(topic) await delete_baustein_daten(topic) + await delete_topic_pipeline(topic) + await delete_guide_content(topic) shutil.rmtree(topic_dir(topic), ignore_errors=True) return {"ok": True} @@ -170,6 +173,7 @@ async def cancel_bausteine_route(topic: str): @router.delete("/bausteine") async def remove_bausteine(topic: str): reset_bausteine(topic) + await delete_topic_pipeline(topic) return {"ok": True} @@ -203,13 +207,13 @@ async def update_bausteine_quelle(req: BausteineQuelleUpdate): @router.get("/bausteine/uebersicht", response_model=list[BausteinUebersicht]) async def get_bausteine_uebersicht(topic: str): - return lade_uebersicht(topic) + return await lade_uebersicht(topic) @router.get("/bausteine/frage-muster") async def get_frage_muster(topic: str, baustein: str): """Vordefinierte Frage-Muster eines Bausteins (leer = Fallback auf Live-Generierung).""" - return {"muster": lade_frage_muster(topic, baustein)} + return {"muster": await lade_frage_muster(topic, baustein)} # --- Baustein-Lernen: Chat, Prüfung --- @@ -316,7 +320,7 @@ async def baustein_pruefung_route(req: BausteinPruefungRequest): frage = await pruefung_frage_variante(req.topic, req.baustein, req.section, kompakt, req.muster, provider=req.provider) else: # Fallback (kein Muster-Sidecar): Live-Generierung mit zufälligem Typ/Subbaustein. - subs = subbausteine_titel(req.topic, req.baustein) # Fokus-Kandidaten (zufällig gewählt) + subs = await subbausteine_titel(req.topic, req.baustein) # Fokus-Kandidaten (zufällig gewählt) frage = await pruefung_frage(req.topic, req.baustein, req.section, kompakt, msgs, subbausteine=subs, vermeide=req.vermeide, provider=req.provider) if frage is None: raise HTTPException(502, "Frage fehlgeschlagen — bitte erneut versuchen") @@ -496,7 +500,10 @@ async def guide_content(guide_id: str): raise HTTPException(404, "Guide nicht gefunden") if guide["status"] != "done": raise HTTPException(404, "Inhalt nicht verfügbar") - path = guide_content_path(guide["topic"], guide["format"]) + stored = await get_guide_content(guide["topic"], guide["format"]) # DB-first + if stored is not None: + return Response(content=stored, media_type="application/json") + path = guide_content_path(guide["topic"], guide["format"]) # Fallback: Datei (Alt-Themen) if not path.exists(): raise HTTPException(404, "Datei nicht gefunden") return FileResponse(path, media_type="application/json") @@ -611,6 +618,7 @@ async def remove(guide_id: str, slots: bool = False): # Content) bleibt fürs Resume erhalten, außer es wird explizit verlangt (slots=1). rest = [g for g in await list_guides() if g["topic"] == guide["topic"] and g["format"] == guide["format"]] if not rest: + await delete_guide_content(guide["topic"], guide["format"]) content = guide_content_path(guide["topic"], guide["format"]) if slots or content.exists(): for p in guide_slot_dateien(content): diff --git a/templates/Prompt/Bausteine-Klaerung.md b/templates/Prompt/Bausteine-Klaerung.md new file mode 100644 index 0000000..3e2a900 --- /dev/null +++ b/templates/Prompt/Bausteine-Klaerung.md @@ -0,0 +1,17 @@ +Zum Thema "{topic}" sind unten Bausteine, die in der Recherche nur EINMAL genannt wurden. Die Quelle wurde vollständig durchsucht. Entscheide jeden Eintrag einzeln. + +BEREITS BESTÄTIGTES INVENTAR (nur Kontext — nicht ändern, nicht duplizieren): +{konsens} + +EINZELN GENANNTE EINTRÄGE (jeweils entscheiden): +{rest} + +Regeln: +- Fachlich valider, eigenständiger Baustein und KEINE Dublette/Variante des Inventars → `aufnehmen`. +- Erfunden, zu vage, oder Dublette/Variante eines Inventar-Bausteins → verwerfen (in keine Liste). +- Übernimm aufgenommene Einträge WÖRTLICH ("Titel — Kurzbeschreibung"), nicht umformulieren.{final} + +Schreibe NUR die JSON-Datei nach: {out_path} + +Format (kein weiterer Text in der Datei): +{{"aufnehmen": ["Titel — Kurzbeschreibung"], "rest": []}} diff --git a/templates/Prompt/Bausteine-Recherche.md b/templates/Prompt/Bausteine-Recherche.md index c2cf381..fd076d6 100644 --- a/templates/Prompt/Bausteine-Recherche.md +++ b/templates/Prompt/Bausteine-Recherche.md @@ -15,4 +15,6 @@ Regeln: Schreibe NUR die Markdown-Datei nach: {bausteine_path} Format: GENAU eine Zeile pro Baustein: `N. Titel — Kurzbeschreibung — Quelle` +Die Quelle (3. Segment) MUSS der exakte Dateiname bzw. die URL der Crawl-Seite sein, aus der der Baustein stammt — sie steuert die Abdeckungs-Prüfung. +{fokus} {extra} \ No newline at end of file diff --git a/templates/Prompt/Subbaustein-Recherche.md b/templates/Prompt/Subbaustein-Recherche.md index c0e169c..e65ac60 100644 --- a/templates/Prompt/Subbaustein-Recherche.md +++ b/templates/Prompt/Subbaustein-Recherche.md @@ -25,4 +25,5 @@ Schreibe NUR die Datei {out_path} — pro Baustein ein baustein-Marker (Titel EX - Zweiter Subbaustein Die Marker-Zeile exakt so schreiben. Kein Text außerhalb der Bausteine. +{bekannt} {extra}