update

backend/bausteine.py

@@ -41,10 +41,12 @@ SUBBAUSTEIN_MAX = 40
 # Einstufen ist billig (kurzes Urteil, keine Websuche) → größere Pakete, weniger Dateien/Agenten.
 STUFE_CHUNK = 100
 
-# Recherche-Loop: 5 Agenten je Runde, Runden bis volle Crawl-Abdeckung / 0 neue / Zeit-Kappe.
+# Recherche: feste Datei-Batches statt Such-Loop → jede Crawl-Seite genau einmal zugeteilt.
-RECHERCHE_AGENTEN = 5
+RECHERCHE_BATCH = 20        # Crawl-Seiten je Batch
-RECHERCHE_QUORUM = 3        # je Runde mind. so viele gültige Slot-Ergebnisse
+RECHERCHE_READERS = 2       # Reader-Agenten je Batch (Konsens ≥2 innerhalb des Batches)
-RECHERCHE_KAPPE = 1800      # Loop-Gesamt-Sekunden (Notbremse, 30 min)
+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)
 SUBBAUSTEIN_KAPPE = 900     # Subbaustein-Finde-Loop je Chunk (15 min)
 KONSOLIDIERUNG_CHUNK = 150  # Konsolidierung ab so vielen Kandidaten chunken
 
@@ -1209,55 +1211,76 @@ async def _set_inventar(topic: str, eintrag: str, status: str) -> None:
     await db.set_baustein_status(topic, norm, status)
 
 
-async def _recherche_loop(ctx: GenContext, set_p, files: dict, q: dict, ordner, instructions: str) -> bool:
+def _sichtung_schema(data) -> list[str] | None:
-    """Füllt DB-Tabelle `bausteine` mit Kandidaten (+ Nennungszähler). Loop: je Runde
+    """{"noise": ["datei.txt", …]} → Liste Dateinamen (leer erlaubt) · sonst None."""
-    RECHERCHE_AGENTEN Agenten; re-promptet mit noch offenen Crawl-Seiten, bis alle gelesen /
+    if not isinstance(data, dict) or not isinstance(data.get("noise"), list):
-    0 neue / Zeit-Kappe. Resume: fertiger Schritt wird übersprungen. → True (ok) / False (Abbruch)."""
+        return None
+    return [s for x in data["noise"] if (s := str(x).strip())]
+
+
+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;
+    alles andere bleibt Content (konservativ). Fehler/Ausfall → Seiten bleiben Content (fail-open).
+    → Menge der Content-Dateinamen (nie leer, solange pages nicht leer)."""
+    topic, is_cancelled = ctx.topic, ctx.is_cancelled
+    pages = list(pages)
+    if not pages:
+        return set()
+    arbeit = files["arbeit"]
+    ordner = Path(ordner)
+
+    def _snippet(fn: str) -> str:
+        zeilen = _read(ordner / fn).splitlines()
+        url = zeilen[0][len("QUELLE:"):].strip() if zeilen and zeilen[0].startswith("QUELLE:") else ""
+        body = " ".join(z.strip() for z in zeilen[1:] if z.strip())[:200]
+        return f"- {fn} · {url} · {body}"
+
+    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(
+            ctx, f"Seiten-Sichtung {ci}",
+            key=f"bausteine-{topic}-sichtung-c{ci}",
+            prompt=_prompt("Bausteine-Seiten-Sichtung", topic=topic, seiten=seiten, out_path=fp),
+            role="judge", capabilities="files",
+            payload=lambda result, p=fp: _sichtung_schema(_json_datei(p)),
+            timeout=_timeout("recherche_mapping", len(chunk)),
+        )
+        if res:  # nur Dateinamen, die wirklich in diesem Chunk sind → Noise
+            cs = set(chunk)
+            return {n for n in res if n in cs}
+        return set()  # Fail/kein Noise → Content behalten (fail-open)
+
+    async def melde(d, t):
+        set_p(f"Sichte Seiten {d}/{t}…")
+
+    set_p("Sichte Seiten…", step=_step_idx(topic, "Recherche"))
+    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)}")
+    return content or set(pages)  # alles als Noise markiert? → fail-open, alle behalten
+
+
+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"]
-    caps = "files" if ordner else "full"
-    crawl_idx = _crawl_index(ordner)
-    alle_seiten = set(crawl_idx.values())
-
     await db.delete_bausteine(topic)
     await db.delete_coverage(topic)
     await db.set_step_status(topic, "Recherche", "laufend")
-    set_p("Recherche läuft (Runde 1)…", step=_step_idx(topic, "Recherche"))
 
-    start = time.monotonic()
+    async def _ingest(text: str) -> None:
-    runde = 0
-    while not is_cancelled():
-        runde += 1
-        vorhandene = {b["titel_norm"] for b in await db.list_bausteine(topic)}
-        gelesen = set(await db.list_coverage(topic))
-        offen = sorted(alle_seiten - gelesen)
-        if runde > 1 and alle_seiten and not offen:
-            break  # volle Abdeckung erreicht
-        fokus = _fokus_text([b["titel"] for b in await db.list_bausteine(topic)], offen)
-        paths = [arbeit / f"recherche-r{runde}-{i}.md" for i in range(1, RECHERCHE_AGENTEN + 1)]
-        for p in paths:
-            p.unlink(missing_ok=True)
-        slots = [{
-            "key": f"bausteine-{topic}-recherche-r{runde}-{i}",
-            "prompt": _build_recherche_prompt(topic, p, instructions, q["type"], ordner, fokus=fokus),
-            "role": "quick", "capabilities": caps,
-            "payload": (lambda result, p=p: _file_payload(p)),
-        } for i, p in enumerate(paths, 1)]
-        set_p(f"Recherche läuft (Runde {runde})…")
-        texte = await _race(topic, f"Recherche R{runde}", slots, RECHERCHE_QUORUM,
-                            _timeout("recherche"), provider, cancelled=is_cancelled, grace=RECHERCHE_GRACE)
-        if is_cancelled():
-            return False
-        if not texte:
-            if runde == 1:
-                _bausteine_errors[topic] = "Recherche fehlgeschlagen (Minimum nicht erreicht)"
-                return False
-            break  # keine neuen Ergebnisse mehr
-        neu, zitiert = 0, set()
-        for text in texte:
-            zitiert |= _cited_sources(text)
         gesehen = set()
         for eintrag in _parse_auswahl(text).values():
             titel = _titel(eintrag)
@@ -1268,19 +1291,75 @@ async def _recherche_loop(ctx: GenContext, set_p, files: dict, q: dict, ordner,
             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 []
-                if norm not in vorhandene:
-                    neu += 1
-                    vorhandene.add(norm)
             await db.upsert_baustein(topic, norm, titel, besch, quelle)
-        await db.mark_quellen_gelesen(topic, sorted(_abgedeckt(zitiert, crawl_idx)))
+
+    pages = sorted(set(_crawl_index(ordner).values()))
+
+    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
+
+    # Content/Noise-Triage: Noise (Events/Clubs/Profile/…) raus, nur Content batchen.
+    content = await _sichte_inhalt(ctx, set_p, files, ordner, pages)
+    if is_cancelled():
+        return False
+    pages = sorted(content)
+
+    # Crawl: feste Batches, je Batch RECHERCHE_READERS Reader, die GENAU diese Dateien lesen.
+    batches = _chunk_nums(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 R{runde}: +{neu} neu (gesamt {gesamt}), Abdeckung {deckung}/{len(alle_seiten) or '?'}")
+    _log(topic, f"Recherche: {gesamt} Kandidaten, Abdeckung {deckung}/{len(pages)} Seiten ({len(batches)} Batches)")
-        if neu == 0:
+    if not gesamt:
-            break
+        _bausteine_errors[topic] = "Recherche fehlgeschlagen (keine Bausteine)"
-        if time.monotonic() - start > RECHERCHE_KAPPE:
+        return False
-            _log(topic, f"Recherche: Zeit-Kappe {RECHERCHE_KAPPE}s erreicht (Runde {runde})")
-            break
     await db.set_step_status(topic, "Recherche", "fertig")
     return True
 
@@ -1307,7 +1386,7 @@ async def _konsolidiere(ctx: GenContext, set_p, files: dict) -> bool:
         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_AGENTEN, eintraege=eintraege, out_path=fp),
+            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)),
@@ -1476,7 +1555,7 @@ async def generate_bausteine(topic: str, instructions: str = "", provider: str =
                 await db.delete_coverage(topic)
 
             # Inventar (DB): Recherche-Loop → Konsolidierung → Klärung.
-            if not await _recherche_loop(ctx, set_p, files, q, ordner, instructions):
+            if not await _recherche_batch(ctx, set_p, files, q, ordner, instructions):
                 if is_cancelled():
                     abgebrochen()
                 return

templates/Prompt/Bausteine-Seiten-Sichtung.md

@@ -0,0 +1,14 @@
+Du sortierst gecrawlte Seiten zum Lern-Thema "{topic}". Markiere NUR die Seiten, die eindeutig KEINEN fachlichen Lerninhalt tragen.
+
+SEITEN (Dateiname · URL · Auszug):
+{seiten}
+
+Regeln:
+- `noise` = eindeutig kein Lerninhalt. Typisch: Veranstaltungs-/Termin-Seiten, Mitglieder-/Profil-Seiten, Vereins-/Community-Seiten, Podcast-/Video-Verzeichnisse, reine Marketing-/Preis-/Werbeseiten, Login/Impressum, leere Navigations- oder Übersichtsseiten ohne Substanz.
+- Im ZWEIFEL NICHT markieren. Alles, was fachlichen Stoff enthalten könnte (Konzepte, Anleitungen, Beispiele, Definitionen), bleibt Content.
+- Nutze NUR die exakten Dateinamen aus der Liste oben.
+
+Schreibe NUR die JSON-Datei nach {out_path} (kein weiterer Text):
+{{"noise": ["exakter-dateiname.txt", "…"]}}
+
+Ist nichts eindeutig themenfremd, schreibe {{"noise": []}}.