diff --git a/src/agents/eu_researcher.py b/src/agents/eu_researcher.py index be68a67..b471624 100644 --- a/src/agents/eu_researcher.py +++ b/src/agents/eu_researcher.py @@ -20,6 +20,7 @@ oder URL ableitbar ist, staan selbst liefert keine Datumsangaben. import asyncio import json import logging +from dataclasses import dataclass from datetime import datetime from config import ( @@ -305,6 +306,17 @@ async def run_eu_research( if not action["queries"]: break + # Ist die Treffer-Obergrenze erreicht, bringen weitere Runden nichts + # mehr. Ohne diesen Abbruch liefe die Schleife bis zur letzten Runde + # weiter und befragte jedes Mal das teuerste Modell, ohne dass danach + # noch eine einzige Suche ausgefuehrt wuerde. + if len(collected) >= MAX_TOTAL_RESULTS: + logger.info( + "EU-Recherche: Suchphase nach Runde %d beendet, Treffer-Obergrenze %d erreicht", + round_no, MAX_TOTAL_RESULTS, + ) + break + total.cost_usd += searches_done * STAAN_COST_PER_QUERY_USD if not collected: @@ -385,6 +397,9 @@ _SEARCH_NOTE_WITH_RESULTS = ( " Nutze stattdessen die unten angehängten ERGEBNISSE DER EUROPÄISCHEN WEBSUCHE " "als unabhängige Zweitquellen und zitiere sie als Evidenz, wenn sie eine " "Behauptung stützen oder widerlegen. Zusätzlich zählen die übergebenen Meldungen." + " WICHTIG: Nenne bei jedem Beleg, auf den du dich stützt, immer die vollständige " + "Adresse (die URL hinter der [S..]-Nummer), nicht nur die Nummer. Ohne Adresse " + "gilt ein Beleg als nicht nachprüfbar." ) _SEARCH_NOTE_NO_RESULTS = ( " Stütze dich ausschließlich auf die übergebenen Meldungen und Quellen." @@ -432,6 +447,21 @@ def _add_usage(total: ClaudeUsage, u: ClaudeUsage | None) -> None: total.duration_ms += u.duration_ms +@dataclass +class Belegsuche: + """Ergebnis einer gezielten Belegsuche. + + block = fertiger Kontextblock fuer den Auftrag, leer wenn nichts gefunden + usage = Verbrauch aus Planung und Suche + queries = tatsaechlich ausgefuehrte Suchanfragen, fuer die Nachrunde + quellen = gefundene Quellen einzeln, fuer die Quellenzuordnung im Faktencheck + """ + block: str + usage: ClaudeUsage + queries: list[str] + quellen: list[dict] + + async def plan_verification_searches( *, title: str, @@ -440,22 +470,23 @@ async def plan_verification_searches( max_queries: int = EU_VERIFY_MAX_QUERIES, avoid_queries: list[str] | None = None, label: str = "Stützsuche", -) -> tuple[str, ClaudeUsage, list[str]]: +) -> Belegsuche: """Plant gezielte Belegsuchen zu den uebergebenen Punkten und fuehrt sie aus. Die Planung erfolgt punktbezogen, das planende Modell ordnet jeder Anfrage zu, welche Punkte sie belegen soll. avoid_queries verhindert, dass eine Nachrunde dieselben Anfragen noch einmal stellt. - Gibt (kontextblock, verbrauch, ausgefuehrte_anfragen) zurueck. Der - Kontextblock ist leer, wenn nichts gefunden wurde. + Gibt eine Belegsuche zurueck. Der Kontextblock ist leer, wenn nichts + gefunden wurde. Die gefundenen Quellen werden zusaetzlich einzeln + zurueckgegeben, damit die Quellenzuordnung im Faktencheck sie kennt. """ from config import CLAUDE_MODEL_FAST total = ClaudeUsage() items = [str(s).strip() for s in (search_items or []) if str(s).strip()][:EU_VERIFY_MAX_ITEMS] if not items: - return "", total, [] + return Belegsuche("", total, [], []) default_market = market_for_language(_iso_from_display(output_language)) avoid_block = "" @@ -479,7 +510,7 @@ async def plan_verification_searches( _add_usage(total, plan_usage) except Exception as e: logger.warning("EU-%s: Planung der Anfragen fehlgeschlagen (%s), fahre ohne Suche fort", label, e) - return "", total, [] + return Belegsuche("", total, [], []) from agents.researcher import _extract_json_object obj = _extract_json_object(plan_text or "") @@ -498,10 +529,11 @@ async def plan_verification_searches( if not queries: logger.warning("EU-%s: Planung lieferte keine brauchbaren Anfragen", label) - return "", total, [] + return Belegsuche("", total, [], []) lines: list[str] = [] ausgefuehrt: list[str] = [] + quellen: list[dict] = [] gesehene_urls: set[str] = set() treffer = 0 for q in queries: @@ -524,6 +556,15 @@ async def plan_verification_searches( if url: gesehene_urls.add(url) treffer += 1 + # Die Quelle wird zusaetzlich einzeln gefuehrt, damit die + # Quellenzuordnung im Faktencheck sie wie eine Meldung behandeln + # kann. Ohne das gelten Fakten als unbelegt, die sich allein auf + # diese Belege stuetzen. + quellen.append({ + "headline": r["title"] or "", + "source": r["hostname"] or "", + "source_url": url, + }) lines.append(f"[S{treffer}] {r['title'] or '(ohne Titel)'} | {r['hostname']} | {r['url']}") if r.get("snippet"): lines.append(f" Auszug: {r['snippet'][:300]}") @@ -534,14 +575,14 @@ async def plan_verification_searches( logger.info("EU-%s: %d Suchen, %d Treffer im Kontextblock", label, len(ausgefuehrt), treffer) if not lines: - return "", total, ausgefuehrt + return Belegsuche("", total, ausgefuehrt, []) block = ( "\n\nERGEBNISSE DER EUROPÄISCHEN WEBSUCHE " "(von unserem System ausgeführt, als unabhängige Zweitquellen nutzbar):\n" + "\n".join(lines) ) - return block, total, ausgefuehrt + return Belegsuche(block, total, ausgefuehrt, quellen) def build_eu_prompt(prompt: str, results_block: str = "") -> str: @@ -571,11 +612,12 @@ async def eu_call_with_search( total = ClaudeUsage() results_block = "" if search_items: - results_block, such_usage, _ = await plan_verification_searches( + suche = await plan_verification_searches( title=title, search_items=search_items, output_language=output_language, max_queries=max_queries, ) - _add_usage(total, such_usage) + results_block = suche.block + _add_usage(total, suche.usage) result, usage = await call_claude(build_eu_prompt(prompt, results_block), tools=None) _add_usage(total, usage) diff --git a/src/agents/factchecker.py b/src/agents/factchecker.py index 90aa922..2450599 100644 --- a/src/agents/factchecker.py +++ b/src/agents/factchecker.py @@ -551,15 +551,18 @@ class FactCheckerAgent: anzahl_anfragen = EU_VERIFY_MAX_QUERIES if max_queries is None else max_queries anzahl_nachhak = EU_VERIFY_FOLLOWUP_QUERIES if followup_queries is None else followup_queries - block, such_usage, gestellte_anfragen = await plan_verification_searches( + suche = await plan_verification_searches( title=title, search_items=search_items, output_language=output_language, max_queries=anzahl_anfragen, ) - _add_usage(gesamt, such_usage) + _add_usage(gesamt, suche.usage) - result, usage = await call_claude(build_eu_prompt(prompt, block), tools=None) + result, usage = await call_claude(build_eu_prompt(prompt, suche.block), tools=None) _add_usage(gesamt, usage) - facts = self._parse_response(result, articles=articles) + # Die Suchbelege zaehlen bei der Quellenzuordnung wie Meldungen. Ohne + # sie gelten Fakten als unbelegt, die sich allein auf die europaeische + # Suche stuetzen, und werden faelschlich herabgestuft. + facts = self._parse_response(result, articles=(articles or []) + suche.quellen) offene = [f for f in facts if f.get("status") in OFFENE_STATUS] if anzahl_nachhak <= 0 or len(offene) < EU_VERIFY_FOLLOWUP_MIN_OPEN: @@ -569,7 +572,7 @@ class FactCheckerAgent: facts, nachhak_usage = await self._eu_nachhaken( facts=facts, offene=offene, title=title, output_language=output_language, incident_type=incident_type, - avoid_queries=gestellte_anfragen, followup_queries=anzahl_nachhak, + avoid_queries=suche.queries, followup_queries=anzahl_nachhak, ) _add_usage(gesamt, nachhak_usage) return facts, gesamt @@ -600,12 +603,13 @@ class FactCheckerAgent: if not claims: return facts, gesamt - block, such_usage, _ = await plan_verification_searches( + suche = await plan_verification_searches( title=title, search_items=claims, output_language=output_language, max_queries=(EU_VERIFY_FOLLOWUP_QUERIES if followup_queries is None else followup_queries), avoid_queries=avoid_queries, label="Nachhak-Suche", ) - _add_usage(gesamt, such_usage) + block = suche.block + _add_usage(gesamt, suche.usage) if not block: logger.info("EU-Nachhak-Runde: keine zusaetzlichen Belege gefunden, Fakten bleiben unveraendert") return facts, gesamt diff --git a/src/services/post_refresh_qc.py b/src/services/post_refresh_qc.py index 25d6d47..e1c30f7 100644 --- a/src/services/post_refresh_qc.py +++ b/src/services/post_refresh_qc.py @@ -25,6 +25,34 @@ STATUS_PRIORITY = { "developing": 1, } +# Schutz gegen Fehlurteile beim Zusammenfassen von Duplikaten. Das Clustering +# entscheidet ein schnelles Modell, das gelegentlich inhaltlich verschiedene +# Fakten in eine Gruppe wirft, etwa wenn viele Behauptungen gleich beginnen. +# Zwei Sicherungen greifen. Eine Gruppe, die einen zu grossen Anteil aller +# Fakten umfasst, wird komplett verworfen. Und jeder einzelne Loeschkandidat +# muss dem behaltenen Fakt auch rechnerisch aehnlich genug sein. +DEDUP_MAX_CLUSTER_ANTEIL = float(os.environ.get("QC_DEDUP_MAX_CLUSTER_ANTEIL", "0.4")) +DEDUP_MIN_AEHNLICHKEIT = float(os.environ.get("QC_DEDUP_MIN_AEHNLICHKEIT", "0.55")) + + +def _aehnlichkeit(claim_a: str, claim_b: str) -> float: + """Rechnerische Aehnlichkeit zweier Behauptungen zwischen 0 und 1. + + Gleiche Gewichtung wie im Vorfilter, damit beide Stufen dasselbe Mass + verwenden. + """ + from agents.factchecker import normalize_claim, _keyword_set + + norm_a = normalize_claim(claim_a or "") + norm_b = normalize_claim(claim_b or "") + if not norm_a or not norm_b: + return 0.0 + kw_a = _keyword_set(claim_a or "") + kw_b = _keyword_set(claim_b or "") + kw_union = kw_a | kw_b + jaccard = len(kw_a & kw_b) / len(kw_union) if kw_union else 0.0 + return 0.7 * SequenceMatcher(None, norm_a, norm_b).ratio() + 0.3 * jaccard + # --------------------------------------------------------------------------- # 1. Faktencheck-Duplikate # --------------------------------------------------------------------------- @@ -177,11 +205,23 @@ async def check_fact_duplicates(db, incident_id: int, incident_title: str) -> in facts_by_id = {f["id"]: f for f in all_facts} ids_to_delete = set() + max_cluster = max(2, int(len(all_facts) * DEDUP_MAX_CLUSTER_ANTEIL)) + for cluster_ids in all_clusters: valid_ids = [cid for cid in cluster_ids if cid in facts_by_id] if len(valid_ids) <= 1: continue + # Sicherung 1: Eine Gruppe, die einen zu grossen Teil aller Fakten + # umfasst, ist fast immer ein Fehlurteil des Clusterings. + if len(valid_ids) > max_cluster: + logger.warning( + "QC Duplikat: Gruppe mit %d von %d Fakten verworfen, das ist " + "unplausibel (Grenze %d)", + len(valid_ids), len(all_facts), max_cluster, + ) + continue + cluster_facts = [facts_by_id[cid] for cid in valid_ids] best = max(cluster_facts, key=lambda f: ( STATUS_PRIORITY.get(f["status"], 0), @@ -190,12 +230,23 @@ async def check_fact_duplicates(db, incident_id: int, incident_title: str) -> in )) for fact in cluster_facts: - if fact["id"] != best["id"]: - ids_to_delete.add(fact["id"]) - logger.info( - "QC Duplikat: ID %d entfernt, behalte ID %d ('%s')", - fact["id"], best["id"], best["claim"][:60], + if fact["id"] == best["id"]: + continue + # Sicherung 2: Nur loeschen, wenn die beiden Behauptungen auch + # rechnerisch nah beieinander liegen. + naehe = _aehnlichkeit(fact["claim"], best["claim"]) + if naehe < DEDUP_MIN_AEHNLICHKEIT: + logger.warning( + "QC Duplikat: ID %d behalten, Aehnlichkeit zu ID %d nur %.2f " + "('%s')", + fact["id"], best["id"], naehe, fact["claim"][:60], ) + continue + ids_to_delete.add(fact["id"]) + logger.info( + "QC Duplikat: ID %d entfernt (Aehnlichkeit %.2f), behalte ID %d ('%s')", + fact["id"], naehe, best["id"], best["claim"][:60], + ) if ids_to_delete: placeholders = ",".join("?" * len(ids_to_delete)) diff --git a/tests/test_eu_belegsuche.py b/tests/test_eu_belegsuche.py index 64c573d..1fab65f 100644 --- a/tests/test_eu_belegsuche.py +++ b/tests/test_eu_belegsuche.py @@ -80,10 +80,11 @@ plan_antwort["queries"] = [ treffer_je_anfrage["Ceuta Opferzahl offiziell"] = treffer(2, "a") treffer_je_anfrage["Ceuta border deaths toll"] = treffer(2, "b") treffer_je_anfrage["Sanchez Ceuta declaracion"] = treffer(2, "c") -block, usage, ausgefuehrt = asyncio.run(eur.plan_verification_searches( +_r = asyncio.run(eur.plan_verification_searches( title="Ceuta", search_items=["57 Tote gemeldet", "Sanchez sprach von Angriff"], output_language="Deutsch", max_queries=7, )) +block, usage, ausgefuehrt = _r.block, _r.usage, _r.queries pruefe("A1 alle geplanten Anfragen ausgefuehrt", len(zustand["suchen"]) == 3, zustand["suchen"]) pruefe("A2 ausgefuehrte Anfragen zurueckgemeldet", len(ausgefuehrt) == 3, ausgefuehrt) pruefe("A3 jeder Punkt steht im Planungsauftrag", @@ -106,9 +107,10 @@ plan_antwort["queries"] = [ ] treffer_je_anfrage["anfrage eins"] = treffer(3, "x") treffer_je_anfrage["anfrage zwei"] = treffer(3, "x") # identische URLs -block_b, _, _ = asyncio.run(eur.plan_verification_searches( +_rb = asyncio.run(eur.plan_verification_searches( title="Ceuta", search_items=["Punkt"], max_queries=7, )) +block_b = _rb.block pruefe("B1 Dubletten entfernt", block_b.count("[S") == 3, block_b.count("[S")) pruefe("B2 Nummerierung bleibt luecklos", all(("[S" + str(i) + "]") in block_b for i in (1, 2, 3))) @@ -120,10 +122,11 @@ plan_antwort["queries"] = [ {"q": "neue anfrage", "market": "de-de"}, ] treffer_je_anfrage["neue anfrage"] = treffer(2, "n") -block_c, _, ausgefuehrt_c = asyncio.run(eur.plan_verification_searches( +_rc = asyncio.run(eur.plan_verification_searches( title="Ceuta", search_items=["Punkt"], max_queries=5, avoid_queries=["alte anfrage"], label="Nachhak-Suche", )) +block_c, ausgefuehrt_c = _rc.block, _rc.queries pruefe("C1 bereits gestellte Anfrage wird verworfen", [s["q"] for s in zustand["suchen"]] == ["neue anfrage"], zustand["suchen"]) pruefe("C2 Sperrliste steht im Planungsauftrag", "alte anfrage" in zustand["plan_prompt"]) @@ -134,9 +137,10 @@ neu() plan_antwort["queries"] = [{"q": "q" + str(i), "market": "de-de"} for i in range(1, 11)] for i in range(1, 11): treffer_je_anfrage["q" + str(i)] = treffer(20, "y" + str(i)) -block_d, _, ausgefuehrt_d = asyncio.run(eur.plan_verification_searches( +_rd = asyncio.run(eur.plan_verification_searches( title="Ceuta", search_items=["Punkt"], max_queries=4, )) +block_d, ausgefuehrt_d = _rd.block, _rd.queries pruefe("D1 hoechstens max_queries Anfragen", len(ausgefuehrt_d) == 4, ausgefuehrt_d) pruefe("D2 Treffer je Anfrage begrenzt", block_d.count("[S") == 4 * EU_VERIFY_HITS_PER_QUERY, block_d.count("[S")) @@ -163,10 +167,11 @@ async def staan_mit_fehler(q, **kw): eur.staan_search = staan_mit_fehler -block_e, usage_e, ausgefuehrt_e = asyncio.run(eur.plan_verification_searches( +_re = asyncio.run(eur.plan_verification_searches( title="Ceuta", search_items=["Punkt"], max_queries=5, )) eur.staan_search = fake_staan +block_e, usage_e, ausgefuehrt_e = _re.block, _re.usage, _re.queries pruefe("E1 gescheiterte Suche uebersprungen", ausgefuehrt_e == ["heil"], ausgefuehrt_e) pruefe("E2 gefundene Belege bleiben erhalten", block_e.count("[S") == 2, block_e.count("[S")) pruefe("E3 nur bezahlte Suchen berechnet", @@ -174,15 +179,48 @@ pruefe("E3 nur bezahlte Suchen berechnet", neu() plan_antwort["queries"] = [] -block_f, _, ausgefuehrt_f = asyncio.run(eur.plan_verification_searches( +_rf = asyncio.run(eur.plan_verification_searches( title="Ceuta", search_items=["Punkt"], max_queries=5, )) +block_f, ausgefuehrt_f = _rf.block, _rf.queries pruefe("E4 leere Planung liefert leeren Block", block_f == "" and ausgefuehrt_f == []) -block_g, _, _ = asyncio.run(eur.plan_verification_searches( +_rg = asyncio.run(eur.plan_verification_searches( title="Ceuta", search_items=[], max_queries=5, )) +block_g = _rg.block pruefe("E5 ohne Pruefpunkte keine Suche", block_g == "") +print("\nF) Gefundene Quellen werden einzeln zurueckgegeben") +neu() +plan_antwort["queries"] = [{"q": "eine anfrage", "market": "de-de"}] +treffer_je_anfrage["eine anfrage"] = treffer(3, "q", host="tagesschau.de") +_rh = asyncio.run(eur.plan_verification_searches( + title="Ceuta", search_items=["Punkt"], max_queries=5, +)) +pruefe("F1 jede Quelle einzeln gemeldet", len(_rh.quellen) == 3, len(_rh.quellen)) +pruefe("F2 Quelle traegt Titel, Herkunft und Adresse", + all(q.get("headline") and q.get("source") and q.get("source_url") for q in _rh.quellen), + _rh.quellen[:1]) +pruefe("F3 Adressen stimmen mit dem Block ueberein", + all(q["source_url"] in _rh.block for q in _rh.quellen)) +neu() +plan_antwort["queries"] = [{"q": "a", "market": "de-de"}, {"q": "b", "market": "de-de"}] +treffer_je_anfrage["a"] = treffer(2, "dup") +treffer_je_anfrage["b"] = treffer(2, "dup") +_ri = asyncio.run(eur.plan_verification_searches( + title="Ceuta", search_items=["Punkt"], max_queries=5, +)) +pruefe("F4 doppelte Quellen nur einmal gemeldet", len(_ri.quellen) == 2, len(_ri.quellen)) + +print("\nG) Der Auftrag verlangt vollstaendige Quellenadressen") +from agents.eu_researcher import build_eu_prompt # noqa: E402 + +mit = build_eu_prompt("Auftrag", "\n\nBELEGE\n[S1] Titel | host | https://x.y/z") +ohne = build_eu_prompt("Auftrag", "") +pruefe("G1 Hinweis auf die Adresspflicht vorhanden", + "vollständige" in mit and "URL" in mit) +pruefe("G2 ohne Belege kein Adresshinweis", "Adresse (die URL" not in ohne) + print("\nErgebnis: " + str(ok) + " bestanden, " + str(fail) + " fehlgeschlagen") sys.exit(1 if fail else 0) diff --git a/tests/test_eu_factcheck.py b/tests/test_eu_factcheck.py index cf61e79..c3e9b13 100644 --- a/tests/test_eu_factcheck.py +++ b/tests/test_eu_factcheck.py @@ -35,6 +35,13 @@ def pruefe(name, bedingung, extra=""): aufrufe = {"such": [], "modell": []} +SUCHQUELLE = { + "headline": "Innenministerkonferenz beschliesst Fahrplan", + "source": "example.org", + "source_url": "https://example.org/a", +} + + async def fake_plan(*, title, search_items, output_language="Deutsch", max_queries=7, avoid_queries=None, label="Stuetzsuche"): aufrufe["such"].append({ @@ -45,7 +52,7 @@ async def fake_plan(*, title, search_items, output_language="Deutsch", }) u = ClaudeUsage(input_tokens=10, output_tokens=5, cost_usd=0.01) block = "\n\nBELEGE [" + label + "]\n[S1] Beleg | example.org | https://example.org/a" - return block, u, [label + "-q1", label + "-q2"] + return eur.Belegsuche(block, u, [label + "-q1", label + "-q2"], [dict(SUCHQUELLE)]) antworten = [] @@ -240,5 +247,33 @@ asyncio.run(fc._eu_pruefen( )) pruefe("I1 followup_queries=0 unterbindet die Nachrunde", len(aufrufe["such"]) == 1, len(aufrufe["such"])) +print("\nJ) Suchbelege zaehlen bei der Quellenzuordnung") +_ai_backend_var.set("bedrock") +neu() +# Behauptung, die zu KEINEM gespeicherten Artikel passt, wohl aber zur +# Suchquelle. Vor der Korrektur wurde so ein Fakt herabgestuft. +NUR_SUCHE = "Die Innenministerkonferenz beschliesst einen Fahrplan" +antworten.append(json_fakten([ + (NUR_SUCHE, "established", "belegt durch die Suche"), + (F_GRENZE, "confirmed", "belegt"), +])) +facts_j, _ = asyncio.run(fc.check("Ceuta", ARTS, "research", "Deutsch")) +treffer_j = [f for f in facts_j if f["claim"].startswith("Die Innenministerkonferenz")] +pruefe("J1 Fakt allein aus der Suche bleibt belegt", + treffer_j and treffer_j[0]["status"] == "established", + [(f["claim"][:35], f["status"]) for f in facts_j]) +pruefe("J2 Adresse der Suchquelle steht in der Evidenz", + treffer_j and "https://example.org/a" in treffer_j[0]["evidence"], + treffer_j[0]["evidence"][:160] if treffer_j else "") + +print("\nK) Ohne Suchbelege bleibt die Zuordnung wie bisher") +_ai_backend_var.set("cli") +neu() +antworten.append(json_fakten([(NUR_SUCHE, "established", "belegt")])) +facts_k, _ = asyncio.run(fc.check("Ceuta", ARTS, "research", "Deutsch")) +pruefe("K1 CLI stuft ohne zuordenbare Quelle weiterhin herab", + facts_k and facts_k[0]["status"] != "established", + [(f["claim"][:35], f["status"]) for f in facts_k]) + print("\nErgebnis: " + str(ok) + " bestanden, " + str(fail) + " fehlgeschlagen") sys.exit(1 if fail else 0) diff --git a/tests/test_qc_und_runden.py b/tests/test_qc_und_runden.py new file mode 100644 index 0000000..a972875 --- /dev/null +++ b/tests/test_qc_und_runden.py @@ -0,0 +1,213 @@ +"""Testet die Absicherung der Duplikatpruefung und den Abbruch der Suchrunden. + +Laeuft ohne Netzzugriff und ohne Kosten, Datenbank, Modell und Suche sind +durch Testdoubles ersetzt. + +Aufruf aus dem Projektstamm: + venv/bin/python tests/test_qc_und_runden.py +""" +import asyncio +import json +import os +import sys + +sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "src")) + +import services.post_refresh_qc as qc # noqa: E402 +import agents.eu_researcher as eur # noqa: E402 +from agents.claude_client import ClaudeUsage # noqa: E402 + +ok = 0 +fail = 0 + + +def pruefe(name, bedingung, extra=""): + global ok, fail + if bedingung: + ok += 1 + print(" OK " + name) + else: + fail += 1 + print(" FEHL " + name + " " + str(extra)) + + +# --------------------------------------------------------------------------- +# Testdouble fuer die Datenbank +# --------------------------------------------------------------------------- +class FakeCursor: + def __init__(self, rows): + self._rows = rows + + async def fetchall(self): + return self._rows + + async def fetchone(self): + return self._rows[0] if self._rows else None + + +class FakeDB: + """Liefert eine feste Faktenliste und merkt sich die Loeschauftraege.""" + + def __init__(self, fakten): + self.fakten = fakten + self.geloescht = [] + + async def execute(self, sql, params=()): + if sql.strip().upper().startswith("DELETE"): + self.geloescht.extend(params) + return FakeCursor([]) + return FakeCursor(list(self.fakten)) + + +def fakt(fid, claim, status="established", quellen=3): + return {"id": fid, "claim": claim, "status": status, "sources_count": quellen, + "evidence": "e", "checked_at": "2026-08-01 10:00:00"} + + +FAHRPLAN = "Die IMK beschloss einen gemeinsamen Bund-Laender-Fahrplan zum Aufbau der zivilen Verteidigungsfaehigkeit bis 2029" +FAHRPLAN_ANDERS = "Die IMK beschloss einen Bund-Laender-Fahrplan fuer die zivile Verteidigung bis zum Jahr 2029" +BUNDESWEHR = "Die IMK beschloss die institutionelle Einbindung der Bundeswehr in die Innenministerkonferenz" +SOZIAL = "Die IMK beschloss Massnahmen zur haerteren Bekaempfung von organisiertem Sozialmissbrauch" +MILLIARDEN = "Die IMK forderte zusaetzliche Milliardeninvestitionen des Bundes in den Zivilschutz" +ABSCHIEBUNG = "Die Innenminister waren sich einig, dass der Bund bei Abschiebungen mehr leisten muss" +GDP = "Die Gewerkschaft der Polizei bewertete die Beschluesse als grundsaetzlich richtig" +SYRER = "In der Frage der Rueckfuehrung von Syrern besteht ein Konflikt zwischen SPD und CDU" +IMK_DATUM = "Die 225. Innenministerkonferenz fand im Juni 2026 in Hamburg statt" +BESCHLUESSE = "Die IMK fasste insgesamt 66 Beschluesse" + +geplante_cluster = [] + + +async def fake_cluster(facts, incident_title): + return geplante_cluster + + +qc._haiku_find_duplicate_clusters = fake_cluster + + +def lauf_qc(fakten, cluster): + global geplante_cluster + geplante_cluster = cluster + db = FakeDB(fakten) + entfernt = asyncio.run(qc.check_fact_duplicates(db, 49, "Innenministerkonferenz 2026")) + return entfernt, db.geloescht + + +print("\nA) Der Fall aus Lage 49, sieben verschiedene Fakten in einer Gruppe") +zehn = [ + fakt(469, FAHRPLAN), fakt(468, IMK_DATUM), fakt(470, BUNDESWEHR), + fakt(471, SOZIAL), fakt(472, MILLIARDEN), fakt(473, ABSCHIEBUNG), + fakt(474, SYRER, "disputed"), fakt(475, BESCHLUESSE), fakt(476, GDP), + fakt(477, "Die IMK befasste sich mit der Umsetzung des europaeischen Asylsystems"), +] +entfernt, geloescht = lauf_qc(zehn, [[468, 469, 470, 471, 472, 473, 475, 476]]) +pruefe("A1 unplausible Gruppe wird komplett verworfen", entfernt == 0, entfernt) +pruefe("A2 kein Fakt geloescht", geloescht == [], geloescht) + +print("\nB) Echte Dubletten werden weiterhin entfernt") +entfernt, geloescht = lauf_qc( + [fakt(1, FAHRPLAN, quellen=8), fakt(2, FAHRPLAN_ANDERS, quellen=2), + fakt(3, BUNDESWEHR), fakt(4, SOZIAL), fakt(5, MILLIARDEN)], + [[1, 2]]) +pruefe("B1 echtes Duplikat entfernt", entfernt == 1, entfernt) +pruefe("B2 der besser belegte Fakt bleibt", geloescht == [2], geloescht) + +print("\nC) Kleine Gruppe mit unaehnlichen Fakten wird nicht geloescht") +entfernt, geloescht = lauf_qc( + [fakt(1, FAHRPLAN), fakt(2, SOZIAL), fakt(3, BUNDESWEHR), fakt(4, MILLIARDEN), + fakt(5, ABSCHIEBUNG), fakt(6, GDP)], + [[1, 2]]) +pruefe("C1 unaehnlicher Fakt bleibt erhalten", entfernt == 0, entfernt) + +print("\nD) Gemischte Gruppe, nur der wirklich aehnliche Fakt faellt weg") +entfernt, geloescht = lauf_qc( + [fakt(1, FAHRPLAN, quellen=9), fakt(2, FAHRPLAN_ANDERS, quellen=1), fakt(3, SOZIAL), + fakt(4, BUNDESWEHR), fakt(5, MILLIARDEN), fakt(6, ABSCHIEBUNG), + fakt(7, GDP), fakt(8, SYRER), fakt(9, BESCHLUESSE), fakt(10, IMK_DATUM)], + [[1, 2, 3]]) +pruefe("D1 genau ein Fakt entfernt", entfernt == 1, entfernt) +pruefe("D2 der aehnliche wurde entfernt", geloescht == [2], geloescht) + +print("\nE) Grenzwerte") +pruefe("E1 Anteilsgrenze gesetzt", 0 < qc.DEDUP_MAX_CLUSTER_ANTEIL <= 1, qc.DEDUP_MAX_CLUSTER_ANTEIL) +pruefe("E2 Aehnlichkeitsgrenze gesetzt", 0 < qc.DEDUP_MIN_AEHNLICHKEIT < 1, qc.DEDUP_MIN_AEHNLICHKEIT) +pruefe("E3 verschiedene Sachverhalte liegen unter der Grenze", + qc._aehnlichkeit(FAHRPLAN, SOZIAL) < qc.DEDUP_MIN_AEHNLICHKEIT, + round(qc._aehnlichkeit(FAHRPLAN, SOZIAL), 2)) +pruefe("E4 echte Dublette liegt ueber der Grenze", + qc._aehnlichkeit(FAHRPLAN, FAHRPLAN_ANDERS) >= qc.DEDUP_MIN_AEHNLICHKEIT, + round(qc._aehnlichkeit(FAHRPLAN, FAHRPLAN_ANDERS), 2)) +pruefe("E5 bei zwei Fakten bleibt die Gruppengrenze nutzbar", + max(2, int(2 * qc.DEDUP_MAX_CLUSTER_ANTEIL)) == 2) + +# --------------------------------------------------------------------------- +# Suchrunden +# --------------------------------------------------------------------------- +print("\nF) Die Suchphase endet, sobald die Treffergrenze erreicht ist") + +runden = {"n": 0} + + +async def fake_bedrock_research(prompt, model=None, raw_text=False, timeout=None): + runden["n"] += 1 + # Jede Runde fordert vier Suchen an, das Modell will nie von selbst enden. + antwort = {"action": "search", "reason": "weiter", + "queries": [{"q": f"anfrage {runden['n']}-{i}", "market": "de-de", + "full_content": False} for i in range(1, 5)]} + return json.dumps(antwort), ClaudeUsage(input_tokens=100, output_tokens=50, cost_usd=0.05) + + +async def fake_staan_research(q, market="de-de", extra_snippets=True, max_snippets=4, + full_content=False, extra_exclude=None, timeout=None): + # Jede Suche liefert 20 neue Treffer, nach zwei Runden ist die Grenze voll. + basis = q.replace(" ", "") + return [{ + "title": "Treffer " + basis + str(i), + "hostname": "example.org", + "url": "https://example.org/" + basis + "/" + str(i), + "snippet": "Auszug", + "extra_snippets": [], + "full_text": "", + } for i in range(1, 21)] + + +letzter_prompt = {"text": ""} + + +async def fake_bedrock_final(prompt, model=None, raw_text=False, timeout=None): + """Ab dem Abschlussauftrag wird eine Auswahl zurueckgegeben.""" + letzter_prompt["text"] = prompt + return "[]", ClaudeUsage(input_tokens=100, output_tokens=50, cost_usd=0.05) + + +async def bedrock_weiche(prompt, model=None, raw_text=False, timeout=None): + # Der Rundenauftrag traegt die Rundennummer im Kopf, der Abschlussauftrag + # nicht. Daran laesst sich beides sicher unterscheiden. + if "Es ist Runde " in prompt: + return await fake_bedrock_research(prompt, model, raw_text, timeout) + return await fake_bedrock_final(prompt, model, raw_text, timeout) + + +eur.call_bedrock = bedrock_weiche +eur.staan_search = fake_staan_research + +runden["n"] = 0 +text, usage = asyncio.run(eur.run_eu_research( + title="Innenministerkonferenz 2026", description="", incident_type="research", + lang_instruction="", existing_context="", preferred_sources_block="", + output_language="Deutsch", excluded_sources=[], research_language_iso="de", +)) +# Die erste Runde fuellt die Grenze von 60 Treffern, danach ist Schluss. +# Ohne die Korrektur liefe die Schleife alle fuenf Runden durch und befragte +# jedes Mal das teuerste Modell, ohne noch zu suchen. +suchrunden = runden["n"] +pruefe("F1 genau eine Suchrunde statt fuenf", suchrunden == 1, suchrunden) +pruefe("F2 vier Leerrunden gespart", eur.EU_RESEARCH_MAX_ROUNDS_RESEARCH - suchrunden == 4, + eur.EU_RESEARCH_MAX_ROUNDS_RESEARCH - suchrunden) +pruefe("F3 Recherche liefert trotzdem ein Ergebnis", text == "[]", text[:40]) +pruefe("F4 Kosten nur fuer die tatsaechlich gelaufene Runde plus Abschluss", + round(usage.cost_usd, 2) == round(0.05 * 2 + 4 * eur.STAAN_COST_PER_QUERY_USD, 2), + round(usage.cost_usd, 4)) + +print("\nErgebnis: " + str(ok) + " bestanden, " + str(fail) + " fehlgeschlagen") +sys.exit(1 if fail else 0)