2276 lines
108 KiB
Python
2276 lines
108 KiB
Python
"""Bausteine-Pipeline: Recherche-Konsens + Klärungs-Loop — reines Inventar, unsortiert.
|
||
|
||
5x Recherche (min. 3, Grace) → Mapping (Konsens/Rest) → Klärungs-Loop (max.
|
||
KONSENS_MAX_RUNDEN Runden): 3 Auswahl-Agenten (min. 2, Grace) entscheiden
|
||
über den strittigen Rest, ein Mapping-Agent sortiert in aufnehmen/verwerfen/
|
||
weiter strittig. Leerer Rest beendet den Loop; die letzte Runde muss alles
|
||
entscheiden. Races nutzen ein Grace-Fenster statt „erste N gewinnen": Nach dem
|
||
ersten gültigen Ergebnis dürfen die übrigen Agenten KONSENS_GRACE Sekunden
|
||
fertig werden. Der Konsens wird im Code akkumuliert — kein Agent re-emittiert
|
||
die Gesamtliste.
|
||
"""
|
||
|
||
import asyncio
|
||
import json
|
||
import logging
|
||
import math
|
||
import re
|
||
import shutil
|
||
import subprocess
|
||
import time
|
||
from pathlib import Path
|
||
|
||
import database as db
|
||
from agents import kill_process, cancel_scope, clear_scope, run_agent
|
||
from config import KONSENS_GRACE, RECHERCHE_GRACE, KONSENS_MAX_RUNDEN, DEFAULT_PROVIDER, CRAWL_KEEP_PATTERNS, CRAWL_NOISE_PATTERNS, CRAWL_MIN_CHARS
|
||
from fsutil import atomic_write_text, atomic_write_json
|
||
from jsonio import read_json_file as _json_datei
|
||
from paths import arbeit_dir, bausteine_path, frage_muster_path, project_dir, subbausteine_path, quelle_path, quelle_crawl_dir, safe_ordner
|
||
from crawl import crawl
|
||
from pipeline import (
|
||
CANCELLED, FAILED, GenContext, _extra, _gather_fortschritt, _log, _prompt, _race, _relevanz_schema,
|
||
_runde_schema, _semaphore, _str_liste, _stufen_schema, _timeout, run_single_slot,
|
||
)
|
||
from textkit import (
|
||
_eindeutige_titel, _lade_bausteine, _norm_titel, _parse_auswahl, _parse_subbausteine, _titel,
|
||
_titel_aufloesen, _titel_index,
|
||
)
|
||
|
||
# Subbausteine (Websuche je Baustein) chunken: 1 Agent je ~10 Bausteine, gedeckelt.
|
||
SUBBAUSTEIN_CHUNK = 10
|
||
SUBBAUSTEIN_MAX = 40
|
||
# Einstufen ist billig (kurzes Urteil, keine Websuche) → größere Pakete, weniger Dateien/Agenten.
|
||
STUFE_CHUNK = 100
|
||
|
||
# Recherche: feste Datei-Batches statt Such-Loop → jede Crawl-Seite genau einmal zugeteilt.
|
||
RECHERCHE_BATCH = 20 # Crawl-Seiten je Batch
|
||
RECHERCHE_READERS = 2 # Reader-Agenten je Batch (Konsens ≥2 innerhalb des Batches)
|
||
RECHERCHE_THEMA_AGENTEN = 5 # Web-Modus (Quelle „thema", kein Crawl-Ordner)
|
||
RECHERCHE_KAPPE = 1800 # Sicherheits-Deckel je Batch-Agent
|
||
# uni/projekt: Skript-Text in Abschnitte ~dieser Größe chunken (gegen Lost-in-the-Middle bei
|
||
# großen Dokumenten). ~12k Zeichen ≈ 3k Token → sicher unter der Recall-Abfall-Schwelle.
|
||
RECHERCHE_ABSCHNITT_ZEICHEN = 12000
|
||
# Sichtung (Content/Noise) ist jetzt ein deterministischer Regel-Filter (config.CRAWL_*).
|
||
SUBBAUSTEIN_KAPPE = 900 # Subbaustein-Finde-Loop je Chunk (15 min)
|
||
KONSOLIDIERUNG_CHUNK = 600 # bis hierher EIN globaler Judge (dedupt alles); darüber chunked + Merge-Pass
|
||
# Frage-Muster-Chunks per LPT nach Sub-Last balancieren (Makespan), statt nach Baustein-Anzahl.
|
||
FRAGE_CHUNK_SUBS = 50 # Ziel-Summe relevanter Subs je Chunk
|
||
FAKTEN_CHUNK_SUBS = 25 # Fakten-Extraktion: kleinere Chunks (Fakten sind umfangreicher als Muster)
|
||
FAKTEN_CHECK_PANEL = 3 # Judges je Chunk im Fakten-Check (Mehrheit beanstandet)
|
||
|
||
log = logging.getLogger("creator.bausteine")
|
||
|
||
_bausteine_progress: dict[str, str] = {}
|
||
_bausteine_errors: dict[str, str] = {}
|
||
_bausteine_cancelled: set[str] = set()
|
||
_bausteine_step: dict[str, int] = {}
|
||
|
||
BAUSTEINE_STEPS = (
|
||
"Recherche", "Konsolidierung", "Klärung",
|
||
"Subbausteine finden", "Subbausteine wählen", "Subbausteine klären",
|
||
"Fakten finden", "Fakten prüfen", "Fakten fix",
|
||
"Stufen finden", "Stufen wählen", "Stufen klären",
|
||
"Relevanz finden", "Relevanz wählen", "Relevanz klären",
|
||
"Gliederung",
|
||
"Fragen finden", "Fragen wählen", "Fragen klären", "Fragen prüfen",
|
||
"Karteikarten", "Beispiele",
|
||
)
|
||
ARTEFAKT_TYPEN = ("karteikarte", "beispiel")
|
||
|
||
|
||
def lade_quelle(topic: str) -> dict:
|
||
"""Persistierte Quellen-Wahl lesen. Fallback (Alt-Themen ohne quelle.json):
|
||
existiert projects/<topic> → projekt, sonst thema."""
|
||
q = _json_datei(quelle_path(topic))
|
||
if isinstance(q, dict) and q.get("type") in ("thema", "projekt", "uni", "link"):
|
||
return q
|
||
if project_dir(topic).is_dir():
|
||
return {"type": "projekt", "ort": f"projects/{topic}", "spec": ""}
|
||
return {"type": "thema", "ort": "", "spec": ""}
|
||
|
||
|
||
def quelle_ordner(topic: str) -> Path | None:
|
||
"""Ordner-Quelle (projekt/uni → Pfad, link → Crawl-Ordner) — sonst None (thema)."""
|
||
q = lade_quelle(topic)
|
||
if q["type"] == "link":
|
||
return quelle_crawl_dir(topic)
|
||
if q["type"] in ("projekt", "uni"):
|
||
return safe_ordner(q.get("ort", ""))
|
||
return None
|
||
|
||
|
||
def _crawl_fertig(topic: str) -> bool:
|
||
return (quelle_crawl_dir(topic) / ".done").exists() # Marker erst bei sauberem Abschluss
|
||
|
||
|
||
# Lernpfad-Stufen (anfaenger/fortgeschritten/experte); alte Schwierigkeits-Werte abwärtskompatibel.
|
||
_STUFEN = ("anfaenger", "fortgeschritten", "experte", "einfach", "mittel", "schwer")
|
||
|
||
|
||
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 []
|
||
return [
|
||
t for s in (sc.get(baustein) or [])
|
||
if isinstance(s, dict) and (t := str(s.get("titel", "")).strip())
|
||
]
|
||
|
||
|
||
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"], "frage": r["frage"]} for r in rows if r["frage"]]
|
||
fm = _json_datei(frage_muster_path(topic))
|
||
if not isinstance(fm, dict):
|
||
return []
|
||
return [
|
||
{"subbaustein": str(e.get("subbaustein", "")).strip(), "frage": frage}
|
||
for e in (fm.get(baustein) or [])
|
||
if isinstance(e, dict) and (frage := str(e.get("frage", "")).strip())
|
||
]
|
||
|
||
|
||
async def subbausteine_frei(topic: str, baustein: str, max_ebene: int) -> list[str]:
|
||
"""Subbaustein-Titel bis zur freigeschalteten Ebene (≤ max_ebene). Fallback ohne
|
||
Ebenen-Wissen (Altbestand/Sidecar): alle Subbaustein-Titel."""
|
||
rows = await db.subs_mit_ebene(topic, baustein)
|
||
if not rows:
|
||
return await subbausteine_titel(topic, baustein)
|
||
return [s["titel"] for s in rows if s["ebene"] <= max_ebene and s["titel"]]
|
||
|
||
|
||
async def lade_frage_muster_frei(topic: str, baustein: str, max_ebene: int) -> list[dict]:
|
||
"""Frage-Muster, gefiltert auf Subbausteine bis zur freigeschalteten Ebene. Ohne
|
||
Ebenen-Wissen (Altbestand/Sidecar) ungefiltert."""
|
||
rows = await db.subs_mit_ebene(topic, baustein)
|
||
if not rows:
|
||
return await lade_frage_muster(topic, baustein)
|
||
frei = {s["norm"] for s in rows if s["ebene"] <= max_ebene}
|
||
return [m for m in await lade_frage_muster(topic, baustein) if _norm_titel(m["subbaustein"]) in frei]
|
||
|
||
|
||
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 "fortgeschritten",
|
||
"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 {}
|
||
out = []
|
||
for num, entry in entries.items():
|
||
titel = _titel(entry)
|
||
teile = entry.split(" — ", 1)
|
||
beschreibung = teile[1].strip() if len(teile) == 2 else ""
|
||
subbausteine = [
|
||
{
|
||
"titel": t,
|
||
"stufe": s.get("stufe") if s.get("stufe") in _STUFEN else "fortgeschritten",
|
||
"relevanz": s.get("relevanz") if s.get("relevanz") in ("relevant", "rand") else None,
|
||
}
|
||
for s in (sidecar.get(titel) or [])
|
||
if isinstance(s, dict) and (t := str(s.get("titel", "")).strip())
|
||
]
|
||
out.append({"num": num, "titel": titel, "beschreibung": beschreibung, "subbausteine": subbausteine})
|
||
return out
|
||
|
||
|
||
def _bausteine_steps(topic: str) -> tuple:
|
||
"""Schritte je Quelle: link bekommt vorne „Quelle laden", projekt zusätzlich „Ergänzung".
|
||
|
||
Subbausteine + Stufen sind je drei Phasen (Finden, Wählen, Klären). Pro Phase
|
||
laufen alle Pakete parallel; der Schritt bleibt, bis das letzte Paket fertig ist.
|
||
"""
|
||
q = lade_quelle(topic)
|
||
base = ("Recherche", "Konsolidierung", "Klärung")
|
||
rest = (
|
||
"Subbausteine finden", "Subbausteine wählen", "Subbausteine klären",
|
||
"Fakten finden", "Fakten prüfen", "Fakten fix",
|
||
"Stufen finden", "Stufen wählen", "Stufen klären",
|
||
"Relevanz finden", "Relevanz wählen", "Relevanz klären",
|
||
"Gliederung",
|
||
"Fragen finden", "Fragen wählen", "Fragen klären", "Fragen prüfen",
|
||
"Karteikarten", "Beispiele",
|
||
)
|
||
mitte = base + (("Ergänzung",) if q["type"] == "projekt" else ()) + rest
|
||
return (("Quelle aufbereiten",) if q["type"] == "link" else ()) + mitte
|
||
|
||
|
||
def _step_idx(topic: str, name: str) -> int:
|
||
return _bausteine_steps(topic).index(name)
|
||
|
||
|
||
def _melde_p(set_p, topic: str, schritt: str):
|
||
"""Async-Melde-Callback für _gather_fortschritt: setzt „<Schritt> d/t…" + Schritt-Index."""
|
||
idx = _step_idx(topic, schritt)
|
||
async def melde(d, t):
|
||
set_p(f"{schritt} {d}/{t}…", step=idx)
|
||
return melde
|
||
|
||
|
||
# Grobe Anzeige-Phasen: bündeln die Feinschritte (intern bleibt alles feingranular).
|
||
# Sonderschritte (Quelle laden, Ergänzung) gehören zur Phase „Inventar".
|
||
PHASEN = (
|
||
("Quelle", ("Quelle aufbereiten",)),
|
||
("Inventar", ("Recherche", "Konsolidierung", "Klärung", "Ergänzung")),
|
||
("Subbausteine", ("Subbausteine finden", "Subbausteine wählen", "Subbausteine klären")),
|
||
("Fakten", ("Fakten finden", "Fakten prüfen", "Fakten fix")),
|
||
("Stufen", ("Stufen finden", "Stufen wählen", "Stufen klären")),
|
||
("Relevanz", ("Relevanz finden", "Relevanz wählen", "Relevanz klären")),
|
||
("Gliederung", ("Gliederung",)),
|
||
("Fragen", ("Fragen finden", "Fragen wählen", "Fragen klären", "Fragen prüfen")),
|
||
("Artefakte", ("Karteikarten", "Beispiele")),
|
||
)
|
||
|
||
|
||
def _phasen(topic: str) -> list[tuple[str, int]]:
|
||
"""[(grob_label, Anzahl vorhandener Feinschritte)] für die aktuelle Quelle."""
|
||
feine = _bausteine_steps(topic)
|
||
return [(label, n) for label, members in PHASEN if (n := sum(f in members for f in feine))]
|
||
|
||
|
||
def _phasen_status(topic: str, current: int | None) -> list[dict]:
|
||
"""Grobe Phasen-Zustände aus dem feinen Fortschritt `current` (None = alles pending,
|
||
len(feine) = alles done). → [{label, state}] mit state done/active/pending."""
|
||
out, start = [], 0
|
||
for label, n in _phasen(topic):
|
||
end = start + n
|
||
if current is None or current < start:
|
||
state = "pending"
|
||
elif current >= end:
|
||
state = "done"
|
||
else:
|
||
state = "active"
|
||
out.append({"label": label, "state": state})
|
||
start = end
|
||
return out
|
||
|
||
|
||
def _bausteine_files(topic: str) -> dict:
|
||
arbeit = arbeit_dir(topic)
|
||
runden = range(1, KONSENS_MAX_RUNDEN + 1)
|
||
return {
|
||
"final": bausteine_path(topic),
|
||
"arbeit": arbeit,
|
||
"recherche": [arbeit / f"recherche-{i}.md" for i in (1, 2, 3, 4, 5)],
|
||
"recherche_mapping": arbeit / "recherche-mapping.json",
|
||
"auswahl": {n: [arbeit / f"auswahl-r{n}-{i}.json" for i in (1, 2, 3)] for n in runden},
|
||
"mapping": {n: arbeit / f"auswahl-mapping-r{n}.json" for n in runden},
|
||
"ergaenzung": arbeit / "ergaenzung.json",
|
||
"sub_roh": arbeit / "subbausteine-roh.json",
|
||
"fakten": arbeit / "subbausteine-fakten.json",
|
||
"sidecar": subbausteine_path(topic),
|
||
"frage_muster": frage_muster_path(topic),
|
||
"gliederung": arbeit / "gliederung.json",
|
||
"gliederung_slots": [arbeit / f"gliederung-{i}.json" for i in (1, 2, 3)],
|
||
"artefakte": arbeit / "artefakte.json",
|
||
}
|
||
|
||
|
||
def _alle_slot_dateien(files: dict) -> list[Path]:
|
||
arbeit = files["arbeit"]
|
||
# Subbaustein-/Stufen-Slots sind pro Chunk dynamisch — per Glob einsammeln.
|
||
dyn = (list(arbeit.glob("subbaustein-*")) + list(arbeit.glob("fakten-*")) + list(arbeit.glob("stufe-*")) + list(arbeit.glob("relevanz-*"))
|
||
+ list(arbeit.glob("frage-muster-*")) + list(arbeit.glob("gliederung-*")) + list(arbeit.glob("artefakt-*"))
|
||
+ list(arbeit.glob("recherche-*")) + list(arbeit.glob("konsolidierung-*"))
|
||
+ list(arbeit.glob("klaerung*"))) if arbeit.is_dir() else []
|
||
return [
|
||
*files["recherche"], files["recherche_mapping"],
|
||
*(p for slots in files["auswahl"].values() for p in slots),
|
||
*files["mapping"].values(), files["ergaenzung"],
|
||
files["sub_roh"], files["sidecar"], files["frage_muster"],
|
||
files["fakten"], files["gliederung"], files["artefakte"], *dyn,
|
||
]
|
||
|
||
|
||
def cancel_bausteine(topic: str) -> bool:
|
||
if topic not in _bausteine_progress:
|
||
return False
|
||
_bausteine_cancelled.add(topic)
|
||
cancel_scope(f"bausteine-{topic}-") # wartende Agenten bailen vorm Spawn
|
||
kill_process(f"bausteine-{topic}-") # laufende Subprozesse killen
|
||
return True
|
||
|
||
|
||
def _resume_step(topic: str) -> int:
|
||
"""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 aufbereiten")
|
||
if not files["final"].exists(): # Inventar (DB-Loop) noch offen
|
||
return _step_idx(topic, "Recherche")
|
||
if q["type"] == "projekt" and not files["ergaenzung"].exists():
|
||
return _step_idx(topic, "Ergänzung")
|
||
sidecar = _json_datei(files["sidecar"])
|
||
if _sidecar_schema(sidecar) is not None:
|
||
# Stufen fertig; nur noch Relevanz offen?
|
||
if not _relevanz_komplett(sidecar):
|
||
return _step_idx(topic, "Relevanz finden")
|
||
# Relevanz fertig; Gliederung (Bausteine-Artefakt für den Guide) offen?
|
||
if not _gliederung_komplett(files):
|
||
return _step_idx(topic, "Gliederung")
|
||
# Gliederung fertig; Frage-Muster offen?
|
||
if not _frage_muster_komplett(topic):
|
||
return _step_idx(topic, "Fragen finden")
|
||
# Fragen fertig; Lern-Artefakte (Karteikarten/Beispiele) offen?
|
||
if not _artefakte_komplett(files):
|
||
return _step_idx(topic, "Karteikarten")
|
||
return len(_bausteine_steps(topic))
|
||
if _sub_roh_schema(_json_datei(files["sub_roh"])) is None:
|
||
return _step_idx(topic, "Subbausteine finden")
|
||
# Subbausteine fertig; Fakten noch offen? (Fakten kommen vor den Stufen.)
|
||
if not _fakten_komplett(files):
|
||
return _step_idx(topic, "Fakten finden")
|
||
return _step_idx(topic, "Stufen finden")
|
||
|
||
|
||
def bausteine_status(topic: str) -> dict:
|
||
# Intern feingranular (Resume/Progress); für die Anzeige zu 5 groben Phasen gebündelt.
|
||
feine = _bausteine_steps(topic)
|
||
ready = bausteine_path(topic).exists()
|
||
generating = topic in _bausteine_progress
|
||
partial = False
|
||
if generating:
|
||
current = _bausteine_step.get(topic)
|
||
elif ready:
|
||
current = len(feine) # alles fertig → alle Phasen done
|
||
else:
|
||
current = _resume_step(topic)
|
||
partial = current > 0
|
||
return {
|
||
"ready": ready,
|
||
"generating": generating,
|
||
"progress": _bausteine_progress.get(topic),
|
||
"error": _bausteine_errors.get(topic),
|
||
"partial": partial,
|
||
"steps": _phasen_status(topic, current),
|
||
}
|
||
|
||
|
||
def active_bausteine() -> list[dict]:
|
||
return [{"topic": t, "progress": p} for t, p in _bausteine_progress.items()]
|
||
|
||
|
||
def reset_bausteine(topic: str) -> None:
|
||
"""„Entfernen": löscht den GESAMTEN Bausteine-Bereich — Crawl, Sichtung, Inventar … Fragen.
|
||
BEHÄLT nur die Themen-Config `quelle.json` (Typ/Link/Spec). Re-Generieren crawlt neu.
|
||
(Crawl/Sichtung gehören zu den Bausteinen; nur die Config ist „Thema".)"""
|
||
files = _bausteine_files(topic)
|
||
files["final"].unlink(missing_ok=True)
|
||
files["sidecar"].unlink(missing_ok=True)
|
||
files["frage_muster"].unlink(missing_ok=True)
|
||
shutil.rmtree(quelle_crawl_dir(topic), ignore_errors=True) # Crawl gehört zu Bausteinen
|
||
shutil.rmtree(files["arbeit"], ignore_errors=True)
|
||
_bausteine_errors.pop(topic, None)
|
||
# quelle.json bleibt bewusst stehen — das ist die Themen-Config.
|
||
|
||
|
||
def _phase_idx(label: str) -> int:
|
||
"""Index der groben Phase in der kanonischen Reihenfolge (Quelle=0 … Fragen=5)."""
|
||
order = [l for l, _ in PHASEN]
|
||
return order.index(label) if label in order else 1
|
||
|
||
|
||
def _reset_ab_phase(topic: str, label: str) -> None:
|
||
"""Datei-Artefakte AB der groben Phase `label` löschen (Quelle/Inventar … Fragen), frühere
|
||
behalten. Kumulativ. quelle.json + Crawl (.done) bleiben immer (re-crawl nur bei Voll-Reset)."""
|
||
files = _bausteine_files(topic)
|
||
arbeit = files["arbeit"]
|
||
idx = _phase_idx(label)
|
||
|
||
def glob_del(pat: str) -> None:
|
||
if arbeit.is_dir():
|
||
for p in arbeit.glob(pat):
|
||
p.unlink(missing_ok=True)
|
||
|
||
# Phasen-Index: Quelle=0 · Inventar=1 · Subbausteine=2 · Fakten=3 · Stufen=4 · Relevanz=5 · Gliederung=6 · Fragen=7 · Artefakte=8
|
||
if idx <= 8: # Artefakte (Karteikarten/Beispiele)
|
||
files["artefakte"].unlink(missing_ok=True)
|
||
glob_del("artefakt-*")
|
||
if idx <= 7: # Fragen
|
||
files["frage_muster"].unlink(missing_ok=True)
|
||
glob_del("frage-muster-*")
|
||
if idx <= 6: # Gliederung
|
||
files["gliederung"].unlink(missing_ok=True)
|
||
glob_del("gliederung-*")
|
||
if idx <= 5: # Relevanz
|
||
glob_del("relevanz-*")
|
||
if idx <= 4: # Stufen + Relevanz teilen die Sidecar → ab Stufen ganz neu
|
||
files["sidecar"].unlink(missing_ok=True)
|
||
glob_del("stufe-*")
|
||
else: # ab Relevanz: Stufen behalten, nur Relevanz-Felder strippen
|
||
sc = _json_datei(files["sidecar"])
|
||
if isinstance(sc, dict):
|
||
for subs in sc.values():
|
||
for s in (subs if isinstance(subs, list) else []):
|
||
if isinstance(s, dict):
|
||
s.pop("relevanz", None)
|
||
atomic_write_json(files["sidecar"], sc, indent=1)
|
||
if idx <= 3: # Fakten (vor den Stufen) — Fakten-Map + Arbeitsdateien weg
|
||
files["fakten"].unlink(missing_ok=True)
|
||
glob_del("fakten-*")
|
||
if idx <= 2: # Subbausteine
|
||
files["sub_roh"].unlink(missing_ok=True)
|
||
glob_del("subbaustein-*")
|
||
if idx <= 1: # Inventar (und Quelle) = Inventar-Dateien + bausteine.md weg
|
||
for p_alt in _alle_slot_dateien(files):
|
||
p_alt.unlink(missing_ok=True)
|
||
files["final"].unlink(missing_ok=True)
|
||
|
||
|
||
def _ergaenzung_schema(data):
|
||
"""{"bausteine": [{"titel", "beschreibung"}]} → Liste (leer erlaubt) · sonst None."""
|
||
if not isinstance(data, dict) or not isinstance(data.get("bausteine"), list):
|
||
return None
|
||
out = []
|
||
for b in data["bausteine"]:
|
||
if not isinstance(b, dict) or not isinstance(b.get("titel"), str) or not isinstance(b.get("beschreibung"), str):
|
||
return None
|
||
titel, beschreibung = b["titel"].strip(), b["beschreibung"].strip()
|
||
if not titel:
|
||
return None
|
||
out.append((titel, beschreibung))
|
||
return out
|
||
|
||
|
||
def _pdfs_konvertieren(project: Path) -> None:
|
||
"""PDFs im Projekt in .txt wandeln (pdftotext) — Agenten lesen Text statt Seiten-Bildern.
|
||
|
||
Wird vor jeder Projekt-Generierung aufgerufen; konvertiert nur, wenn die
|
||
.txt fehlt oder älter als das PDF ist. Das Original bleibt unangetastet.
|
||
Fehlt pdftotext und das Projekt enthält PDFs → harter Fehler statt
|
||
unzuverlässigem Direkt-Lese-Modus (MiniMax-Bilderlimit, Vision-Kosten).
|
||
"""
|
||
pdfs = list(project.rglob("*.pdf"))
|
||
if not pdfs:
|
||
return
|
||
if shutil.which("pdftotext") is None:
|
||
raise RuntimeError("pdftotext fehlt (poppler-utils installieren) — PDFs im Projekt können nicht gelesen werden")
|
||
for pdf in pdfs:
|
||
txt = pdf.with_suffix(".txt")
|
||
if txt.exists() and txt.stat().st_mtime >= pdf.stat().st_mtime:
|
||
continue
|
||
try:
|
||
subprocess.run(["pdftotext", "-layout", str(pdf), str(txt)], check=True, timeout=120)
|
||
_log(project.name, f"PDF konvertiert: {pdf.name} → {txt.name}")
|
||
except Exception as e:
|
||
raise RuntimeError(f"PDF-Konvertierung fehlgeschlagen ({pdf.name}): {e}") from e
|
||
|
||
|
||
_QUELLE_TEMPLATE = {"projekt": "Bausteine-Quelle-Projekt", "uni": "Bausteine-Quelle-Uni", "link": "Bausteine-Quelle-Link"}
|
||
|
||
|
||
def _text_abschnitte(text: str, ziel: int = RECHERCHE_ABSCHNITT_ZEICHEN) -> list[str]:
|
||
"""Text an Absatz-/Zeilengrenzen in Abschnitte ~`ziel` Zeichen splitten (gegen Lost-in-the-Middle
|
||
bei großen Dokumenten). Kleiner Text bleibt EIN Abschnitt. Inhalt bleibt vollständig — nur
|
||
Trenn-Whitespace fällt weg."""
|
||
text = text.strip()
|
||
if len(text) <= ziel:
|
||
return [text] if text else []
|
||
abschnitte: list[str] = []
|
||
buf = ""
|
||
|
||
def flush():
|
||
nonlocal buf
|
||
if buf.strip():
|
||
abschnitte.append(buf.strip())
|
||
buf = ""
|
||
|
||
for block in re.split(r"\n\s*\n", text): # an Absatz-Grenzen
|
||
block = block.strip()
|
||
if not block:
|
||
continue
|
||
if len(block) > ziel: # einzelner Riesen-Absatz → hart an Zeilen schneiden
|
||
flush()
|
||
for zeile in block.split("\n"):
|
||
if buf and len(buf) + len(zeile) + 1 > ziel:
|
||
flush()
|
||
buf += zeile + "\n"
|
||
flush()
|
||
elif buf and len(buf) + len(block) + 2 > ziel:
|
||
flush()
|
||
buf = block
|
||
else:
|
||
buf = (buf + "\n\n" + block) if buf else block
|
||
flush()
|
||
return abschnitte
|
||
|
||
|
||
def _build_recherche_prompt(topic: str, out_path: Path, instructions: str, typ: str, ordner: Path | None, fokus: str = "", abschnitt: str = "") -> str:
|
||
if abschnitt:
|
||
# Abschnitt-Modus (uni/projekt): Text direkt im Prompt → kleiner Kontext, kein Datei-Lesen.
|
||
source = abschnitt
|
||
elif 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, fokus=fokus, extra=_extra(instructions),
|
||
)
|
||
|
||
|
||
def _file_payload(path: Path):
|
||
"""Gültig, wenn die Slot-Datei existiert und nummerierte Einträge enthält."""
|
||
if not path.exists():
|
||
return None
|
||
text = path.read_text(encoding="utf-8")
|
||
return text if _parse_auswahl(text) else None
|
||
|
||
|
||
def _mapping_schema(data):
|
||
"""{"bausteine": [str, ≥1], "rest": [str]} → (bausteine, rest) · sonst None."""
|
||
if not isinstance(data, dict):
|
||
return None
|
||
bausteine = _str_liste(data.get("bausteine"))
|
||
rest = _str_liste(data.get("rest"))
|
||
if not bausteine or rest is None:
|
||
return None
|
||
return bausteine, rest
|
||
|
||
|
||
def _sub_roh_schema(data):
|
||
"""{Baustein-Titel: [Subbaustein, …]} → dict · sonst None (Zwischenstand Block B)."""
|
||
if not isinstance(data, dict) or not data:
|
||
return None
|
||
out: dict[str, list[str]] = {}
|
||
for k, v in data.items():
|
||
subs = _str_liste(v) if isinstance(v, list) else None
|
||
if not isinstance(k, str) or not k.strip() or not subs:
|
||
return None
|
||
out[k] = subs
|
||
return out
|
||
|
||
|
||
def _sidecar_schema(data):
|
||
"""{Baustein-Titel: [{titel, stufe}, …]} → dict · sonst None (Sidecar mit Stufen)."""
|
||
if not isinstance(data, dict) or not data:
|
||
return None
|
||
for v in data.values():
|
||
if not isinstance(v, list) or not v:
|
||
return None
|
||
for s in v:
|
||
if not isinstance(s, dict) or not str(s.get("titel", "")).strip() or s.get("stufe") not in _STUFEN:
|
||
return None
|
||
return data
|
||
|
||
|
||
def _relevanz_komplett(data) -> bool:
|
||
"""Jeder Subbaustein der Sidecar trägt eine gültige Relevanz (relevant/rand)?"""
|
||
if not isinstance(data, dict) or not data:
|
||
return False
|
||
return all(
|
||
isinstance(s, dict) and s.get("relevanz") in ("relevant", "rand")
|
||
for v in data.values() if isinstance(v, list)
|
||
for s in v
|
||
)
|
||
|
||
|
||
|
||
|
||
def _frage_muster_chunk_schema(data) -> list[dict] | None:
|
||
"""{"muster": [{baustein, subbaustein, frage}, …]} → Liste valider Einträge · sonst None.
|
||
|
||
Ein Muster je Subbaustein (kein Typ-Kreuzprodukt — die Schwierigkeit kommt erst bei der
|
||
Prüfung aus dem Lerner-Niveau). Ungültige Einzel-Einträge werden übersprungen."""
|
||
if not isinstance(data, dict) or not isinstance(data.get("muster"), list):
|
||
return None
|
||
out = []
|
||
for e in data["muster"]:
|
||
if not isinstance(e, dict):
|
||
continue
|
||
bau = str(e.get("baustein", "")).strip()
|
||
sub = str(e.get("subbaustein", "")).strip()
|
||
frage = str(e.get("frage", "")).strip()
|
||
if not bau or not sub or not frage:
|
||
continue
|
||
out.append({"baustein": bau, "subbaustein": sub, "frage": frage})
|
||
return out or None
|
||
|
||
|
||
def _frage_muster_komplett(topic: str) -> bool:
|
||
"""Frage-Muster-Sidecar existiert (Build gelaufen)? Einzelne leere Bausteine
|
||
fallen zur Prüfungszeit auf Live-Generierung zurück — daher genügt die Datei."""
|
||
return isinstance(_json_datei(frage_muster_path(topic)), dict)
|
||
|
||
|
||
def _read(p: Path) -> str:
|
||
return p.read_text(encoding="utf-8") if p.exists() else ""
|
||
|
||
|
||
def _chunk_nums(items: list, n: int) -> list[list]:
|
||
"""Teilt eine flache Liste in n möglichst gleich große Chunks."""
|
||
n = max(1, n)
|
||
size = max(1, math.ceil(len(items) / n))
|
||
return [items[i:i + size] for i in range(0, len(items), size)]
|
||
|
||
|
||
def _n_chunks(count: int, size: int = SUBBAUSTEIN_CHUNK) -> int:
|
||
return min(SUBBAUSTEIN_MAX, max(1, math.ceil(count / size)))
|
||
|
||
|
||
def _lpt_chunks(gewichte: list[int], target: int) -> list[list[int]]:
|
||
"""Indizes lastbalanciert auf Chunks verteilen (LPT, Makespan-minimal). Gewicht = Kosten je Index.
|
||
K = ceil(Gesamtgewicht/target); schwerste zuerst in den jeweils leichtesten Bin. → Index-Listen."""
|
||
if not gewichte:
|
||
return []
|
||
K = max(1, math.ceil(sum(gewichte) / max(1, target)))
|
||
bins: list[list[int]] = [[] for _ in range(K)]
|
||
last = [0] * K
|
||
for i in sorted(range(len(gewichte)), key=lambda x: gewichte[x], reverse=True):
|
||
j = min(range(K), key=lambda b: last[b])
|
||
bins[j].append(i)
|
||
last[j] += gewichte[i]
|
||
return [b for b in bins if b]
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
async def _subbausteine_block(ctx: GenContext, set_p, files: dict, entries: dict, instructions: str) -> dict | None:
|
||
"""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"]
|
||
ordner = quelle_ordner(topic)
|
||
caps = "files" if ordner else "full"
|
||
# Quelle für die Beleg-Prüfung im Klär-Schritt (verwirft erfundene/unbelegbare Subs).
|
||
_typ = lade_quelle(topic).get("type", "thema")
|
||
source = _prompt(_QUELLE_TEMPLATE[_typ], project=ordner) if _typ in _QUELLE_TEMPLATE else _prompt("Bausteine-Quelle-Thema", topic=topic)
|
||
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)
|
||
|
||
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"<!-- baustein: {titel_by_num[num]} -->\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)
|
||
|
||
# Phase „Subbausteine finden": je Paket Loop bis 0 neue Subs / Zeit-Kappe.
|
||
async def _finde(c, chunk):
|
||
zuteilung = "\n".join(f"- {entries[num]}" for num in chunk)
|
||
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)], 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": ≥2 Nennungen = Konsens, 1× verworfen (Code).
|
||
set_p(f"Subbausteine wählen ({n} Pakete)…", step=_step_idx(topic, "Subbausteine wählen"))
|
||
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 bereinigt die Konsens-Liste.
|
||
async def _klaere(c, chunk):
|
||
fp = arbeit / f"subbaustein-final-c{c}.md"
|
||
if _parse_subbausteine(_read(fp)):
|
||
return
|
||
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, source=source, bausteine="\n\n".join(bloecke), out_path=fp, extra=_extra(instructions)),
|
||
role="judge", capabilities=caps,
|
||
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 — Konsens übernommen")
|
||
|
||
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
|
||
|
||
# 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):
|
||
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
|
||
return roh
|
||
|
||
|
||
async def _stufen_block(ctx: GenContext, set_p, files: dict, roh: dict, instructions: str) -> dict | None:
|
||
"""Block C: drei Phasen mit Barriere — Finden (einstufen), Wählen (Vote), Klären.
|
||
Lokale IDs 1..n pro Paket, hinterher auf globale gid gemappt.
|
||
→ {Baustein-Titel: [{titel, stufe}, …]} oder None."""
|
||
topic, provider, is_cancelled = ctx.topic, ctx.provider, ctx.is_cancelled
|
||
arbeit = files["arbeit"]
|
||
items = [(titel, sub) for titel, subs in roh.items() for sub in subs] # globale id = index+1
|
||
if not items:
|
||
return {titel: [] for titel in roh}
|
||
# Chunks aus GANZEN Bausteinen packen (keinen Baustein splitten) → Rater sieht je Baustein
|
||
# alle Subs und kann relativ einstufen. Item-Indizes je Baustein in roh-Reihenfolge.
|
||
chunks, cur, i = [], [], 0
|
||
for _titel_b, subs in roh.items():
|
||
g = list(range(i, i + len(subs)))
|
||
i += len(subs)
|
||
if cur and len(cur) + len(g) > STUFE_CHUNK:
|
||
chunks.append(cur)
|
||
cur = []
|
||
cur.extend(g)
|
||
if cur:
|
||
chunks.append(cur)
|
||
n = len(chunks)
|
||
|
||
def rater_paths(c):
|
||
return [arbeit / f"stufe-c{c}-{i}.json" for i in (1, 2, 3)]
|
||
|
||
def lset(item_idxs):
|
||
return set(range(1, len(item_idxs) + 1))
|
||
|
||
# Phase „Stufen finden": pro Paket 3 Rater (min. 2), lokale IDs.
|
||
async def _rate(c, item_idxs):
|
||
local_set = lset(item_idxs)
|
||
paths = rater_paths(c)
|
||
vorhanden = sum(1 for p in paths if _stufen_schema(_json_datei(p), local_set))
|
||
if vorhanden >= 2:
|
||
return True
|
||
enum_zeilen, cur_b = [], None
|
||
for k, j in enumerate(item_idxs, 1):
|
||
b, sub = items[j]
|
||
if b != cur_b:
|
||
enum_zeilen.append(f"\nBAUSTEIN: {b}")
|
||
cur_b = b
|
||
enum_zeilen.append(f"{k}. {sub}")
|
||
enum = "\n".join(enum_zeilen).strip()
|
||
offen = [(i, p) for i, p in enumerate(paths, 1) if not _stufen_schema(_json_datei(p), local_set)]
|
||
slots = [{
|
||
"key": f"bausteine-{topic}-stufe-c{c}-{i}",
|
||
"prompt": _prompt("Stufen-Recherche", topic=topic, subbausteine=enum, out_path=p, extra=_extra(instructions)),
|
||
"role": "fast", "capabilities": "files",
|
||
"payload": (lambda result, p=p, ids=local_set: _stufen_schema(_json_datei(p), ids)),
|
||
} for i, p in offen]
|
||
neu = await _race(topic, f"Stufen Paket {c}", slots, 2 - vorhanden, _timeout("stufe", len(item_idxs)), provider, cancelled=is_cancelled, grace=KONSENS_GRACE)
|
||
return not is_cancelled() and neu is not None
|
||
|
||
oks = await _gather_fortschritt([_rate(c, idxs) for c, idxs in enumerate(chunks, 1)], len(chunks), _melde_p(set_p, topic, "Stufen finden"))
|
||
if is_cancelled():
|
||
return None
|
||
if not all(ok is True for ok in oks):
|
||
_bausteine_errors[topic] = "Einstufung fehlgeschlagen (Recherche)"
|
||
return None
|
||
|
||
# Phase „Stufen wählen": Code-Vote je Paket → (ergebnis, strittig).
|
||
set_p(f"Stufen wählen ({n} Pakete)…", step=_step_idx(topic, "Stufen wählen"))
|
||
vote_by_c = {}
|
||
for c, item_idxs in enumerate(chunks, 1):
|
||
local_set = lset(item_idxs)
|
||
rater = [d for p in rater_paths(c) if (d := _stufen_schema(_json_datei(p), local_set))]
|
||
ergebnis: dict[int, str] = {}
|
||
strittig: dict[int, list[str]] = {}
|
||
for k in range(1, len(item_idxs) + 1):
|
||
stimmen = [d[k] for d in rater if k in d]
|
||
zaehler: dict[str, int] = {}
|
||
for s in stimmen:
|
||
zaehler[s] = zaehler.get(s, 0) + 1
|
||
best = max(zaehler.values(), default=0)
|
||
gewinner = [s for s, v in zaehler.items() if v == best]
|
||
if len(gewinner) == 1 and best >= 2:
|
||
ergebnis[k] = gewinner[0]
|
||
else:
|
||
strittig[k] = stimmen
|
||
vote_by_c[c] = (ergebnis, strittig)
|
||
|
||
# Phase „Stufen klären": Judge je Paket mit Strittigem, alle parallel.
|
||
async def _klaere(c, item_idxs):
|
||
ergebnis, strittig = vote_by_c[c]
|
||
if strittig:
|
||
judge_path = arbeit / f"stufe-final-c{c}.json"
|
||
entsch = _stufen_schema(_json_datei(judge_path), set(strittig))
|
||
if entsch is None:
|
||
strittig_block = "\n".join(
|
||
f"{k}. [{items[item_idxs[k - 1]][0]}] {items[item_idxs[k - 1]][1]} — Stimmen: {', '.join(stimmen) or 'keine'}"
|
||
for k, stimmen in strittig.items()
|
||
)
|
||
status, entsch = await run_single_slot(
|
||
ctx, f"Stufen-Klärung {c}",
|
||
key=f"bausteine-{topic}-stufe-final-c{c}",
|
||
prompt=_prompt("Stufen-Mapping", topic=topic, strittig=strittig_block, out_path=judge_path, extra=_extra(instructions)),
|
||
role="judge", capabilities="files",
|
||
payload=lambda result, p=judge_path, ids=set(strittig): _stufen_schema(_json_datei(p), ids),
|
||
timeout=_timeout("stufe_check", len(strittig)),
|
||
)
|
||
if status == FAILED:
|
||
_log(topic, f"Stufen-Klärung Paket {c} fehlgeschlagen — Default 'fortgeschritten'")
|
||
entsch = entsch if isinstance(entsch, dict) else {}
|
||
# Strittige ohne Entscheid → 'fortgeschritten'; Vote-Gewinner bleiben; Judge überschreibt.
|
||
ergebnis = {**{k: "fortgeschritten" for k in strittig}, **ergebnis, **entsch}
|
||
return {item_idxs[k - 1] + 1: stufe for k, stufe in ergebnis.items()}
|
||
|
||
parts = await _gather_fortschritt([_klaere(c, idxs) for c, idxs in enumerate(chunks, 1)], len(chunks), _melde_p(set_p, topic, "Stufen klären"))
|
||
if is_cancelled():
|
||
return None
|
||
stufe_by_id: dict[int, str] = {}
|
||
for c, part in enumerate(parts, 1):
|
||
if not isinstance(part, dict):
|
||
# Klärung ist nicht fatal: Vote-Ergebnis + Default 'fortgeschritten' für Strittige.
|
||
if isinstance(part, BaseException):
|
||
_log(topic, f"Stufen-Klärung Paket {c}: {type(part).__name__}: {part}")
|
||
ergebnis, strittig = vote_by_c[c]
|
||
item_idxs = chunks[c - 1]
|
||
merged = {**{k: "fortgeschritten" for k in strittig}, **ergebnis}
|
||
part = {item_idxs[k - 1] + 1: s for k, s in merged.items()}
|
||
stufe_by_id.update(part)
|
||
|
||
# Sidecar zusammensetzen — gleiche Reihenfolge wie items → gid stimmt
|
||
sidecar: dict[str, list[dict]] = {}
|
||
gid = 0
|
||
for titel, subs in roh.items():
|
||
lst = []
|
||
for sub in subs:
|
||
gid += 1
|
||
lst.append({"titel": sub, "stufe": stufe_by_id.get(gid, "fortgeschritten")})
|
||
sidecar[titel] = lst
|
||
return sidecar
|
||
|
||
|
||
_FAKTEN_FELDER = ("kernpunkte", "voraussetzungen", "huerden", "belegte_fakten", "beispiel_idee")
|
||
|
||
|
||
def _fakten_schema(data) -> list[dict] | None:
|
||
"""{"fakten": [{baustein, subbaustein, …}]} → valide Liste · sonst None.
|
||
Trennt belegte_fakten (mit Quelle) hart von beispiel_idee (generativ)."""
|
||
if not isinstance(data, dict) or not isinstance(data.get("fakten"), list):
|
||
return None
|
||
out = []
|
||
for e in data["fakten"]:
|
||
if not isinstance(e, dict):
|
||
continue
|
||
bau = str(e.get("baustein", "")).strip()
|
||
sub = str(e.get("subbaustein", "")).strip()
|
||
if not bau or not sub:
|
||
continue
|
||
bf = [{"text": t, "quelle": str(f.get("quelle", "")).strip()}
|
||
for f in (e.get("belegte_fakten") or []) if isinstance(f, dict) and (t := str(f.get("text", "")).strip())]
|
||
out.append({
|
||
"baustein": bau, "subbaustein": sub,
|
||
"kernpunkte": [k for x in (e.get("kernpunkte") or []) if (k := str(x).strip())],
|
||
"voraussetzungen": str(e.get("voraussetzungen", "")).strip(),
|
||
"huerden": str(e.get("huerden", "")).strip(),
|
||
"belegte_fakten": bf,
|
||
"beispiel_idee": str(e.get("beispiel_idee", "")).strip(),
|
||
})
|
||
return out or None
|
||
|
||
|
||
def _fakten_check_schema(data) -> list[tuple[str, bool]] | None:
|
||
"""Fakten-Check → [(sub_norm, verwerfen)] je Beanstandung · {ok:true}→[] · None bei ungültig.
|
||
verwerfen=True: Sub inhaltlich nicht belegbar (entfernen). verwerfen=False: nur Fakt korrigieren."""
|
||
if not isinstance(data, dict):
|
||
return None
|
||
if data.get("ok") is True:
|
||
return []
|
||
pr = data.get("probleme")
|
||
if not isinstance(pr, list):
|
||
return None
|
||
return [(sn, bool(p.get("verwerfen")))
|
||
for p in pr if isinstance(p, dict) and (sn := _norm_titel(str(p.get("subbaustein", ""))))]
|
||
|
||
|
||
def _fakten_zeilen(fk: dict) -> str:
|
||
z = []
|
||
if fk.get("kernpunkte"):
|
||
z.append("Kernpunkte: " + " · ".join(fk["kernpunkte"]))
|
||
if fk.get("voraussetzungen"):
|
||
z.append("Voraussetzung: " + fk["voraussetzungen"])
|
||
if fk.get("huerden"):
|
||
z.append("Hürde: " + fk["huerden"])
|
||
for bf in fk.get("belegte_fakten", []):
|
||
z.append(f"FAKT: {bf['text']} (Quelle: {bf.get('quelle', '?')})")
|
||
if fk.get("beispiel_idee"):
|
||
z.append("Beispiel: " + fk["beispiel_idee"])
|
||
return "\n".join(z)
|
||
|
||
|
||
def _fakten_komplett(files: dict) -> bool:
|
||
"""Fakten-Map existiert (Block durch)? {Baustein: {sub_norm: {...}}}."""
|
||
d = _json_datei(files["fakten"])
|
||
return isinstance(d, dict) and bool(d)
|
||
|
||
|
||
async def _fakten_block(ctx, set_p, files: dict, roh: dict, q: dict, ordner, instructions: str) -> tuple | None:
|
||
"""Block: je Sub Quell-Fakten extrahieren (finden) → verifizieren (prüfen) → korrigieren/verwerfen (fix).
|
||
Extract-once-Grounding: das Ergebnis nährt Stufe/Relevanz/Fragen/Guide.
|
||
→ (fakten_map, verworfen_map) — fakten_map {Baustein: {sub_norm: fakten}}, verworfen_map
|
||
{Baustein: {sub_norm}} (unbelegbare Subs zum Entfernen) — oder None bei Abbruch/Fehler."""
|
||
topic, provider, is_cancelled = ctx.topic, ctx.provider, ctx.is_cancelled
|
||
arbeit = files["arbeit"]
|
||
caps = "files" if ordner else "full"
|
||
typ = q.get("type", "thema")
|
||
source = _prompt(_QUELLE_TEMPLATE[typ], project=ordner) if typ in _QUELLE_TEMPLATE else _prompt("Bausteine-Quelle-Thema", topic=topic)
|
||
bausteine = [(titel, [str(s).strip() for s in subs if str(s).strip()]) for titel, subs in roh.items() if subs]
|
||
if not bausteine:
|
||
return {}, {}
|
||
chunks = _lpt_chunks([len(subs) for _, subs in bausteine], FAKTEN_CHUNK_SUBS)
|
||
|
||
def roh_path(ci): return arbeit / f"fakten-c{ci}.json"
|
||
def chk_path(ci, j): return arbeit / f"fakten-check-c{ci}-j{j}.json"
|
||
def fix_path(ci): return arbeit / f"fakten-fix-c{ci}.json"
|
||
def ctitel(idxs): return [bausteine[i][0] for i in idxs]
|
||
def block_text(idxs):
|
||
return "\n\n".join(
|
||
f"BAUSTEIN: {bausteine[i][0]}\nSUBBAUSTEINE:\n" + "\n".join(f"- {s}" for s in bausteine[i][1])
|
||
for i in idxs)
|
||
|
||
# Roh-Fakten eines Chunks → {Baustein: {sub_norm: {sub, …felder}}}, auf Chunk-Titel gematcht.
|
||
def roh_map(ci, path):
|
||
idxs = chunks[ci]
|
||
rel_by = {bausteine[i][0]: bausteine[i][1] for i in idxs}
|
||
ct = ctitel(idxs)
|
||
out: dict[str, dict] = {}
|
||
for e in _fakten_schema(_json_datei(path)) or []:
|
||
bt = _match_sub(e["baustein"], ct)
|
||
if bt not in rel_by:
|
||
continue
|
||
sub = _match_sub(e["subbaustein"], rel_by[bt])
|
||
out.setdefault(bt, {})[_norm_titel(sub)] = {"sub": sub, **{k: e[k] for k in _FAKTEN_FELDER}}
|
||
return out
|
||
|
||
# Phase „Fakten finden": 1 Generator je Chunk.
|
||
async def _finde(ci, idxs):
|
||
fp = roh_path(ci)
|
||
if _fakten_schema(_json_datei(fp)):
|
||
return True
|
||
subs_total = sum(len(bausteine[i][1]) for i in idxs)
|
||
status, _r = await run_single_slot(
|
||
ctx, f"Fakten {ci}", key=f"bausteine-{topic}-fakten-c{ci}",
|
||
prompt=_prompt("Fakten-Recherche", topic=topic, source=source, bausteine=block_text(idxs), out_path=fp, extra=_extra(instructions)),
|
||
role="guide", capabilities=caps,
|
||
payload=lambda result, p=fp: _fakten_schema(_json_datei(p)),
|
||
timeout=_timeout("inhalt", subs_total))
|
||
return status != FAILED and _fakten_schema(_json_datei(fp)) is not None
|
||
|
||
oks = await _gather_fortschritt([_finde(ci, idxs) for ci, idxs in enumerate(chunks)], len(chunks), _melde_p(set_p, topic, "Fakten finden"))
|
||
if is_cancelled():
|
||
return None
|
||
if not any(ok is True for ok in oks):
|
||
_bausteine_errors[topic] = "Fakten-Extraktion fehlgeschlagen"
|
||
return None
|
||
|
||
# Phase „Fakten prüfen": FAKTEN_CHECK_PANEL Judges je Chunk. Zwei Mehrheits-Mengen:
|
||
# beanstandet (Fakt ungenau → korrigieren) und verwerfen (Sub nicht belegbar → entfernen).
|
||
async def _pruefe(ci, idxs):
|
||
per = roh_map(ci, roh_path(ci))
|
||
if not per:
|
||
return ci, set(), set()
|
||
fakten_text = "\n\n".join(f"SUBBAUSTEIN: {fk['sub']}\n{_fakten_zeilen(fk)}" for fm in per.values() for fk in fm.values())
|
||
offen = [j for j in (1, 2, 3)[:FAKTEN_CHECK_PANEL] if _fakten_check_schema(_json_datei(chk_path(ci, j))) is None]
|
||
await asyncio.gather(*[
|
||
run_agent(f"bausteine-{topic}-fakten-check-c{ci}-j{j}",
|
||
_prompt("Fakten-Check", topic=topic, source=source, fakten=fakten_text, out_path=chk_path(ci, j), extra=_extra(instructions)),
|
||
_timeout("inhalt_check", len(per)), provider=provider, role="judge", capabilities=caps)
|
||
for j in offen], return_exceptions=True)
|
||
outs = [s for j in (1, 2, 3)[:FAKTEN_CHECK_PANEL] if (s := _fakten_check_schema(_json_datei(chk_path(ci, j)))) is not None]
|
||
bvotes: dict[str, int] = {}
|
||
vvotes: dict[str, int] = {}
|
||
for s in outs: # s = [(sub_norm, verwerfen)] eines Judges
|
||
gb, gv = set(), set()
|
||
for sn, verw in s:
|
||
if sn not in gb:
|
||
gb.add(sn); bvotes[sn] = bvotes.get(sn, 0) + 1
|
||
if verw and sn not in gv:
|
||
gv.add(sn); vvotes[sn] = vvotes.get(sn, 0) + 1
|
||
schwelle = len(outs) / 2 if outs else 99
|
||
beanstandet = {sn for sn, v in bvotes.items() if v > schwelle}
|
||
# Verwerfen ist irreversibel → strenger als Beanstanden: Mehrheit UND ≥2 zustimmende Judges
|
||
# (verhindert Löschung durch eine Einzelstimme, wenn das Panel degradiert ist).
|
||
verwerfen = {sn for sn, v in vvotes.items() if v > schwelle and v >= 2}
|
||
return ci, beanstandet, verwerfen
|
||
|
||
pruef = await _gather_fortschritt([_pruefe(ci, idxs) for ci, idxs in enumerate(chunks)], len(chunks), _melde_p(set_p, topic, "Fakten prüfen"))
|
||
if is_cancelled():
|
||
return None
|
||
beanstandet: dict[int, set] = {}
|
||
verwerfen: dict[int, set] = {}
|
||
for r in pruef:
|
||
if isinstance(r, tuple) and len(r) == 3:
|
||
ci, b, v = r
|
||
beanstandet[ci] = b
|
||
verwerfen[ci] = v
|
||
|
||
# Phase „Fakten fix": nur KORRIGIERBARE (beanstandet ohne verwerfen) neu extrahieren.
|
||
korrigieren = {ci: (beanstandet.get(ci, set()) - verwerfen.get(ci, set())) for ci in beanstandet}
|
||
n_problem = sum(len(s) for s in korrigieren.values())
|
||
if n_problem:
|
||
set_p(f"Fakten korrigieren ({n_problem})…", step=_step_idx(topic, "Fakten fix"))
|
||
async def _fix(ci):
|
||
subs_norm = korrigieren.get(ci, set())
|
||
if not subs_norm or _fakten_schema(_json_datei(fix_path(ci))):
|
||
return
|
||
idxs = chunks[ci]
|
||
rel_by = {bausteine[i][0]: bausteine[i][1] for i in idxs}
|
||
# nur die korrigierbaren Subs als Block
|
||
ziel = []
|
||
for bt, subs in rel_by.items():
|
||
betroffen = [s for s in subs if _norm_titel(s) in subs_norm]
|
||
if betroffen:
|
||
ziel.append(f"BAUSTEIN: {bt}\nSUBBAUSTEINE:\n" + "\n".join(f"- {s}" for s in betroffen))
|
||
if not ziel:
|
||
return
|
||
await run_single_slot(
|
||
ctx, f"Fakten-Fix {ci}", key=f"bausteine-{topic}-fakten-fix-c{ci}",
|
||
prompt=_prompt("Fakten-Recherche", topic=topic, source=source, bausteine="\n\n".join(ziel), out_path=fix_path(ci), extra=_extra(instructions)),
|
||
role="guide", capabilities=caps,
|
||
payload=lambda result, p=fix_path(ci): _fakten_schema(_json_datei(p)),
|
||
timeout=_timeout("inhalt", len(subs_norm)))
|
||
await _gather_fortschritt([_fix(ci) for ci in korrigieren], len(korrigieren), _melde_p(set_p, topic, "Fakten fix"))
|
||
if is_cancelled():
|
||
return None
|
||
|
||
# Zusammensetzen: roh + Fix-Overrides für korrigierte. Verworfene Subs raus (+ je Baustein melden).
|
||
ergebnis: dict[str, dict] = {}
|
||
verworfen_map: dict[str, set] = {}
|
||
for ci in range(len(chunks)):
|
||
per = roh_map(ci, roh_path(ci))
|
||
fix = roh_map(ci, fix_path(ci)) if fix_path(ci).exists() else {}
|
||
verw = verwerfen.get(ci, set())
|
||
for bt, fm in per.items():
|
||
for sn, fk in fm.items():
|
||
if sn in verw:
|
||
verworfen_map.setdefault(bt, set()).add(sn)
|
||
continue
|
||
gewinner = fix.get(bt, {}).get(sn, fk) if sn in korrigieren.get(ci, set()) else fk
|
||
ergebnis.setdefault(bt, {})[sn] = {k: gewinner[k] for k in _FAKTEN_FELDER}
|
||
if verworfen_map:
|
||
_log(topic, f"Fakten-Check verwirft {sum(len(s) for s in verworfen_map.values())} unbelegbare Subbausteine")
|
||
return ergebnis, verworfen_map
|
||
|
||
|
||
async def _relevanz_block(ctx: GenContext, set_p, files: dict, sidecar: dict, instructions: str) -> dict | None:
|
||
"""Block D: drei Phasen mit Barriere — Finden (relevant/rand), Wählen (Vote), Klären.
|
||
Items aus der Sidecar; lokale IDs 1..n pro Paket → globale gid.
|
||
→ {gid: relevanz} oder None bei Abbruch/Recherche-Fehler. Default bei Lücke/Streit: 'relevant'."""
|
||
topic, provider, is_cancelled = ctx.topic, ctx.provider, ctx.is_cancelled
|
||
arbeit = files["arbeit"]
|
||
items = [(titel, sub["titel"]) for titel, subs in sidecar.items() for sub in subs] # globale id = index+1
|
||
if not items:
|
||
return {}
|
||
chunks = _chunk_nums(list(range(len(items))), _n_chunks(len(items), STUFE_CHUNK))
|
||
n = len(chunks)
|
||
|
||
def rater_paths(c):
|
||
return [arbeit / f"relevanz-c{c}-{i}.json" for i in (1, 2, 3)]
|
||
|
||
def lset(item_idxs):
|
||
return set(range(1, len(item_idxs) + 1))
|
||
|
||
# Phase „Relevanz finden": pro Paket 3 Rater (min. 2), lokale IDs.
|
||
async def _rate(c, item_idxs):
|
||
local_set = lset(item_idxs)
|
||
paths = rater_paths(c)
|
||
vorhanden = sum(1 for p in paths if _relevanz_schema(_json_datei(p), local_set))
|
||
if vorhanden >= 2:
|
||
return True
|
||
enum = "\n".join(f"{k}. [{items[j][0]}] {items[j][1]}" for k, j in enumerate(item_idxs, 1))
|
||
offen = [(i, p) for i, p in enumerate(paths, 1) if not _relevanz_schema(_json_datei(p), local_set)]
|
||
slots = [{
|
||
"key": f"bausteine-{topic}-relevanz-c{c}-{i}",
|
||
"prompt": _prompt("Relevanz-Recherche", topic=topic, subbausteine=enum, out_path=p, extra=_extra(instructions)),
|
||
"role": "fast", "capabilities": "files",
|
||
"payload": (lambda result, p=p, ids=local_set: _relevanz_schema(_json_datei(p), ids)),
|
||
} for i, p in offen]
|
||
neu = await _race(topic, f"Relevanz Paket {c}", slots, 2 - vorhanden, _timeout("relevanz", len(item_idxs)), provider, cancelled=is_cancelled, grace=KONSENS_GRACE)
|
||
return not is_cancelled() and neu is not None
|
||
|
||
oks = await _gather_fortschritt([_rate(c, idxs) for c, idxs in enumerate(chunks, 1)], len(chunks), _melde_p(set_p, topic, "Relevanz finden"))
|
||
if is_cancelled():
|
||
return None
|
||
if not all(ok is True for ok in oks):
|
||
_bausteine_errors[topic] = "Relevanz fehlgeschlagen (Recherche)"
|
||
return None
|
||
|
||
# Phase „Relevanz wählen": Code-Vote je Paket → (ergebnis, strittig).
|
||
set_p(f"Relevanz wählen ({n} Pakete)…", step=_step_idx(topic, "Relevanz wählen"))
|
||
vote_by_c = {}
|
||
for c, item_idxs in enumerate(chunks, 1):
|
||
local_set = lset(item_idxs)
|
||
rater = [d for p in rater_paths(c) if (d := _relevanz_schema(_json_datei(p), local_set))]
|
||
ergebnis: dict[int, str] = {}
|
||
strittig: dict[int, list[str]] = {}
|
||
for k in range(1, len(item_idxs) + 1):
|
||
stimmen = [d[k] for d in rater if k in d]
|
||
zaehler: dict[str, int] = {}
|
||
for s in stimmen:
|
||
zaehler[s] = zaehler.get(s, 0) + 1
|
||
best = max(zaehler.values(), default=0)
|
||
gewinner = [s for s, v in zaehler.items() if v == best]
|
||
if len(gewinner) == 1 and best >= 2:
|
||
ergebnis[k] = gewinner[0]
|
||
else:
|
||
strittig[k] = stimmen
|
||
vote_by_c[c] = (ergebnis, strittig)
|
||
|
||
# Phase „Relevanz klären": Judge je Paket mit Strittigem, alle parallel.
|
||
async def _klaere(c, item_idxs):
|
||
ergebnis, strittig = vote_by_c[c]
|
||
if strittig:
|
||
judge_path = arbeit / f"relevanz-final-c{c}.json"
|
||
entsch = _relevanz_schema(_json_datei(judge_path), set(strittig))
|
||
if entsch is None:
|
||
strittig_block = "\n".join(
|
||
f"{k}. [{items[item_idxs[k - 1]][0]}] {items[item_idxs[k - 1]][1]} — Stimmen: {', '.join(stimmen) or 'keine'}"
|
||
for k, stimmen in strittig.items()
|
||
)
|
||
status, entsch = await run_single_slot(
|
||
ctx, f"Relevanz-Klärung {c}",
|
||
key=f"bausteine-{topic}-relevanz-final-c{c}",
|
||
prompt=_prompt("Relevanz-Mapping", topic=topic, strittig=strittig_block, out_path=judge_path, extra=_extra(instructions)),
|
||
role="judge", capabilities="files",
|
||
payload=lambda result, p=judge_path, ids=set(strittig): _relevanz_schema(_json_datei(p), ids),
|
||
timeout=_timeout("relevanz_check", len(strittig)),
|
||
)
|
||
if status == FAILED:
|
||
_log(topic, f"Relevanz-Klärung Paket {c} fehlgeschlagen — Default 'relevant'")
|
||
entsch = entsch if isinstance(entsch, dict) else {}
|
||
# Strittige ohne Entscheid → 'relevant' (nie versehentlich ausschließen).
|
||
ergebnis = {**{k: "relevant" for k in strittig}, **ergebnis, **entsch}
|
||
return {item_idxs[k - 1] + 1: rel for k, rel in ergebnis.items()}
|
||
|
||
parts = await _gather_fortschritt([_klaere(c, idxs) for c, idxs in enumerate(chunks, 1)], len(chunks), _melde_p(set_p, topic, "Relevanz klären"))
|
||
if is_cancelled():
|
||
return None
|
||
relevanz_by_id: dict[int, str] = {}
|
||
for c, part in enumerate(parts, 1):
|
||
if not isinstance(part, dict):
|
||
# Klärung ist nicht fatal: Vote-Ergebnis + Default 'relevant' für Strittige.
|
||
if isinstance(part, BaseException):
|
||
_log(topic, f"Relevanz-Klärung Paket {c}: {type(part).__name__}: {part}")
|
||
ergebnis, strittig = vote_by_c[c]
|
||
item_idxs = chunks[c - 1]
|
||
merged = {**{k: "relevant" for k in strittig}, **ergebnis}
|
||
part = {item_idxs[k - 1] + 1: s for k, s in merged.items()}
|
||
relevanz_by_id.update(part)
|
||
return relevanz_by_id
|
||
|
||
|
||
def _norm_frage(t: str) -> str:
|
||
return " ".join(str(t or "").lower().split())
|
||
|
||
|
||
def _match_sub(agent_sub: str, rel: list[str]) -> str:
|
||
"""Subbaustein-Titel des Agenten auf den passenden relevanten Titel mappen — exakt,
|
||
dann normalisiert, dann Teilstring (der Agent lässt z. B. das Präfix „Frage: " weg).
|
||
Kein Treffer → Agent-Titel behalten. So geht KEIN Muster durch Titel-Abweichung verloren."""
|
||
if agent_sub in rel:
|
||
return agent_sub
|
||
an = _norm_titel(agent_sub)
|
||
for r in rel:
|
||
rn = _norm_titel(r)
|
||
if an and rn and (an == rn or an in rn or rn in an):
|
||
return r
|
||
return agent_sub
|
||
|
||
|
||
async def _frage_muster_block(ctx: GenContext, set_p, files: dict, sidecar: dict, instructions: str) -> dict | None:
|
||
"""Block E (10er-Chunks): Finden (1 Generator je ~10 Bausteine, parallel), Wählen (Code:
|
||
je Baustein gruppieren + Dedup), Klären (1 Kritiker je Chunk), Prüfen (Nachrunde).
|
||
Zuordnung je Eintrag über das `baustein`-Feld (Chunk-Datei trägt mehrere Bausteine).
|
||
→ {Baustein-Titel: [{subbaustein, frage}, …]} oder None bei Abbruch."""
|
||
topic, is_cancelled = ctx.topic, ctx.is_cancelled
|
||
arbeit = files["arbeit"]
|
||
# ALLE Subbausteine (auch rand) bekommen ein Muster — Rand ist in der FGuide-Ebene prüfbar.
|
||
bausteine = []
|
||
for titel, subs in sidecar.items():
|
||
alle = [s["titel"] for s in subs if isinstance(s, dict) and str(s.get("titel", "")).strip()]
|
||
if alle:
|
||
bausteine.append((titel, alle))
|
||
if not bausteine:
|
||
return {}
|
||
chunks = _lpt_chunks([len(rel) for _, rel in bausteine], FRAGE_CHUNK_SUBS) # lastbalanciert nach Sub-Zahl
|
||
|
||
def roh_path(ci):
|
||
return arbeit / f"frage-muster-c{ci}.json"
|
||
|
||
def final_path(ci):
|
||
return arbeit / f"frage-muster-final-c{ci}.json"
|
||
|
||
def _chunk_titel(idxs):
|
||
return [bausteine[i][0] for i in idxs]
|
||
|
||
# Phase „Fragen finden": je Chunk 1 Generator, alle parallel.
|
||
async def _finde(ci, idxs):
|
||
fp = roh_path(ci)
|
||
if _frage_muster_chunk_schema(_json_datei(fp)):
|
||
return # Resume
|
||
block = "\n\n".join(
|
||
f"BAUSTEIN: {bausteine[i][0]}\nSUBBAUSTEINE:\n" + "\n".join(f"- {s}" for s in bausteine[i][1])
|
||
for i in idxs
|
||
)
|
||
subs_total = sum(len(bausteine[i][1]) for i in idxs)
|
||
status, _ = await run_single_slot(
|
||
ctx, f"Frage-Muster {ci}",
|
||
key=f"bausteine-{topic}-frage-muster-c{ci}",
|
||
prompt=_prompt("Frage-Muster-Recherche", topic=topic, bausteine=block,
|
||
out_path=fp, extra=_extra(instructions)),
|
||
role="fast", capabilities="files",
|
||
payload=lambda result, p=fp: _frage_muster_chunk_schema(_json_datei(p)),
|
||
timeout=_timeout("frage_muster", subs_total),
|
||
)
|
||
if status == FAILED:
|
||
_log(topic, f"Frage-Muster Chunk {ci} fehlgeschlagen — Bausteine im Fallback (Nachrunde/Live)")
|
||
|
||
async def finde_alle(ci_list):
|
||
ci_list = list(ci_list)
|
||
await _gather_fortschritt([_finde(ci, chunks[ci]) for ci in ci_list], len(ci_list), _melde_p(set_p, topic, "Fragen finden"))
|
||
|
||
await finde_alle(range(len(chunks)))
|
||
if is_cancelled():
|
||
return None
|
||
|
||
# Phase „Fragen wählen": Code — Chunk-Dateien je Baustein gruppieren, Dubletten raus,
|
||
# Baustein-/Subbaustein-Titel locker auf die Vorgaben mappen (nichts wegen Abweichung verwerfen).
|
||
def _waehle_chunk(ci):
|
||
idxs = chunks[ci]
|
||
ctitel = _chunk_titel(idxs)
|
||
rel_by = {bausteine[i][0]: bausteine[i][1] for i in idxs}
|
||
out, gesehen = {}, {}
|
||
for e in _frage_muster_chunk_schema(_json_datei(roh_path(ci))) or []:
|
||
titel = _match_sub(e["baustein"], ctitel)
|
||
if titel not in rel_by:
|
||
continue # nicht zuordenbar → verwerfen
|
||
sub = _match_sub(e["subbaustein"], rel_by[titel])
|
||
seen = gesehen.setdefault(titel, set())
|
||
if sub in seen:
|
||
continue # genau ein Muster je Subbaustein
|
||
seen.add(sub)
|
||
out.setdefault(titel, []).append({"subbaustein": sub, "frage": e["frage"]})
|
||
return out
|
||
|
||
def _waehle_all(ci_list):
|
||
roh = {}
|
||
for ci in ci_list:
|
||
for titel, eintraege in _waehle_chunk(ci).items():
|
||
roh.setdefault(titel, []).extend(eintraege)
|
||
return roh
|
||
|
||
set_p("Fragen wählen…", step=_step_idx(topic, "Fragen wählen"))
|
||
roh_by_titel = _waehle_all(range(len(chunks)))
|
||
|
||
# Phase „Fragen klären": je Chunk 1 Kritiker bereinigt die Tabellen (nach Baustein gruppiert).
|
||
async def _klaere(ci, idxs):
|
||
fp = final_path(ci)
|
||
if _frage_muster_chunk_schema(_json_datei(fp)):
|
||
return # Resume
|
||
bloecke = []
|
||
for i in idxs:
|
||
t = bausteine[i][0]
|
||
eintraege = roh_by_titel.get(t) or []
|
||
if not eintraege:
|
||
continue
|
||
zeilen = "\n".join(f"- ({e['subbaustein']}) {e['frage']}" for e in eintraege)
|
||
bloecke.append(f"BAUSTEIN: {t}\n{zeilen}")
|
||
if not bloecke:
|
||
return # nichts zu klären in diesem Chunk
|
||
subs_total = sum(len(bausteine[i][1]) for i in idxs)
|
||
status, _ = await run_single_slot(
|
||
ctx, f"Frage-Muster-Klärung {ci}",
|
||
key=f"bausteine-{topic}-frage-muster-final-c{ci}",
|
||
prompt=_prompt("Frage-Muster-Kritik", topic=topic, tabelle="\n\n".join(bloecke), out_path=fp, extra=_extra(instructions)),
|
||
role="judge", capabilities="files",
|
||
payload=lambda result, p=fp: _frage_muster_chunk_schema(_json_datei(p)),
|
||
timeout=_timeout("frage_muster_check", subs_total),
|
||
)
|
||
if status == FAILED:
|
||
_log(topic, f"Frage-Muster-Klärung Chunk {ci} fehlgeschlagen — Roh-Muster übernommen")
|
||
|
||
async def klaere_alle(ci_list):
|
||
ci_list = list(ci_list)
|
||
await _gather_fortschritt([_klaere(ci, chunks[ci]) for ci in ci_list], len(ci_list), _melde_p(set_p, topic, "Fragen klären"))
|
||
|
||
await klaere_alle(range(len(chunks)))
|
||
if is_cancelled():
|
||
return None
|
||
|
||
# Geklärte Chunk-Tabelle je Baustein, Fallback auf Roh-Muster. Titel locker mappen.
|
||
def _final_by_titel(ci_list):
|
||
out = {}
|
||
for ci in ci_list:
|
||
idxs = chunks[ci]
|
||
ctitel = _chunk_titel(idxs)
|
||
rel_by = {bausteine[i][0]: bausteine[i][1] for i in idxs}
|
||
for e in _frage_muster_chunk_schema(_json_datei(final_path(ci))) or []:
|
||
titel = _match_sub(e["baustein"], ctitel)
|
||
if titel not in rel_by:
|
||
continue
|
||
out.setdefault(titel, []).append(
|
||
{"subbaustein": _match_sub(e["subbaustein"], rel_by[titel]), "frage": e["frage"]})
|
||
return out
|
||
|
||
final_by_titel = _final_by_titel(range(len(chunks)))
|
||
ergebnis = {t: (final_by_titel.get(t) or roh_by_titel.get(t) or []) for t, _ in bausteine}
|
||
|
||
# Phase „Fragen prüfen": Chunks mit ≥1 leeren Baustein eine Runde nachholen.
|
||
set_p("Fragen prüfen…", step=_step_idx(topic, "Fragen prüfen"))
|
||
leer_titel = {t for t, _ in bausteine if not ergebnis.get(t)}
|
||
if leer_titel:
|
||
nach = [ci for ci, idxs in enumerate(chunks) if any(bausteine[i][0] in leer_titel for i in idxs)]
|
||
_log(topic, f"Frage-Muster: {len(leer_titel)} Baustein(e) ohne Muster — Nachrunde über {len(nach)} Chunk(s)")
|
||
for ci in nach:
|
||
roh_path(ci).unlink(missing_ok=True)
|
||
final_path(ci).unlink(missing_ok=True)
|
||
await finde_alle(nach)
|
||
if is_cancelled():
|
||
return None
|
||
for titel, eintraege in _waehle_all(nach).items():
|
||
roh_by_titel[titel] = eintraege
|
||
await klaere_alle(nach)
|
||
if is_cancelled():
|
||
return None
|
||
for titel, eintraege in _final_by_titel(nach).items():
|
||
final_by_titel[titel] = eintraege
|
||
for t, _ in bausteine:
|
||
if not ergebnis.get(t):
|
||
ergebnis[t] = final_by_titel.get(t) or roh_by_titel.get(t) or []
|
||
rest = [t for t, _ in bausteine if not ergebnis.get(t)]
|
||
if rest:
|
||
_log(topic, f"Frage-Muster: {len(rest)} Baustein(e) bleiben leer (Fallback Live): {rest[:5]}")
|
||
return ergebnis
|
||
|
||
|
||
# ── Inventar in der DB: Recherche-Loop · Konsolidierung · Klärung ────────────
|
||
|
||
|
||
|
||
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
|
||
|
||
|
||
|
||
|
||
|
||
|
||
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)
|
||
|
||
|
||
def _sichte_regeln(ordner, pages: list[str]) -> tuple[list[str], list[str]]:
|
||
"""Deterministischer Content/Noise-Filter (config.CRAWL_*). Substring-Match (klein) gegen
|
||
URL + Dateiname. Reihenfolge: keep > noise > min_chars > behalten. → (content, noise)."""
|
||
ordner = Path(ordner)
|
||
content, noise = [], []
|
||
for fn in pages:
|
||
zeilen = _read(ordner / fn).splitlines()
|
||
url = zeilen[0][len("QUELLE:"):].strip() if zeilen and zeilen[0].startswith("QUELLE:") else ""
|
||
body = "\n".join(zeilen[1:]).strip()
|
||
hay = f"{url}\n{fn}".lower()
|
||
if any(p in hay for p in CRAWL_KEEP_PATTERNS):
|
||
content.append(fn)
|
||
elif any(p in hay for p in CRAWL_NOISE_PATTERNS):
|
||
noise.append(fn)
|
||
elif len(body) < CRAWL_MIN_CHARS:
|
||
noise.append(fn)
|
||
else:
|
||
content.append(fn) # Default: behalten — alles mit Inhalt bleibt
|
||
return content, noise
|
||
|
||
|
||
async def _quelle_aufbereiten(ctx: GenContext, set_p, files: dict, q: dict, ordner) -> bool:
|
||
"""Schritt „Quelle aufbereiten": Crawl (link) + PDF-Konvert + Content/Noise-Sichtung.
|
||
Persistiert die Sichtung in der Coverage-Tabelle (inhalt). → True (ok) / False (Abbruch/Fehler).
|
||
thema: nichts. projekt/uni: nur PDFs (kuratierter Ordner, keine Sichtung)."""
|
||
topic, is_cancelled = ctx.topic, ctx.is_cancelled
|
||
if not ordner:
|
||
return True # thema → keine Quelle aufzubereiten
|
||
if q["type"] != "link":
|
||
await asyncio.to_thread(_pdfs_konvertieren, ordner) # projekt/uni: nur PDFs, keine Sichtung
|
||
return True
|
||
if await db.get_step_status(topic, "Quelle aufbereiten") == "fertig":
|
||
return True
|
||
if not _crawl_fertig(topic):
|
||
set_p("Quelle laden (Crawl)…", step=_step_idx(topic, "Quelle aufbereiten"))
|
||
n = await asyncio.to_thread(crawl, q["ort"], ordner, cancelled=is_cancelled)
|
||
if is_cancelled():
|
||
return False
|
||
if not n:
|
||
_bausteine_errors[topic] = "Crawl ergab keine Inhalte — Link/Domain prüfen"
|
||
return False
|
||
await asyncio.to_thread(_pdfs_konvertieren, ordner)
|
||
pages = sorted(set(_crawl_index(ordner).values()))
|
||
if pages:
|
||
set_p("Sichte Seiten…", step=_step_idx(topic, "Quelle aufbereiten"))
|
||
await db.delete_coverage(topic)
|
||
content, noise = _sichte_regeln(ordner, pages) # deterministischer Regel-Filter
|
||
await db.mark_inhalt(topic, sorted(content), sorted(noise))
|
||
_log(topic, f"Sichtung: {len(content)} Content / {len(noise)} Noise von {len(pages)} (Regeln)")
|
||
await db.set_step_status(topic, "Quelle aufbereiten", "fertig")
|
||
return True
|
||
|
||
|
||
async def _recherche_batch(ctx: GenContext, set_p, files: dict, q: dict, ordner, instructions: str) -> bool:
|
||
"""Befüllt DB-Tabelle `bausteine` mit Kandidaten (+ Nennungszähler). FESTE Datei-Batches:
|
||
jede Crawl-Seite wird genau einem Batch zugeteilt und von RECHERCHE_READERS Agenten gelesen
|
||
(Konsens ≥2 im Batch). Alle zugeteilten Seiten werden als gelesen markiert → 100 % Abdeckung.
|
||
Ohne Crawl-Ordner (Quelle „thema") → freie Web-Recherche, eine Runde. → True/False."""
|
||
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"]
|
||
await db.delete_bausteine(topic) # Coverage/inhalt gehört der Sichtung — NICHT löschen
|
||
await db.set_step_status(topic, "Recherche", "laufend")
|
||
|
||
async def _ingest(text: str) -> None:
|
||
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 []
|
||
await db.upsert_baustein(topic, norm, titel, besch, quelle)
|
||
|
||
pages = await db.list_content(topic) # von der Sichtung als Content markierte Seiten
|
||
if not pages and ordner:
|
||
pages = sorted(set(_crawl_index(ordner).values())) # Fallback (projekt/uni: keine Sichtung)
|
||
|
||
if not pages:
|
||
# Quelle „thema" (oder kein Crawl): freie Web-Recherche, eine Runde.
|
||
set_p("Recherche läuft…", step=_step_idx(topic, "Recherche"))
|
||
caps = "files" if ordner else "full"
|
||
paths = [arbeit / f"recherche-{i}.md" for i in range(1, RECHERCHE_THEMA_AGENTEN + 1)]
|
||
for p in paths:
|
||
p.unlink(missing_ok=True)
|
||
slots = [{
|
||
"key": f"bausteine-{topic}-recherche-{i}",
|
||
"prompt": _build_recherche_prompt(topic, p, instructions, q["type"], ordner),
|
||
"role": "quick", "capabilities": caps,
|
||
"payload": (lambda result, p=p: _file_payload(p)),
|
||
} for i, p in enumerate(paths, 1)]
|
||
texte = await _race(topic, "Recherche", slots, 3, _timeout("recherche"), provider,
|
||
cancelled=is_cancelled, grace=RECHERCHE_GRACE)
|
||
if is_cancelled():
|
||
return False
|
||
if not texte:
|
||
_bausteine_errors[topic] = "Recherche fehlgeschlagen (Minimum nicht erreicht)"
|
||
return False
|
||
for text in texte:
|
||
await _ingest(text)
|
||
await db.set_step_status(topic, "Recherche", "fertig")
|
||
return True
|
||
|
||
# uni/projekt: kuratierte, oft GROSSE Dateien (Skript). Statt alle am Stück zu lesen
|
||
# (Lost-in-the-Middle), in Abschnitte chunken und JEDEN gründlich von 2 Readern lesen —
|
||
# Text direkt im Prompt (kleiner Kontext), Nennungen akkumulieren zu Konsens.
|
||
if q["type"] in ("uni", "projekt"):
|
||
eintraege: list[tuple[str, str]] = [] # (dateiname, abschnitt-text)
|
||
for fn in sorted(pages):
|
||
for absch in _text_abschnitte(_read(ordner / fn)):
|
||
eintraege.append((fn, absch))
|
||
if not eintraege:
|
||
_bausteine_errors[topic] = "Recherche: Quelle leer"
|
||
return False
|
||
set_p(f"Recherche ({len(eintraege)} Abschnitte)…", step=_step_idx(topic, "Recherche"))
|
||
|
||
async def _lese_abschnitt(ei: int, fn: str, absch: str) -> None:
|
||
block = (f"ARBEITE AUSSCHLIESSLICH MIT DIESEM TEXTABSCHNITT (Quelle: {fn}). Lies ihn "
|
||
f"VOLLSTÄNDIG, überspringe nichts. Notiere `{fn}` als Quelle jedes Bausteins. "
|
||
f"Suche NICHT im Web — nur dieser Abschnitt zählt.\n\n-----\n{absch}\n-----")
|
||
paths = [arbeit / f"recherche-a{ei}-{i}.md" for i in range(1, RECHERCHE_READERS + 1)]
|
||
for p in paths:
|
||
p.unlink(missing_ok=True)
|
||
if is_cancelled():
|
||
return
|
||
slots = [{
|
||
"key": f"bausteine-{topic}-recherche-a{ei}-{i}",
|
||
"prompt": _build_recherche_prompt(topic, p, instructions, q["type"], ordner, abschnitt=block),
|
||
"role": "quick", "capabilities": "files",
|
||
"payload": (lambda result, p=p: _file_payload(p)),
|
||
} for i, p in enumerate(paths, 1)]
|
||
# Quorum 2: beide Reader pro Abschnitt sollen durch (mehr Augen = mehr Konzepte +
|
||
# echter Konsens); nach Timeout fällt _race auf das Vorhandene zurück.
|
||
texte = await _race(topic, f"Recherche Abschnitt {ei}", slots, 2, _timeout("recherche", 1),
|
||
provider, cancelled=is_cancelled, grace=RECHERCHE_GRACE)
|
||
for text in (texte or []):
|
||
await _ingest(text)
|
||
|
||
await _gather_fortschritt([_lese_abschnitt(ei, fn, a) for ei, (fn, a) in enumerate(eintraege, 1)],
|
||
len(eintraege), _melde_p(set_p, topic, "Recherche"))
|
||
if is_cancelled():
|
||
return False
|
||
await db.mark_quellen_gelesen(topic, sorted(pages))
|
||
gesamt = len(await db.list_bausteine(topic))
|
||
_log(topic, f"Recherche (uni/projekt): {gesamt} Kandidaten aus {len(eintraege)} Abschnitten ({len(pages)} Dateien)")
|
||
if not gesamt:
|
||
_bausteine_errors[topic] = "Recherche fehlgeschlagen (keine Bausteine)"
|
||
return False
|
||
await db.set_step_status(topic, "Recherche", "fertig")
|
||
return True
|
||
|
||
# Crawl/Link: viele kleine Content-Seiten (Sichtung im Schritt „Quelle aufbereiten").
|
||
# Feste Batches, je Batch RECHERCHE_READERS Reader, die GENAU diese Dateien lesen.
|
||
batches = _chunk_nums(sorted(pages), max(1, math.ceil(len(pages) / RECHERCHE_BATCH)))
|
||
|
||
async def _lese_batch(bi: int, batch: list[str]) -> bool:
|
||
liste = "\n".join(f"- {p}" for p in batch)
|
||
fokus = ("WICHTIG — feste Zuteilung: Bearbeite AUSSCHLIESSLICH diese Dateien und lies JEDE "
|
||
f"vollständig. Ignoriere alle anderen Dateien im Ordner:\n{liste}")
|
||
paths = [arbeit / f"recherche-b{bi}-{i}.md" for i in range(1, RECHERCHE_READERS + 1)]
|
||
for p in paths:
|
||
p.unlink(missing_ok=True)
|
||
if not is_cancelled():
|
||
slots = [{
|
||
"key": f"bausteine-{topic}-recherche-b{bi}-{i}",
|
||
"prompt": _build_recherche_prompt(topic, p, instructions, q["type"], ordner, fokus=fokus),
|
||
"role": "quick", "capabilities": "files",
|
||
"payload": (lambda result, p=p: _file_payload(p)),
|
||
} for i, p in enumerate(paths, 1)]
|
||
texte = await _race(topic, f"Recherche Batch {bi}", slots, 1, _timeout("recherche", len(batch)),
|
||
provider, cancelled=is_cancelled, grace=RECHERCHE_GRACE)
|
||
for text in (texte or []):
|
||
await _ingest(text)
|
||
await db.mark_quellen_gelesen(topic, batch) # alle zugeteilten Seiten abhaken (auch ohne Treffer)
|
||
return not is_cancelled()
|
||
|
||
await _gather_fortschritt([_lese_batch(bi, b) for bi, b in enumerate(batches, 1)],
|
||
len(batches), _melde_p(set_p, topic, "Recherche"))
|
||
if is_cancelled():
|
||
return False
|
||
gesamt = len(await db.list_bausteine(topic))
|
||
deckung = len(await db.list_coverage(topic))
|
||
_log(topic, f"Recherche: {gesamt} Kandidaten, Abdeckung {deckung}/{len(pages)} Seiten ({len(batches)} Batches)")
|
||
if not gesamt:
|
||
_bausteine_errors[topic] = "Recherche fehlgeschlagen (keine Bausteine)"
|
||
return False
|
||
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_READERS, 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
|
||
# Bei mehreren Chunks: ein globaler Merge-Pass über die vereinten Konsens-Einträge,
|
||
# damit Dubletten über Chunk-Grenzen (DAL×4, PHPUnit×5 …) verschmelzen.
|
||
if len(chunks) > 1 and konsens:
|
||
fp = arbeit / "konsolidierung-merge.json"
|
||
fp.unlink(missing_ok=True)
|
||
eintraege = "\n".join(f"{i}. {t} (2× genannt)" for i, t in enumerate(konsens, 1))
|
||
status, mapping = await run_single_slot(
|
||
ctx, "Konsolidierung Merge",
|
||
key=f"bausteine-{topic}-konsolidierung-merge",
|
||
prompt=_prompt("Bausteine-Recherche-Mapping", topic=topic, n=RECHERCHE_READERS, eintraege=eintraege, out_path=fp),
|
||
role="judge", capabilities="files",
|
||
payload=lambda result, p=fp: _mapping_schema(_json_datei(p)),
|
||
timeout=_timeout("recherche_mapping", len(konsens)),
|
||
)
|
||
if status == CANCELLED:
|
||
return False
|
||
if status != FAILED and mapping:
|
||
konsens, r2 = mapping
|
||
rest += r2 # vom Merge zurückgestufte Einträge in den Rest
|
||
# 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
|
||
|
||
|
||
# --- Gliederung (Bausteine-Artefakt: Kapitel-Struktur, vom Guide nur gelesen) ---
|
||
|
||
def _gliederung_komplett(files: dict) -> bool:
|
||
"""Gliederung steht (Kapitel-Liste vorhanden)?"""
|
||
d = _json_datei(files["gliederung"])
|
||
return isinstance(d, dict) and isinstance(d.get("kapitel"), list) and bool(d.get("kapitel"))
|
||
|
||
|
||
def _gliederung_schema(data, valid: set[int]):
|
||
"""{"kapitel":[{titel,nummern}]} → bereinigt (valide Nummern, je genau einmal) ·
|
||
None bei <80 % Abdeckung (Agent/Judge hat zu viel weggelassen)."""
|
||
if not isinstance(data, dict) or not isinstance(data.get("kapitel"), list):
|
||
return None
|
||
out, seen = [], set()
|
||
for ch in data["kapitel"]:
|
||
if not isinstance(ch, dict):
|
||
continue
|
||
titel = str(ch.get("titel", "")).strip() or "Kapitel"
|
||
nums = []
|
||
for n in (ch.get("nummern") or []):
|
||
try:
|
||
n = int(n)
|
||
except (ValueError, TypeError):
|
||
continue
|
||
if n in valid and n not in seen:
|
||
seen.add(n)
|
||
nums.append(n)
|
||
if nums:
|
||
out.append({"titel": titel, "nummern": nums})
|
||
if not out or len(seen) < 0.8 * len(valid):
|
||
return None
|
||
return {"kapitel": out}
|
||
|
||
|
||
async def _gliederung_block(ctx: GenContext, set_p, files: dict, entries: dict, instructions: str) -> dict:
|
||
"""Format-agnostische Gliederung über ALLE Bausteine — 3 Vorschläge → Judge merged.
|
||
Bricht nie ab: 0 gültige → ein Kapitel mit allem; fehlende Bausteine landen in „Weitere".
|
||
→ {"kapitel":[{titel,nummern}]} (auch in files["gliederung"])."""
|
||
topic, is_cancelled = ctx.topic, ctx.is_cancelled
|
||
valid = set(entries)
|
||
liste = "\n".join(f"{n}. {t}" for n, t in entries.items())
|
||
step = _step_idx(topic, "Gliederung")
|
||
set_p("Gliederung — Vorschläge…", step=step)
|
||
|
||
async def _vorschlag(i, path):
|
||
if _gliederung_schema(_json_datei(path), valid):
|
||
return True
|
||
await run_single_slot(
|
||
ctx, f"Gliederung {i}", key=f"bausteine-{topic}-gliederung-{i}",
|
||
prompt=_prompt("Guide-Gliederung", topic=topic, bausteine=liste, out_path=path, extra=_extra(instructions)),
|
||
role="guide", capabilities="files",
|
||
payload=lambda result, p=path: _gliederung_schema(_json_datei(p), valid),
|
||
timeout=_timeout("plan", len(entries)))
|
||
return _gliederung_schema(_json_datei(path), valid) is not None
|
||
|
||
slots = files["gliederung_slots"]
|
||
await _gather_fortschritt([_vorschlag(i, p) for i, p in enumerate(slots, 1)], len(slots), _melde_p(set_p, topic, "Gliederung"))
|
||
if is_cancelled():
|
||
return {}
|
||
vorschlaege = [v for p in slots if (v := _gliederung_schema(_json_datei(p), valid))]
|
||
|
||
if not vorschlaege:
|
||
plan = {"kapitel": [{"titel": "Inhalte", "nummern": list(entries)}]}
|
||
elif len(vorschlaege) == 1:
|
||
plan = vorschlaege[0]
|
||
else:
|
||
set_p("Gliederung zusammenführen…", step=step)
|
||
bloecke = "\n\n".join(
|
||
f"### Vorschlag {i}\n" + "\n".join(
|
||
f"KAPITEL: {ch['titel']}\n Nummern: {', '.join(str(n) for n in ch['nummern'])}" for ch in v["kapitel"])
|
||
for i, v in enumerate(vorschlaege, 1))
|
||
await run_single_slot(
|
||
ctx, "Gliederung-Judge", key=f"bausteine-{topic}-gliederung-judge",
|
||
prompt=_prompt("Guide-Gliederung-Judge", topic=topic, format_name="den Guide",
|
||
zweck="alle Bausteine in einem roten Faden", n=len(vorschlaege),
|
||
bausteine=liste, gliederungen=bloecke, out_path=files["gliederung"], extra=_extra(instructions)),
|
||
role="judge", capabilities="files",
|
||
payload=lambda result: _gliederung_schema(_json_datei(files["gliederung"]), valid),
|
||
timeout=_timeout("plan_judge", len(entries)))
|
||
plan = _gliederung_schema(_json_datei(files["gliederung"]), valid) or vorschlaege[0]
|
||
|
||
# Vollständigkeit: jeder Baustein kommt vor — fehlende in „Weitere" (gegen weglassende Agenten/Judge).
|
||
drin = {n for ch in plan["kapitel"] for n in ch["nummern"]}
|
||
fehlen = [n for n in entries if n not in drin]
|
||
if fehlen:
|
||
plan["kapitel"].append({"titel": "Weitere", "nummern": fehlen})
|
||
atomic_write_json(files["gliederung"], plan, indent=1)
|
||
return plan
|
||
|
||
|
||
# --- Lern-Artefakte (Karteikarten/Beispiele aus den Fakten) ---
|
||
|
||
def _karten_schema(data):
|
||
"""{"karten":[{baustein,subbaustein,frage,antwort}]} → Liste (auch leer) · None bei kaputt."""
|
||
if not isinstance(data, dict) or not isinstance(data.get("karten"), list):
|
||
return None
|
||
out = []
|
||
for e in data["karten"]:
|
||
if isinstance(e, dict) and (f := str(e.get("frage", "")).strip()) and (a := str(e.get("antwort", "")).strip()):
|
||
out.append({"baustein": str(e.get("baustein", "")).strip(), "subbaustein": str(e.get("subbaustein", "")).strip(),
|
||
"frage": f, "antwort": a})
|
||
return out
|
||
|
||
|
||
def _beispiel_schema(data):
|
||
"""{"beispiele":[{baustein,subbaustein,problem,schritte,ergebnis}]} → Liste (auch leer) · None bei kaputt."""
|
||
if not isinstance(data, dict) or not isinstance(data.get("beispiele"), list):
|
||
return None
|
||
out = []
|
||
for e in data["beispiele"]:
|
||
if not isinstance(e, dict):
|
||
continue
|
||
problem = str(e.get("problem", "")).strip()
|
||
schritte = [s for x in (e.get("schritte") or []) if (s := str(x).strip())]
|
||
if problem and schritte:
|
||
out.append({"baustein": str(e.get("baustein", "")).strip(), "subbaustein": str(e.get("subbaustein", "")).strip(),
|
||
"problem": problem, "schritte": schritte, "ergebnis": str(e.get("ergebnis", "")).strip()})
|
||
return out
|
||
|
||
|
||
_ARTEFAKT_SCHEMA = {"karteikarte": _karten_schema, "beispiel": _beispiel_schema}
|
||
_ARTEFAKT_PROMPT = {"karteikarte": "Artefakt-Karteikarte", "beispiel": "Artefakt-Beispiel"}
|
||
_ARTEFAKT_SCHRITT = {"karteikarte": "Karteikarten", "beispiel": "Beispiele"}
|
||
|
||
|
||
def _artefakte_komplett(files: dict) -> bool:
|
||
"""Artefakt-Map steht (alle Typen erzeugt)? Werte dürfen leer sein (content-aware)."""
|
||
d = _json_datei(files["artefakte"])
|
||
return isinstance(d, dict) and all(t in d for t in ARTEFAKT_TYPEN)
|
||
|
||
|
||
async def _artefakte_block(ctx: GenContext, set_p, files: dict, sidecar: dict, instructions: str) -> dict | None:
|
||
"""Lern-Artefakte je Typ aus den gespeicherten Fakten erzeugen — ein Generierungs-Durchlauf
|
||
je Typ über Chunks, keine Verifikation (Bonus-Material). → {typ: [eintraege]} (auch in files)."""
|
||
topic, is_cancelled = ctx.topic, ctx.is_cancelled
|
||
arbeit = files["arbeit"]
|
||
# Bausteine mit Subs + Fakten-Zeilen als Input-Block (extract-once aus den Fakten).
|
||
bausteine = []
|
||
for btitel, subs in sidecar.items():
|
||
if not isinstance(subs, list):
|
||
continue
|
||
zeilen = []
|
||
for s in subs:
|
||
if not isinstance(s, dict) or not (st := str(s.get("titel", "")).strip()):
|
||
continue
|
||
fk = s.get("fakten") if isinstance(s.get("fakten"), dict) else {}
|
||
zeile = f"- {st}"
|
||
if fk and (fk_text := _fakten_zeilen(fk)):
|
||
zeile += "\n" + "\n".join(" " + l for l in fk_text.split("\n"))
|
||
zeilen.append(zeile)
|
||
if zeilen:
|
||
bausteine.append((btitel, zeilen))
|
||
if not bausteine:
|
||
leer = {t: [] for t in ARTEFAKT_TYPEN}
|
||
atomic_write_json(files["artefakte"], leer, indent=1)
|
||
return leer
|
||
|
||
chunks = _lpt_chunks([len(z) for _, z in bausteine], FAKTEN_CHUNK_SUBS)
|
||
def block_text(idxs):
|
||
return "\n\n".join(f"BAUSTEIN: {bausteine[i][0]}\nSUBBAUSTEINE:\n" + "\n".join(bausteine[i][1]) for i in idxs)
|
||
|
||
ergebnis: dict[str, list] = {}
|
||
for typ in ARTEFAKT_TYPEN:
|
||
schema = _ARTEFAKT_SCHEMA[typ]
|
||
def apath(ci, t=typ): return arbeit / f"artefakt-{t}-c{ci}.json"
|
||
|
||
async def _gen(ci, idxs, t=typ, schema=schema):
|
||
p = apath(ci, t)
|
||
if schema(_json_datei(p)) is not None:
|
||
return True
|
||
await run_single_slot(
|
||
ctx, f"{_ARTEFAKT_SCHRITT[t]} {ci}", key=f"bausteine-{topic}-artefakt-{t}-c{ci}",
|
||
prompt=_prompt(_ARTEFAKT_PROMPT[t], topic=topic, bausteine=block_text(idxs), out_path=p, extra=_extra(instructions)),
|
||
role="guide", capabilities="files",
|
||
payload=lambda result, p=p, schema=schema: schema(_json_datei(p)),
|
||
timeout=_timeout("inhalt", sum(len(bausteine[i][1]) for i in idxs)))
|
||
return schema(_json_datei(p)) is not None
|
||
|
||
await _gather_fortschritt([_gen(ci, idxs) for ci, idxs in enumerate(chunks)], len(chunks), _melde_p(set_p, topic, _ARTEFAKT_SCHRITT[typ]))
|
||
if is_cancelled():
|
||
return None
|
||
eintraege: list = []
|
||
for ci in range(len(chunks)):
|
||
eintraege += schema(_json_datei(apath(ci))) or []
|
||
ergebnis[typ] = eintraege
|
||
atomic_write_json(files["artefakte"], ergebnis, indent=1)
|
||
return ergebnis
|
||
|
||
|
||
async def _mirror_artefakte_db(topic: str, sidecar: dict, artefakte: dict) -> None:
|
||
"""Artefakte in die DB spiegeln. Karteikarte/Beispiel je Sub (sub_norm)."""
|
||
await db.delete_sub_artefakte(topic)
|
||
btitel_list = list(sidecar.keys())
|
||
for typ in ARTEFAKT_TYPEN:
|
||
for e in artefakte.get(typ, []):
|
||
bt = _match_sub(e.get("baustein", ""), btitel_list)
|
||
bnorm, sn = _norm_titel(bt), _norm_titel(e.get("subbaustein", ""))
|
||
if not bnorm or not sn:
|
||
continue
|
||
daten = json.dumps({k: v for k, v in e.items() if k not in ("baustein", "subbaustein")}, ensure_ascii=False)
|
||
await db.put_sub_artefakt(topic, bnorm, sn, typ, daten, bt, e.get("subbaustein", ""))
|
||
|
||
|
||
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
|
||
fakten = json.dumps(s["fakten"], ensure_ascii=False) if isinstance(s.get("fakten"), dict) else None
|
||
await db.put_subbaustein(topic, bnorm, sn, btitel, st,
|
||
stufe=s.get("stufe"), relevanz=s.get("relevanz"),
|
||
fakten=fakten, status="konsens")
|
||
|
||
|
||
async def _mirror_frage_muster_db(topic: str, muster: dict) -> None:
|
||
"""Frage-Muster {Baustein-Titel: [{subbaustein, 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)
|
||
frage = str(e.get("frage", "")).strip()
|
||
if not (sn and frage):
|
||
continue
|
||
await db.upsert_frage_muster(topic, bnorm, sn, btitel, sub, frage)
|
||
|
||
|
||
async def _reset_db_ab_phase(topic: str, label: str) -> None:
|
||
"""DB-Inhalt der Phasen ≥ `label` verwerfen (kanonische Reihenfolge Quelle…Artefakte)."""
|
||
idx = _phase_idx(label)
|
||
if idx <= 8: # Artefakte (Karteikarten/Beispiele)
|
||
await db.delete_sub_artefakte(topic)
|
||
if idx <= 7: # Fragen
|
||
await db.delete_frage_muster(topic)
|
||
if idx <= 6: # Gliederung
|
||
await db.delete_gliederung(topic)
|
||
if idx <= 2: # Subbausteine (Fakten/Stufen/Relevanz greifen über Sidecar→Mirror)
|
||
await db.delete_subbausteine(topic)
|
||
if idx <= 1: # Inventar: Inventar + Recherche-Schritte — Sichtung bleibt
|
||
await db.delete_bausteine(topic)
|
||
await db.delete_pipeline_state(topic, ["Recherche", "Konsolidierung", "Klärung"])
|
||
if idx <= 0: # Quelle: Sichtung neu (Coverage/inhalt + Schritt)
|
||
await db.delete_coverage(topic)
|
||
await db.delete_pipeline_state(topic, ["Quelle aufbereiten"])
|
||
|
||
|
||
async def generate_bausteine(topic: str, instructions: str = "", provider: str = DEFAULT_PROVIDER, ab_phase: int | None = None) -> None:
|
||
if topic in _bausteine_progress:
|
||
return
|
||
_bausteine_progress[topic] = "Wartend…"
|
||
_bausteine_errors.pop(topic, None)
|
||
|
||
files = _bausteine_files(topic)
|
||
final_path = files["final"]
|
||
q = lade_quelle(topic)
|
||
ordner = quelle_ordner(topic) # projekt/uni/link → Ordner, thema → None
|
||
instructions = q.get("spec") or instructions # persistierte Spezifikation bevorzugen (auch bei Resume)
|
||
|
||
def set_p(msg: str, step: int | None = None) -> None:
|
||
_bausteine_progress[topic] = msg
|
||
if step is not None:
|
||
_bausteine_step[topic] = step
|
||
|
||
def is_cancelled() -> bool:
|
||
return topic in _bausteine_cancelled
|
||
|
||
def abgebrochen() -> None:
|
||
_bausteine_errors[topic] = "Abgebrochen — Fortschritt bleibt erhalten"
|
||
|
||
ctx = GenContext(topic=topic, provider=provider, is_cancelled=is_cancelled)
|
||
|
||
try:
|
||
async with _semaphore:
|
||
files["arbeit"].mkdir(parents=True, exist_ok=True)
|
||
# Re-Run ab gewählter Phase: Artefakte ab dort löschen; der Frischstart-Block
|
||
# unten wird übersprungen (er würde bei erhaltener Sidecar sonst alles wischen).
|
||
if ab_phase is not None:
|
||
phasen = _phasen(topic)
|
||
label = phasen[ab_phase - 1][0] if 1 <= ab_phase <= len(phasen) else "Inventar"
|
||
_reset_ab_phase(topic, label)
|
||
await _reset_db_ab_phase(topic, label)
|
||
# Schritt „Quelle aufbereiten": Crawl (link) + PDFs + Content/Noise-Sichtung.
|
||
if not await _quelle_aufbereiten(ctx, set_p, files, q, ordner):
|
||
if is_cancelled():
|
||
abgebrochen()
|
||
return
|
||
# „Neu erstellen": NUR wenn wirklich alles fertig ist (bausteine.md UND
|
||
# Sidecar) → kompletter Frischstart. Liegt bausteine.md ohne Sidecar vor,
|
||
# ist das ein Teil-Stand (Block B/C offen) → Resume, nicht wischen.
|
||
# Bei explizitem Re-Run (ab_phase) hat _reset_ab_phase das schon erledigt.
|
||
fertig = ab_phase is None and final_path.exists() and _sidecar_schema(_json_datei(files["sidecar"])) is not None
|
||
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)
|
||
await db.delete_gliederung(topic)
|
||
await db.delete_sub_artefakte(topic)
|
||
|
||
# Inventar (DB): Recherche-Loop → Konsolidierung → Klärung.
|
||
if not await _recherche_batch(ctx, set_p, files, q, ordner, instructions):
|
||
if is_cancelled():
|
||
abgebrochen()
|
||
return
|
||
if not await _konsolidiere(ctx, set_p, files):
|
||
if is_cancelled():
|
||
abgebrochen()
|
||
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].
|
||
if q["type"] == "projekt":
|
||
set_p("Ergänze Themenfeld…", step=_step_idx(topic, "Ergänzung"))
|
||
erg_path = files["ergaenzung"]
|
||
ergaenzungen = _ergaenzung_schema(_json_datei(erg_path))
|
||
if ergaenzungen is None:
|
||
erg_path.unlink(missing_ok=True)
|
||
status, ergaenzungen = await run_single_slot(
|
||
ctx, "Ergänzung",
|
||
key=f"bausteine-{topic}-ergaenzung-1",
|
||
prompt=_prompt(
|
||
"Bausteine-Ergaenzung",
|
||
topic=topic, bausteine="\n".join(f"- {t}" for t in entries.values()),
|
||
out_path=erg_path, extra=_extra(instructions),
|
||
),
|
||
role="quick", capabilities="full",
|
||
payload=lambda result: _ergaenzung_schema(_json_datei(erg_path)),
|
||
timeout=_timeout("ergaenzung"),
|
||
)
|
||
if status == CANCELLED:
|
||
abgebrochen()
|
||
return
|
||
if status == FAILED:
|
||
_bausteine_errors[topic] = "Ergänzung fehlgeschlagen (kein gültiges Ergebnis)"
|
||
return
|
||
idx = _titel_index(entries)
|
||
neu = [(t, b) for t, b in ergaenzungen if _titel_aufloesen(idx, t) is None]
|
||
if neu:
|
||
_log(topic, f"Ergänzung: {len(neu)} Baustein(e) aus dem Themenfeld ergänzt")
|
||
start = max(entries, default=0) + 1
|
||
for off, (t, b) in enumerate(neu):
|
||
entries[start + off] = f"{t} — {b} [Ergänzung]"
|
||
|
||
# Titel eindeutig machen und unsortiertes Inventar schreiben
|
||
entries = _eindeutige_titel(entries)
|
||
atomic_write_text(final_path, "\n".join(f"{i}. {t}" for i, t in entries.items()) + "\n")
|
||
|
||
# Block B + C: Subbausteine je Baustein + Stufen → Sidecar subbausteine.json.
|
||
# Nicht-destruktiv: bausteine.md steht schon; fehlt die Sidecar, wird beim
|
||
# nächsten Lauf nur dieser Teil neu versucht. Guide fällt ohne Sidecar zurück.
|
||
if _sidecar_schema(_json_datei(files["sidecar"])) is None:
|
||
roh = _sub_roh_schema(_json_datei(files["sub_roh"]))
|
||
if roh is None:
|
||
roh = await _subbausteine_block(ctx, set_p, files, entries, instructions)
|
||
if is_cancelled():
|
||
abgebrochen()
|
||
return
|
||
if roh is None:
|
||
return # Fehler ist gesetzt
|
||
atomic_write_json(files["sub_roh"], roh, indent=1)
|
||
# Fakten je Sub (VOR der Stufe): Quell-Fakten extrahieren + verifizieren → fakten.json.
|
||
# Extract-once-Grounding — Stufe/Relevanz/Fragen/Guide nähren sich daraus.
|
||
if not _fakten_komplett(files):
|
||
res = await _fakten_block(ctx, set_p, files, roh, q, ordner, instructions)
|
||
if is_cancelled():
|
||
abgebrochen()
|
||
return
|
||
if res is None:
|
||
return # Fehler ist gesetzt
|
||
fakten_map, verworfen = res
|
||
# Verworfene (unbelegbare) Subs aus roh streichen — ZUERST (Resume-robust), dann
|
||
# fakten.json. So sehen Stufen/Relevanz/Gliederung/Fragen/Guide sie nicht mehr.
|
||
if verworfen:
|
||
for bt, sns in verworfen.items():
|
||
if bt in roh:
|
||
roh[bt] = [s for s in roh[bt] if _norm_titel(s) not in sns]
|
||
roh = {bt: subs for bt, subs in roh.items() if subs} # leere Bausteine raus (_sub_roh_schema verlangt ≥1)
|
||
atomic_write_json(files["sub_roh"], roh, indent=1)
|
||
atomic_write_json(files["fakten"], fakten_map, indent=1)
|
||
sidecar = await _stufen_block(ctx, set_p, files, roh, instructions)
|
||
if is_cancelled():
|
||
abgebrochen()
|
||
return
|
||
if sidecar is None:
|
||
return
|
||
# Fakten in die Sidecar-Subs mergen (DB-Spiegel + Guide-Nutzung).
|
||
fakten_map = _json_datei(files["fakten"])
|
||
if isinstance(fakten_map, dict):
|
||
for btitel, subs in sidecar.items():
|
||
fm = fakten_map.get(btitel, {})
|
||
for sub in subs:
|
||
if (fk := fm.get(_norm_titel(sub["titel"]))):
|
||
sub["fakten"] = fk
|
||
atomic_write_json(files["sidecar"], sidecar, indent=1)
|
||
|
||
# Block D: Relevanz je Subbaustein (relevant/rand) → in die Sidecar mergen.
|
||
# Eigene Phase nach den Stufen; treibt das ProGuide-Format (alle Bausteine
|
||
# mit ≥1 relevantem Subbaustein) und filtert Rand-Subs aus den Guides.
|
||
sidecar = _json_datei(files["sidecar"])
|
||
if _sidecar_schema(sidecar) is not None and not _relevanz_komplett(sidecar):
|
||
relevanz_by_id = await _relevanz_block(ctx, set_p, files, sidecar, instructions)
|
||
if is_cancelled():
|
||
abgebrochen()
|
||
return
|
||
if relevanz_by_id is None:
|
||
return # Fehler ist gesetzt
|
||
gid = 0
|
||
for subs in sidecar.values():
|
||
for sub in subs:
|
||
gid += 1
|
||
sub["relevanz"] = relevanz_by_id.get(gid, "relevant")
|
||
atomic_write_json(files["sidecar"], sidecar, indent=1)
|
||
|
||
# Block D.5: Gliederung (Bausteine-Artefakt) — Kapitel-Struktur über ALLE Bausteine,
|
||
# vom Guide nur noch gelesen. Format-agnostisch; der Guide filtert je Format.
|
||
if not _gliederung_komplett(files):
|
||
await _gliederung_block(ctx, set_p, files, entries, instructions)
|
||
if is_cancelled():
|
||
abgebrochen()
|
||
return
|
||
|
||
# Block E: Frage-Muster je relevantem Subbaustein × Typ → eigenes Sidecar.
|
||
# Zur Prüfungszeit zieht jeder Agent ein Muster ohne Zurücklegen und formuliert
|
||
# daraus eine Frage — distinkte Saat verhindert die Doppelfragen der Live-Generierung.
|
||
sidecar = _json_datei(files["sidecar"])
|
||
if _sidecar_schema(sidecar) is not None and _relevanz_komplett(sidecar) and not _frage_muster_komplett(topic):
|
||
muster = await _frage_muster_block(ctx, set_p, files, sidecar, instructions)
|
||
if is_cancelled():
|
||
abgebrochen()
|
||
return
|
||
if muster is None:
|
||
return # Abbruch
|
||
atomic_write_json(files["frage_muster"], muster, indent=1)
|
||
|
||
# Block F: Lern-Artefakte (Karteikarten/Beispiele) aus den Fakten — Bonus,
|
||
# vom Frontend präsentiert. Bricht den Lauf nicht ab (Artefakte sind optional).
|
||
sidecar = _json_datei(files["sidecar"])
|
||
if _sidecar_schema(sidecar) is not None and not _artefakte_komplett(files):
|
||
artefakte = await _artefakte_block(ctx, set_p, files, sidecar, instructions)
|
||
if is_cancelled():
|
||
abgebrochen()
|
||
return
|
||
if artefakte is None:
|
||
return # Abbruch (Fehler/Cancel)
|
||
|
||
# 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)
|
||
# Gliederung (Titel-basiert, robust gegen Nummern-Drift) → DB.
|
||
plan = _json_datei(files["gliederung"])
|
||
if isinstance(plan, dict) and plan.get("kapitel"):
|
||
kapitel = [
|
||
{"titel": ch.get("titel", "Kapitel"),
|
||
"bausteine": [_titel(entries[n]) for n in ch.get("nummern", []) if n in entries]}
|
||
for ch in plan["kapitel"]
|
||
]
|
||
await db.set_gliederung(topic, json.dumps({"kapitel": kapitel}, ensure_ascii=False))
|
||
# Artefakte → DB (Karteikarte/Beispiel je Sub, Diagramm je Baustein).
|
||
artefakte = _json_datei(files["artefakte"])
|
||
if isinstance(artefakte, dict) and _sidecar_schema(sidecar) is not None:
|
||
await _mirror_artefakte_db(topic, sidecar, artefakte)
|
||
except Exception as e:
|
||
log.exception("[%s] Bausteine-Generierung fehlgeschlagen", topic)
|
||
_bausteine_errors[topic] = str(e)[:2000]
|
||
finally:
|
||
# Kein Datei-Cleanup: Zwischendateien bleiben für Resume bzw. Nachvollziehbarkeit.
|
||
_bausteine_progress.pop(topic, None)
|
||
_bausteine_step.pop(topic, None)
|
||
_bausteine_cancelled.discard(topic)
|
||
clear_scope(f"bausteine-{topic}-") # Scope leeren → Neustart blockiert nicht
|