update

backend/bausteine.py

@@ -33,10 +33,9 @@ from textkit import (
     _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
+# 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
 
@@ -474,6 +473,27 @@ def _frage_muster_schema(data) -> list[dict] | None:
     return out or None
 
 
+def _frage_muster_chunk_schema(data) -> list[dict] | None:
+    """{"muster": [{baustein, subbaustein, typ, frage}, …]} → Liste valider Einträge · sonst None.
+
+    Wie _frage_muster_schema, aber mit `baustein` (Zuordnung im 10er-Chunk). Ungültige
+    Einzel-Einträge werden übersprungen (nicht die ganze Liste verworfen)."""
+    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()
+        typ = str(e.get("typ", "")).strip().casefold()
+        frage = str(e.get("frage", "")).strip()
+        if not bau or not sub or typ not in FRAGETYPEN or not frage:
+            continue
+        out.append({"baustein": bau, "subbaustein": sub, "typ": typ, "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."""
@@ -877,9 +897,9 @@ def _match_sub(agent_sub: str, rel: list[str]) -> str:
 
 
 async def _frage_muster_block(ctx: GenContext, set_p, files: dict, sidecar: dict, instructions: str) -> dict | None:
-    """Block E: drei Phasen — Finden (1 Generator je Baustein, parallel), Wählen (Code:
-    relevante Subs + Dedup), Klären (Kritiker je Baustein bereinigt die Tabelle).
-    Generativ statt Vote: Muster sind Text, kein 3-Rater-Konsens sinnvoll.
+    """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, typ, frage}, …]} oder None bei Abbruch."""
     topic, is_cancelled = ctx.topic, ctx.is_cancelled
     arbeit = files["arbeit"]
@@ -892,99 +912,153 @@ async def _frage_muster_block(ctx: GenContext, set_p, files: dict, sidecar: dict
     if not bausteine:
         return {}
     typen_block = "\n".join(f"- {k}: {v}" for k, v in FRAGETYPEN.items())
+    chunks = _chunk_nums(list(range(len(bausteine))), _n_chunks(len(bausteine)))  # ~10 Bausteine/Chunk
 
-    def roh_path(c):
-        return arbeit / f"frage-muster-c{c}.json"
+    def roh_path(ci):
+        return arbeit / f"frage-muster-c{ci}.json"
 
-    def final_path(c):
-        return arbeit / f"frage-muster-final-c{c}.json"
+    def final_path(ci):
+        return arbeit / f"frage-muster-final-c{ci}.json"
 
-    # Phase „Fragen finden": pro Baustein 1 Generator, alle parallel.
-    async def _finde(c, titel, rel):
-        fp = roh_path(c)
-        if _frage_muster_schema(_json_datei(fp)):
+    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
-        sub_block = "\n".join(f"- {s}" for s in rel)
+        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 {c}",
-            key=f"bausteine-{topic}-frage-muster-c{c}",
-            prompt=_prompt("Frage-Muster-Recherche", topic=topic, baustein=titel,
-                           subbausteine=sub_block, typen=typen_block, out_path=fp, extra=_extra(instructions)),
+            ctx, f"Frage-Muster {ci}",
+            key=f"bausteine-{topic}-frage-muster-c{ci}",
+            prompt=_prompt("Frage-Muster-Recherche", topic=topic, bausteine=block,
+                           typen=typen_block, out_path=fp, extra=_extra(instructions)),
             role="fast", capabilities="files",
-            payload=lambda result, p=fp: _frage_muster_schema(_json_datei(p)),
-            timeout=_timeout("frage_muster", len(rel)),
+            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 Baustein {c} fehlgeschlagen — kein Muster (Fallback Live)")
+            _log(topic, f"Frage-Muster Chunk {ci} fehlgeschlagen — Bausteine im Fallback (Nachrunde/Live)")
 
-    await _gather_fortschritt([_finde(c, t, r) for c, (t, r) in enumerate(bausteine, 1)], len(bausteine), _melde_p(set_p, topic, "Fragen finden"))
+    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 — Dubletten je Baustein raus, Subbaustein-Titel locker
-    # auf die relevanten mappen (nichts wegen Titel-Abweichung verwerfen).
-    def _waehle(c, rel):
-        eintraege = _frage_muster_schema(_json_datei(roh_path(c))) or []
-        gesehen, sauber = set(), []
-        for e in eintraege:
+    # 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
             norm = _norm_frage(e["frage"])
-            if norm in gesehen:
+            seen = gesehen.setdefault(titel, set())
+            if norm in seen:
                 continue
-            gesehen.add(norm)
-            sauber.append({**e, "subbaustein": _match_sub(e["subbaustein"], rel)})
-        return sauber
+            seen.add(norm)
+            out.setdefault(titel, []).append(
+                {"subbaustein": _match_sub(e["subbaustein"], rel_by[titel]), "typ": e["typ"], "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_c = {c: _waehle(c, rel) for c, (titel, rel) in enumerate(bausteine, 1)}
+    roh_by_titel = _waehle_all(range(len(chunks)))
 
-    # Phase „Fragen klären": Kritiker je Baustein bereinigt die Tabelle (eindeutig, distinkt).
-    async def _klaere(c, titel):
-        roh = roh_by_c[c]
-        fp = final_path(c)
-        if _frage_muster_schema(_json_datei(fp)) or not roh:
-            return
-        tabelle = "\n".join(f"{i}. [{e['typ']}] ({e['subbaustein']}) {e['frage']}" for i, e in enumerate(roh, 1))
+    # 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['typ']}] ({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 {c}",
-            key=f"bausteine-{topic}-frage-muster-final-c{c}",
-            prompt=_prompt("Frage-Muster-Kritik", topic=topic, baustein=titel, tabelle=tabelle, out_path=fp, extra=_extra(instructions)),
+            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_schema(_json_datei(p)),
-            timeout=_timeout("frage_muster_check", len(roh)),
+            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 Baustein {c} fehlgeschlagen — Roh-Muster übernommen")
+            _log(topic, f"Frage-Muster-Klärung Chunk {ci} fehlgeschlagen — Roh-Muster übernommen")
 
-    await _gather_fortschritt([_klaere(c, t) for c, (t, _) in enumerate(bausteine, 1)], len(bausteine), _melde_p(set_p, topic, "Fragen klären"))
+    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 Tabelle je Baustein, Fallback auf Roh-Muster. Subbaustein-Titel locker mappen.
-    def _finalisiere(c, rel):
-        final = _frage_muster_schema(_json_datei(final_path(c))) or roh_by_c[c]
-        return [{**e, "subbaustein": _match_sub(e["subbaustein"], rel)} for e in final]
+    # 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]), "typ": e["typ"], "frage": e["frage"]})
+        return out
 
-    ergebnis = {titel: _finalisiere(c, rel) for c, (titel, rel) in enumerate(bausteine, 1)}
+    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": Bausteine mit relevanten Subs aber 0 Mustern eine Runde nachholen.
+    # 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 = [(c, t, r) for c, (t, r) in enumerate(bausteine, 1) if not ergebnis.get(t)]
-    if leer:
-        _log(topic, f"Frage-Muster: {len(leer)} Baustein(e) ohne Muster — Nachrunde")
-        for c, t, r in leer:
-            roh_path(c).unlink(missing_ok=True)
-            final_path(c).unlink(missing_ok=True)
-        await _gather_fortschritt([_finde(c, t, r) for c, t, r in leer], len(leer), _melde_p(set_p, 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 c, t, r in leer:
-            roh_by_c[c] = _waehle(c, r)
-        await _gather_fortschritt([_klaere(c, t) for c, t, r in leer], len(leer), _melde_p(set_p, topic, "Fragen prüfen"))
+        for titel, eintraege in _waehle_all(nach).items():
+            roh_by_titel[titel] = eintraege
+        await klaere_alle(nach)
         if is_cancelled():
             return None
-        for c, t, r in leer:
-            ergebnis[t] = _finalisiere(c, r)
-        rest = [t for c, t, r in leer if not ergebnis.get(t)]
+        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

backend/guide.py

@@ -43,8 +43,9 @@ GUIDE_STEPS = ("Auswahl", "Gliederung", "Inhalte", "Inhalts-Check", "Schreiben",
 # Writer skalieren mit der Section-Zahl: 1 Writer je ~30 Sections (gedeckelt).
 # Kleine Pakete vermeiden Lazy-Output bei langen Listen und begrenzen den Schaden
 # eines fehlgeschlagenen Writers.
-WRITER_SECTIONS = 30
-WRITER_MAX = 20
+# Inhalte/Inhalts-Check/Lese-Prüfung laufen in Paketen von ~GUIDE_CHUNK Bausteinen je Agent.
+# Nur der Writer (Schreiben) bleibt 1 Agent je Baustein (variable Längen, kein Kürzen).
+GUIDE_CHUNK = 10
 
 # Lese-Prüfung: nur EINE Runde (Check + Fix). Folgerunden brachten kaum Mehrwert
 # (1 Agent je Baustein prüft ohnehin fein), kosten aber extra Agenten.
@@ -584,10 +585,10 @@ async def _generate_sections(
     plan = _mit_resten(plan, sel_entries)
     _set_fertig(content_path, 1)  # Gliederung steht
 
-    # Chunks festlegen: 1 Chunk = 1 Baustein. So läuft JEDER Section-Schritt (Inhalte,
-    # Inhalts-Prüfung, Writer, Lese-Check) mit genau einem Agent je Baustein. Der Writer
-    # rationiert sein Output-Budget dann nicht über viele Sections → längere, variable Texte.
-    chunks = [[{"title": ch["title"], "nums": [num]}] for ch in plan for num in ch["nums"]]
+    # Grobe Chunks (~GUIDE_CHUNK Bausteine je Agent) für Inhalte, Inhalts-Check und Lese-Prüfung.
+    # Der Writer baut darunter pro Baustein (eigene feine Chunks, s.u.) → variable Längen.
+    total_sections = sum(len(c["nums"]) for c in plan)
+    chunks = _split_chunks(plan, max(1, math.ceil(total_sections / GUIDE_CHUNK)))
     # Subbausteine je Baustein, je Format gefiltert: Guide=relevant · FullGuide=alle · Rest=Rand.
     # Subs ohne Relevanz-Feld (Alt-Themen) zählen als „nicht Rand" (→ Guide/FullGuide).
     if format_name == "FullGuide":
@@ -706,11 +707,14 @@ async def _generate_sections(
     _set_fertig(content_path, 3)  # Inhalts-Check durch
 
     # Schritt 4: Schreiben — Writer formuliert die geprüften Inhalte aus (Resume).
+    # FEINE Chunks: genau 1 Baustein je Writer → variable Längen, kein Budget-Rationieren.
     def inhalte_text(chunk) -> str:
         nums = [num for ch in chunk for num in ch["nums"] if num in inhalt_by_num]
         return "\n\n".join(f"<!-- section: {_titel(entries[num])} -->\n{inhalt_by_num[num]}" for num in nums)
 
-    paths = [content_path.parent / f"{content_path.stem}.chunk-{i}.md" for i in range(1, writer_count + 1)]
+    w_chunks = [[{"title": ch["title"], "nums": [num]}] for ch in plan for num in ch["nums"]]
+    w_zuteil = [_zuteilung_subs(c, entries, subs_by_titel) for c in w_chunks]
+    paths = [content_path.parent / f"{content_path.stem}.chunk-{i}.md" for i in range(1, len(w_chunks) + 1)]
     offen = [i for i, p in enumerate(paths) if not p.exists()]
     if offen:
         async def melde(d, t): await _set_step(guide_id, 4, f"Schreibe Sections {d}/{t}…")
@@ -719,14 +723,14 @@ async def _generate_sections(
                 f"{guide_id}-w{i + 1}",
                 _prompt(
                     "Guide-Writer",
-                    topic=topic, format_name=format_name, zuteilung=zuteilungen[i],
-                    inhalte=inhalte_text(chunks[i]),
+                    topic=topic, format_name=format_name, zuteilung=w_zuteil[i],
+                    inhalte=inhalte_text(w_chunks[i]),
                     facts=facts, spec=spec, out_path=paths[i], extra=_extra(instructions),
                 ),
-                _timeout("writer", chunk_sizes[i]), provider=provider, role="guide", capabilities="full",
+                _timeout("writer", 1), provider=provider, role="guide", capabilities="full",
             )
             for i in offen
-        ], writer_count, melde, start=writer_count - len(offen))
+        ], len(w_chunks), melde, start=len(w_chunks) - len(offen))
         if is_cancelled():
             return None
         for i, r in zip(offen, results):