This commit is contained in:
team3
2026-06-27 11:08:36 +02:00
parent 2266606ca2
commit 8c69c44b37
3 changed files with 172 additions and 33 deletions

View File

@@ -14,6 +14,7 @@ import asyncio
import json
import logging
import math
import re
import shutil
import subprocess
import time
@@ -46,6 +47,9 @@ 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
@@ -478,8 +482,47 @@ def _pdfs_konvertieren(project: Path) -> None:
_QUELLE_TEMPLATE = {"projekt": "Bausteine-Quelle-Projekt", "uni": "Bausteine-Quelle-Uni", "link": "Bausteine-Quelle-Link"}
def _build_recherche_prompt(topic: str, out_path: Path, instructions: str, typ: str, ordner: Path | None, fokus: str = "") -> str:
if typ in _QUELLE_TEMPLATE:
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)
@@ -615,7 +658,11 @@ async def _subbausteine_block(ctx: GenContext, set_p, files: dict, entries: dict
{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"]
caps = "files" if quelle_ordner(topic) else "full"
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)
@@ -711,8 +758,8 @@ async def _subbausteine_block(ctx: GenContext, set_p, files: dict, entries: dict
status, _ = await run_single_slot(
ctx, f"Subbaustein-Klärung {c}",
key=f"bausteine-{topic}-subbaustein-final-c{c}",
prompt=_prompt("Subbaustein-Mapping", topic=topic, bausteine="\n\n".join(bloecke), out_path=fp, extra=_extra(instructions)),
role="judge", capabilities="files",
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)),
)
@@ -917,8 +964,9 @@ def _fakten_schema(data) -> list[dict] | None:
return out or None
def _fakten_check_schema(data) -> list[str] | None:
"""Fakten-Check → beanstandete sub_norms · {ok:true}→[] · None bei ungültig."""
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:
@@ -926,7 +974,8 @@ def _fakten_check_schema(data) -> list[str] | None:
pr = data.get("probleme")
if not isinstance(pr, list):
return None
return [sn for p in pr if isinstance(p, dict) and (sn := _norm_titel(str(p.get("subbaustein", ""))))]
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:
@@ -950,10 +999,11 @@ def _fakten_komplett(files: dict) -> bool:
return isinstance(d, dict) and bool(d)
async def _fakten_block(ctx, set_p, files: dict, roh: dict, q: dict, ordner, instructions: str) -> dict | None:
"""Block: je Sub Quell-Fakten extrahieren (finden) → verifizieren (prüfen) → korrigieren (fix).
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.
{Baustein-Titel: {sub_norm: fakten-dict}} oder None bei Abbruch/Fehler."""
(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"
@@ -961,7 +1011,7 @@ async def _fakten_block(ctx, set_p, files: dict, roh: dict, q: dict, ordner, ins
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 {}
return {}, {}
chunks = _lpt_chunks([len(subs) for _, subs in bausteine], FAKTEN_CHUNK_SUBS)
def roh_path(ci): return arbeit / f"fakten-c{ci}.json"
@@ -1008,11 +1058,12 @@ async def _fakten_block(ctx, set_p, files: dict, roh: dict, q: dict, ordner, ins
_bausteine_errors[topic] = "Fakten-Extraktion fehlgeschlagen"
return None
# Phase „Fakten prüfen": FAKTEN_CHECK_PANEL Judges je Chunk → beanstandete sub_norms (Mehrheit).
# 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()
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(*[
@@ -1021,29 +1072,45 @@ async def _fakten_block(ctx, set_p, files: dict, roh: dict, q: dict, ordner, ins
_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]
votes: dict[str, int] = {}
for s in outs:
for sn in set(s):
votes[sn] = votes.get(sn, 0) + 1
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
return ci, {sn for sn, v in votes.items() if v > schwelle}
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 = {ci: subs for r in pruef if isinstance(r, tuple) for ci, subs in [r]}
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": je Chunk mit beanstandeten Subs → diese neu extrahieren (frische Quelle).
n_problem = sum(len(s) for s in beanstandet.values())
# 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 = beanstandet.get(ci, set())
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 beanstandeten Subs als Block
# 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]
@@ -1057,20 +1124,27 @@ async def _fakten_block(ctx, set_p, files: dict, roh: dict, q: dict, ordner, ins
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 beanstandet], len(beanstandet), _melde_p(set_p, topic, "Fakten fix"))
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 beanstandete (Vollständigkeits-Guard: fehlt der Fix, bleibt roh).
# 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():
gewinner = fix.get(bt, {}).get(sn, fk) if sn in beanstandet.get(ci, set()) else fk
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}
return ergebnis
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:
@@ -1511,7 +1585,55 @@ async def _recherche_batch(ctx: GenContext, set_p, files: dict, q: dict, ordner,
await db.set_step_status(topic, "Recherche", "fertig")
return True
# Content-Seiten stehen schon fest (Sichtung im Schritt „Quelle aufbereiten").
# 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)))
@@ -2040,12 +2162,21 @@ async def generate_bausteine(topic: str, instructions: str = "", provider: str =
# 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):
fakten_map = await _fakten_block(ctx, set_p, files, roh, q, ordner, instructions)
res = await _fakten_block(ctx, set_p, files, roh, q, ordner, instructions)
if is_cancelled():
abgebrochen()
return
if fakten_map is None:
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():