diff --git a/src/agents/eu_researcher.py b/src/agents/eu_researcher.py index 84f3261..be68a67 100644 --- a/src/agents/eu_researcher.py +++ b/src/agents/eu_researcher.py @@ -26,6 +26,9 @@ from config import ( CLAUDE_MODEL_STANDARD, EU_RESEARCH_MAX_ROUNDS_ADHOC, EU_RESEARCH_MAX_ROUNDS_RESEARCH, + EU_VERIFY_HITS_PER_QUERY, + EU_VERIFY_MAX_ITEMS, + EU_VERIFY_MAX_QUERIES, STAAN_COST_PER_QUERY_USD, TIMEZONE, ) @@ -390,15 +393,25 @@ _SEARCH_NOTE_NO_RESULTS = ( _VERIFY_QUERY_PROMPT = """Du planst Verifikations-Suchanfragen für einen OSINT-Faktencheck. Heute ist der {today}. Lage: {title} -ZU PRÜFENDE PUNKTE: +ZU PRÜFENDE PUNKTE (nummeriert): {items_block} -Erzeuge maximal {max_queries} Websuche-Anfragen, mit denen sich die wichtigsten dieser -Punkte gegen unabhängige, aktuelle Quellen prüfen lassen. Konkrete Ereignisse, Zahlen -und Akteure zuerst. Keine Jahreszahlen anhängen. +AUFGABE: +Erzeuge bis zu {max_queries} Websuche-Anfragen. Plane sie GEZIELT je Punkt, nicht +pauschal für alle zusammen. Eine gute Anfrage macht genau einen dieser Punkte +überprüfbar, mit seinen konkreten Zahlen, Namen und Orten. -Antworte NUR mit JSON: {{"queries": [{{"q": "suchbegriffe", "market": "{default_market}"}}]}} -"market" ist "de-de", "en-us" oder "fr-fr".""" +REGELN: +- Ordne jeder Anfrage über "for_items" zu, welche Punkte sie belegen 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). +- 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]}}]}} +"market" ist "de-de", "en-us" oder "fr-fr". Fremdsprachige Anfragen funktionieren mit "en-us".""" def _iso_from_display(output_language: str) -> str: @@ -407,98 +420,163 @@ def _iso_from_display(output_language: str) -> str: return mapping.get((output_language or "").strip().lower(), "en") +def _add_usage(total: ClaudeUsage, u: ClaudeUsage | None) -> None: + """Zaehlt einen Einzelverbrauch auf die Gesamtsumme.""" + if not u: + return + total.input_tokens += u.input_tokens + total.output_tokens += u.output_tokens + total.cache_creation_tokens += u.cache_creation_tokens + total.cache_read_tokens += u.cache_read_tokens + total.cost_usd += u.cost_usd + total.duration_ms += u.duration_ms + + +async def plan_verification_searches( + *, + title: str, + search_items: list[str], + output_language: str = "Deutsch", + max_queries: int = EU_VERIFY_MAX_QUERIES, + avoid_queries: list[str] | None = None, + label: str = "Stützsuche", +) -> tuple[str, ClaudeUsage, list[str]]: + """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. + """ + 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, [] + + default_market = market_for_language(_iso_from_display(output_language)) + avoid_block = "" + if avoid_queries: + avoid_block = ( + "- Diese Anfragen wurden bereits gestellt und brachten nicht genug, stelle ANDERE:\n" + + "\n".join(f" {q[:120]}" for q in avoid_queries[:10]) + "\n" + ) + + 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, + default_market=default_market, + avoid_block=avoid_block, + ) + + try: + plan_text, plan_usage = await call_bedrock(plan_prompt, model=CLAUDE_MODEL_FAST, timeout=120) + _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, [] + + from agents.researcher import _extract_json_object + obj = _extract_json_object(plan_text or "") + queries: list[dict] = [] + gesehen: set[str] = {(q or "").strip().lower() for q in (avoid_queries or [])} + for q in (obj or {}).get("queries", []) or []: + if not isinstance(q, dict): + continue + text = str(q.get("q", "")).strip()[:400] + if not text or text.lower() in gesehen: + continue + gesehen.add(text.lower()) + queries.append({"q": text, "market": str(q.get("market", "")).strip().lower()}) + if len(queries) >= max_queries: + break + + if not queries: + logger.warning("EU-%s: Planung lieferte keine brauchbaren Anfragen", label) + return "", total, [] + + lines: list[str] = [] + ausgefuehrt: list[str] = [] + gesehene_urls: set[str] = set() + treffer = 0 + for q in queries: + 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, + ) + ausgefuehrt.append(q["q"]) + except StaanError as e: + logger.warning("EU-%s: Suche fehlgeschlagen (%s)", label, e) + continue + for r in res[:EU_VERIFY_HITS_PER_QUERY]: + # Mehrere Anfragen finden oft dieselbe Seite, jede Quelle nur einmal + # in den Kontextblock legen. + url = (r.get("url") or "").strip() + if url and url in gesehene_urls: + continue + if url: + gesehene_urls.add(url) + treffer += 1 + 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]}") + for chunk in (r.get("extra_snippets") or [])[:1]: + lines.append(f" Auszug: {chunk[:300]}") + + total.cost_usd += len(ausgefuehrt) * STAAN_COST_PER_QUERY_USD + logger.info("EU-%s: %d Suchen, %d Treffer im Kontextblock", label, len(ausgefuehrt), treffer) + + if not lines: + return "", 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 + + +def build_eu_prompt(prompt: str, results_block: str = "") -> str: + """Haengt den Werkzeug-Hinweis und einen etwaigen Belegblock an.""" + hint = _EU_TOOL_HINT.format( + search_note=_SEARCH_NOTE_WITH_RESULTS if results_block else _SEARCH_NOTE_NO_RESULTS + ) + return prompt + hint + results_block + + async def eu_call_with_search( prompt: str, *, title: str = "", search_items: list[str] | None = None, output_language: str = "Deutsch", - max_queries: int = 3, + max_queries: int = EU_VERIFY_MAX_QUERIES, ) -> tuple[str, ClaudeUsage]: """EU-Ersatz für call_claude mit Default-Tools (Faktencheck/Lagebild). - Mit search_items läuft vorab die staan-Stützsuche, ohne wird nur der - Werkzeug-Hinweis angehängt. Der Modellaufruf geht werkzeuglos über + Mit search_items läuft vorab die gezielte staan-Belegsuche, ohne wird nur + der Werkzeug-Hinweis angehängt. Der Modellaufruf geht werkzeuglos über call_claude und damit (aktives Bedrock-Backend vorausgesetzt) über Frankfurt. Gibt (result_text, gesamt_usage) zurück.""" from agents.claude_client import call_claude - from config import CLAUDE_MODEL_FAST total = ClaudeUsage() - - def _add(u: ClaudeUsage | None): - if not u: - return - total.input_tokens += u.input_tokens - total.output_tokens += u.output_tokens - total.cache_creation_tokens += u.cache_creation_tokens - total.cache_read_tokens += u.cache_read_tokens - total.cost_usd += u.cost_usd - total.duration_ms += u.duration_ms - results_block = "" if search_items: - items = [str(s).strip() for s in search_items if str(s).strip()][:12] - today = datetime.now(TIMEZONE).strftime("%d.%m.%Y") - default_market = market_for_language(_iso_from_display(output_language)) - plan_prompt = _VERIFY_QUERY_PROMPT.format( - today=today, - title=title or "(ohne Titel)", - items_block="\n".join(f"- {s[:200]}" for s in items), - max_queries=max_queries, - default_market=default_market, + results_block, such_usage, _ = await plan_verification_searches( + title=title, search_items=search_items, + output_language=output_language, max_queries=max_queries, ) - try: - plan_text, plan_usage = await call_bedrock(plan_prompt, model=CLAUDE_MODEL_FAST, timeout=120) - _add(plan_usage) - except Exception as e: - logger.warning("EU-Stützsuche: Query-Planung fehlgeschlagen (%s), fahre ohne Suche fort", e) - plan_text = "" - queries = [] - if plan_text: - from agents.researcher import _extract_json_object - obj = _extract_json_object(plan_text) - for q in (obj or {}).get("queries", []) or []: - if isinstance(q, dict) and str(q.get("q", "")).strip(): - queries.append({ - "q": str(q["q"]).strip()[:400], - "market": str(q.get("market", "")).strip().lower(), - }) - searches = 0 - lines = [] - n = 0 - for q in queries[:max_queries]: - try: - res = await staan_search( - q["q"], - market=q["market"] or market_for_language(_iso_from_display(output_language)), - extra_snippets=True, max_snippets=3, full_content=False, - ) - searches += 1 - except StaanError as e: - logger.warning("EU-Stützsuche: Suche fehlgeschlagen (%s)", e) - continue - for r in res[:6]: - n += 1 - lines.append(f"[S{n}] {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]}") - total.cost_usd += searches * STAAN_COST_PER_QUERY_USD - if lines: - results_block = ( - "\n\nERGEBNISSE DER EUROPÄISCHEN WEBSUCHE " - "(von unserem System ausgeführt, als unabhängige Zweitquellen nutzbar):\n" - + "\n".join(lines) - ) - logger.info("EU-Stützsuche: %d Suchen, %d Treffer im Kontextblock", searches, n) + _add_usage(total, such_usage) - hint = _EU_TOOL_HINT.format( - search_note=_SEARCH_NOTE_WITH_RESULTS if results_block else _SEARCH_NOTE_NO_RESULTS - ) - final_prompt = prompt + hint + results_block - - result, usage = await call_claude(final_prompt, tools=None) - _add(usage) + result, usage = await call_claude(build_eu_prompt(prompt, results_block), tools=None) + _add_usage(total, usage) return result, total diff --git a/src/agents/factchecker.py b/src/agents/factchecker.py index 66e65be..90aa922 100644 --- a/src/agents/factchecker.py +++ b/src/agents/factchecker.py @@ -330,6 +330,38 @@ STATUS_PRIORITY = { MAX_FACTS_PER_VERIFY_GROUP = 8 # Max Fakten pro Verifikationsgruppe TWOPHASE_MIN_FACTS = 25 # Ab dieser Anzahl bestehender Fakten wird der Zwei-Phasen-Ansatz genutzt +# Status, die als "noch nicht ausreichend belegt" gelten und im EU-Modus eine +# Nachhak-Runde ausloesen koennen. +OFFENE_STATUS = ("unconfirmed", "unverified", "developing") + +EU_FOLLOWUP_PROMPT_TEMPLATE = """Du bist ein OSINT-Faktenchecker. +AUSGABESPRACHE: {output_language} +- KEINE Gedankenstriche verwenden, stattdessen Kommas oder neue Saetze. +- Verwende IMMER echte UTF-8-Umlaute (ä, ö, ü, ß), NIEMALS Umschreibungen. + +LAGE: {title} + +Die folgenden Behauptungen konnten im ersten Durchgang NICHT ausreichend belegt werden. +Fuer genau diese Punkte wurden zusaetzliche Suchergebnisse beschafft, sie stehen unten. + +OFFENE BEHAUPTUNGEN: +{open_claims_text} + +AUFGABE: +Pruefe jede dieser Behauptungen erneut, ausschliesslich gegen die unten angehaengten +Suchergebnisse. + +REGELN: +- Stufe eine Behauptung nur dann hoch, wenn die neuen Belege sie tatsaechlich stuetzen. +- Widersprechen die Belege der Behauptung, setze den Status auf "contradicted". +- Findest du keinen tragfaehigen Beleg, lass den Status unveraendert. +- Nenne in "evidence" die stuetzende Quelle mit ihrer [S..]-Nummer UND ihrer vollstaendigen + URL, damit der Beleg nachpruefbar bleibt, und fasse kurz zusammen, was sie belegt. +- Erfinde nichts und uebernimm nichts aus deinem Gedaechtnis. + +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..]"}}]""" + def normalize_claim(claim: str) -> str: """Normalisiert einen Claim fuer Aehnlichkeitsvergleich.""" @@ -480,6 +512,157 @@ class FactCheckerAgent: lines.append(line) return "\n".join(lines) + @staticmethod + def _ist_eu_modus() -> bool: + from agents.claude_client import _ai_backend_var + from config import AI_BACKEND + return (_ai_backend_var.get(None) or AI_BACKEND) == "bedrock" + + async def _eu_pruefen( + self, + prompt: str, + *, + title: str, + articles: list[dict], + output_language: str, + search_items: list[str], + incident_type: str = "adhoc", + max_queries: int | None = None, + followup_queries: int | None = None, + ) -> tuple[list[dict], ClaudeUsage]: + """Faktencheck im EU-Modus, mit gezielter Belegsuche und Nachhak-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. + """ + from agents.claude_client import call_claude + from agents.eu_researcher import ( + plan_verification_searches, build_eu_prompt, _add_usage, + ) + from config import ( + EU_VERIFY_MAX_QUERIES, EU_VERIFY_FOLLOWUP_QUERIES, EU_VERIFY_FOLLOWUP_MIN_OPEN, + ) + + gesamt = ClaudeUsage() + 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( + title=title, search_items=search_items, + output_language=output_language, max_queries=anzahl_anfragen, + ) + _add_usage(gesamt, such_usage) + + result, usage = await call_claude(build_eu_prompt(prompt, block), tools=None) + _add_usage(gesamt, usage) + facts = self._parse_response(result, articles=articles) + + 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: + 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=gestellte_anfragen, followup_queries=anzahl_nachhak, + ) + _add_usage(gesamt, nachhak_usage) + return facts, gesamt + + async def _eu_nachhaken( + self, + *, + facts: list[dict], + offene: list[dict], + title: str, + output_language: str, + incident_type: str, + avoid_queries: list[str], + followup_queries: int | None = None, + ) -> tuple[list[dict], ClaudeUsage]: + """Sucht gezielt Belege fuer die offenen Behauptungen und bewertet nur + diese erneut. Gibt die zusammengefuehrte Faktenliste zurueck.""" + from agents.claude_client import call_claude + from agents.eu_researcher import ( + plan_verification_searches, build_eu_prompt, _add_usage, + ) + from config import EU_VERIFY_FOLLOWUP_QUERIES + from json_utils import extract_json_array + + gesamt = ClaudeUsage() + kandidaten = offene[:MAX_FACTS_PER_VERIFY_GROUP] + claims = [f.get("claim", "") for f in kandidaten if f.get("claim")] + if not claims: + return facts, gesamt + + block, such_usage, _ = 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) + if not block: + logger.info("EU-Nachhak-Runde: keine zusaetzlichen Belege gefunden, Fakten bleiben unveraendert") + return facts, gesamt + + status_werte = ("established|unverified|disputed|developing" + if incident_type == "research" + else "confirmed|unconfirmed|contradicted|developing") + prompt = EU_FOLLOWUP_PROMPT_TEMPLATE.format( + output_language=output_language, + title=title, + open_claims_text="\n".join(f"- {c}" for c in claims), + status_werte=status_werte, + ) + + try: + result, usage = await call_claude(build_eu_prompt(prompt, block), tools=None) + _add_usage(gesamt, usage) + except TimeoutError: + logger.warning("EU-Nachhak-Runde: Timeout, Fakten bleiben unveraendert") + return facts, gesamt + except Exception as e: + logger.warning("EU-Nachhak-Runde fehlgeschlagen (%s), Fakten bleiben unveraendert", e) + return facts, gesamt + + aktualisierungen = extract_json_array(result or "") + if not isinstance(aktualisierungen, list): + logger.warning("EU-Nachhak-Runde: Antwort nicht auswertbar, Fakten bleiben unveraendert") + return facts, gesamt + + hochgestuft = 0 + for eintrag in aktualisierungen: + 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", "developing") + if STATUS_PRIORITY.get(neuer_status, 0) <= STATUS_PRIORITY.get(alt, 0): + continue + treffer["status"] = neuer_status + belege = str(eintrag.get("evidence", "")).strip() + if belege: + bisher = (treffer.get("evidence") or "").strip() + treffer["evidence"] = f"{bisher} Nachrecherche: {belege}".strip() if bisher else belege + hochgestuft += 1 + logger.info("EU-Nachhak-Runde: '%s...' %s -> %s", claim[:60], alt, neuer_status) + + logger.info( + "EU-Nachhak-Runde abgeschlossen: %d von %d offenen Behauptungen belegt", + hochgestuft, len(kandidaten), + ) + return facts, gesamt + async def check(self, title: str, articles: list[dict], incident_type: str = "adhoc", output_language: str = "Deutsch") -> tuple[list[dict], ClaudeUsage | None]: """Führt vollständigen Faktencheck durch (erster Refresh).""" if not articles: @@ -495,19 +678,17 @@ class FactCheckerAgent: ) try: - # EU-Modus (Phase 3): staan-Stützsuche statt WebSearch, Aufruf über Bedrock. - from agents.claude_client import _ai_backend_var - from config import AI_BACKEND - if (_ai_backend_var.get(None) or AI_BACKEND) == "bedrock": - from agents.eu_researcher import eu_call_with_search - headlines = [a.get("headline_de") or a.get("headline", "") for a in articles[:8]] - result, usage = await eu_call_with_search( - prompt, title=title, output_language=output_language, - search_items=[title] + headlines, max_queries=3, + # EU-Modus: gezielte staan-Belegsuche plus Nachhak-Runde statt WebSearch. + if self._ist_eu_modus(): + headlines = [a.get("headline_de") or a.get("headline", "") for a in articles[:10]] + facts, usage = await self._eu_pruefen( + prompt, title=title, articles=articles, + output_language=output_language, incident_type=incident_type, + search_items=[title] + headlines, ) else: result, usage = await call_claude(prompt) - facts = self._parse_response(result, articles=articles) + facts = self._parse_response(result, articles=articles) logger.info(f"Faktencheck: {len(facts)} Fakten geprüft") return facts, usage except TimeoutError: @@ -548,23 +729,21 @@ class FactCheckerAgent: ) try: - # EU-Modus (Phase 3): staan-Stützsuche zu offenen Fakten statt WebSearch. - from agents.claude_client import _ai_backend_var - from config import AI_BACKEND - if (_ai_backend_var.get(None) or AI_BACKEND) == "bedrock": - from agents.eu_researcher import eu_call_with_search + # EU-Modus: gezielte Belegsuche zu offenen Fakten plus Nachhak-Runde. + if self._ist_eu_modus(): open_claims = [ f.get("claim", "") for f in existing_facts if f.get("status") in ("unconfirmed", "developing", "disputed", "unverified") ][:8] - headlines = [a.get("headline_de") or a.get("headline", "") for a in new_articles[:4]] - result, usage = await eu_call_with_search( - prompt, title=title, output_language=output_language, - search_items=open_claims + headlines, max_queries=3, + headlines = [a.get("headline_de") or a.get("headline", "") for a in new_articles[:6]] + facts, usage = await self._eu_pruefen( + prompt, title=title, articles=new_articles, + output_language=output_language, incident_type=incident_type, + search_items=open_claims + headlines, ) else: result, usage = await call_claude(prompt) - facts = self._parse_response(result, articles=new_articles) + facts = self._parse_response(result, articles=new_articles) logger.info(f"Inkrementeller Faktencheck: {len(facts)} Fakten (neu + aktualisiert)") return facts, usage except TimeoutError: @@ -671,19 +850,24 @@ class FactCheckerAgent: ) try: - # EU-Modus (Phase 3): gezielte Stützsuche je Faktengruppe. - from agents.claude_client import _ai_backend_var - from config import AI_BACKEND - if (_ai_backend_var.get(None) or AI_BACKEND) == "bedrock": - from agents.eu_researcher import eu_call_with_search + # EU-Modus: gezielte Belegsuche je Faktengruppe plus Nachhak-Runde. + if self._ist_eu_modus(): + from config import ( + EU_VERIFY_GROUP_QUERIES, EU_VERIFY_GROUP_FOLLOWUP_QUERIES, + ) claims = [f.get("claim", "") for f in group_facts] + list(group_claims) - result, v_usage = await eu_call_with_search( - prompt, title=f"{title} ({theme})", output_language=output_language, - search_items=claims[:8], max_queries=2, + # Kleinere Suchmenge je Gruppe, da mehrere Gruppen parallel + # laufen und jede nur ein Teilthema abdeckt. + facts, v_usage = await self._eu_pruefen( + prompt, title=f"{title} ({theme})", articles=None, + output_language=output_language, incident_type=incident_type, + search_items=claims[:8], + max_queries=EU_VERIFY_GROUP_QUERIES, + followup_queries=EU_VERIFY_GROUP_FOLLOWUP_QUERIES, ) else: result, v_usage = await call_claude(prompt) - facts = self._parse_response(result) + facts = self._parse_response(result) logger.info(f"Gruppe '{theme}': {len(facts)} Fakten geprüft") return facts, v_usage except TimeoutError: diff --git a/src/config.py b/src/config.py index f7d588a..3a43975 100644 --- a/src/config.py +++ b/src/config.py @@ -90,6 +90,23 @@ STAAN_EXCLUDE_DOMAINS = [ # ohne die Abschlussrunde). research bildet die 4-Phasen-Tiefenrecherche nach. EU_RESEARCH_MAX_ROUNDS_ADHOC = int(os.environ.get("EU_RESEARCH_MAX_ROUNDS_ADHOC", "3")) EU_RESEARCH_MAX_ROUNDS_RESEARCH = int(os.environ.get("EU_RESEARCH_MAX_ROUNDS_RESEARCH", "5")) +# --- Belegsuche des Faktenchecks (EU-Modus) ----------------------------------- +# Vorher pauschal 3 Anfragen fuer alle Behauptungen zusammen. Der Vergleich am +# 31.07.2026 zeigte, dass die EU-Fassung dadurch systematisch weniger Fakten +# bestaetigen konnte als der heutige Weg, der je Behauptung nachsuchen darf. +# Jetzt gezielt je Behauptung geplant, breiter, und mit einer Nachrunde fuer +# die Punkte, die im ersten Durchgang offen geblieben sind. +EU_VERIFY_MAX_QUERIES = int(os.environ.get("EU_VERIFY_MAX_QUERIES", "7")) +EU_VERIFY_MAX_ITEMS = int(os.environ.get("EU_VERIFY_MAX_ITEMS", "14")) +EU_VERIFY_HITS_PER_QUERY = int(os.environ.get("EU_VERIFY_HITS_PER_QUERY", "6")) +# Nachhak-Runde: Anzahl zusaetzlicher Anfragen und Obergrenze offener +# Behauptungen, die eine Nachrunde ausloesen. 0 schaltet die Nachrunde ab. +EU_VERIFY_FOLLOWUP_QUERIES = int(os.environ.get("EU_VERIFY_FOLLOWUP_QUERIES", "5")) +EU_VERIFY_FOLLOWUP_MIN_OPEN = int(os.environ.get("EU_VERIFY_FOLLOWUP_MIN_OPEN", "2")) +# Faktenpruefung in Themengruppen. Mehrere Gruppen laufen parallel und jede +# 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")) # --- 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 new file mode 100644 index 0000000..64c573d --- /dev/null +++ b/tests/test_eu_belegsuche.py @@ -0,0 +1,188 @@ +"""Testet die gezielte Belegsuche der EU-Fassung. + +Laeuft ohne Netzzugriff und ohne Kosten, das planende Modell und die +europaeische Suche sind durch Testdoubles ersetzt. + +Aufruf aus dem Projektstamm: + venv/bin/python tests/test_eu_belegsuche.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 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, +) + +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)) + + +zustand = {"plan_prompt": "", "suchen": []} +plan_antwort = {"queries": []} + + +async def fake_bedrock(prompt, model=None, raw_text=False, timeout=None): + zustand["plan_prompt"] = prompt + return json.dumps(plan_antwort), ClaudeUsage(input_tokens=10, output_tokens=5, cost_usd=0.002) + + +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}) + return treffer_je_anfrage.get(q, []) + + +def treffer(n, praefix="t", host="example.org"): + return [{ + "title": praefix + " Titel " + str(i), + "hostname": host, + "url": "https://" + host + "/" + praefix + str(i), + "snippet": "Auszug " + str(i), + "extra_snippets": ["Zusatz " + str(i)], + } for i in range(1, n + 1)] + + +eur.call_bedrock = fake_bedrock +eur.staan_search = fake_staan + + +def neu(): + zustand["suchen"].clear() + treffer_je_anfrage.clear() + + +print("\nA) Planung ist punktbezogen und fuehrt die Anfragen aus") +neu() +plan_antwort["queries"] = [ + {"q": "Ceuta Opferzahl offiziell", "market": "de-de", "for_items": [1]}, + {"q": "Ceuta border deaths toll", "market": "en-us", "for_items": [1, 2]}, + {"q": "Sanchez Ceuta declaracion", "market": "es-es", "for_items": [2]}, +] +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( + title="Ceuta", search_items=["57 Tote gemeldet", "Sanchez sprach von Angriff"], + output_language="Deutsch", max_queries=7, +)) +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", + "57 Tote gemeldet" in zustand["plan_prompt"] and "Sanchez sprach von Angriff" in zustand["plan_prompt"]) +pruefe("A4 Planungsauftrag verlangt Punktzuordnung", "for_items" in zustand["plan_prompt"]) +pruefe("A5 Obergrenze steht im Auftrag", "7" in zustand["plan_prompt"]) +pruefe("A6 alle Treffer nummeriert im Block", block.count("[S") == 6, block.count("[S")) +pruefe("A7 Auszuege uebernommen", block.count("Auszug") >= 6, block.count("Auszug")) +pruefe("A8 unbekannter Markt faellt auf die Vorgabe zurueck", + [s["market"] for s in zustand["suchen"]] == ["de-de", "en-us", "de-de"], + [s["market"] for s in zustand["suchen"]]) +erwartet = round(0.002 + 3 * STAAN_COST_PER_QUERY_USD, 6) +pruefe("A9 Suchkosten eingerechnet", round(usage.cost_usd, 6) == erwartet, usage.cost_usd) + +print("\nB) Dieselbe Quelle steht nur einmal im Belegblock") +neu() +plan_antwort["queries"] = [ + {"q": "anfrage eins", "market": "de-de"}, + {"q": "anfrage zwei", "market": "de-de"}, +] +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( + title="Ceuta", search_items=["Punkt"], max_queries=7, +)) +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))) + +print("\nC) Nachrunde meidet bereits gestellte Anfragen") +neu() +plan_antwort["queries"] = [ + {"q": "alte anfrage", "market": "de-de"}, + {"q": "neue anfrage", "market": "de-de"}, +] +treffer_je_anfrage["neue anfrage"] = treffer(2, "n") +block_c, _, ausgefuehrt_c = asyncio.run(eur.plan_verification_searches( + title="Ceuta", search_items=["Punkt"], max_queries=5, + avoid_queries=["alte anfrage"], label="Nachhak-Suche", +)) +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"]) +pruefe("C3 nur die neue Anfrage gemeldet", ausgefuehrt_c == ["neue anfrage"], ausgefuehrt_c) + +print("\nD) Grenzen werden eingehalten") +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( + title="Ceuta", search_items=["Punkt"], max_queries=4, +)) +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")) +neu() +plan_antwort["queries"] = [{"q": "q1", "market": "de-de"}] +treffer_je_anfrage["q1"] = treffer(1, "z") +asyncio.run(eur.plan_verification_searches( + title="Ceuta", search_items=["Punkt " + str(i) for i in range(1, 40)], max_queries=2, +)) +gezaehlt = sum(1 for i in range(1, 40) if ("Punkt " + str(i)) in zustand["plan_prompt"]) +pruefe("D3 Zahl der Pruefpunkte begrenzt", gezaehlt == EU_VERIFY_MAX_ITEMS, gezaehlt) + +print("\nE) Ausfaelle fuehren nicht zum Abbruch") +neu() +plan_antwort["queries"] = [{"q": "kaputt", "market": "de-de"}, {"q": "heil", "market": "de-de"}] +treffer_je_anfrage["heil"] = treffer(2, "h") + + +async def staan_mit_fehler(q, **kw): + zustand["suchen"].append({"q": q, "market": kw.get("market")}) + if q == "kaputt": + raise eur.StaanError("Suche nicht erreichbar") + return treffer_je_anfrage.get(q, []) + + +eur.staan_search = staan_mit_fehler +block_e, usage_e, ausgefuehrt_e = asyncio.run(eur.plan_verification_searches( + title="Ceuta", search_items=["Punkt"], max_queries=5, +)) +eur.staan_search = fake_staan +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", + round(usage_e.cost_usd, 6) == round(0.002 + STAAN_COST_PER_QUERY_USD, 6), usage_e.cost_usd) + +neu() +plan_antwort["queries"] = [] +block_f, _, ausgefuehrt_f = asyncio.run(eur.plan_verification_searches( + title="Ceuta", search_items=["Punkt"], max_queries=5, +)) +pruefe("E4 leere Planung liefert leeren Block", block_f == "" and ausgefuehrt_f == []) + +block_g, _, _ = asyncio.run(eur.plan_verification_searches( + title="Ceuta", search_items=[], max_queries=5, +)) +pruefe("E5 ohne Pruefpunkte keine Suche", block_g == "") + +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 new file mode 100644 index 0000000..cf61e79 --- /dev/null +++ b/tests/test_eu_factcheck.py @@ -0,0 +1,244 @@ +"""Testet die EU-Nachhak-Schleife im Faktencheck und sichert den CLI-Weg ab. + +Laeuft ohne Netzzugriff und ohne Kosten, die Belegsuche und der Modellaufruf +sind durch Testdoubles ersetzt. + +Aufruf aus dem Projektstamm: + venv/bin/python tests/test_eu_factcheck.py +""" +import asyncio +import os +import sys + +sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "src")) + +from agents.factchecker import FactCheckerAgent # noqa: E402 +import agents.factchecker as fcmod # noqa: E402 +import agents.eu_researcher as eur # noqa: E402 +import agents.claude_client as cc # noqa: E402 +from agents.claude_client import _ai_backend_var, 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)) + + +aufrufe = {"such": [], "modell": []} + + +async def fake_plan(*, title, search_items, output_language="Deutsch", + max_queries=7, avoid_queries=None, label="Stuetzsuche"): + aufrufe["such"].append({ + "label": label, + "items": list(search_items), + "max": max_queries, + "avoid": list(avoid_queries or []), + }) + 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"] + + +antworten = [] + + +async def fake_call_claude(prompt, tools="WebSearch,WebFetch", model=None, + raw_text=False, timeout=None): + aufrufe["modell"].append({"tools": tools, "hat_belege": "BELEGE" in prompt}) + if not antworten: + raise AssertionError("unerwarteter Modellaufruf") + return antworten.pop(0), ClaudeUsage(input_tokens=100, output_tokens=50, cost_usd=0.05) + + +eur.plan_verification_searches = fake_plan +fcmod.call_claude = fake_call_claude # CLI-Zweig nutzt das Modul-Symbol +cc.call_claude = fake_call_claude # EU-Zweig importiert erst zur Laufzeit + +fc = FactCheckerAgent() +ARTS = [ + {"headline": "Ceuta Grenze Tote gemeldet", "source": "tagesschau", + "source_url": "https://tagesschau.de/ceuta", "content_original": "t"}, + {"headline": "Sanchez spricht ueber Angriff in Ceuta", "source": "elpais", + "source_url": "https://elpais.com/ceuta", "content_original": "t"}, +] + +F_TOTE = "In Ceuta wurden 57 Tote gemeldet" +F_SANCHEZ = "Sanchez sprach in Ceuta von einem Angriff" +F_GRENZE = "Die Grenze in Ceuta wurde geschlossen" + + +def neu(): + aufrufe["such"].clear() + aufrufe["modell"].clear() + antworten.clear() + + +def json_fakten(eintraege): + teile = [] + for claim, status, ev in eintraege: + teile.append('{"claim": "%s", "status": "%s", "evidence": "%s"}' % (claim, status, ev)) + return "[" + ", ".join(teile) + "]" + + +print("\nA) EU-Modus, Nachhak-Runde stuft offene Behauptung hoch") +_ai_backend_var.set("bedrock") +neu() +antworten.append(json_fakten([ + (F_TOTE, "unconfirmed", "nur eine Quelle"), + (F_SANCHEZ, "unconfirmed", "unklar"), + (F_GRENZE, "confirmed", "belegt"), +])) +antworten.append(json_fakten([ + (F_TOTE, "confirmed", "laut [S1] https://example.org/a bestaetigt"), +])) +facts, usage = asyncio.run(fc.check("Ceuta", ARTS, "adhoc", "Deutsch")) +status = {f.get("claim", "")[:20]: f.get("status") for f in facts} +pruefe("A1 Erstsuche plant 7 Anfragen", aufrufe["such"][0]["max"] == 7) +pruefe("A2 Erstsuche kennt Titel und Schlagzeilen", + len(aufrufe["such"][0]["items"]) == 3, aufrufe["such"][0]["items"]) +pruefe("A3 Nachhak-Runde ausgeloest", + len(aufrufe["such"]) == 2 and aufrufe["such"][1]["label"] == "Nachhak-Suche") +pruefe("A4 Nachhak nur fuer die 2 offenen Punkte", + len(aufrufe["such"][1]["items"]) == 2, aufrufe["such"][1]["items"]) +pruefe("A5 Nachhak meidet die alten Anfragen", + aufrufe["such"][1]["avoid"] == ["Stuetzsuche-q1", "Stuetzsuche-q2"], + aufrufe["such"][1]["avoid"]) +pruefe("A6 belegter Fakt hochgestuft", status.get(F_TOTE[:20]) == "confirmed", status) +pruefe("A7 unbelegter Fakt bleibt offen", status.get(F_SANCHEZ[:20]) == "unconfirmed", status) +pruefe("A8 bestaetigter Fakt unangetastet", status.get(F_GRENZE[:20]) == "confirmed", status) +ev = [f["evidence"] for f in facts if f["claim"].startswith("In Ceuta")][0] +pruefe("A9 Nachrecherche mit URL in der Evidenz", + "Nachrecherche" in ev and "https://example.org/a" in ev, ev[:160]) +pruefe("A10 kein Faktenverlust", len(facts) == 3, len(facts)) +pruefe("A11 beide Modellaufrufe ohne Werkzeuge", + [m["tools"] for m in aufrufe["modell"]] == [None, None], aufrufe["modell"]) +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") +neu() +antworten.append(json_fakten([ + (F_TOTE, "confirmed", "belegt"), + (F_GRENZE, "confirmed", "belegt"), +])) +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"])) + +print("\nC) EU-Modus, nur ein offener Punkt, Schwelle nicht erreicht") +neu() +antworten.append(json_fakten([ + (F_TOTE, "unconfirmed", "unklar"), + (F_GRENZE, "confirmed", "belegt"), +])) +asyncio.run(fc.check("Ceuta", ARTS, "adhoc", "Deutsch")) +pruefe("C1 Schwelle von 2 offenen Punkten greift", len(aufrufe["such"]) == 1, len(aufrufe["such"])) + +print("\nD) Regression, CLI-Weg unveraendert") +_ai_backend_var.set("cli") +neu() +antworten.append(json_fakten([ + (F_TOTE, "unconfirmed", "unklar"), + (F_SANCHEZ, "unconfirmed", "unklar"), +])) +facts_d, _ = asyncio.run(fc.check("Ceuta", ARTS, "adhoc", "Deutsch")) +pruefe("D1 CLI macht keine Belegsuche", len(aufrufe["such"]) == 0, aufrufe["such"]) +pruefe("D2 CLI genau ein Modellaufruf", len(aufrufe["modell"]) == 1, len(aufrufe["modell"])) +pruefe("D3 CLI behaelt die WebSearch-Werkzeuge", + aufrufe["modell"][0]["tools"] == "WebSearch,WebFetch", aufrufe["modell"]) +pruefe("D4 CLI bekommt keinen Belegblock", aufrufe["modell"][0]["hat_belege"] is False) +pruefe("D5 CLI liefert Fakten", len(facts_d) == 2, len(facts_d)) + +print("\nE) EU-Modus, inkrementelle Pruefung") +_ai_backend_var.set("bedrock") +neu() +antworten.append(json_fakten([ + (F_TOTE, "unconfirmed", "unklar"), + (F_SANCHEZ, "unconfirmed", "unklar"), +])) +antworten.append(json_fakten([ + (F_SANCHEZ, "confirmed", "[S1] https://example.org/a belegt die Aussage"), +])) +alt = [{"claim": "Alter offener Punkt aus Ceuta", "status": "unconfirmed", "evidence": "e"}] +facts_e, _ = asyncio.run(fc.check_incremental("Ceuta", ARTS, alt, "adhoc", "Deutsch")) +pruefe("E1 inkrementell nutzt die Belegsuche", len(aufrufe["such"]) >= 1, aufrufe["such"]) +pruefe("E2 alte offene Punkte fliessen in die Suche", + any("Alter offener Punkt" in i for i in aufrufe["such"][0]["items"]), + aufrufe["such"][0]["items"]) +pruefe("E3 inkrementell hakt nach", len(aufrufe["such"]) == 2, len(aufrufe["such"])) +pruefe("E4 inkrementelle Hochstufung greift", + any(f["claim"].startswith("Sanchez") and f["status"] == "confirmed" for f in facts_e), + [(f["claim"][:25], f["status"]) for f in facts_e]) + +print("\nF) Regression, CLI-Weg inkrementell unveraendert") +_ai_backend_var.set("cli") +neu() +antworten.append(json_fakten([(F_TOTE, "unconfirmed", "unklar")])) +asyncio.run(fc.check_incremental("Ceuta", ARTS, alt, "adhoc", "Deutsch")) +pruefe("F1 CLI inkrementell ohne Belegsuche", len(aufrufe["such"]) == 0, aufrufe["such"]) +pruefe("F2 CLI inkrementell mit Werkzeugen", + aufrufe["modell"][0]["tools"] == "WebSearch,WebFetch", aufrufe["modell"]) + +print("\nG) EU-Modus, Nachhak-Runde faellt aus, Fakten bleiben erhalten") +_ai_backend_var.set("bedrock") +neu() +antworten.append(json_fakten([ + (F_TOTE, "unconfirmed", "unklar"), + (F_SANCHEZ, "unconfirmed", "unklar"), +])) +antworten.append("Das ist kein JSON, sondern Fliesstext.") +facts_g, _ = asyncio.run(fc.check("Ceuta", ARTS, "adhoc", "Deutsch")) +pruefe("G1 Fakten ueberleben eine unbrauchbare Nachhak-Antwort", len(facts_g) == 2, len(facts_g)) +pruefe("G2 Status bleibt offen", + all(f["status"] in ("unconfirmed", "unverified") for f in facts_g), + [(f["claim"][:25], f["status"]) for f in facts_g]) + +print("\nH) EU-Modus, Gruppenpruefung nutzt die kleinere Suchmenge") +from config import EU_VERIFY_GROUP_QUERIES, EU_VERIFY_GROUP_FOLLOWUP_QUERIES # noqa: E402 + +_ai_backend_var.set("bedrock") +neu() +antworten.append(json_fakten([ + (F_TOTE, "unconfirmed", "unklar"), + (F_SANCHEZ, "unconfirmed", "unklar"), +])) +antworten.append(json_fakten([(F_TOTE, "confirmed", "[S1] https://example.org/a belegt")])) +facts_h, _ = asyncio.run(fc._eu_pruefen( + "Auftrag", title="Ceuta (Opfer)", articles=None, output_language="Deutsch", + search_items=[F_TOTE, F_SANCHEZ], incident_type="adhoc", + max_queries=EU_VERIFY_GROUP_QUERIES, + followup_queries=EU_VERIFY_GROUP_FOLLOWUP_QUERIES, +)) +pruefe("H1 Gruppe plant die kleinere Suchmenge", + aufrufe["such"][0]["max"] == EU_VERIFY_GROUP_QUERIES, aufrufe["such"][0]["max"]) +pruefe("H2 Gruppe hakt mit kleinerer Menge nach", + len(aufrufe["such"]) == 2 and aufrufe["such"][1]["max"] == EU_VERIFY_GROUP_FOLLOWUP_QUERIES, + [a["max"] for a in aufrufe["such"]]) +pruefe("H3 Gruppe stuft belegten Fakt hoch", + any(f["claim"].startswith("In Ceuta") and f["status"] == "confirmed" for f in facts_h), + [(f["claim"][:25], f["status"]) for f in facts_h]) + +print("\nI) Nachhaken abschaltbar") +neu() +antworten.append(json_fakten([ + (F_TOTE, "unconfirmed", "unklar"), + (F_SANCHEZ, "unconfirmed", "unklar"), +])) +asyncio.run(fc._eu_pruefen( + "Auftrag", title="Ceuta", articles=None, output_language="Deutsch", + search_items=[F_TOTE], incident_type="adhoc", followup_queries=0, +)) +pruefe("I1 followup_queries=0 unterbindet die Nachrunde", len(aufrufe["such"]) == 1, len(aufrufe["such"])) + +print("\nErgebnis: " + str(ok) + " bestanden, " + str(fail) + " fehlgeschlagen") +sys.exit(1 if fail else 0)