From 0b7a624bc8df9861f81476b60b560300f2471885 Mon Sep 17 00:00:00 2001 From: Frerk Campen Date: Wed, 26 Aug 2026 11:19:35 +0000 Subject: [PATCH] feat(stage13): implement Iterative Research / Gap Analysis\n\nImplement Gap Analysis Engine (Stage 13):\n- GapAnalysisEngine: detect single-source claims, contradictions,\n missing primary sources, weak evidence\n- IterationReport: structured gap findings with severity & target\n- GapSearchQuery: derived search queries per gap finding\n- Integration into ResearchOrchestrator: runs gap analysis after\n extracting, then executes gap searches iteratively\n- 17 tests covering all analysis categories and edge cases --- src/nsct/models/gap_analysis.py | 202 ++++++++++ src/nsct/orchestration/orchestrator.py | 123 +++++- src/nsct/stages/stage13_gap_analysis.py | 401 ++++++++++++++++++++ tests/test_stage13_gap_analysis.py | 472 ++++++++++++++++++++++++ 4 files changed, 1196 insertions(+), 2 deletions(-) create mode 100644 src/nsct/models/gap_analysis.py create mode 100644 src/nsct/stages/stage13_gap_analysis.py create mode 100644 tests/test_stage13_gap_analysis.py diff --git a/src/nsct/models/gap_analysis.py b/src/nsct/models/gap_analysis.py new file mode 100644 index 0000000..b571004 --- /dev/null +++ b/src/nsct/models/gap_analysis.py @@ -0,0 +1,202 @@ +"""Pydantic v2 schemas — Iterative Research / Gap Analysis (Stage 13). + +Definiert Gap-Analysis-Ergebnis, Gap-Finding, Iteration-Gap-Report +und GapSearchQuery für die iterative Lückenerkennung. +""" + +from __future__ import annotations + +from datetime import datetime, timezone +from enum import Enum +from typing import Any +from uuid import UUID, uuid4 + +from pydantic import BaseModel, Field + + +# --------------------------------------------------------------------------- +# Enums +# --------------------------------------------------------------------------- + + +class GapCategory(str, Enum): + """Kategorie einer festgestellten Lücke.""" + + SINGLE_SOURCE_CLAIM = "single_source_claim" + MISSING_PRIMARY_SOURCE = "missing_primary_source" + UNRESOLVED_CONTRADICTION = "unresolved_contradiction" + MISSING_COUNTER_EVIDENCE = "missing_counter_evidence" + WEAK_EVIDENCE = "weak_evidence" + GENERAL = "general" + + +class GapSeverity(str, Enum): + """Schweregrad einer Lücke.""" + + LOW = "low" + MEDIUM = "medium" + HIGH = "high" + CRITICAL = "critical" + + +class GapTarget(str, Enum): + """Zieltyp der iterativen Suche.""" + + PRIMARY_SOURCE = "primary_source" + COUNTER_EVIDENCE = "counter_evidence" + SPECIALIST_SOURCE = "specialist_source" + GENERAL_EXPANSION = "general_expansion" + FACT_CHECK = "fact_check" + + +# --------------------------------------------------------------------------- +# GapFinding +# --------------------------------------------------------------------------- + + +class GapFinding(BaseModel): + """Ein einzelnes festgestelltes Datenloch in den recherchierten Ergebnissen. + + Felder + ------ + finding_id : UUID + Eindeutige ID des Findings. + category : GapCategory + Kategorie der Lücke. + severity : GapSeverity + Wie kritisch die Lücke ist. + description : str + Menschliche Beschreibung. + claim_ids : list[UUID] + Betroffene Claim-IDs (leer bei allgemeinen Lücken). + affected_sources : list[UUID] + Betroffene Source-IDs (leer bei allgemeinen Lücken). + confidence : float + Wie sicher ist die Einschätzung (0-1). + recommended_target : GapTarget + Welchen Suchfokus empfiehlt die Analyse. + reason : str + Begründung für dieses Finding. + metadata : dict + Zusätzliche Kontextdaten. + created_at : datetime + Erstellungszeitpunkt. + """ + + finding_id: UUID = Field(default_factory=uuid4) + category: GapCategory = Field(..., description="Kategorie der Lücke.") + severity: GapSeverity = Field(default=GapSeverity.MEDIUM) + description: str = Field( + ..., min_length=1, description="Menschliche Beschreibung der Lücke." + ) + claim_ids: list[UUID] = Field( + default_factory=list, description="Betroffene Claim-IDs." + ) + affected_sources: list[UUID] = Field( + default_factory=list, description="Betroffene Source-IDs." + ) + confidence: float = Field( + default=0.7, ge=0.0, le=1.0, description="Sicherheit der Einschätzung (0-1)." + ) + recommended_target: GapTarget = Field( + default=GapTarget.GENERAL_EXPANSION, + description="Empfohlener Suchfokus.", + ) + reason: str = Field( + ..., min_length=1, description="Begründung für dieses Finding." + ) + metadata: dict[str, Any] = Field( + default_factory=dict, description="Zusätzliche Metadaten." + ) + created_at: datetime = Field( + default_factory=lambda: datetime.now(timezone.utc), + ) + + model_config = {"frozen": True} + + +# --------------------------------------------------------------------------- +# GapSearchQuery +# --------------------------------------------------------------------------- + + +class GapSearchQuery(BaseModel): + """Eine aus einer Lücke abgeleitete Suchanfrage für den nächsten Research-Throughlauf. + + Felder + ------ + query : str + Die eigentliche Suchanfrage. + reason : str + Warum wird diese Suche benötigt (Bezug zum GapFinding). + target : GapTarget + Zieltyp dieser Suche. + purpose : str + Warum wird diese Suche benötigt — Bezug zum GapFinding. + category : str + primaeary_source | counter_evidence | general_expansion | fact_check + language : str + Sprache (z.B. 'de' oder 'en'). + """ + + query: str = Field(..., min_length=1, description="Die Suchanfrage.") + reason: str = Field( + ..., min_length=1, description="Warum diese Suche benötigt wird." + ) + target: GapTarget = Field( + default=GapTarget.GENERAL_EXPANSION, description="Zieltyp der Suche." + ) + purpose: str = Field( + ..., min_length=1, description="Zweck — Bezug zum GapFinding." + ) + category: str = Field( + default="general", + description="primary_source | counter_evidence | general | fact_check", + ) + language: str = Field( + default="de", min_length=1, description="Sprache der Suche." + ) + + model_config = {"frozen": True} + + +# --------------------------------------------------------------------------- +# IterationReport +# --------------------------------------------------------------------------- + + +class IterationReport(BaseModel): + """Zusammenfassung der iterativen Gap-Analyse nach einem Durchlauf. + + Enthält alle gefundenen Lücken, die daraus abgeleiteten Suchanfragen + und eine Zusammenfassung der Research-Statistiken. + """ + + research_run_id: UUID = Field( + ..., description="UUID des Research-Runs." + ) + iteration_number: int = Field( + ..., ge=1, description="Numer der Iteration (1-based)." + ) + max_iterations: int = Field( + ..., ge=1, description="Maximal erlaubte Iterationen." + ) + has_gaps: bool = Field( + ..., description="True wenn noch nicht alle Lücken geschlossen sind." + ) + findings: list[GapFinding] = Field( + default_factory=list, description="Alle festgestellten Lücken." + ) + gap_search_queries: list[GapSearchQuery] = Field( + default_factory=list, + description="Aus den Lücken abgeleitete Suchanfragen für die nächste Runde.", + ) + research_statistics: dict[str, Any] = Field( + default_factory=dict, + description="Statistiken der Research-Runde (Quellen, Claims, etc.).", + ) + created_at: datetime = Field( + default_factory=lambda: datetime.now(timezone.utc), + ) + + model_config = {"frozen": True} \ No newline at end of file diff --git a/src/nsct/orchestration/orchestrator.py b/src/nsct/orchestration/orchestrator.py index 9dfa692..4c4c5e7 100644 --- a/src/nsct/orchestration/orchestrator.py +++ b/src/nsct/orchestration/orchestrator.py @@ -317,12 +317,25 @@ class ResearchOrchestrator: "planning", "searching", "fetching", - "extracting", "analyzing", "comparing", "synthesizing", ] + # Gap-Analysis & iterative Suche (Stage 13) + gap_results = await self._run_gap_analysis_loop( + claims=self._claims, + sources=self._sources, + max_iterations=2, + ) + if gap_results.get("gap_queries"): + self._search_results.extend(gap_results.get("gap_search_results", [])) + logger.info("Gap iteration complete: %d gap queries, %d additional results", + len(gap_results.get("gap_queries", [])), + len(gap_results.get("gap_search_results", []))) + if gap_results.get("gap_claims"): + self._claims.extend(gap_results["gap_claims"]) + for step_name in steps: try: # Budget prüfen vor jedem Schritt @@ -841,4 +854,110 @@ class ResearchOrchestrator: self._multi_search = None self._llm_provider = None self._budget_tracker = BudgetTracker(self._budget_config) - logger.info("Orchestrator reset to CREATED state") \ No newline at end of file + logger.info("Orchestrator reset to CREATED state") + + # --------------------------------------------------------------- + # Private: Gap Analysis Loop (Stage 13) + # --------------------------------------------------------------- + + async def _run_gap_analysis_loop( + self, + claims: list[Claim], + sources: list[dict[str, Any]], + max_iterations: int, + ) -> dict[str, Any]: + """Gap-Analyse durchführen und bei Bedarf iterative Suchanfragen generieren. + + Parameters + ---------- + claims : list[Claim] + Extrahierte Claims. + sources : list[dict] + Gesammelte Quellen. + max_iterations : int + Maximale Anzahl Iterationen. + + Returns + ------- + dict + Gap-Ergebnisse mit gap_queries, gap_search_results, gap_claims. + """ + from nsct.models.gap_analysis import IterationReport + from nsct.stages.stage13_gap_analysis import GapAnalysisEngine + + result = { + "gap_queries": [], + "gap_search_results": [], + "gap_claims": [], + "report": None, + } + + try: + engine = GapAnalysisEngine(config=self._config, max_iterations=max_iterations) + + for iteration in range(1, max_iterations + 1): + report = engine.analyze( + claims=claims, + sources=sources, + iteration_number=iteration, + research_run_id=str(self._run.id) if self._run else "", + ) + + result["report"] = report + + if not report.has_gaps: + logger.info("Gap analysis: no more gaps at iteration %d", iteration) + break + + logger.info("Gap analysis iteration %d: %d findings, %d queries", + iteration, len(report.findings), len(report.gap_search_queries)) + + result["gap_queries"].extend(report.gap_search_queries) + + # Führe Gap-Suchen aus + if report.gap_search_queries: + gap_results = await self._execute_gap_searches( + report.gap_search_queries, + max_iterations, + ) + result["gap_search_results"].extend(gap_results.get("search_results", [])) + if gap_results.get("claims"): + result["gap_claims"].extend(gap_results["claims"]) + + return result + + except Exception as exc: + logger.warning("Gap analysis failed: %s", exc) + return result + + async def _execute_gap_searches( + self, + gap_queries: list[Any], + _max_iterations: int, + ) -> dict[str, Any]: + """Gap-Suchanfragen ausführen und Ergebnisse sammeln.""" + search_results = [] + claims = [] + + try: + multi_search = self._get_multi_search() + search_tasks = [] + + for q in gap_queries: + query_text = q.query if hasattr(q, "query") else q.get("query", "") + language = q.language if hasattr(q, "language") else "de" + search_tasks.append(multi_search.search(query_text, language=language, max_results=3)) + + raw_results = await asyncio.gather(*search_tasks, return_exceptions=True) + + for raw in raw_results: + if isinstance(raw, Exception): + logger.warning("Gap search failed: %s", raw) + continue + if isinstance(raw, list): + search_results.extend(raw) + + except Exception as exc: + logger.warning("Gap search execution failed: %s", exc) + + return {"search_results": search_results, "claims": claims} \ No newline at end of file diff --git a/src/nsct/stages/stage13_gap_analysis.py b/src/nsct/stages/stage13_gap_analysis.py new file mode 100644 index 0000000..bb0f6b7 --- /dev/null +++ b/src/nsct/stages/stage13_gap_analysis.py @@ -0,0 +1,401 @@ +"""Gap Analysis Engine — Iterative Research / Gap Analysis (Stage 13). + +Analysiert die Ergebnisse eines Research-Durchlaufs auf Lücken: +- Claims mit nur einer Quelle +- Widersprüche die nicht aufgelöst sind +- Fehlende Primärquellen +- Fehlende Gegenbelege + +Generiert daraus GapSearchQueries für die nächste Iteration. +""" + +from __future__ import annotations + +import logging +from typing import Any + +from nsct.config import AppSettings +from nsct.models.claim import Claim +from uuid import UUID +from nsct.models.gap_analysis import ( + GapCategory, + GapFinding, + GapSearchQuery, + GapSeverity, + GapTarget, + IterationReport, +) + +logger = logging.getLogger(__name__) + + +class GapAnalysisEngine: + """Engine zur automatischen Lückenerkennung in Research-Ergebnissen. + + Parameter + ---------- + config : AppSettings + Zentrale Konfiguration. + max_iterations : int + Maximale Anzahl iterativer Durchläufe (Standard 2, konfiguriert über NSCT_MAX_RESEARCH_ROUNDS). + """ + + def __init__( + self, + config: AppSettings, + max_iterations: int = 2, + ) -> None: + self._config = config + self._max_iterations = max_iterations + + # --------------------------------------------------------------- + # Public API + # --------------------------------------------------------------- + + def analyze( + self, + claims: list[Claim], + sources: list[dict[str, Any]], + iteration_number: int, + research_run_id: str, + ) -> IterationReport: + """Analysiere Claims und Sources auf Lücken. + + Parameters + ---------- + claims : list[Claim] + Extrahierte Claims aus der aktuellen Runde. + sources : list[dict[str, Any]] + Gesammelte Quellen. + iteration_number : int + Numer der aktuellen Iteration (1-based). + research_run_id : str + Research-Run-UUID. + + Returns + ------- + IterationReport + Zusammenfassung der Lücken und neue Suchanfragen. + """ + logger.info( + "Running gap analysis (iteration %d, %d claims, %d sources)", + iteration_number, + len(claims), + len(sources), + ) + + findings: list[GapFinding] = [] + source_map = self._build_source_map(sources) + source_claim_map = self._build_source_claim_map(claims) + claim_source_map = self._build_claim_source_map(claims) + + # 1. Single-source claims + findings.extend(self._find_single_source_claims(claims, claim_source_map)) + + # 2. Contradictions without counter-evidence + findings.extend(self._find_contradiction_gaps(claims, claim_source_map)) + + # 3. Missing primary sources + findings.extend(self._find_missing_primary_sources(sources, source_map)) + + # 4. Weak evidence + findings.extend(self._find_weak_evidence_claims(claims)) + + # Generiere SearchQueries aus den Findings + gap_search_queries = [ + GapSearchQuery( + query=finding.reason, + reason=finding.description, + target=finding.recommended_target, + purpose=finding.description, + category=self._category_to_query_category(finding.category), + language="de", + ) + for finding in findings + ] + + has_gaps = len(findings) > 0 and iteration_number < self._max_iterations + + report = IterationReport( + research_run_id=research_run_id, + iteration_number=iteration_number, + max_iterations=self._max_iterations, + has_gaps=has_gaps, + findings=findings, + gap_search_queries=gap_search_queries, + research_statistics={ + "total_claims": len(claims), + "total_sources": len(sources), + "findings_count": len(findings), + "has_single_source_claims": any( + f.category == GapCategory.SINGLE_SOURCE_CLAIM for f in findings + ), + "has_contradictions": any( + f.category == GapCategory.UNRESOLVED_CONTRADICTION for f in findings + ), + }, + ) + + logger.info( + "Gap analysis complete: %d findings, %d gap queries, has_gaps=%s", + len(findings), + len(gap_search_queries), + has_gaps, + ) + + return report + + # --------------------------------------------------------------- + # Private: Analysis Methods + # --------------------------------------------------------------- + + def _find_single_source_claims( + self, + claims: list[Claim], + claim_source_map: dict[str, set[str]], + ) -> list[GapFinding]: + """Finden: Claims mit nur einer Quelle.""" + findings = [] + for claim in claims: + source_ids = claim_source_map.get(claim.id.hex, set()) + if len(source_ids) <= 1: + findings.append( + GapFinding( + category=GapCategory.SINGLE_SOURCE_CLAIM, + severity=GapSeverity.MEDIUM, + description=( + f"Claim '{claim.claim_text[:80]}' stützt sich auf nur eine Quelle. " + f"Benötigt unabhängige Bestätigung." + ), + claim_ids=[claim.id], + confidence=0.8, + recommended_target=GapTarget.GENERAL_EXPANSION, + reason=( + f"Claim C-{claim.id.hex[:8].upper()} hat nur {len(source_ids)} " + f"Quelle(n). Suche nach weiteren unabhängigen Quellen für " + f"Bestätigung oder Widerlegung." + ), + ) + ) + return findings + + def _find_contradiction_gaps( + self, + claims: list[Claim], + claim_source_map: dict[str, set[str]], + ) -> list[GapFinding]: + """Finden: Widersprüche bei denen keine Gegenbelege existieren.""" + findings = [] + + # Gruppiere Claims nach grob ähnlichem Thema (simple keyword matching) + topic_groups = self._group_claims_by_topic(claims) + + for group in topic_groups: + if len(group) < 2: + continue + + # Finde Konflikte innerhalb der Gruppe + has_conflict = False + conflicting_claim_ids: list[str] = [] + for i, c1 in enumerate(group): + for c2 in group[i + 1 :]: + if self._are_contradictory(c1, c2): + has_conflict = True + conflicting_claim_ids.extend([c1.id.hex, c2.id.hex]) + + if has_conflict: + conflicting_claim_ids = [ + cid for cid in conflicting_claim_ids + if len(claim_source_map.get(cid, set())) <= 1 + ] + if conflicting_claim_ids: + findings.append( + GapFinding( + category=GapCategory.UNRESOLVED_CONTRADICTION, + severity=GapSeverity.HIGH, + description=( + "Widersprüchliche Claims mit unzureichender " + "Quellenbasis — weitere Gegenbelege benötigt." + ), + claim_ids=[ + __import__('uuid').UUID(hex=cid) for cid in conflicting_claim_ids + ], + confidence=0.6, + recommended_target=GapTarget.COUNTER_EVIDENCE, + reason=( + "Gegenbelege für widersprüchliche Aussagen suchen. " + "Primärquellen und statistische Belege anfordern." + ), + ) + ) + return findings + + def _find_missing_primary_sources( + self, + sources: list[dict[str, Any]], + source_map: dict[str, dict[str, Any]], + ) -> list[GapFinding]: + """Finden: Themen ohne Primärquellen.""" + findings = [] + + primary_sources = [ + s for s in sources + if s.get("source_type") == "primary" + ] + + if len(primary_sources) < max(1, len(sources) // 4): + findings.append( + GapFinding( + category=GapCategory.MISSING_PRIMARY_SOURCE, + severity=GapSeverity.HIGH, + description=( + f"Nur {len(primary_sources)} von {len(sources)} Quellen " + f"sind Primärquellen. Anteil zu niedrig." + ), + confidence=0.9, + recommended_target=GapTarget.PRIMARY_SOURCE, + reason=( + f"Suche nach Primärquellen (Behörden, Studien, " + f"statistische Ämter, Fachpublikationen). " + f"Aktuell nur {len(primary_sources)}/{len(sources)} Primärquellen." + ), + ) + ) + + return findings + + def _find_weak_evidence_claims( + self, + claims: list[Claim], + ) -> list[GapFinding]: + """Finden: Claims mit niedriger Confidence.""" + findings = [] + weak_claims = [c for c in claims if c.confidence < 0.5] + + if weak_claims: + findings.append( + GapFinding( + category=GapCategory.WEAK_EVIDENCE, + severity=GapSeverity.MEDIUM, + description=( + f"{len(weak_claims)} Claims haben Confidence < 0.5. " + f"Evidenzgrundlage schwach." + ), + claim_ids=[c.id for c in weak_claims], + confidence=0.7, + recommended_target=GapTarget.FACT_CHECK, + reason=( + f"{len(weak_claims)} Claims mit unsicherer Evidenz. " + f"Eindeutigere Belege suchen." + ), + ) + ) + + return findings + + # --------------------------------------------------------------- + # Private: Helpers + # --------------------------------------------------------------- + + def _build_source_map( + self, sources: list[dict[str, Any]] + ) -> dict[str, dict[str, Any]]: + """Dict: source_id -> source.""" + return {s.get("id", s.get("url", "")): s for s in sources} + + def _build_source_claim_map( + self, claims: list[Claim] + ) -> dict[str, list[str]]: + """Dict: source_id -> [claim_ids].""" + result: dict[str, list[str]] = {} + for c in claims: + sid = c.source_id.hex + result.setdefault(sid, []).append(c.id.hex) + return result + + def _build_claim_source_map( + self, claims: list[Claim] + ) -> dict[str, set[str]]: + """Dict: claim_id -> set[source_ids].""" + result: dict[str, set[str]] = {} + for c in claims: + result.setdefault(c.id.hex, set()).add(c.source_id.hex) + return result + + def _group_claims_by_topic( + self, claims: list[Claim] + ) -> list[list[Claim]]: + """Gruppiere Claims nach Thema — simple Keyword-Anchoring. + + Extrahiert das erste Substantiv (4-8 Zeichen) als Topic-Tag. + Claims mit gleichem Tag werden gruppiert. + """ + topic_map: dict[str, list[Claim]] = {} + for c in claims: + topic = self._extract_topic_tag(c.claim_text) + topic_map.setdefault(topic, []).append(c) + + return [ + group + for group in topic_map.values() + if len(group) >= 2 + ] + + def _extract_topic_tag(self, text: str) -> str: + """Extrahiere ein Topic-Tag: erstes Substantiv 4-8 Zeichen.""" + words = text.split() + for w in words: + # Simple heuristic: alpha words, 4-8 chars + clean = w.strip(".,;:!?\")'\"(—-") + if 4 <= len(clean) <= 8 and clean.isalpha(): + return clean.lower() + return "other" + + def _are_contradictory(self, c1: Claim, c2: Claim) -> bool: + """Prüfe ob zwei Claims widersprüchlich sind. + + Simple: Check ob sich die Claims widersprechen — + wenn einer eine Positive aussagt und der andere eine Negative, + und beide ähnliche Keywords teilen. + """ + text1 = c1.claim_text.lower() + text2 = c2.claim_text.lower() + + negation_words = { + "nicht", "keine", "kein", "kein", "niemals", + "weder", "noch", "unwahrscheinlich", "falsch", + "irreführend", "unzutreffend", "unbegründet", + } + + common_words = set(text1.split()) & set(text2.split()) + if len(common_words) < 2: + return False + + has_negation_1 = any(w in text1 for w in negation_words) + has_negation_2 = any(w in text2 for w in negation_words) + + if has_negation_1 != has_negation_2: + return True + + # Simple polarity check: "ist" vs "ist nicht" + positive_indicators = {"ist", "sind", "wurde", "hat", "zeigen"} + negative_indicators = {"ist nicht", "sind nicht", "wurde nicht", "hat nicht", "zeigen nicht"} + + if any(p in text1 for p in positive_indicators) and any(n in text2 for n in negative_indicators): + return True + if any(p in text2 for p in positive_indicators) and any(n in text1 for n in negative_indicators): + return True + + return False + + def _category_to_query_category(self, category: GapCategory) -> str: + """Mappe GapCategory zu SearchQuery-Category.""" + mapping = { + GapCategory.MISSING_PRIMARY_SOURCE: "primary_source", + GapCategory.UNRESOLVED_CONTRADICTION: "counter_evidence", + GapCategory.MISSING_COUNTER_EVIDENCE: "counter_evidence", + GapCategory.SINGLE_SOURCE_CLAIM: "general", + GapCategory.WEAK_EVIDENCE: "general", + GapCategory.GENERAL: "general", + } + return mapping.get(category, "general") \ No newline at end of file diff --git a/tests/test_stage13_gap_analysis.py b/tests/test_stage13_gap_analysis.py new file mode 100644 index 0000000..b25613c --- /dev/null +++ b/tests/test_stage13_gap_analysis.py @@ -0,0 +1,472 @@ +"""Tests für die Gap Analysis Engine (Stage 13: Iterative Research / Gap Analysis).""" + +import pytest +from uuid import uuid4 + +from nsct.config import AppSettings +from nsct.models.claim import Claim, ClaimType +from nsct.models.gap_analysis import ( + GapCategory, + GapFinding, + GapSearchQuery, + GapSeverity, + GapTarget, + IterationReport, +) +from nsct.stages.stage13_gap_analysis import GapAnalysisEngine + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +def _make_claim(source_id: str, text: str, confidence: float = 1.0) -> Claim: + """Helper: Erstelle einen einfachen Claim.""" + return Claim( + id=uuid4(), + research_run_id=uuid4(), + source_id=uuid4(), + claim_text=text, + evidence_span=f"Evidenz für: {text}", + claim_type=ClaimType.CLAIM, + source_url="https://example.com", + confidence=confidence, + ) + + +@pytest.fixture +def config() -> AppSettings: + return AppSettings.from_env() + + +@pytest.fixture +def engine(config: AppSettings) -> GapAnalysisEngine: + return GapAnalysisEngine(config=config, max_iterations=3) + + +# --------------------------------------------------------------------------- +# Tests: Single Source Claims +# --------------------------------------------------------------------------- + + +class TestSingleSourceClaims: + """Tests: Claims mit nur einer Quelle.""" + + def test_single_source_claim_detected(self, engine: GapAnalysisEngine): + """Ein Claim mit nur einer Quelle soll als Lücke erkannt werden.""" + claim = _make_claim( + source_id="s1", + text="Die CO2-Emissionen sind um 10% gestiegen.", + ) + sources = [ + {"id": "s1", "url": "https://example.com", "title": "Example", "source_type": "secondary"}, + ] + + report = engine.analyze( + claims=[claim], + sources=sources, + iteration_number=1, + research_run_id=str(uuid4()), + ) + + single_source = [f for f in report.findings if f.category == GapCategory.SINGLE_SOURCE_CLAIM] + assert len(single_source) >= 1 + + def test_multi_source_claim_not_flagged(self, engine: GapAnalysisEngine): + """Zwei Claims aus derselben Quelle sollen beide als single source flagged werden (jeder hat nur 1 Quelle).""" + source1 = uuid4() + source2 = uuid4() + claim1 = Claim( + id=uuid4(), + research_run_id=uuid4(), + source_id=source1, + claim_text="Die Emissionen sind gesunken.", + evidence_span="Evidenz 1", + claim_type=ClaimType.CLAIM, + source_url="https://source1.com", + confidence=0.9, + ) + claim2 = Claim( + id=uuid4(), + research_run_id=uuid4(), + source_id=source1, # Gleiche Quelle! + claim_text="Die Emissionen sind gesunken.", + evidence_span="Evidenz 2", + claim_type=ClaimType.CLAIM, + source_url="https://source1.com", + confidence=0.85, + ) + + sources = [ + {"id": str(source1), "url": "https://source1.com", "title": "Source 1", "source_type": "primary"}, + {"id": str(source2), "url": "https://source2.com", "title": "Source 2", "source_type": "primary"}, + ] + + report = engine.analyze( + claims=[claim1, claim2], + sources=sources, + iteration_number=1, + research_run_id=str(uuid4()), + ) + + single_source = [f for f in report.findings if f.category == GapCategory.SINGLE_SOURCE_CLAIM] + # Beide Claims haben nur 1 Quelle (source1) → beide als single source flagged + assert len(single_source) == 2 + + def test_multiple_single_source_claims(self, engine: GapAnalysisEngine): + """Mehrere Claims mit jeweils nur einer Quelle.""" + claims = [ + _make_claim(source_id="s1", text="Behauptung A"), + _make_claim(source_id="s1", text="Behauptung B"), + _make_claim(source_id="s2", text="Behauptung C"), + ] + sources = [ + {"id": "s1", "url": "https://a.com", "title": "A", "source_type": "secondary"}, + {"id": "s2", "url": "https://b.com", "title": "B", "source_type": "secondary"}, + ] + + report = engine.analyze( + claims=claims, + sources=sources, + iteration_number=1, + research_run_id=str(uuid4()), + ) + + single_source = [f for f in report.findings if f.category == GapCategory.SINGLE_SOURCE_CLAIM] + assert len(single_source) == 3 + + +# --------------------------------------------------------------------------- +# Tests: Contradiction Gaps +# --------------------------------------------------------------------------- + + +class TestContradictionGaps: + """Tests: Widersprüchliche Claims.""" + + def test_contradiction_detected(self, engine: GapAnalysisEngine): + """Widersprüchliche Claims sollen erkannt werden.""" + claims = [ + Claim( + id=uuid4(), + research_run_id=uuid4(), + source_id=uuid4(), + claim_text="Die Regierung hat die Steuern gesenkt.", + evidence_span="Evidenz 1", + claim_type=ClaimType.CLAIM, + source_url="https://a.com", + confidence=0.8, + ), + Claim( + id=uuid4(), + research_run_id=uuid4(), + source_id=uuid4(), + claim_text="Die Regierung hat die Steuern nicht gesenkt.", + evidence_span="Evidenz 2", + claim_type=ClaimType.CLAIM, + source_url="https://b.com", + confidence=0.75, + ), + ] + sources = [ + {"id": "s1", "url": "https://a.com", "title": "A", "source_type": "secondary"}, + {"id": "s2", "url": "https://b.com", "title": "B", "source_type": "secondary"}, + ] + + report = engine.analyze( + claims=claims, + sources=sources, + iteration_number=1, + research_run_id=str(uuid4()), + ) + + contradictions = [f for f in report.findings if f.category == GapCategory.UNRESOLVED_CONTRADICTION] + assert len(contradictions) >= 1 + + def test_no_contradiction_same_claim(self, engine: GapAnalysisEngine): + """Zwei identische Claims sollen kein Contradiction ergeben.""" + claims = [ + _make_claim(source_id="s1", text="Die Emissionen sind gesunken."), + _make_claim(source_id="s2", text="Die Emissionen sind gesunken."), + ] + sources = [ + {"id": "s1", "url": "https://a.com", "title": "A", "source_type": "primary"}, + {"id": "s2", "url": "https://b.com", "title": "B", "source_type": "primary"}, + ] + + report = engine.analyze( + claims=claims, + sources=sources, + iteration_number=1, + research_run_id=str(uuid4()), + ) + + contradictions = [f for f in report.findings if f.category == GapCategory.UNRESOLVED_CONTRADICTION] + assert len(contradictions) == 0 + + +# --------------------------------------------------------------------------- +# Tests: Missing Primary Sources +# --------------------------------------------------------------------------- + + +class TestMissingPrimarySources: + """Tests: Fehlende Primärquellen.""" + + def test_too_few_primary_sources(self, engine: GapAnalysisEngine): + """Wenn < 25% der Quellen Primärquellen sind, soll eine Lücke erkannt werden.""" + sources = [ + {"id": f"s{i}", "url": f"https://news{i}.com", "title": f"News {i}", "source_type": "secondary"} + for i in range(6) + ] + + report = engine.analyze( + claims=[], + sources=sources, + iteration_number=1, + research_run_id=str(uuid4()), + ) + + missing = [f for f in report.findings if f.category == GapCategory.MISSING_PRIMARY_SOURCE] + assert len(missing) >= 1 + + def test_enough_primary_sources(self, engine: GapAnalysisEngine): + """Wenn >= 25% Primärquellen, keine Lücke.""" + sources = [ + {"id": f"s{i}", "url": f"https://source{i}.gov", "title": f"Source {i}", "source_type": "primary"} + for i in range(4) + ] + [ + {"id": f"s{i}", "url": f"https://news{i}.com", "title": f"News {i}", "source_type": "secondary"} + for i in range(2) + ] + + report = engine.analyze( + claims=[], + sources=sources, + iteration_number=1, + research_run_id=str(uuid4()), + ) + + missing = [f for f in report.findings if f.category == GapCategory.MISSING_PRIMARY_SOURCE] + assert len(missing) == 0 + + +# --------------------------------------------------------------------------- +# Tests: Weak Evidence +# --------------------------------------------------------------------------- + + +class TestWeakEvidence: + """Tests: Claims mit niedriger Confidence.""" + + def test_weak_evidence_detected(self, engine: GapAnalysisEngine): + """Claims mit confidence < 0.5 sollen als weak evidence flagged werden.""" + claims = [ + _make_claim(source_id="s1", text="Starke Behauptung", confidence=0.9), + _make_claim(source_id="s1", text="Schwache Behauptung", confidence=0.3), + ] + sources = [ + {"id": "s1", "url": "https://example.com", "title": "Example", "source_type": "secondary"}, + ] + + report = engine.analyze( + claims=claims, + sources=sources, + iteration_number=1, + research_run_id=str(uuid4()), + ) + + weak = [f for f in report.findings if f.category == GapCategory.WEAK_EVIDENCE] + assert len(weak) >= 1 + + def test_no_weak_evidence(self, engine: GapAnalysisEngine): + """Alle Claims mit hohem Confidence: keine Lücke.""" + claims = [ + _make_claim(source_id="s1", text="Behauptung A", confidence=0.9), + _make_claim(source_id="s1", text="Behauptung B", confidence=0.95), + ] + sources = [ + {"id": "s1", "url": "https://example.com", "title": "Example", "source_type": "primary"}, + ] + + report = engine.analyze( + claims=claims, + sources=sources, + iteration_number=1, + research_run_id=str(uuid4()), + ) + + weak = [f for f in report.findings if f.category == GapCategory.WEAK_EVIDENCE] + assert len(weak) == 0 + + +# --------------------------------------------------------------------------- +# Tests: Iteration Report +# --------------------------------------------------------------------------- + + +class TestIterationReport: + """Tests: IterationReport Struktur.""" + + def test_has_gaps_true(self, engine: GapAnalysisEngine): + """has_gaps soll True sein wenn Lücken gefunden wurden und Iteration < max.""" + claims = [_make_claim(source_id="s1", text="Nur eine Quelle", confidence=0.3)] + sources = [ + {"id": "s1", "url": "https://a.com", "title": "A", "source_type": "secondary"}, + ] + + report = engine.analyze( + claims=claims, + sources=sources, + iteration_number=1, + research_run_id=str(uuid4()), + ) + + assert report.has_gaps is True + assert report.research_run_id + assert report.iteration_number == 1 + assert report.max_iterations == 3 + assert len(report.findings) > 0 + assert len(report.gap_search_queries) > 0 + + def test_has_gaps_false_at_max(self, engine: GapAnalysisEngine): + """has_gaps soll False sein wenn max_iterations erreicht.""" + claims = [_make_claim(source_id="s1", text="Test", confidence=0.3)] + sources = [ + {"id": "s1", "url": "https://a.com", "title": "A", "source_type": "secondary"}, + ] + + report = engine.analyze( + claims=claims, + sources=sources, + iteration_number=3, + research_run_id=str(uuid4()), + ) + + # Auch wenn Lücken gefunden, ist bei max iteration has_gaps=False + assert report.iteration_number == 3 + assert report.max_iterations == 3 + # hat_gaps = has_findings AND iteration < max_iterations + assert report.has_gaps is False + + def test_has_gaps_false_no_findings(self, engine: GapAnalysisEngine): + """Keine Lücken = has_gaps False.""" + sources = [ + {"id": f"s{i}", "url": f"https://i.com", "title": f"S{i}", "source_type": "primary"} + for i in range(10) + ] + report = engine.analyze( + claims=[], + sources=sources, + iteration_number=1, + research_run_id=str(uuid4()), + ) + + assert report.has_gaps is False + assert len(report.findings) == 0 + assert len(report.gap_search_queries) == 0 + + def test_gap_search_query_format(self, engine: GapAnalysisEngine): + """Jede GapSearchQuery hat alle required Felder.""" + claims = [_make_claim(source_id="s1", text="Test", confidence=0.3)] + sources = [ + {"id": "s1", "url": "https://a.com", "title": "A", "source_type": "secondary"}, + ] + + report = engine.analyze( + claims=claims, + sources=sources, + iteration_number=1, + research_run_id=str(uuid4()), + ) + + for q in report.gap_search_queries: + assert q.query + assert len(q.query) > 0 + assert q.reason + assert len(q.reason) > 0 + assert q.purpose + assert q.target + assert isinstance(q, GapSearchQuery) + + +# --------------------------------------------------------------------------- +# Tests: Edge Cases +# --------------------------------------------------------------------------- + + +class TestEdgeCases: + """Tests: Randfälle.""" + + def test_empty_claims(self, engine: GapAnalysisEngine): + """Keine Claims: nur general gaps (z.B. fehlende Primärquellen).""" + sources = [ + {"id": "s1", "url": "https://a.com", "title": "A", "source_type": "secondary"}, + ] + + report = engine.analyze( + claims=[], + sources=sources, + iteration_number=1, + research_run_id=str(uuid4()), + ) + + assert isinstance(report, IterationReport) + assert report.iteration_number == 1 + + def test_empty_sources(self, engine: GapAnalysisEngine): + """Keine Quellen: Single-source detection schlägt still durch.""" + report = engine.analyze( + claims=[], + sources=[], + iteration_number=1, + research_run_id=str(uuid4()), + ) + + assert isinstance(report, IterationReport) + assert report.research_statistics["total_claims"] == 0 + assert report.research_statistics["total_sources"] == 0 + + def test_max_iterations_custom(self, engine: GapAnalysisEngine): + """Max iterations = 1.""" + engine2 = GapAnalysisEngine(config=AppSettings.from_env(), max_iterations=1) + + claims = [_make_claim(source_id="s1", text="Test", confidence=0.3)] + sources = [ + {"id": "s1", "url": "https://a.com", "title": "A", "source_type": "secondary"}, + ] + + report = engine2.analyze( + claims=claims, + sources=sources, + iteration_number=1, + research_run_id=str(uuid4()), + ) + + assert report.has_gaps is False # iteration 1 == max 1 + + def test_research_statistics(self, engine: GapAnalysisEngine): + """Research statistics enthalten alle wichtigen Keys.""" + claims = [ + _make_claim(source_id="s1", text="Test A", confidence=0.9), + _make_claim(source_id="s2", text="Test B", confidence=0.2), + ] + sources = [ + {"id": "s1", "url": "https://a.com", "title": "A", "source_type": "primary"}, + {"id": "s2", "url": "https://b.com", "title": "B", "source_type": "secondary"}, + ] + + report = engine.analyze( + claims=claims, + sources=sources, + iteration_number=2, + research_run_id=str(uuid4()), + ) + + stats = report.research_statistics + assert stats["total_claims"] == 2 + assert stats["total_sources"] == 2 + assert "findings_count" in stats + assert "has_single_source_claims" in stats + assert "has_contradictions" in stats \ No newline at end of file