diff --git a/src/agents/eu_researcher.py b/src/agents/eu_researcher.py index 01726ce..0bf3476 100644 --- a/src/agents/eu_researcher.py +++ b/src/agents/eu_researcher.py @@ -28,6 +28,9 @@ from config import ( EU_RESEARCH_MAX_NEW_PER_ROUND, EU_RESEARCH_MAX_ROUNDS_ADHOC, EU_RESEARCH_MAX_ROUNDS_RESEARCH, + EU_VERIFY_COUNTER_SHARE, + EU_VERIFY_FULLTEXT_CHARS, + EU_VERIFY_FULLTEXT_QUERIES, EU_VERIFY_HITS_PER_QUERY, EU_VERIFY_MAX_ITEMS, EU_VERIFY_MAX_QUERIES, @@ -448,6 +451,13 @@ _SEARCH_NOTE_WITH_RESULTS = ( " 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." + " EBENSO WICHTIG, so prüfst du einen Beleg: Ein Treffer, der nur das Thema " + "erwähnt, ist KEIN Beleg. Die konkrete Aussage mit ihrer Zahl, ihrem Namen oder " + "ihrem Datum muss im Titel, im Auszug oder im Artikeltext tatsächlich vorkommen. " + "Steht dort nur, dass es die Lage gibt, oder ist die Angabe vager als deine " + "Behauptung, dann trägt dieser Treffer sie nicht. Zähle ihn dann nicht mit und " + "sage das in der Begründung ehrlich. Lieber ein unbestätigter Punkt mit klarer " + "Begründung als eine Bestätigung, die einer Nachprüfung nicht standhält." ) _SEARCH_NOTE_NO_RESULTS = ( " Stütze dich ausschließlich auf die übergebenen Meldungen und Quellen." @@ -464,16 +474,30 @@ Erzeuge bis zu {max_queries} Websuche-Anfragen. Plane sie GEZIELT je Punkt, nich pauschal für alle zusammen. Eine gute Anfrage macht genau einen dieser Punkte überprüfbar, mit seinen konkreten Zahlen, Namen und Orten. +Davon sind {counter_queries} Anfragen ausdrücklich GEGENPROBEN. Eine Gegenprobe sucht +nicht nach Bestätigung, sondern nach allem, was der Behauptung widerspricht. Also nach +Dementis, Richtigstellungen, abweichenden Zahlen anderer Stellen, nach der Darstellung +der Gegenseite und nach Berichten, die den Vorgang anders einordnen. Wer nur nach +Bestätigung sucht, findet auch nur Bestätigung, und der Faktencheck wird wertlos. +Beispiele für Gegenproben: "Behörde dementiert Angaben zu X", "X Zahlen umstritten +Kritik", "X widerspricht Darstellung". Markiere sie mit "gegenprobe": true. + REGELN: -- Ordne jeder Anfrage über "for_items" zu, welche Punkte sie belegen soll. +- Ordne jeder Anfrage über "for_items" zu, welche Punkte sie prüfen soll. - Decke möglichst viele verschiedene Punkte ab, statt denselben mehrfach zu suchen. - Priorisiere Punkte mit harten, überprüfbaren Angaben (Zahlen, Daten, Zitate, Beschlüsse). +- Wähle für die Gegenproben die Punkte aus, bei denen ein Irrtum am schwersten wiegt, + also Opferzahlen, Zuschreibungen von Verantwortung, Beschlüsse und Zitate. +- Setze "full_content": true bei den Anfragen, deren Punkt sich nur im Artikeltext + wirklich prüfen lässt, höchstens bei {fulltext_queries} Anfragen. Bei diesen bekommst + du den Artikeltext statt nur der Überschrift. - Nutze die Sprache, in der die Berichterstattung zu erwarten ist. Für Ereignisse in einem Land sind Quellen in dessen Landessprache oft ergiebiger. - Keine Jahreszahlen anhängen, keine Wiederholung derselben Anfrage. {avoid_block} Antworte NUR mit JSON: -{{"queries": [{{"q": "suchbegriffe", "market": "{default_market}", "for_items": [1, 2]}}]}} +{{"queries": [{{"q": "suchbegriffe", "market": "{default_market}", "for_items": [1, 2], +"gegenprobe": false, "full_content": false}}]}} "market" ist "de-de", "en-us" oder "fr-fr". Fremdsprachige Anfragen funktionieren mit "en-us".""" @@ -518,6 +542,7 @@ async def plan_verification_searches( max_queries: int = EU_VERIFY_MAX_QUERIES, avoid_queries: list[str] | None = None, label: str = "Stützsuche", + nur_gegenproben: bool = False, ) -> Belegsuche: """Plant gezielte Belegsuchen zu den uebergebenen Punkten und fuehrt sie aus. @@ -544,11 +569,22 @@ async def plan_verification_searches( + "\n".join(f" {q[:120]}" for q in avoid_queries[:10]) + "\n" ) + # Ein fester Anteil der Anfragen sucht nach Widerspruch statt nach + # Bestaetigung. Ohne das sieht der Faktencheck nur stuetzendes Material und + # bestaetigt praktisch alles. + if nur_gegenproben: + anzahl_gegenproben = max_queries + else: + anzahl_gegenproben = max(1, round(max_queries * EU_VERIFY_COUNTER_SHARE)) if max_queries >= 3 else 0 + anzahl_volltexte = min(EU_VERIFY_FULLTEXT_QUERIES, max_queries) + plan_prompt = _VERIFY_QUERY_PROMPT.format( today=datetime.now(TIMEZONE).strftime("%d.%m.%Y"), title=title or "(ohne Titel)", items_block="\n".join(f"{i}. {s[:220]}" for i, s in enumerate(items, 1)), max_queries=max_queries, + counter_queries=anzahl_gegenproben, + fulltext_queries=anzahl_volltexte, default_market=default_market, avoid_block=avoid_block, ) @@ -571,7 +607,12 @@ async def plan_verification_searches( if not text or text.lower() in gesehen: continue gesehen.add(text.lower()) - queries.append({"q": text, "market": str(q.get("market", "")).strip().lower()}) + queries.append({ + "q": text, + "market": str(q.get("market", "")).strip().lower(), + "gegenprobe": nur_gegenproben or bool(q.get("gegenprobe")), + "full_content": bool(q.get("full_content")), + }) if len(queries) >= max_queries: break @@ -584,14 +625,21 @@ async def plan_verification_searches( quellen: list[dict] = [] gesehene_urls: set[str] = set() treffer = 0 + volltexte_genutzt = 0 + gegenproben_gelaufen = 0 for q in queries: + will_volltext = q["full_content"] and volltexte_genutzt < anzahl_volltexte try: res = await staan_search( q["q"], market=q["market"] if q["market"] in ("de-de", "en-us", "fr-fr") else default_market, - extra_snippets=True, max_snippets=3, full_content=False, + extra_snippets=True, max_snippets=3, full_content=will_volltext, ) ausgefuehrt.append(q["q"]) + if will_volltext: + volltexte_genutzt += 1 + if q["gegenprobe"]: + gegenproben_gelaufen += 1 except StaanError as e: logger.warning("EU-%s: Suche fehlgeschlagen (%s)", label, e) continue @@ -613,23 +661,40 @@ async def plan_verification_searches( "source": r["hostname"] or "", "source_url": url, }) - lines.append(f"[S{treffer}] {r['title'] or '(ohne Titel)'} | {r['hostname']} | {r['url']}") + marke = " [GEGENPROBE]" if q["gegenprobe"] else "" + lines.append( + f"[S{treffer}]{marke} {r['title'] or '(ohne Titel)'} | {r['hostname']} | {r['url']}") if r.get("snippet"): lines.append(f" Auszug: {r['snippet'][:300]}") for chunk in (r.get("extra_snippets") or [])[:1]: lines.append(f" Auszug: {chunk[:300]}") + # Volltext, damit das Modell pruefen kann, ob die Quelle die + # Behauptung wirklich traegt, statt aus der Ueberschrift zu schliessen. + volltext = (r.get("full_text") or "").strip() + if volltext: + lines.append(f" Artikeltext: {volltext[:EU_VERIFY_FULLTEXT_CHARS]}") total.cost_usd += len(ausgefuehrt) * STAAN_COST_PER_QUERY_USD - logger.info("EU-%s: %d Suchen, %d Treffer im Kontextblock", label, len(ausgefuehrt), treffer) + logger.info( + "EU-%s: %d Suchen (davon %d Gegenproben, %d mit Artikeltext), %d Treffer im Kontextblock", + label, len(ausgefuehrt), gegenproben_gelaufen, volltexte_genutzt, treffer) if not lines: return Belegsuche("", total, ausgefuehrt, []) - block = ( + kopf = ( "\n\nERGEBNISSE DER EUROPÄISCHEN WEBSUCHE " - "(von unserem System ausgeführt, als unabhängige Zweitquellen nutzbar):\n" - + "\n".join(lines) + "(von unserem System ausgeführt, als unabhängige Zweitquellen nutzbar).\n" ) + if gegenproben_gelaufen: + kopf += ( + "Ein Teil dieser Suchen war eine GEGENPROBE, sie suchte gezielt nach " + "Widerspruch, Dementi oder abweichenden Angaben. Diese Treffer sind mit " + "[GEGENPROBE] gekennzeichnet. Findet sich dort nichts Widersprechendes, " + "stützt das die Behauptung zusätzlich. Findet sich etwas, muss es in die " + "Bewertung einfließen.\n" + ) + block = kopf + "\n".join(lines) return Belegsuche(block, total, ausgefuehrt, quellen) diff --git a/src/agents/factchecker.py b/src/agents/factchecker.py index f457d22..4d565cd 100644 --- a/src/agents/factchecker.py +++ b/src/agents/factchecker.py @@ -414,6 +414,53 @@ TWOPHASE_MIN_FACTS = 25 # Ab dieser Anzahl bestehender Fakten wird der # Nachhak-Runde ausloesen koennen. OFFENE_STATUS = ("unconfirmed", "unverified", "developing") +# Status, die eine Bestaetigung behaupten. Genau diese werden in der +# Belastungsprobe noch einmal angegriffen. +BESTAETIGTE_STATUS = ("confirmed", "established") + +# Status, die einen Widerspruch oder eine Widerlegung ausdruecken. Sie duerfen +# eine Bestaetigung auch zuruecknehmen, sonst gaebe es nur den Weg nach oben. +WIDERSPRUCH_STATUS = ("contradicted", "disputed", "false") + +EU_CHALLENGE_PROMPT_TEMPLATE = """Du bist ein OSINT-Faktenchecker und fuehrst eine BELASTUNGSPROBE durch. +AUSGABESPRACHE: {output_language} +- KEINE Gedankenstriche verwenden, stattdessen Kommas oder neue Saetze. +- Verwende IMMER echte UTF-8-Umlaute (ä, ö, ü, ß), NIEMALS Umschreibungen. + +LAGE: {title} + +Im ersten Durchgang wurden die folgenden Behauptungen als belegt eingestuft. Das ist +verdaechtig, denn bei einer laufenden Lage ist selten alles gesichert. Fuer genau diese +Punkte wurden jetzt GEGENPROBEN gesucht, also gezielt nach Dementis, Richtigstellungen, +abweichenden Angaben und der Darstellung der Gegenseite. Die Ergebnisse stehen unten. + +ALS BELEGT EINGESTUFTE BEHAUPTUNGEN: +{claims_text} + +AUFGABE: +Pruefe jede dieser Behauptungen kritisch gegen die unten angehaengten Suchergebnisse. +Deine Aufgabe ist NICHT, die Bestaetigung zu wiederholen, sondern zu pruefen, ob sie +haelt. + +Stufe eine Behauptung zurueck, wenn einer dieser Punkte zutrifft: +- Eine Quelle widerspricht ihr in der Sache, dann "{widerspruch}". +- Eine Stelle hat sie ausdruecklich dementiert oder richtiggestellt, dann "false". +- Die Belege nennen die konkrete Angabe gar nicht, sondern nur das Thema, dann "{offen}". +- Die Angabe ist praeziser formuliert als die Quellen sie hergeben, etwa eine genaue + Zahl, wo die Quellen von einer Schaetzung oder einer Spanne sprechen, dann "{offen}". +- Nur eine einzige Stelle berichtet es und keine zweite bestaetigt es, dann "{offen}". + +Lass sie unveraendert, wenn die Gegenprobe nichts Belastendes gebracht hat. Das ist ein +gutes Ergebnis und keine Niederlage. + +Erfinde keinen Zweifel. Ein Zurueckstufen braucht einen konkreten Beleg oder eine +konkrete Luecke, die du benennen kannst. + +Antworte NUR mit einem JSON-Array, ein Objekt je oben genannter Behauptung: +[{{"claim": "Behauptung im Wortlaut wie oben", "status": "{status_werte}", +"evidence": "Begruendung mit [S..] und vollstaendiger Adresse", "unveraendert": true}}] +Setze "unveraendert" auf true, wenn die Behauptung haelt, sonst auf false.""" + EU_FOLLOWUP_PROMPT_TEMPLATE = """Du bist ein OSINT-Faktenchecker. AUSGABESPRACHE: {output_language} - KEINE Gedankenstriche verwenden, stattdessen Kommas oder neue Saetze. @@ -675,14 +722,21 @@ class FactCheckerAgent: max_queries: int | None = None, followup_queries: int | None = None, ) -> tuple[list[dict], ClaudeUsage]: - """Faktencheck im EU-Modus, mit gezielter Belegsuche und Nachhak-Runde. + """Faktencheck im EU-Modus, mit gezielter Belegsuche und zweiter Runde. Ablauf. Erstens werden zu den uebergebenen Punkten gezielt Belege - gesucht und in den Auftrag gelegt. Zweitens laeuft der eigentliche - Faktencheck. Drittens, falls danach noch Behauptungen unbelegt sind, - wird fuer genau diese noch einmal gesucht und nur sie werden erneut - bewertet. Das bildet das Nachhaken nach, das der heutige Weg mit seiner - eingebauten Websuche von sich aus macht. + gesucht und in den Auftrag gelegt, ein Teil davon als Gegenprobe. + Zweitens laeuft der eigentliche Faktencheck. Drittens folgt eine zweite + Runde, und zwar in beide Richtungen. Sind Behauptungen offen geblieben, + wird fuer genau diese noch einmal gesucht und nachbelegt. Sieht dagegen + alles bestaetigt aus, laeuft eine Belastungsprobe, die gezielt nach + Widerspruch zu den staerksten Behauptungen sucht. + + Der zweite Fall ist der wichtigere. Der EU-Weg bekommt seine Belege + vorab beschafft und sieht deshalb nie, dass etwas fehlt. Ohne + Belastungsprobe bestaetigte er acht Laeufe in Folge jede einzelne + Behauptung, waehrend der Weg ueber die Anthropic-CLI bei derselben Lage + rund ein Drittel offen liess. """ from agents.claude_client import call_claude from agents.eu_researcher import ( @@ -709,17 +763,143 @@ class FactCheckerAgent: # 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: + if anzahl_nachhak <= 0: return facts, gesamt - logger.info("EU-Faktencheck: %d von %d Behauptungen offen, starte Nachhak-Runde", len(offene), len(facts)) - facts, nachhak_usage = await self._eu_nachhaken( - facts=facts, offene=offene, title=title, - output_language=output_language, incident_type=incident_type, - avoid_queries=suche.queries, followup_queries=anzahl_nachhak, + # Die zweite Runde ist eine Verbesserung, keine Voraussetzung. Faellt sie + # aus, bleibt das Ergebnis des ersten Durchgangs erhalten, statt dass der + # ganze Faktencheck verloren geht. + try: + offene = [f for f in facts if f.get("status") in OFFENE_STATUS] + if len(offene) >= EU_VERIFY_FOLLOWUP_MIN_OPEN: + logger.info( + "EU-Faktencheck: %d von %d Behauptungen offen, starte Nachhak-Runde", + len(offene), len(facts)) + facts, zweite_usage = await self._eu_nachhaken( + facts=facts, offene=offene, title=title, + output_language=output_language, incident_type=incident_type, + avoid_queries=suche.queries, followup_queries=anzahl_nachhak, + ) + else: + bestaetigt = [f for f in facts if f.get("status") in BESTAETIGTE_STATUS] + if not bestaetigt: + return facts, gesamt + logger.info( + "EU-Faktencheck: %d von %d Behauptungen bestaetigt, starte Belastungsprobe", + len(bestaetigt), len(facts)) + facts, zweite_usage = await self._eu_belastungsprobe( + facts=facts, bestaetigt=bestaetigt, title=title, + output_language=output_language, incident_type=incident_type, + avoid_queries=suche.queries, queries=anzahl_nachhak, + ) + _add_usage(gesamt, zweite_usage) + except asyncio.CancelledError: + raise + except Exception as e: + logger.warning( + "EU-Faktencheck: zweite Runde fehlgeschlagen (%s), Ergebnis des ersten " + "Durchgangs bleibt bestehen", e) + return facts, gesamt + + async def _eu_belastungsprobe( + self, + *, + facts: list[dict], + bestaetigt: list[dict], + title: str, + output_language: str, + incident_type: str, + avoid_queries: list[str], + queries: int, + ) -> tuple[list[dict], ClaudeUsage]: + """Greift die als belegt eingestuften Behauptungen gezielt an. + + Sucht ausschliesslich nach Widerspruch und laesst das Modell die + Bestaetigungen daran messen. Nur so kann der EU-Weg ueberhaupt zu einem + unbestaetigten Ergebnis kommen, denn seine normale Belegsuche ist auf + Bestaetigung ausgelegt und findet deshalb auch nur Bestaetigung. + """ + from agents.claude_client import call_claude + from agents.eu_researcher import ( + plan_verification_searches, build_eu_prompt, _add_usage, + ) + from json_utils import extract_json_arrays + + gesamt = ClaudeUsage() + kandidaten = bestaetigt[:MAX_FACTS_PER_VERIFY_GROUP] + claims = [f.get("claim", "") for f in kandidaten if f.get("claim")] + if not claims: + return facts, gesamt + + suche = await plan_verification_searches( + title=title, search_items=claims, output_language=output_language, + max_queries=queries, avoid_queries=avoid_queries, + label="Belastungsprobe", nur_gegenproben=True, + ) + _add_usage(gesamt, suche.usage) + if not suche.block: + logger.info("EU-Belastungsprobe: keine Gegenbelege gefunden, Bewertungen bleiben bestehen") + return facts, gesamt + + forschung = incident_type == "research" + prompt = EU_CHALLENGE_PROMPT_TEMPLATE.format( + output_language=output_language, + title=title, + claims_text="\n".join(f"- {c}" for c in claims), + widerspruch="disputed" if forschung else "contradicted", + offen="unverified" if forschung else "unconfirmed", + status_werte=("established|unverified|disputed|false" + if forschung else "confirmed|unconfirmed|contradicted|false"), + ) + bewertungsregeln() + + try: + result, usage = await call_claude(build_eu_prompt(prompt, suche.block), tools=None) + _add_usage(gesamt, usage) + except TimeoutError: + logger.warning("EU-Belastungsprobe: Timeout, Bewertungen bleiben bestehen") + return facts, gesamt + except Exception as e: + logger.warning("EU-Belastungsprobe fehlgeschlagen (%s), Bewertungen bleiben bestehen", e) + return facts, gesamt + + fassungen = extract_json_arrays(result or "") + if not fassungen: + logger.warning("EU-Belastungsprobe: Antwort nicht auswertbar, Bewertungen bleiben bestehen") + return facts, gesamt + + zurueckgestuft = 0 + for eintrag in max(fassungen, key=len): + if not isinstance(eintrag, dict): + continue + claim = str(eintrag.get("claim", "")).strip() + neuer_status = str(eintrag.get("status", "")).strip().lower() + if not claim or neuer_status not in STATUS_PRIORITY: + continue + treffer = find_matching_claim(claim, kandidaten) + if not treffer: + continue + alt = treffer.get("status", "") + if neuer_status == alt: + continue + # Die Belastungsprobe darf nur zurueckstufen. Eine Bestaetigung + # laesst sich nicht dadurch erhaerten, dass die Gegenprobe nichts + # gefunden hat. + if neuer_status not in OFFENE_STATUS + WIDERSPRUCH_STATUS: + continue + treffer["status"] = neuer_status + begruendung = str(eintrag.get("evidence", "")).strip() + if begruendung: + bisher = (treffer.get("evidence") or "").strip() + treffer["evidence"] = ( + f"{bisher} Belastungsprobe: {begruendung}".strip() if bisher else begruendung) + zurueckgestuft += 1 + logger.info( + "EU-Belastungsprobe: '%s...' %s -> %s", claim[:60], alt, neuer_status) + + logger.info( + "EU-Belastungsprobe abgeschlossen: %d von %d Bestaetigungen zurueckgenommen", + zurueckgestuft, len(kandidaten), ) - _add_usage(gesamt, nachhak_usage) return facts, gesamt async def _eu_nachhaken( @@ -799,6 +979,20 @@ class FactCheckerAgent: if not treffer: continue alt = treffer.get("status", "developing") + # Widerspricht die Nachrecherche der Behauptung, gilt das auch dann, + # wenn der neue Status in der Rangfolge nicht hoeher steht. Ohne + # diese Ausnahme kennt die Runde nur den Weg nach oben. + if neuer_status in WIDERSPRUCH_STATUS and neuer_status != alt: + treffer["status"] = neuer_status + widerspruch = str(eintrag.get("evidence", "")).strip() + if widerspruch: + bisher = (treffer.get("evidence") or "").strip() + treffer["evidence"] = ( + f"{bisher} Nachrecherche: {widerspruch}".strip() if bisher else widerspruch) + logger.info( + "EU-Nachhak-Runde: '%s...' %s -> %s, Beleg widerspricht", + claim[:60], alt, neuer_status) + continue if STATUS_PRIORITY.get(neuer_status, 0) <= STATUS_PRIORITY.get(alt, 0): continue # "developing" heisst: der Sachverhalt selbst ist noch offen. Das diff --git a/src/config.py b/src/config.py index ac6c0d5..95f642a 100644 --- a/src/config.py +++ b/src/config.py @@ -130,6 +130,15 @@ EU_VERIFY_FOLLOWUP_MIN_OPEN = int(os.environ.get("EU_VERIFY_FOLLOWUP_MIN_OPEN", # deckt nur ein Teilthema ab, deshalb kleinere Suchmengen als beim Gesamtlauf. EU_VERIFY_GROUP_QUERIES = int(os.environ.get("EU_VERIFY_GROUP_QUERIES", "4")) EU_VERIFY_GROUP_FOLLOWUP_QUERIES = int(os.environ.get("EU_VERIFY_GROUP_FOLLOWUP_QUERIES", "3")) +# Anteil der Belegsuchen, der gezielt nach Widerspruch sucht statt nach +# Bestaetigung. Ohne Gegenproben sieht der Faktencheck nur stuetzendes Material +# und bestaetigt praktisch jede Behauptung. +EU_VERIFY_COUNTER_SHARE = float(os.environ.get("EU_VERIFY_COUNTER_SHARE", "0.3")) +# Anfragen, fuer die der Artikeltext geholt wird. Nur damit kann das Modell +# pruefen, ob eine Quelle die Behauptung wirklich traegt, statt aus der +# Ueberschrift zu schliessen. +EU_VERIFY_FULLTEXT_QUERIES = int(os.environ.get("EU_VERIFY_FULLTEXT_QUERIES", "3")) +EU_VERIFY_FULLTEXT_CHARS = int(os.environ.get("EU_VERIFY_FULLTEXT_CHARS", "1800")) # --- Verbrauchssaetze (Credits je Aktion) ------------------------------------- # Beschlossen 2026-07-23, der Verbrauchsrechner im Verwaltungsportal nutzt diff --git a/tests/test_eu_belegsuche.py b/tests/test_eu_belegsuche.py index 1fab65f..7043949 100644 --- a/tests/test_eu_belegsuche.py +++ b/tests/test_eu_belegsuche.py @@ -16,7 +16,8 @@ sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), ".." import agents.eu_researcher as eur # noqa: E402 from agents.claude_client import ClaudeUsage # noqa: E402 from config import ( # noqa: E402 - EU_VERIFY_HITS_PER_QUERY, EU_VERIFY_MAX_ITEMS, STAAN_COST_PER_QUERY_USD, + EU_VERIFY_FULLTEXT_CHARS, EU_VERIFY_FULLTEXT_QUERIES, EU_VERIFY_HITS_PER_QUERY, + EU_VERIFY_MAX_ITEMS, STAAN_COST_PER_QUERY_USD, ) ok = 0 @@ -47,7 +48,7 @@ treffer_je_anfrage = {} async def fake_staan(q, market="de-de", extra_snippets=True, max_snippets=3, full_content=False, extra_exclude=None, timeout=None): - zustand["suchen"].append({"q": q, "market": market}) + zustand["suchen"].append({"q": q, "market": market, "full_content": full_content}) return treffer_je_anfrage.get(q, []) @@ -222,5 +223,86 @@ 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("\nH) Ein Teil der Anfragen ist eine Gegenprobe") +neu() +plan_antwort["queries"] = [ + {"q": "ceuta opferzahl offiziell", "market": "de-de", "for_items": [1]}, + {"q": "ceuta opferzahl dementiert kritik", "market": "de-de", "for_items": [1], + "gegenprobe": True}, +] +treffer_je_anfrage["ceuta opferzahl offiziell"] = treffer(2, "a") +treffer_je_anfrage["ceuta opferzahl dementiert kritik"] = treffer(2, "b") +_rj = asyncio.run(eur.plan_verification_searches( + title="Ceuta", search_items=["57 Tote gemeldet"], max_queries=7, +)) +def markierte_treffer(block): + """Zaehlt nur Trefferzeilen, nicht die Erklaerung im Kopf.""" + return sum(1 for z in block.splitlines() + if z.startswith("[S") and "[GEGENPROBE]" in z) + + +pruefe("H1 Auftrag verlangt Gegenproben", "GEGENPROBEN" in zustand["plan_prompt"]) +pruefe("H2 Gegenprobe-Treffer sind gekennzeichnet", + markierte_treffer(_rj.block) == 2, markierte_treffer(_rj.block)) +pruefe("H3 Kopf erklaert die Gegenprobe", "GEGENPROBE" in _rj.block.split("[S1]")[0]) +pruefe("H4 stuetzende Treffer bleiben unmarkiert", + _rj.block.count("\n[S") == 4, _rj.block.count("\n[S")) + +print("\nI) Belastungsprobe sucht ausschliesslich Gegenproben") +neu() +plan_antwort["queries"] = [ + {"q": "eine anfrage", "market": "de-de"}, + {"q": "andere anfrage", "market": "de-de"}, +] +treffer_je_anfrage["eine anfrage"] = treffer(1, "c") +treffer_je_anfrage["andere anfrage"] = treffer(1, "d") +_rk = asyncio.run(eur.plan_verification_searches( + title="Ceuta", search_items=["Punkt"], max_queries=4, + label="Belastungsprobe", nur_gegenproben=True, +)) +pruefe("I1 alle Treffer als Gegenprobe gekennzeichnet", + markierte_treffer(_rk.block) == 2, markierte_treffer(_rk.block)) +pruefe("I2 Auftrag fordert so viele Gegenproben wie Anfragen", + "Davon sind 4 Anfragen" in zustand["plan_prompt"]) + +print("\nJ) Artikeltexte werden geholt und begrenzt") +neu() +plan_antwort["queries"] = [ + {"q": "mit text", "market": "de-de", "full_content": True}, + {"q": "ohne text", "market": "de-de", "full_content": False}, +] +treffer_je_anfrage["mit text"] = [{ + "title": "Titel mit Text", "hostname": "example.org", + "url": "https://example.org/volltext", "snippet": "Auszug", + "extra_snippets": [], "full_text": "W" * 5000, +}] +treffer_je_anfrage["ohne text"] = treffer(1, "e") +_rl = asyncio.run(eur.plan_verification_searches( + title="Ceuta", search_items=["Punkt"], max_queries=5, +)) +pruefe("J1 Volltext-Anfrage wurde mit Volltext gestellt", + zustand["suchen"][0].get("full_content") is True, zustand["suchen"][0]) +pruefe("J2 andere Anfrage ohne Volltext", + zustand["suchen"][1].get("full_content") is False, zustand["suchen"][1]) +pruefe("J3 Artikeltext steht im Belegblock", "Artikeltext:" in _rl.block) +pruefe("J4 Artikeltext ist begrenzt", + _rl.block.count("W") <= EU_VERIFY_FULLTEXT_CHARS + 10, _rl.block.count("W")) + +neu() +plan_antwort["queries"] = [ + {"q": f"anfrage {i}", "market": "de-de", "full_content": True} for i in range(1, 7) +] +for i in range(1, 7): + treffer_je_anfrage[f"anfrage {i}"] = [{ + "title": "T", "hostname": "example.org", "url": f"https://example.org/{i}", + "snippet": "s", "extra_snippets": [], "full_text": "Text", + }] +asyncio.run(eur.plan_verification_searches( + title="Ceuta", search_items=["Punkt"], max_queries=6, +)) +mit_volltext = sum(1 for s in zustand["suchen"] if s.get("full_content")) +pruefe("J5 Zahl der Volltext-Anfragen ist gedeckelt", + mit_volltext == EU_VERIFY_FULLTEXT_QUERIES, mit_volltext) + 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 21ee072..ce624e5 100644 --- a/tests/test_eu_factcheck.py +++ b/tests/test_eu_factcheck.py @@ -48,12 +48,14 @@ SUCHQUELLE = { async def fake_plan(*, title, search_items, output_language="Deutsch", - max_queries=7, avoid_queries=None, label="Stuetzsuche"): + max_queries=7, avoid_queries=None, label="Stuetzsuche", + nur_gegenproben=False): aufrufe["such"].append({ "label": label, "items": list(search_items), "max": max_queries, "avoid": list(avoid_queries or []), + "nur_gegenproben": nur_gegenproben, }) 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" @@ -137,24 +139,34 @@ pruefe("A12 beide Modellaufrufe mit Belegblock", all(m["hat_belege"] for m in aufrufe["modell"])) pruefe("A13 Kosten aufsummiert", round(usage.cost_usd, 3) == 0.12, usage.cost_usd) -print("\nB) EU-Modus, alles belegt, keine Nachrunde") +print("\nB) Ist alles belegt, folgt keine Nachbelegung sondern eine Belastungsprobe") neu() antworten.append(json_fakten([ (F_TOTE, "confirmed", "belegt"), (F_GRENZE, "confirmed", "belegt"), ])) +antworten.append(json_fakten([(F_TOTE, "confirmed", "haelt der Gegenprobe stand")])) asyncio.run(fc.check("Ceuta", ARTS, "adhoc", "Deutsch")) -pruefe("B1 keine Nachrunde bei belegten Fakten", len(aufrufe["such"]) == 1, len(aufrufe["such"])) -pruefe("B2 nur ein Modellaufruf", len(aufrufe["modell"]) == 1, len(aufrufe["modell"])) +pruefe("B1 keine Nachbelegung, sondern Belastungsprobe", + [a["label"] for a in aufrufe["such"]] == ["Stuetzsuche", "Belastungsprobe"], + [a["label"] for a in aufrufe["such"]]) +pruefe("B2 Belastungsprobe sucht nur Gegenproben", + aufrufe["such"][1]["nur_gegenproben"] is True) +pruefe("B3 zwei Modellaufrufe", len(aufrufe["modell"]) == 2, len(aufrufe["modell"])) -print("\nC) EU-Modus, nur ein offener Punkt, Schwelle nicht erreicht") +print("\nC) Ein einzelner offener Punkt loest keine Nachbelegung aus") neu() antworten.append(json_fakten([ (F_TOTE, "unconfirmed", "unklar"), (F_GRENZE, "confirmed", "belegt"), ])) +antworten.append(json_fakten([(F_GRENZE, "confirmed", "haelt")])) asyncio.run(fc.check("Ceuta", ARTS, "adhoc", "Deutsch")) -pruefe("C1 Schwelle von 2 offenen Punkten greift", len(aufrufe["such"]) == 1, len(aufrufe["such"])) +pruefe("C1 Schwelle von zwei offenen Punkten greift weiterhin", + aufrufe["such"][1]["label"] == "Belastungsprobe", + [a["label"] for a in aufrufe["such"]]) +pruefe("C2 nur der belegte Punkt wird angegriffen", + aufrufe["such"][1]["items"] == [F_GRENZE], aufrufe["such"][1]["items"]) print("\nD) Regression, CLI-Weg unveraendert") _ai_backend_var.set("cli") @@ -282,5 +294,101 @@ 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("\nL) Belastungsprobe greift, wenn alles bestaetigt aussieht") +_ai_backend_var.set("bedrock") +neu() +antworten.append(json_fakten([ + (F_TOTE, "confirmed", "belegt"), + (F_GRENZE, "confirmed", "belegt"), + (F_SANCHEZ, "confirmed", "belegt"), +])) +antworten.append(json_fakten([ + (F_TOTE, "unconfirmed", "[S1] https://example.org/a nennt nur eine Schaetzung"), +])) +facts_l, usage_l = asyncio.run(fc.check("Ceuta", ARTS, "adhoc", "Deutsch")) +status_l = {f["claim"][:20]: f["status"] for f in facts_l} +pruefe("L1 zweite Runde ist eine Belastungsprobe", + len(aufrufe["such"]) == 2 and aufrufe["such"][1]["label"] == "Belastungsprobe", + [a["label"] for a in aufrufe["such"]]) +pruefe("L2 Belastungsprobe sucht ausschliesslich Gegenproben", + aufrufe["such"][1]["nur_gegenproben"] is True) +pruefe("L3 Belastungsprobe meidet die schon gestellten Anfragen", + aufrufe["such"][1]["avoid"] == ["Stuetzsuche-q1", "Stuetzsuche-q2"], + aufrufe["such"][1]["avoid"]) +pruefe("L4 nicht haltbare Bestaetigung wird zurueckgenommen", + status_l.get(F_TOTE[:20]) == "unconfirmed", status_l) +pruefe("L5 haltbare Bestaetigungen bleiben", + status_l.get(F_GRENZE[:20]) == "confirmed" and status_l.get(F_SANCHEZ[:20]) == "confirmed", + status_l) +pruefe("L6 kein Faktenverlust", len(facts_l) == 3, len(facts_l)) +pruefe("L7 Begruendung vermerkt die Belastungsprobe", + any("Belastungsprobe" in (f.get("evidence") or "") for f in facts_l), + [f.get("evidence", "")[:70] for f in facts_l]) + +print("\nM) Die Belastungsprobe kann nur zurueckstufen") +neu() +antworten.append(json_fakten([ + (F_TOTE, "unconfirmed", "unklar"), + (F_GRENZE, "confirmed", "belegt"), + (F_SANCHEZ, "confirmed", "belegt"), +])) +antworten.append(json_fakten([ + (F_GRENZE, "confirmed", "haelt"), + (F_SANCHEZ, "established", "sogar noch besser belegt"), +])) +facts_m, _ = asyncio.run(fc.check("Ceuta", ARTS, "adhoc", "Deutsch")) +status_m = {f["claim"][:20]: f["status"] for f in facts_m} +pruefe("M1 Belastungsprobe laeuft auch bei einem offenen Punkt", + len(aufrufe["such"]) == 2 and aufrufe["such"][1]["label"] == "Belastungsprobe", + [a["label"] for a in aufrufe["such"]]) +pruefe("M2 keine Hochstufung durch die Belastungsprobe", + status_m.get(F_SANCHEZ[:20]) == "confirmed", status_m) +pruefe("M3 offener Punkt bleibt offen", status_m.get(F_TOTE[:20]) == "unconfirmed", status_m) + +print("\nN) Widerspruch in der Nachhak-Runde stuft zurueck") +neu() +antworten.append(json_fakten([ + (F_TOTE, "unconfirmed", "unklar"), + (F_SANCHEZ, "unconfirmed", "unklar"), + (F_GRENZE, "confirmed", "belegt"), +])) +antworten.append(json_fakten([ + (F_TOTE, "contradicted", "[S1] https://example.org/a nennt eine andere Zahl"), +])) +facts_n, _ = asyncio.run(fc.check("Ceuta", ARTS, "adhoc", "Deutsch")) +status_n = {f["claim"][:20]: f["status"] for f in facts_n} +pruefe("N1 Nachhak-Runde nimmt den Widerspruch auf", + status_n.get(F_TOTE[:20]) == "contradicted", status_n) + +print("\nO) Faellt die zweite Runde aus, bleibt der erste Durchgang erhalten") +neu() +antworten.append(json_fakten([ + (F_TOTE, "confirmed", "belegt"), + (F_GRENZE, "confirmed", "belegt"), +])) + + +async def plan_kaputt(**kw): + raise RuntimeError("Suche nicht erreichbar") + + +eur.plan_verification_searches = fake_plan +_alt_plan = eur.plan_verification_searches + + +async def plan_erst_gut_dann_kaputt(**kw): + if kw.get("label") == "Belastungsprobe": + raise RuntimeError("Suche nicht erreichbar") + return await fake_plan(**kw) + + +eur.plan_verification_searches = plan_erst_gut_dann_kaputt +facts_o, _ = asyncio.run(fc.check("Ceuta", ARTS, "adhoc", "Deutsch")) +eur.plan_verification_searches = fake_plan +pruefe("O1 Fakten ueberleben den Ausfall", len(facts_o) == 2, len(facts_o)) +pruefe("O2 Bewertungen bleiben unveraendert", + all(f["status"] == "confirmed" for f in facts_o), + [(f["claim"][:25], f["status"]) for f in facts_o]) + print("\nErgebnis: " + str(ok) + " bestanden, " + str(fail) + " fehlgeschlagen") sys.exit(1 if fail else 0)