799 lines
36 KiB
Python
799 lines
36 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 logging
|
||
import math
|
||
import shutil
|
||
import subprocess
|
||
from pathlib import Path
|
||
|
||
from agents import kill_process
|
||
from config import KONSENS_GRACE, KONSENS_MAX_RUNDEN, DEFAULT_PROVIDER
|
||
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, project_dir, subbausteine_path
|
||
from pipeline import (
|
||
CANCELLED, FAILED, GenContext, _extra, _log, _prompt, _race, _rest_schema,
|
||
_runde_schema, _semaphore, _str_liste, _stufen_schema, _timeout, run_single_slot,
|
||
)
|
||
from textkit import (
|
||
_eindeutige_titel, _norm_titel, _parse_auswahl, _parse_subbausteine, _titel,
|
||
_titel_aufloesen, _titel_index, _vormerge,
|
||
)
|
||
|
||
# Subbausteine + Stufen entstehen pro Baustein → wie die Writer chunken:
|
||
# 1 Agent je ~30 Bausteine, gedeckelt.
|
||
SUBBAUSTEIN_CHUNK = 30
|
||
SUBBAUSTEIN_MAX = 20
|
||
# Einstufen ist billig (kurzes Urteil, keine Websuche) → größere Pakete, weniger Dateien/Agenten.
|
||
STUFE_CHUNK = 100
|
||
|
||
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",
|
||
"Stufen finden", "Stufen wählen", "Stufen klären",
|
||
)
|
||
|
||
|
||
def _bausteine_steps(topic: str) -> tuple:
|
||
"""Projekte haben einen zusätzlichen Schritt (Ergänzung), nach der Klärung eingefügt.
|
||
|
||
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.
|
||
"""
|
||
base = ("Recherche", "Konsolidierung", "Klärung")
|
||
rest = (
|
||
"Subbausteine finden", "Subbausteine wählen", "Subbausteine klären",
|
||
"Stufen finden", "Stufen wählen", "Stufen klären",
|
||
)
|
||
if project_dir(topic).is_dir():
|
||
return base + ("Ergänzung",) + rest
|
||
return base + rest
|
||
|
||
|
||
def _step_idx(topic: str, name: str) -> int:
|
||
return _bausteine_steps(topic).index(name)
|
||
|
||
|
||
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",
|
||
"sidecar": subbausteine_path(topic),
|
||
}
|
||
|
||
|
||
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("stufe-*"))) 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"], *dyn,
|
||
]
|
||
|
||
|
||
def cancel_bausteine(topic: str) -> bool:
|
||
if topic not in _bausteine_progress:
|
||
return False
|
||
_bausteine_cancelled.add(topic)
|
||
kill_process(f"bausteine-{topic}-")
|
||
return True
|
||
|
||
|
||
def _resume_step(topic: str) -> int:
|
||
"""Erster noch offener Schritt anhand der persistierten Zwischendateien."""
|
||
files = _bausteine_files(topic)
|
||
if sum(p.exists() for p in files["recherche"]) < 3:
|
||
return 0
|
||
if not files["recherche_mapping"].exists():
|
||
return 1
|
||
mapping = _mapping_schema(_json_datei(files["recherche_mapping"]))
|
||
geklaert = mapping is not None and (
|
||
not mapping[1] # kein strittiger Rest
|
||
or any((r := _runde_schema(_json_datei(p))) is not None and not r[1] for p in files["mapping"].values())
|
||
)
|
||
if not geklaert:
|
||
return 2
|
||
if project_dir(topic).is_dir() and not files["ergaenzung"].exists():
|
||
return _step_idx(topic, "Ergänzung")
|
||
if _sidecar_schema(_json_datei(files["sidecar"])) is not None:
|
||
return len(_bausteine_steps(topic))
|
||
if _sub_roh_schema(_json_datei(files["sub_roh"])) is None:
|
||
return _step_idx(topic, "Subbausteine finden")
|
||
return _step_idx(topic, "Stufen finden")
|
||
|
||
|
||
def bausteine_status(topic: str) -> dict:
|
||
steps = _bausteine_steps(topic)
|
||
ready = bausteine_path(topic).exists()
|
||
generating = topic in _bausteine_progress
|
||
partial = False
|
||
if generating:
|
||
current = _bausteine_step.get(topic)
|
||
states = [
|
||
"pending" if current is None else "done" if i < current else "active" if i == current else "pending"
|
||
for i in range(len(steps))
|
||
]
|
||
elif ready:
|
||
states = ["done"] * len(steps)
|
||
else:
|
||
nxt = _resume_step(topic)
|
||
partial = nxt > 0
|
||
states = ["done" if i < nxt else "pending" for i in range(len(steps))]
|
||
return {
|
||
"ready": ready,
|
||
"generating": generating,
|
||
"progress": _bausteine_progress.get(topic),
|
||
"error": _bausteine_errors.get(topic),
|
||
"partial": partial,
|
||
"steps": [{"label": label, "state": s} for label, s in zip(steps, states)],
|
||
}
|
||
|
||
|
||
def active_bausteine() -> list[dict]:
|
||
return [{"topic": t, "progress": p} for t, p in _bausteine_progress.items()]
|
||
|
||
|
||
def reset_bausteine(topic: str) -> None:
|
||
files = _bausteine_files(topic)
|
||
files["final"].unlink(missing_ok=True)
|
||
files["sidecar"].unlink(missing_ok=True) # liegt im Themen-Root, nicht in arbeit/
|
||
shutil.rmtree(files["arbeit"], ignore_errors=True)
|
||
_bausteine_errors.pop(topic, None)
|
||
|
||
|
||
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
|
||
|
||
|
||
def _build_recherche_prompt(topic: str, out_path: Path, instructions: str = "", project: Path | None = None) -> str:
|
||
if project:
|
||
source = _prompt("Bausteine-Quelle-Projekt", project=project)
|
||
else:
|
||
source = _prompt("Bausteine-Quelle-Thema", topic=topic)
|
||
return _prompt(
|
||
"Bausteine-Recherche",
|
||
topic=topic, source=source, bausteine_path=out_path, 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 (finale Sidecar)."""
|
||
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 ("einfach", "mittel", "schwer"):
|
||
return None
|
||
return data
|
||
|
||
|
||
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 _merge_finder(num: int, idx: dict, finder: list[dict]) -> tuple[list[str], list[str]]:
|
||
"""Subbausteine eines Bausteins über die Finder mergen → (konsens ≥2, rest ==1)."""
|
||
counts: dict[str, int] = {}
|
||
repr_text: dict[str, str] = {}
|
||
order: list[str] = []
|
||
for d in finder:
|
||
# die Marker-Liste dieses Finders für genau diesen Baustein
|
||
subs = next((s for marker, s in d.items() if _titel_aufloesen(idx, marker) == num), [])
|
||
gesehen: set[str] = set()
|
||
for sub in subs:
|
||
key = _norm_titel(sub)
|
||
if not key or key in gesehen:
|
||
continue
|
||
gesehen.add(key)
|
||
if key not in counts:
|
||
counts[key], repr_text[key] = 0, sub
|
||
order.append(key)
|
||
counts[key] += 1
|
||
konsens = [repr_text[k] for k in order if counts[k] >= 2]
|
||
rest = [repr_text[k] for k in order if counts[k] == 1]
|
||
return konsens, rest
|
||
|
||
|
||
def _final_text(chunk: list[int], entries: dict, daten: dict) -> str:
|
||
"""Marker-Datei aus reinem Konsens (wenn kein Judge nötig)."""
|
||
teile = []
|
||
for num in chunk:
|
||
konsens = daten[num][0]
|
||
if konsens:
|
||
teile.append(f"<!-- baustein: {_titel(entries[num])} -->\n" + "\n".join(f"- {s}" for s in konsens))
|
||
return "\n".join(teile) + "\n"
|
||
|
||
|
||
def _judge_block(chunk: list[int], entries: dict, daten: dict) -> str:
|
||
"""Eingabe für den Subbaustein-Judge: pro Baustein Konsens + Strittiges."""
|
||
lines = []
|
||
for num in chunk:
|
||
konsens, rest = daten[num]
|
||
lines.append(f"BAUSTEIN: {_titel(entries[num])}")
|
||
lines.append("Konsens:")
|
||
lines.extend(f"- {s}" for s in konsens)
|
||
if not konsens:
|
||
lines.append("- (noch keiner)")
|
||
if rest:
|
||
lines.append("Strittig:")
|
||
lines.extend(f"- {s}" for s in rest)
|
||
lines.append("")
|
||
return "\n".join(lines)
|
||
|
||
|
||
async def _subbausteine_block(ctx: GenContext, set_p, files: dict, entries: dict, instructions: str) -> dict | None:
|
||
"""Block B: drei Phasen mit Barriere — Finden, Wählen (Code-Merge), Klären.
|
||
Pro Phase laufen alle Pakete parallel; der Schritt bleibt, bis das letzte fertig ist.
|
||
→ {Baustein-Titel: [Subbaustein, …]} oder None bei Abbruch/Fehler."""
|
||
topic, provider, is_cancelled = ctx.topic, ctx.provider, ctx.is_cancelled
|
||
arbeit = files["arbeit"]
|
||
idx = _titel_index(entries)
|
||
caps = "files" if project_dir(topic).is_dir() else "full"
|
||
nums = list(entries)
|
||
chunks = _chunk_nums(nums, _n_chunks(len(nums)))
|
||
n = len(chunks)
|
||
|
||
def finder_paths(c):
|
||
return [arbeit / f"subbaustein-c{c}-{i}.md" for i in (1, 2, 3)]
|
||
|
||
def final_path(c):
|
||
return arbeit / f"subbaustein-final-c{c}.md"
|
||
|
||
# Phase „Subbausteine finden": pro Paket 3 Finder (min. 2), alle Pakete parallel.
|
||
async def _finde(c, chunk):
|
||
paths = finder_paths(c)
|
||
vorhanden = sum(1 for p in paths if _parse_subbausteine(_read(p)))
|
||
if vorhanden >= 2:
|
||
return True
|
||
zuteilung = "\n".join(f"- {entries[num]}" for num in chunk)
|
||
offen = [(i, p) for i, p in enumerate(paths, 1) if not _parse_subbausteine(_read(p))]
|
||
slots = [{
|
||
"key": f"bausteine-{topic}-subbaustein-c{c}-{i}",
|
||
"prompt": _prompt("Subbaustein-Recherche", topic=topic, zuteilung=zuteilung, out_path=p, extra=_extra(instructions)),
|
||
"role": "quick", "capabilities": caps,
|
||
"payload": (lambda result, p=p: _parse_subbausteine(_read(p)) or None),
|
||
} for i, p in offen]
|
||
neu = await _race(topic, f"Subbausteine Paket {c}", slots, 2 - vorhanden, _timeout("subbaustein", len(chunk)), provider, cancelled=is_cancelled, grace=KONSENS_GRACE)
|
||
return not is_cancelled() and neu is not None
|
||
|
||
set_p(f"Subbausteine finden ({n} Pakete)…", step=_step_idx(topic, "Subbausteine finden"))
|
||
oks = await asyncio.gather(*[_finde(c, chunk) for c, chunk in enumerate(chunks, 1)], return_exceptions=True)
|
||
if is_cancelled():
|
||
return None
|
||
if not all(ok is True for ok in oks):
|
||
_bausteine_errors[topic] = "Subbausteine fehlgeschlagen (Recherche)"
|
||
return None
|
||
|
||
# Phase „Subbausteine wählen": Code-Merge je Paket (instant, kein Agent).
|
||
set_p(f"Subbausteine wählen ({n} Pakete)…", step=_step_idx(topic, "Subbausteine wählen"))
|
||
daten_by_c = {}
|
||
for c, chunk in enumerate(chunks, 1):
|
||
finder = [d for p in finder_paths(c) if (d := _parse_subbausteine(_read(p)))]
|
||
daten_by_c[c] = {num: _merge_finder(num, idx, finder) for num in chunk}
|
||
|
||
# Phase „Subbausteine klären": Judge je Paket mit Strittigem, alle parallel.
|
||
async def _klaere(c, chunk):
|
||
daten = daten_by_c[c]
|
||
fp = final_path(c)
|
||
if _parse_subbausteine(_read(fp)):
|
||
return
|
||
if not any(daten[num][1] for num in chunk):
|
||
atomic_write_text(fp, _final_text(chunk, entries, daten))
|
||
return
|
||
status, _ = await run_single_slot(
|
||
ctx, f"Subbaustein-Klärung {c}",
|
||
key=f"bausteine-{topic}-subbaustein-final-c{c}",
|
||
prompt=_prompt("Subbaustein-Mapping", topic=topic, bausteine=_judge_block(chunk, entries, daten), out_path=fp, extra=_extra(instructions)),
|
||
role="judge", capabilities="files",
|
||
payload=lambda result, p=fp: _parse_subbausteine(_read(p)) or None,
|
||
timeout=_timeout("subbaustein_check", len(chunk)),
|
||
)
|
||
if status == FAILED:
|
||
_log(topic, f"Subbaustein-Klärung Paket {c} fehlgeschlagen — nur Konsens übernommen")
|
||
|
||
set_p(f"Subbausteine klären ({n} Pakete)…", step=_step_idx(topic, "Subbausteine klären"))
|
||
await asyncio.gather(*[_klaere(c, chunk) for c, chunk in enumerate(chunks, 1)], return_exceptions=True)
|
||
if is_cancelled():
|
||
return None
|
||
|
||
# Fehlende finale Dateien → Konsens-Fallback; dann alle parsen.
|
||
roh: dict[str, list[str]] = {}
|
||
for c, chunk in enumerate(chunks, 1):
|
||
fp = final_path(c)
|
||
if not _parse_subbausteine(_read(fp)):
|
||
atomic_write_text(fp, _final_text(chunk, entries, daten_by_c[c]))
|
||
for marker, subs in (_parse_subbausteine(_read(fp)) or {}).items():
|
||
num = _titel_aufloesen(idx, marker)
|
||
if num is not None:
|
||
roh[_titel(entries[num])] = subs
|
||
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 = _chunk_nums(list(range(len(items))), _n_chunks(len(items), STUFE_CHUNK))
|
||
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 = "\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 _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
|
||
|
||
set_p(f"Stufen finden ({n} Pakete)…", step=_step_idx(topic, "Stufen finden"))
|
||
oks = await asyncio.gather(*[_rate(c, idxs) for c, idxs in enumerate(chunks, 1)], return_exceptions=True)
|
||
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 'mittel'")
|
||
entsch = entsch if isinstance(entsch, dict) else {}
|
||
# Strittige ohne Entscheid → 'mittel'; Vote-Gewinner bleiben; Judge überschreibt.
|
||
ergebnis = {**{k: "mittel" for k in strittig}, **ergebnis, **entsch}
|
||
return {item_idxs[k - 1] + 1: stufe for k, stufe in ergebnis.items()}
|
||
|
||
set_p(f"Stufen klären ({n} Pakete)…", step=_step_idx(topic, "Stufen klären"))
|
||
parts = await asyncio.gather(*[_klaere(c, idxs) for c, idxs in enumerate(chunks, 1)], return_exceptions=True)
|
||
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 'mittel' 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: "mittel" 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, "mittel")})
|
||
sidecar[titel] = lst
|
||
return sidecar
|
||
|
||
|
||
async def generate_bausteine(topic: str, instructions: str = "", provider: str = DEFAULT_PROVIDER) -> None:
|
||
if topic in _bausteine_progress:
|
||
return
|
||
_bausteine_progress[topic] = "Wartend…"
|
||
_bausteine_errors.pop(topic, None)
|
||
|
||
files = _bausteine_files(topic)
|
||
final_path = files["final"]
|
||
project = project_dir(topic) if project_dir(topic).is_dir() else None
|
||
|
||
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)
|
||
if project:
|
||
await asyncio.to_thread(_pdfs_konvertieren, project)
|
||
# „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.
|
||
fertig = 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)
|
||
|
||
# Schritt 1: 5 Recherche-Agenten, min. 3 mit Grace-Fenster — alle gültigen
|
||
# Slot-Dateien fließen ins Mapping (kein Kappen mehr bei 3)
|
||
recherchen: list[str] = []
|
||
offen = []
|
||
for i, path in enumerate(files["recherche"], 1):
|
||
text = _file_payload(path)
|
||
if text is not None:
|
||
recherchen.append(text)
|
||
else:
|
||
offen.append((i, path))
|
||
vorhanden = len(recherchen)
|
||
set_p(f"Recherche läuft ({vorhanden} gültig, min. 3)…", step=0)
|
||
if vorhanden < 3:
|
||
caps = "files" if project else "full"
|
||
slots = [
|
||
{
|
||
"key": f"bausteine-{topic}-recherche-{i}",
|
||
"prompt": _build_recherche_prompt(topic, path, instructions, project),
|
||
"role": "quick", "capabilities": caps,
|
||
"payload": (lambda result, p=path: _file_payload(p)),
|
||
}
|
||
for i, path in offen
|
||
]
|
||
neue = await _race(
|
||
topic, "Recherche", slots, 3 - vorhanden, _timeout("recherche"), provider,
|
||
on_update=lambda c: set_p(f"Recherche läuft ({vorhanden + c} gültig, min. 3)…"),
|
||
cancelled=is_cancelled, grace=KONSENS_GRACE,
|
||
)
|
||
if is_cancelled():
|
||
abgebrochen()
|
||
return
|
||
if neue is None:
|
||
_bausteine_errors[topic] = "Recherche fehlgeschlagen (Minimum nicht erreicht)"
|
||
return
|
||
recherchen += neue
|
||
|
||
# Schritt 2: Recherche-Mapping — Code-Vormerge (exakte Titel) + 1 Agent
|
||
# für semantische Dubletten und Konsens/Rest-Teilung (fatal)
|
||
mapping = _mapping_schema(_json_datei(files["recherche_mapping"]))
|
||
if mapping is None:
|
||
set_p("Konsolidiere Recherche…", step=1)
|
||
files["recherche_mapping"].unlink(missing_ok=True)
|
||
gemergt = _vormerge([_parse_auswahl(t) for t in recherchen])
|
||
eintraege = "\n".join(f"{i}. {text} ({n}× genannt)" for i, (text, n) in enumerate(gemergt, 1))
|
||
status, mapping = await run_single_slot(
|
||
ctx, "Recherche-Mapping",
|
||
key=f"bausteine-{topic}-recherche-mapping",
|
||
prompt=_prompt(
|
||
"Bausteine-Recherche-Mapping",
|
||
topic=topic, n=len(recherchen), eintraege=eintraege,
|
||
out_path=files["recherche_mapping"],
|
||
),
|
||
role="judge", capabilities="files",
|
||
payload=lambda result: _mapping_schema(_json_datei(files["recherche_mapping"])),
|
||
timeout=_timeout("recherche_mapping", len(gemergt)),
|
||
)
|
||
if status == CANCELLED:
|
||
abgebrochen()
|
||
return
|
||
if status == FAILED:
|
||
_bausteine_errors[topic] = "Recherche-Mapping fehlgeschlagen"
|
||
return
|
||
konsens, rest = mapping
|
||
|
||
# Klärungs-Loop: 3 Auswahl-Agenten entscheiden über den Rest, ein
|
||
# Mapping-Agent sortiert in aufnehmen/verwerfen/weiter strittig.
|
||
# Leerer Rest beendet den Loop; Runde KONSENS_MAX_RUNDEN muss
|
||
# alles entscheiden. Der Konsens wächst nur hier im Code.
|
||
runde = 0
|
||
while rest and runde < KONSENS_MAX_RUNDEN:
|
||
runde += 1
|
||
final_runde = runde == KONSENS_MAX_RUNDEN
|
||
set_p(f"Klärung läuft (Runde {runde}/{KONSENS_MAX_RUNDEN})…", step=2)
|
||
mapping_path = files["mapping"][runde]
|
||
|
||
# Resume: fertiges Runden-Mapping wird direkt übernommen
|
||
ergebnis = _runde_schema(_json_datei(mapping_path), final=final_runde)
|
||
if ergebnis is None:
|
||
mapping_path.unlink(missing_ok=True)
|
||
konsens_block = "\n".join(f"- {t}" for t in konsens)
|
||
rest_block = "\n".join(f"- {t}" for t in rest)
|
||
|
||
# 3 Auswahl-Agenten, min. 2 mit Grace-Fenster
|
||
entscheidungen = []
|
||
offen = []
|
||
for i, path in enumerate(files["auswahl"][runde], 1):
|
||
res = _rest_schema(_json_datei(path))
|
||
if res is not None:
|
||
entscheidungen.append(res)
|
||
else:
|
||
offen.append((i, path))
|
||
if len(entscheidungen) < 2:
|
||
slots = [
|
||
{
|
||
"key": f"bausteine-{topic}-auswahl-r{runde}-{i}",
|
||
"prompt": _prompt(
|
||
"Bausteine-Auswahl",
|
||
topic=topic, konsens=konsens_block, rest=rest_block, out_path=path,
|
||
),
|
||
"role": "fast", "capabilities": "files",
|
||
"payload": (lambda result, p=path: _rest_schema(_json_datei(p))),
|
||
}
|
||
for i, path in offen
|
||
]
|
||
neue = await _race(
|
||
topic, f"Auswahl r{runde}", slots, 2 - len(entscheidungen),
|
||
_timeout("auswahl", len(rest)), provider,
|
||
cancelled=is_cancelled, grace=KONSENS_GRACE,
|
||
)
|
||
if is_cancelled():
|
||
abgebrochen()
|
||
return
|
||
if neue is None:
|
||
_bausteine_errors[topic] = f"Auswahl fehlgeschlagen (Runde {runde}, Minimum nicht erreicht)"
|
||
return
|
||
entscheidungen += neue
|
||
|
||
# Votum pro Rest-Eintrag deterministisch zählen
|
||
indizes = [_titel_index(dict(enumerate(e, 1))) for e in entscheidungen]
|
||
voten = "\n".join(
|
||
f"{i}. {text} (von {sum(1 for idx in indizes if _titel_aufloesen(idx, text) is not None)}"
|
||
f"/{len(entscheidungen)} Agenten übernommen)"
|
||
for i, text in enumerate(rest, 1)
|
||
)
|
||
final_zusatz = (
|
||
"\n- LETZTE RUNDE: Es gibt keine weitere Runde. `rest` MUSS leer sein"
|
||
" — entscheide JEDEN Eintrag selbst: aufnehmen oder verwerfen."
|
||
if final_runde else ""
|
||
)
|
||
status, ergebnis = await run_single_slot(
|
||
ctx, f"Auswahl-Mapping r{runde}",
|
||
key=f"bausteine-{topic}-auswahl-mapping-r{runde}",
|
||
prompt=_prompt(
|
||
"Bausteine-Auswahl-Mapping",
|
||
topic=topic, n=len(entscheidungen), konsens=konsens_block,
|
||
rest=voten, final=final_zusatz, out_path=mapping_path,
|
||
),
|
||
role="judge", capabilities="files",
|
||
payload=lambda result, p=mapping_path, f=final_runde: _runde_schema(_json_datei(p), final=f),
|
||
timeout=_timeout("auswahl_mapping", len(rest)),
|
||
)
|
||
if status == CANCELLED:
|
||
abgebrochen()
|
||
return
|
||
if status == FAILED:
|
||
_bausteine_errors[topic] = f"Auswahl-Mapping fehlgeschlagen (Runde {runde})"
|
||
return
|
||
|
||
aufnehmen, rest = ergebnis
|
||
_log(topic, f"Klärung Runde {runde}: {len(aufnehmen)} aufgenommen, {len(rest)} weiter strittig")
|
||
konsens = konsens + aufnehmen
|
||
|
||
entries = {i: t for i, t in enumerate(konsens, 1)}
|
||
|
||
# Nur Projekte: Themenfeld-Ergänzung — Skript/Projekt ist ein Ausschnitt,
|
||
# ein Web-Agent ergänzt kanonisch fehlende Bausteine, markiert mit [Ergänzung].
|
||
if project:
|
||
set_p("Ergänze Themenfeld…", step=3)
|
||
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)
|
||
sidecar = await _stufen_block(ctx, set_p, files, roh, instructions)
|
||
if is_cancelled():
|
||
abgebrochen()
|
||
return
|
||
if sidecar is None:
|
||
return
|
||
atomic_write_json(files["sidecar"], sidecar, indent=1)
|
||
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)
|