AegisSight/AegisSight-Monitor
3
0
Fork 0
Code Issues Pull-Requests Actions Pakete Projekte Releases Wiki Aktivität
Dateien
58aec2edd516087a6bd0f0e6f66f510cb3e9e2d4
AegisSight-Monitor/tests/test_eu_belegsuche.py
claude-dev 0f39154e39 fix(eu): drei Fehler beheben, die den Ertrag des Faktenchecks aufzehrten 
Der Vergleichslauf zur Innenministerkonferenz (Lage 49) zeigte, dass der
verbesserte Faktencheck zwar zehn Fakten lieferte, davon aber nur drei
uebrig blieben. Die Ursachen lagen in drei nachgelagerten Stellen.

1. Quellenzuordnung stufte alle zehn Fakten herab.
   Durch den groesseren Belegblock stuetzt das Modell seine Fakten jetzt
   staerker auf die europaeische Suche als auf die gespeicherten Meldungen.
   Es zitiert dabei die Belegnummern statt vollstaendiger Adressen. Die
   Quellenzuordnung suchte daraufhin vergeblich nach passenden Meldungen und
   stufte jeden Fakt herab. Im Vergleich: Lauf 45 hatte null Herabstufungen,
   Lauf 44 eine, Lauf 49 zehn von zehn.
   Behoben an zwei Stellen. Die Belegsuche gibt die gefundenen Quellen jetzt
   einzeln zurueck, der Faktencheck reicht sie an die Zuordnung durch, und
   der Auftrag verlangt ausdruecklich die vollstaendige Adresse je Beleg.

2. Duplikatpruefung loeschte sieben von zehn Fakten.
   Das Clustering warf inhaltlich verschiedene Beschluesse in eine Gruppe,
   weil viele Behauptungen gleich beginnen. Es gab keinerlei Absicherung, bei
   kleinen Faktenmengen entschied das Modell sogar voellig allein.
   Zwei Sicherungen greifen jetzt. Eine Gruppe, die mehr als 40 Prozent aller
   Fakten umfasst, wird verworfen. Und jeder Loeschkandidat muss dem
   behaltenen Fakt auch rechnerisch aehnlich sein, Grenze 0,55 im selben Mass
   wie im Vorfilter. Beide Werte sind ueber Umgebungsvariablen einstellbar.

3. Leerrunden in der Recherche.
   War die Treffer-Obergrenze erreicht, brach nur die innere Schleife ab. Die
   Rundenschleife lief bis zum Ende weiter und befragte jedes Mal das teuerste
   Modell, ohne noch eine einzige Suche auszufuehren. Das kostete rund 0,17
   USD je Durchgang, bei drei Durchgaengen etwa 0,50 USD pro Aktualisierung.

Der Weg ueber die Anthropic-CLI bleibt unveraendert. Neu sind 16 Pruefungen,
darunter der nachgebildete Fall aus Lage 49, insgesamt laufen jetzt 80
Pruefungen ohne Netzzugriff und ohne Kosten.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-01 13:26:32 +00:00

227 Zeilen
8.9 KiB
Python
Originalformat Blame Verlauf

"""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")
_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",
"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
_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)))
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")
_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"])
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))
_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"))
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
_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",
round(usage_e.cost_usd, 6) == round(0.002 + STAAN_COST_PER_QUERY_USD, 6), usage_e.cost_usd)
neu()
plan_antwort["queries"] = []
_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 == [])
_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)
In neuem Issue referenzieren Git Blame ansehen Permalink kopieren