update

backend/bausteine.py

@@ -46,7 +46,8 @@ 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
-SICHTUNG_CHUNK = 80         # Crawl-Seiten je Sichtungs-Agent (Content/Noise-Triage)
+SICHTUNG_CHUNK = 80         # Crawl-Seiten je Sichtungs-Chunk (Content/Noise-Triage)
+SICHTUNG_READERS = 3        # Sichter je Chunk; Merge per Mehrheit, Tie-Break = Content
 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
 
@@ -1226,10 +1227,10 @@ def _sichtung_schema(data) -> list[str] | None:
 
 
 async def _sichte_inhalt(ctx: GenContext, set_p, files: dict, ordner, pages: list[str]) -> set[str]:
-    """Content/Noise-Triage nach dem Crawl. Ein Agent je Chunk markiert eindeutigen Noise;
+    """Content/Noise-Triage nach dem Crawl. Je Chunk SICHTUNG_READERS Sichter; Merge je Seite
-    alles andere bleibt Content (konservativ). Fehler/Ausfall → Seiten bleiben Content (fail-open).
+    per Mehrheit, Tie-Break = Content (konservativ). Fehler/Ausfall → Content (fail-open).
     → Menge der Content-Dateinamen (nie leer, solange pages nicht leer)."""
-    topic, is_cancelled = ctx.topic, ctx.is_cancelled
+    topic, provider, is_cancelled = ctx.topic, ctx.provider, ctx.is_cancelled
     pages = list(pages)
     if not pages:
         return set()
@@ -1245,33 +1246,38 @@ async def _sichte_inhalt(ctx: GenContext, set_p, files: dict, ordner, pages: lis
     chunks = _chunk_nums(sorted(pages), max(1, math.ceil(len(pages) / SICHTUNG_CHUNK)))
 
     async def _chunk(ci: int, chunk: list[str]) -> set[str]:
-        fp = arbeit / f"sichtung-c{ci}.json"
-        fp.unlink(missing_ok=True)
         seiten = "\n".join(_snippet(fn) for fn in chunk)
-        status, res = await run_single_slot(
+        paths = [arbeit / f"sichtung-c{ci}-{i}.json" for i in range(1, SICHTUNG_READERS + 1)]
-            ctx, f"Seiten-Sichtung {ci}",
+        for p in paths:
-            key=f"bausteine-{topic}-sichtung-c{ci}",
+            p.unlink(missing_ok=True)
-            prompt=_prompt("Bausteine-Seiten-Sichtung", topic=topic, seiten=seiten, out_path=fp),
+        slots = [{
-            role="judge", capabilities="files",
+            "key": f"bausteine-{topic}-sichtung-c{ci}-{i}",
-            payload=lambda result, p=fp: _sichtung_schema(_json_datei(p)),
+            "prompt": _prompt("Bausteine-Seiten-Sichtung", topic=topic, seiten=seiten, out_path=p),
-            timeout=_timeout("recherche_mapping", len(chunk)),
+            "role": "judge", "capabilities": "files",
-        )
+            "payload": (lambda result, p=p: _sichtung_schema(_json_datei(p))),
-        if res:  # nur Dateinamen, die wirklich in diesem Chunk sind → Noise
+        } for i, p in enumerate(paths, 1)]
-            cs = set(chunk)
+        ergebnisse = await _race(topic, f"Seiten-Sichtung {ci}", slots, 1,
-            return {n for n in res if n in cs}
+                                 _timeout("recherche_mapping", len(chunk)), provider,
-        return set()  # Fail/kein Noise → Content behalten (fail-open)
+                                 cancelled=is_cancelled, grace=KONSENS_GRACE)
+        cs = set(chunk)
+        stimmen = [set(r) & cs for r in (ergebnisse or [])]  # Noise-Menge je Reader (nur Chunk-Seiten)
+        R = len(stimmen)
+        if not R:
+            return set()  # fail-open: kein Urteil → alles Content
+        # Seite = Noise nur bei strenger Mehrheit; Gleichstand/Minderheit → Content.
+        return {page for page in chunk if 2 * sum(page in s for s in stimmen) > R}
 
     async def melde(d, t):
         set_p(f"Sichte Seiten {d}/{t}…")
 
-    set_p("Sichte Seiten…", step=_step_idx(topic, "Recherche"))
+    set_p("Sichte Seiten…", step=_step_idx(topic, "Quelle aufbereiten"))
     teile = await _gather_fortschritt([_chunk(ci, c) for ci, c in enumerate(chunks, 1)], len(chunks), melde)
     if is_cancelled():
         return set(pages)
     noise = set().union(*(t for t in teile if isinstance(t, set))) if teile else set()
     content = set(pages) - noise
-    _log(topic, f"Sichtung: {len(content)} Content / {len(noise)} Noise von {len(pages)}")
+    _log(topic, f"Sichtung: {len(content)} Content / {len(noise)} Noise von {len(pages)} ({len(chunks)} Chunks × {SICHTUNG_READERS} Sichter)")
-    return content or set(pages)  # alles als Noise markiert? → fail-open, alle behalten
+    return content or set(pages)  # alles Noise? → fail-open, alle behalten
 
 
 async def _quelle_aufbereiten(ctx: GenContext, set_p, files: dict, q: dict, ordner) -> bool: