AegisSight/AegisSight-Monitor
3
0
Fork 0
Code Issues Pull-Requests Actions Pakete Projekte Releases Wiki Aktivität

feat(eu): Faktencheck der EU-Fassung sucht breiter, gezielter und hakt nach

Quellcode durchsuchen 
Der Faktencheck ueber Bedrock und staan lieferte spuerbar weniger belegte
Fakten als der bisherige Weg ueber die Anthropic-CLI. Ursache war die Bauart.
Der bisherige Weg sucht waehrend der Pruefung selbst und kann bei duenner
Beleglage nachfassen, die EU-Fassung bekam dagegen einen festen, vorab
beschafften Stapel Belege und hatte genau einen Versuch.

Drei Aenderungen schliessen die Luecke.

1. Mehr Belege je Pruefung. Statt 3 Anfragen mit je 3 Treffern laufen jetzt
   7 Anfragen mit je 6 Treffern, und bis zu 14 statt 8 Pruefpunkte gehen in
   die Planung ein. Gleiche Quellen werden nur einmal in den Kontext gelegt.

2. Punktbezogene Planung. Das planende Modell ordnet jeder Suchanfrage zu,
   welche Behauptung sie belegen soll, statt allgemein zum Thema zu suchen.

3. Echte Nachhak-Runde. Bleiben nach der Pruefung mindestens zwei
   Behauptungen unbelegt, wird fuer genau diese noch einmal gesucht, mit
   anderen Anfragen als zuvor, und nur diese Punkte werden neu bewertet.
   Ein Status wird dabei nur angehoben, nie gesenkt, und faellt die Runde
   aus, bleiben die Fakten unveraendert.

Die Nachhak-Runde greift in allen drei Pruefwegen, also bei der Erstpruefung,
der inkrementellen Pruefung und der Pruefung in Themengruppen. Gruppen nutzen
kleinere Suchmengen, da sie parallel laufen und je nur ein Teilthema abdecken.
Alle Mengen sind ueber Umgebungsvariablen einstellbar.

Der Weg ueber die Anthropic-CLI bleibt unveraendert, das sichern sechs
Regressionstests ab. Neu sind zwei Testdateien mit zusammen 55 Pruefungen,
die ohne Netzzugriff und ohne Kosten laufen.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Dieser Commit ist enthalten in:
claude-dev
2026-08-01 12:59:26 +00:00
Ursprung 768150c302
Commit c4e19777f9
5 geänderte Dateien mit 822 neuen und 111 gelöschten Zeilen
Patch-Datei herunterladen Vergleichs-Datei herunterladen Alle Dateien ausklappen Alle Dateien einklappen

242
src/agents/eu_researcher.py
Datei anzeigen

@@ -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