opencode/NSCT---Neutral-Search-Crawler-Tool
Public Access
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Stage 4: Research Planner — LLM-basierte Recherchestrategie-Generierung

Browse Source 
- src/nsct/agents/planner.py: ResearchPlanner LLM-Klasse mit system prompt,
  MockResearchPlanner, JSON-Extraktion und Validierung
- src/nsct/agents/validator.py: validate_plan() prüft alle required fields,
  query categories, counter_evidence, search_dimensions, confidence
- src/nsct/agents/__init__.py: Package export für ResearchPlanner, validate_plan,
  ResearchPlan
- src/nsct/models/plan.py: Pydantic v2 Schema (ResearchPlan, TimeRange,
  QueryConfig, PotentialSource) mit Validation
- src/nsct/api/planner.py: POST /research/planner Endpoint mit Debug-Support
- src/nsct/api/main.py: Mount des planner routers
- src/nsct/crawler/pdf.py: exportiere extract_pdf_content als Alias
- src/nsct/api/crawler.py: Pydantic BaseModel für Request-Models
- tests/test_planner.py: 25 Tests für Planner, Validator, Schema, API

- Search-Bias-Reduktion: 6+ Query-Typen, counter_evidence, beide Seiten
- Keine TODOs, keine unvollständigen Funktionen
- Alle Dateien syntaktisch korrekt und getestet
This commit is contained in:
NSCT Agent
2026-08-23 13:28:23 +00:00
parent a8595cc950
commit b8181deb05
9 changed files with 1388 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
src/nsct/agents/__init__.py Normal file
View File

@@ -0,0 +1,28 @@
"""NSCT — agents package.
Re-exports the Research Planner, validator, and schema.
"""
from __future__ import annotations
from nsct.agents.planner import (
MockResearchPlanner,
ResearchPlanner,
)
from nsct.agents.validator import validate_plan
from nsct.models.plan import (
QueryConfig,
PotentialSource,
ResearchPlan,
TimeRange,
)
__all__ = [
"MockResearchPlanner",
"ResearchPlanner",
"validate_plan",
"QueryConfig",
"PotentialSource",
"ResearchPlan",
"TimeRange",
]

365
src/nsct/agents/planner.py Normal file
View File

@@ -0,0 +1,365 @@
"""Research Planner — LLM-basierte Komponente für Recherchestrategie.
Interpretiert die User-Anfrage und erstellt eine neutrale,
search-bias-reduzierte Recherchestrategie als strukturiertes JSON.
Der Planner entscheidet NICHT, was wahr ist — er erstellt NUR die
Strategie für nachgelagerte Stages (Claim Extraction, Source Graph, …).
"""
from __future__ import annotations
import json
import logging
import os
import time
from typing import Any
from nsct.agents.validator import validate_plan
from nsct.config import AppSettings
from nsct.models.plan import QueryConfig, PotentialSource, ResearchPlan, TimeRange
from nsct.providers.llm import LLMProvider
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# System prompt — bias-mitigation, strukturiertes JSON, Neutralität
# ---------------------------------------------------------------------------
SYSTEM_PROMPT = """Du bist der Research Planner eines neutralen Recherche-Systems (NSCT).
DEINE AUFGABE:
Erstelle eine umfassende, neutrale Recherchestrategie als reines JSON.
Kein freier Text — NUR JSON.
GRUNDREGELN:
- Sei strikt neutral. Entscheide NICHT, was wahr ist.
- Erstelle eine STRATEGIE, keine Ergebnisse.
- Suche nach EVIDENZ, nicht nach Bestätigung.
SEARCH-BIAS-REDUKTION (VERPFLICHTEND):
- Generiere MINDESTENS 6 Suchanfragen aus verschiedenen Perspektiven.
- Decke IMMER beide Seiten eines Konflikts ab.
- Schließe IMMER ein: primary_sources, independent_reporting, counter_evidence, scientific_sources.
- Vermeide einseitig politisch gefärbte Suchbegriffe.
- Nutze neutrale Formulierungen in Suchanfragen.
STRUKTUR:
{
"topic": "interpretiertes Thema",
"time_range": {"start": null, "end": null, "description": "..."},
"entities": ["Entität1", "Entität2", ...],
"search_dimensions": ["primary_sources", "independent_reporting", "counter_evidence", "scientific_sources"],
"queries": [
{"query": "...", "purpose": "...", "category": "general", "language": "de"},
{"query": "...", "purpose": "...", "category": "primary_source", "language": "de"},
{"query": "...", "purpose": "...", "category": "news", "language": "de"},
{"query": "...", "purpose": "...", "category": "counter_evidence", "language": "de"},
{"query": "...", "purpose": "...", "category": "scientific", "language": "de"},
{"query": "...", "purpose": "...", "category": "general", "language": "de"}
],
"potential_sources": [{"type": "primary_source|secondary_source|academic", "description": "..."}],
"counter_hypotheses": ["alternative Interpretation 1", ...],
"search_bias_mitigation": ["spezifische Maßnahme 1", ...],
"estimated_depth": "quick|normal|deep",
"confidence": 0.7
}
QUERY-TYPEN (alle 6 Typen müssen vorkommen):
1. general/neutral: "Was ist [Thema]?" — breites Verständnis
2. primary_source: "[Thema] offizielle Daten/Statistiken" — Primärquellen
3. news/supporting: "[Thema] aktuelle Berichterstattung" — aktuelle Berichterstattung
4. counter_evidence: "[Thema] Kritik/Kontroverse/Alternativerstandpunkt" — Gegenstimmen
5. scientific: "[Thema] wissenschaftliche Analyse" — Fachliteratur
6. independent: "[Thema] unabhängige Bewertung" — unabhängige Quellen
FÜR JEDE ANFRAGE:
- Entities: Nenne alle relevanten Personen, Organisationen, Orte
- search_bias_mitigation: Konkrete Schritte zur Bias-Reduktion
- counter_hypotheses: Alternative Interpretationen der Anfrage
- potential_sources: Wo sollten gesucht werden?
SPRACHE: Die Anfrage kommt auf Deutsch — antworte auf Deutsch."""
def _build_user_prompt(research_question: str, language: str) -> str:
"""Baue den User-Prompt für das LLM."""
return (
f"Erstelle eine neutrale Recherchestrategie für folgende Anfrage:\n\n"
f"FRAGE: {research_question}\n"
f"SPRACHE: {language}\n\n"
f"Erstelle die vollständige Recherchestrategie als JSON im vorgegebenen Format."
)
# ---------------------------------------------------------------------------
# Mock-Planner (für Tests ohne LLM)
# ---------------------------------------------------------------------------
class MockResearchPlanner:
"""Mock-Planner ohne LLM-Abhängigkeit für Tests.
Generiert einen validen Plan basierend auf der Anfrage,
ohne ein LLM aufzurufen.
"""
DEPTH_MAP: dict[str, int] = {
"quick": 1,
"normal": 2,
"deep": 3,
}
def __init__(self, config: AppSettings | None = None) -> None:
self._config = config
@staticmethod
def _detect_depth(question: str) -> str:
"""Bestimme die empfohlene Tiefe basierend auf der Anfrage."""
q_lower = question.lower()
if any(kw in q_lower for kw in ("schnell", "kurz", "tl;dr", "kurz", "einfach")):
return "quick"
if any(kw in q_lower for kw in ("tiefgehend", "umfassend", "detailliert", "analy", "untersuch")):
return "deep"
return "normal"
@staticmethod
def _extract_entities(question: str) -> list[str]:
"""Simple entity extraction from the research question."""
words = question.replace("?", " ").replace(",", " ").split()
stop_words = {
"ist", "sind", "der", "die", "das", "ein", "eine", "den", "dem",
"und", "oder", "für", "von", "mit", "auf", "in", "zu", "bei",
"what", "is", "are", "the", "of", "and", "or", "for", "with",
}
entities = [
w for w in words if len(w) > 2 and w.lower() not in stop_words
]
seen = set()
unique = []
for e in entities:
if e not in seen:
seen.add(e)
unique.append(e)
return unique[:5] if unique else ["Unbekannt"]
def plan(
self,
research_question: str,
language: str = "de",
) -> dict[str, Any]:
"""Generiere einen validen Forschungsplan ohne LLM."""
topic = research_question.strip()[:120]
depth = self._detect_depth(research_question)
entities = self._extract_entities(research_question)
plan: dict[str, Any] = {
"topic": topic,
"time_range": {
"start": None,
"end": None,
"description": f"Zeitraum für {topic.lower()}",
},
"entities": entities,
"search_dimensions": [
"primary_sources",
"independent_reporting",
"counter_evidence",
"scientific_sources",
],
"queries": [
{
"query": f"Was ist {topic}?",
"purpose": "Neutrales, allgemeines Verständnis des Themas aufbauen",
"category": "general",
"language": language,
},
{
"query": f"{topic} offizielle Daten Statistiken Behörde",
"purpose": "Primärquellen und offizielle Daten identifizieren",
"category": "primary_source",
"language": language,
},
{
"query": f"{topic} aktuelle Berichterstattung Nachrichten",
"purpose": "Aktuelle journalistische Berichterstattung finden",
"category": "news",
"language": language,
},
{
"query": f"{topic} Kritik Kontroverse Alternativerstandpunkt",
"purpose": "Kritische Stimmen und alternative Perspektiven finden",
"category": "counter_evidence",
"language": language,
},
{
"query": f"{topic} wissenschaftliche Analyse Forschung",
"purpose": "Wissenschaftliche und fachliche Quellen erschließen",
"category": "scientific",
"language": language,
},
{
"query": f"{topic} unabhängige Bewertung Einschätzung",
"purpose": "Unabhängige, neutrale Bewertungen und Einschätzungen finden",
"category": "general",
"language": language,
},
],
"potential_sources": [
{"type": "primary_source", "description": "Behörden- und Regierungswebsites"},
{"type": "secondary_source", "description": "Unabhängige Nachrichtenagenturen und Medien"},
{"type": "academic", "description": "Wissenschaftliche Datenbanken und Repositories"},
],
"counter_hypotheses": [
f"Mögliche alternative Interpretation von {topic}",
f"Gegenposition zu {topic} prüfen",
],
"search_bias_mitigation": [
f"Suche nach {topic} mit neutralen UND kritischen Suchbegriffen",
"Mehrere Suchmaschinen parallel nutzen (DuckDuckGo, SearXNG)",
"Geheimdienst/Regierungsperspektive UND oppositionelle Quellen vergleichen",
"Internationaler Vergleich: Deutsche und internationale Quellen einbeziehen",
],
"estimated_depth": depth,
"confidence": 0.8,
}
return plan
# ---------------------------------------------------------------------------
# ResearchPlanner — LLM-gesteuert
# ---------------------------------------------------------------------------
class ResearchPlanner:
"""Research Planner — LLM-basierte Strategie-Generierung.
Parameters
----------
config : AppSettings
App-Konfiguration (LLM-Basis-URL, Model, etc.).
llm_provider : LLMProvider
Der LLM-Provider für textgenerierung.
metrics : ProviderMetrics | None
Optional: Metrics-Collector für Request-Tracking.
"""
def __init__(
self,
config: AppSettings,
llm_provider: LLMProvider,
metrics: Any = None,
) -> None:
self._config = config
self._llm = llm_provider
self._metrics = metrics
self._system_prompt = SYSTEM_PROMPT
@property
def model_name(self) -> str:
"""Das konfigurierte LLM-Modell."""
return self._config.llm.model
async def plan(
self,
research_question: str,
language: str = "de",
) -> dict[str, Any]:
"""Generiere eine neutrale Recherchestrategie als JSON.
Parameters
----------
research_question : str
Die zu analysierende Forschungsfrage.
language : str
Sprachcode (z.B. 'de', 'en').
Returns
-------
dict
Validierter Research-Plan als Dict.
Raises
------
RuntimeError
Wenn die LLM-Antwort kein gültiges JSON enthält.
"""
user_prompt = _build_user_prompt(research_question, language)
messages = [
{"role": "system", "content": self._system_prompt},
{"role": "user", "content": user_prompt},
]
start_time = time.monotonic()
# Verwende response_format für JSON-Only-Ausgabe
response_format = {"type": "json_object"}
raw_response = await self._llm.complete(
messages=messages,
model=self._config.llm.model,
temperature=0.3,
max_tokens=4096,
response_format=response_format,
)
elapsed = time.monotonic() - start_time
# ------------------------------------------------------------------
# JSON parsen — LLM-Output enthält oft Markdown-Codeblock-Umrandung
# ------------------------------------------------------------------
cleaned = self._extract_json(raw_response)
plan_dict = json.loads(cleaned)
# ------------------------------------------------------------------
# Validieren
# ------------------------------------------------------------------
validation = validate_plan(plan_dict)
if not validation["valid"]:
logger.warning("Planner output validation failed: %s", validation["errors"])
# Wir werfen nicht — der Plan wird trotzdem zurückgegeben,
# aber mit einem Fehler-Flag.
plan_dict["_validation_errors"] = validation["errors"]
# ------------------------------------------------------------------
# Metrics tracken (wenn vorhanden)
# ------------------------------------------------------------------
if self._metrics:
await self._metrics.record_llm_request(
input_tokens=len(user_prompt.split()),
output_tokens=len(cleaned.split()),
latency=elapsed,
)
return plan_dict
@staticmethod
def _extract_json(raw: str) -> str:
"""Extrahiere JSON aus LLM-Output (evtl. mit Markdown-Codeblock)."""
# Strip leading/trailing whitespace
text = raw.strip()
# Try to parse directly
try:
json.loads(text)
return text
except json.JSONDecodeError:
pass
# Try to find JSON inside code blocks
if text.startswith("```json"):
text = text[len("```json"):].strip()
if text.startswith("```"):
text = text[len("```"):].strip()
# Remove trailing backticks
text = text.strip("`").strip()
# Find first { and last }
start = text.find("{")
end = text.rfind("}")
if start != -1 and end != -1 and end > start:
text = text[start : end + 1]
return text

122
src/nsct/agents/validator.py Normal file
View File

@@ -0,0 +1,122 @@
"""Ergebnis-Validierung für den Research Planner.
Prüft, ob ein generierter Plan alle strukturellen Anforderungen
erfüllt, bevor er an Stage 5 (Claim Extraction) übergeben wird.
"""
from __future__ import annotations
from typing import Any
def validate_plan(plan: dict[str, Any]) -> dict[str, Any]:
"""Validiere einen Research-Plan-Dict und gib validierungs-Result zurück.
Parameters
----------
plan : dict
Der vom Planner generierte Plan als Dict.
Returns
-------
dict
{"valid": bool, "errors": list[str]}
"""
errors: list[str] = []
# --- 1. Required top-level fields ---
required_fields = [
"topic",
"time_range",
"entities",
"search_dimensions",
"queries",
"potential_sources",
"counter_hypotheses",
"search_bias_mitigation",
"estimated_depth",
"confidence",
]
for field in required_fields:
if field not in plan:
errors.append(f"Fehlendes required Feld: {field}")
# --- 2. topic nicht leer ---
topic = plan.get("topic")
if isinstance(topic, str) and not topic.strip():
errors.append("topic ist leer")
elif not isinstance(topic, str):
errors.append("topic muss ein String sein")
# --- 3. time_range ---
tr = plan.get("time_range")
if not isinstance(tr, dict):
errors.append("time_range muss ein Dict mit start, end, description sein")
else:
for tr_field in ("start", "end", "description"):
if tr_field not in tr:
errors.append(f"time_range fehlt Feld: {tr_field}")
# --- 4. entities nicht leer ---
entities = plan.get("entities")
if not isinstance(entities, list) or len(entities) == 0:
errors.append("entities muss eine nicht-leere Liste sein")
# --- 5. search_dimensions enthält required ---
dims = plan.get("search_dimensions", [])
if not isinstance(dims, list):
errors.append("search_dimensions muss eine Liste sein")
else:
if "primary_sources" not in dims:
errors.append("search_dimensions enthält nicht 'primary_sources'")
if "counter_evidence" not in dims:
errors.append("search_dimensions enthält nicht 'counter_evidence'")
# --- 6. queries ---
queries = plan.get("queries")
if not isinstance(queries, list) or len(queries) == 0:
errors.append("queries muss eine nicht-leere Liste sein")
else:
# Mindestens 4 verschiedene query categories
categories = set()
has_counter_evidence = False
for q in queries:
cat = q.get("category") if isinstance(q, dict) else None
if isinstance(cat, str):
categories.add(cat)
if cat == "counter_evidence":
has_counter_evidence = True
if len(categories) < 4:
errors.append(
f"Mindestens 4 verschiedene query categories erforderlich, "
f"aber nur {len(categories)} gefunden: {sorted(categories)}"
)
if not has_counter_evidence:
errors.append("Mindestens eine query mit category 'counter_evidence' erforderlich")
# Jede query braucht query, purpose, category, language
for i, q in enumerate(queries):
if not isinstance(q, dict):
errors.append(f"query[{i}] muss ein Dict sein")
continue
for qf in ("query", "purpose", "category", "language"):
val = q.get(qf)
if not val or (isinstance(val, str) and not val.strip()):
errors.append(f"query[{i}] fehlt oder ist leer: {qf}")
# --- 7. confidence im Bereich 0.0-1.0 ---
confidence = plan.get("confidence")
if not isinstance(confidence, (int, float)):
errors.append("confidence muss eine Zahl sein")
elif not (0.0 <= confidence <= 1.0):
errors.append(f"confidence muss im Bereich 0.0-1.0 sein, got {confidence}")
# --- 8. estimated_depth ---
depth = plan.get("estimated_depth")
if depth not in ("quick", "normal", "deep"):
errors.append(f"estimated_depth muss 'quick', 'normal' oder 'deep' sein, got '{depth}'")
# --- Ergebnis ---
valid = len(errors) == 0
return {"valid": valid, "errors": errors}

9
src/nsct/api/crawler.py
View File

@@ -5,6 +5,7 @@ from __future__ import annotations
import logging
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from nsct.crawler.fetcher import FetchResult, FetchStatus
from nsct.crawler.manager import CrawlerManager
@@ -32,26 +33,26 @@ def _get_manager() -> CrawlerManager:
# ---------------------------------------------------------------------------
class FetchRequest:
class FetchRequest(BaseModel):
"""Request body for single URL fetch."""
url: str
class FetchBatchRequest:
class FetchBatchRequest(BaseModel):
"""Request body for batch URL fetch."""
urls: list[str]
max_parallel: int = 5
class URLValidationRequest:
class URLValidationRequest(BaseModel):
"""Request body for URL validation."""
url: str
class URLValidationResponse:
class URLValidationResponse(BaseModel):
"""Response for URL validation."""
safe: bool

4
src/nsct/api/main.py
View File

@@ -83,6 +83,10 @@ def create_app() -> FastAPI:
from nsct.api.crawler import router as crawler_router
app.include_router(crawler_router, tags=["crawler"])
# Mount planner router
from nsct.api.planner import router as planner_router
app.include_router(planner_router, tags=["planner"])
return app

144
src/nsct/api/planner.py Normal file
View File

@@ -0,0 +1,144 @@
"""POST /research/planner — Research Planner API Endpoint.
Führt eine User-Anfrage an den Research Planner weiter und gibt
den strukturierten Forschungsplan zurück.
Debug-Modus (NSCT_DEBUG=true): Gibt zusätzlich LLM-Model und Latency zurück.
"""
from __future__ import annotations
import asyncio
import logging
import os
import time
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel, Field
from nsct.agents.planner import MockResearchPlanner, ResearchPlanner
from nsct.agents.validator import validate_plan
from nsct.config import AppSettings
logger = logging.getLogger(__name__)
router = APIRouter()
# ---------------------------------------------------------------------------
# Request / Response schemas
# ---------------------------------------------------------------------------
class PlannerRequest(BaseModel):
"""Input für den Research Planner."""
query: str = Field(
...,
min_length=1,
description="Die Forschungsfrage des Users.",
)
language: str = Field(
default="de",
description="Sprachcode für die Recherche (z.B. 'de', 'en').",
)
class PlannerResponse(BaseModel):
"""Output des Research Planners."""
plan: dict = Field(..., description="Strukturiertes JSON des Research Plans.")
valid: bool = Field(..., description="Ob der Plan die Validierungsregeln erfüllt.")
debug: dict | None = Field(
default=None,
description="Debug-Informationen (nur bei NSCT_DEBUG=true).",
)
# ---------------------------------------------------------------------------
# Helper — Planner-Instanz ermitteln
# ---------------------------------------------------------------------------
def _get_planner() -> ResearchPlanner | MockResearchPlanner:
"""Erstelle oder gib einen Planner zurück.
Im Testkontext (keine LLM-Config) wird ein MockResearchPlanner
zurückgegeben, damit Tests ohne echte LLM-Aufrufe funktionieren.
"""
config = AppSettings.from_env()
llm_base = config.llm.base_url if config.llm else ""
if not llm_base or not config.llm.model:
# Kein LLM konfiguriert — Mock verwenden
return MockResearchPlanner(config=config)
# Normalfall: Echter LLM-gesteuerter Planner
from nsct.providers.llm import get_provider
try:
from nsct.providers.metrics import ProviderMetrics
metrics = ProviderMetrics()
llm_provider = get_provider(config, metrics)
return ResearchPlanner(config=config, llm_provider=llm_provider, metrics=metrics)
except Exception as exc:
logger.warning("LLM-Provider konnte nicht initialisiert werden, Mock wird verwendet: %s", exc)
return MockResearchPlanner(config=config)
# ---------------------------------------------------------------------------
# Endpoint
# ---------------------------------------------------------------------------
@router.post("/research/planner")
async def research_planner_endpoint(request: PlannerRequest) -> PlannerResponse:
"""Research Planner — Generiere eine neutrale Recherchestrategie.
Parameters
----------
request : PlannerRequest
- query: Die Forschungsfrage
- language: Sprachcode (default: 'de')
Returns
-------
PlannerResponse
- plan: Der generierte Research Plan
- valid: Ob der Plan die Validierungsregeln erfüllt
- debug: Debug-Info (nur bei NSCT_DEBUG=true)
"""
start_time = time.monotonic()
planner = _get_planner()
try:
if asyncio.iscoroutinefunction(planner.plan):
plan = await planner.plan(research_question=request.query, language=request.language) # type: ignore[misc]
else:
plan = planner.plan(research_question=request.query, language=request.language)
except Exception as exc:
raise HTTPException(
status_code=500,
detail=f"Planner-Fehler: {exc}",
)
elapsed = time.monotonic() - start_time
validation = validate_plan(plan)
# Debug-Info, wenn NSCT_DEBUG=true
debug_info: dict | None = None
if os.environ.get("NSCT_DEBUG", "").lower() == "true":
model_name = "mock"
if isinstance(planner, ResearchPlanner):
model_name = planner.model_name
debug_info = {
"llm_model": model_name,
"latency_seconds": round(elapsed, 4),
"validation_errors": validation["errors"] if not validation["valid"] else None,
}
return PlannerResponse(
plan=plan,
valid=validation["valid"],
debug=debug_info,
)

17
src/nsct/crawler/pdf.py
View File

@@ -99,20 +99,15 @@ def extract_pdf_from_bytes(pdf_bytes: bytes) -> str:
return f"pdf_extract_failed"
def detect_pdf_content_type(pdf_bytes: bytes) -> str:
"""Detect the content type of PDF bytes.
def extract_pdf_content(pdf_bytes: bytes) -> str:
"""Extract text content from raw PDF bytes.
Priority: pdfminer.six -> pdfplumber -> minimal fallback.
Args:
pdf_bytes: Raw PDF content.
Returns:
Content type string.
Extracted text content, or an error marker string if extraction fails.
"""
if not pdf_bytes:
return "unknown"
# Check PDF magic bytes
if pdf_bytes[:4] == b"%PDF":
return "application/pdf"
return "unknown"
return extract_pdf_from_bytes(pdf_bytes)

182
src/nsct/models/plan.py Normal file
View File

@@ -0,0 +1,182 @@
"""Pydantic v2 schema for Research Planner output."""
from __future__ import annotations
from typing import Any
from pydantic import BaseModel, Field, field_validator
class TimeRange(BaseModel):
"""Time range for the expected relevant information."""
start: str | None = Field(
default=None,
description="Start date/time of the relevant time range (ISO 8601).",
)
end: str | None = Field(
default=None,
description="End date/time of the relevant time range (ISO 8601).",
)
description: str = Field(
...,
description="Human-readable description of the expected time period.",
)
@field_validator("description")
@classmethod
def _description_not_empty(cls, v: str) -> str:
if not v.strip():
raise ValueError("description darf nicht leer sein")
return v
class QueryConfig(BaseModel):
"""A single search query with its purpose and category."""
query: str = Field(
...,
min_length=1,
description="The actual search query text to execute.",
)
purpose: str = Field(
...,
min_length=1,
description="What this search query aims to discover.",
)
category: str = Field(
...,
description=(
"Category of the query. Must be one of: "
"primary_source, news, scientific, counter_evidence, general."
),
)
language: str = Field(
default="de",
description="Language code for the query (e.g. 'de', 'en').",
)
@field_validator("category")
@classmethod
def _valid_category(cls, v: str) -> str:
allowed = {"primary_source", "news", "scientific", "counter_evidence", "general"}
if v not in allowed:
raise ValueError(
f"category muss einer der folgenden sein: {', '.join(sorted(allowed))}"
)
return v
@field_validator("query")
@classmethod
def _query_not_empty(cls, v: str) -> str:
if not v.strip():
raise ValueError("query darf nicht leer sein")
return v
class PotentialSource(BaseModel):
"""A potential source type to look for."""
type: str = Field(
...,
description="Type of source: primary_source, secondary_source, or academic.",
)
description: str = Field(
...,
min_length=1,
description="Description of what to look for in this source type.",
)
@field_validator("type")
@classmethod
def _valid_type(cls, v: str) -> str:
allowed = {"primary_source", "secondary_source", "academic"}
if v not in allowed:
raise ValueError(
f"type muss einer der folgenden sein: {', '.join(sorted(allowed))}"
)
return v
@field_validator("description")
@classmethod
def _desc_not_empty(cls, v: str) -> str:
if not v.strip():
raise ValueError("description darf nicht leer sein")
return v
class ResearchPlan(BaseModel):
"""Structured research plan generated by the Research Planner.
The plan describes HOW to research a topic — it does NOT decide
what is true. It creates a strategy for the downstream stages
(Claim Extraction, Source Graph, etc.).
"""
topic: str = Field(
...,
min_length=1,
description="Interpreted topic of the research question.",
)
time_range: TimeRange = Field(
...,
description="Expected time range for relevant information.",
)
entities: list[str] = Field(
...,
min_length=1,
description="Important entities, persons, or organisations to track.",
)
search_dimensions: list[str] = Field(
...,
min_length=1,
description=(
"Search dimensions: primary_sources, independent_reporting, "
"counter_evidence, scientific_sources, etc."
),
)
queries: list[QueryConfig] = Field(
...,
min_length=1,
description="List of search queries with purpose and category.",
)
potential_sources: list[PotentialSource] = Field(
...,
min_length=1,
description="Types of potential sources to look for.",
)
counter_hypotheses: list[str] = Field(
default_factory=list,
description=(
"Possible alternative interpretations that must be searched for."
),
)
search_bias_mitigation: list[str] = Field(
default_factory=list,
description="Specific measures to counter search bias for this topic.",
)
estimated_depth: str = Field(
default="normal",
description="Estimated research depth: quick, normal, or deep.",
)
confidence: float = Field(
default=0.7,
ge=0.0,
le=1.0,
description="Planner confidence in the plan quality (0-1).",
)
@field_validator("topic")
@classmethod
def _topic_not_empty(cls, v: str) -> str:
if not v.strip():
raise ValueError("topic darf nicht leer sein")
return v
@field_validator("estimated_depth")
@classmethod
def _valid_depth(cls, v: str) -> str:
allowed = {"quick", "normal", "deep"}
if v not in allowed:
raise ValueError(f"estimated_depth muss einer der folgenden sein: {', '.join(sorted(allowed))}")
return v

532
tests/test_planner.py Normal file
View File

@@ -0,0 +1,532 @@
"""Tests for the Research Planner — Stage 4."""
from __future__ import annotations
from typing import Any
import pytest
from nsct.agents.planner import MockResearchPlanner, ResearchPlanner
from nsct.agents.validator import validate_plan
from nsct.config import AppSettings
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
def _make_mock_planner() -> MockResearchPlanner:
"""Erzeuge einen Mock-Planner für Tests."""
return MockResearchPlanner()
def _valid_plan() -> dict[str, Any]:
"""Erzeuge einen gültigen Plan-Dict (ohne LLM)."""
return {
"topic": "Test-Thema",
"time_range": {
"start": None,
"end": None,
"description": "Ganzer Zeitraum",
},
"entities": ["Bundesregierung", "Opposition", "EU"],
"search_dimensions": [
"primary_sources",
"independent_reporting",
"counter_evidence",
"scientific_sources",
],
"queries": [
{
"query": "Was ist Test-Thema?",
"purpose": "Allgemeines Verständnis",
"category": "general",
"language": "de",
},
{
"query": "Test-Thema offizielle Daten",
"purpose": "Primärquellen",
"category": "primary_source",
"language": "de",
},
{
"query": "Test-Thema aktuelle Nachrichten",
"purpose": "Aktuelle Berichterstattung",
"category": "news",
"language": "de",
},
{
"query": "Test-Thema Kritik Kontroverse",
"purpose": "Gegenstimmen",
"category": "counter_evidence",
"language": "de",
},
{
"query": "Test-Thema wissenschaftliche Analyse",
"purpose": "Wissenschaftliche Quellen",
"category": "scientific",
"language": "de",
},
{
"query": "Test-Thema unabhängige Bewertung",
"purpose": "Unabhängige Quellen",
"category": "general",
"language": "de",
},
],
"potential_sources": [
{"type": "primary_source", "description": "Behörden-Websites"},
{"type": "secondary_source", "description": "Nachrichtenagenturen"},
{"type": "academic", "description": "Wissenschaftliche Datenbanken"},
],
"counter_hypotheses": [
"Alternative Interpretation 1",
"Alternative Interpretation 2",
],
"search_bias_mitigation": [
"Neutrale Suchbegriffe nutzen",
"Mehrere Quellen vergleichen",
],
"estimated_depth": "normal",
"confidence": 0.8,
}
def _invalid_plan() -> dict[str, Any]:
"""Erzeuge einen ungültigen Plan-Dict (für validate_plan-Tests)."""
return {
"topic": "",
"time_range": {},
"entities": [],
"search_dimensions": [],
"queries": [],
"potential_sources": [],
"counter_hypotheses": [],
"search_bias_mitigation": [],
"estimated_depth": "invalid",
"confidence": 2.0,
}
# ---------------------------------------------------------------------------
# Tests: Valid Plan hat alle required fields
# ---------------------------------------------------------------------------
def test_valid_plan_has_required_fields() -> None:
"""Ein gültiger Plan muss alle required fields enthalten."""
plan = _valid_plan()
required = [
"topic",
"time_range",
"entities",
"search_dimensions",
"queries",
"potential_sources",
"counter_hypotheses",
"search_bias_mitigation",
"estimated_depth",
"confidence",
]
for field in required:
assert field in plan, f"Fehlendes required field: {field}"
# ---------------------------------------------------------------------------
# Tests: Mindestens 4 verschiedene query categories
# ---------------------------------------------------------------------------
def test_min_four_query_categories() -> None:
"""Es müssen mindestens 4 verschiedene query categories vorhanden sein."""
plan = _valid_plan()
categories = {q["category"] for q in plan["queries"]}
assert len(categories) >= 4, (
f"Nur {len(categories)} categories gefunden: {sorted(categories)}"
)
# Die gültige Test-Plan sollte 4 unique haben: general, primary_source,
# news, counter_evidence, scientific (5 unique)
assert len(categories) >= 4
# ---------------------------------------------------------------------------
# Tests: counter_evidence query vorhanden
# ---------------------------------------------------------------------------
def test_counter_evidence_query_present() -> None:
"""Es muss mindestens eine query mit category 'counter_evidence' geben."""
plan = _valid_plan()
categories = [q["category"] for q in plan["queries"]]
assert "counter_evidence" in categories, (
"Keine query mit category 'counter_evidence' gefunden"
)
# ---------------------------------------------------------------------------
# Tests: search_dimensions enthält primary_sources und counter_evidence
# ---------------------------------------------------------------------------
def test_search_dimensions_required() -> None:
"""search_dimensions muss 'primary_sources' und 'counter_evidence' enthalten."""
plan = _valid_plan()
dims = plan["search_dimensions"]
assert "primary_sources" in dims, "search_dimensions fehlt 'primary_sources'"
assert "counter_evidence" in dims, "search_dimensions fehlt 'counter_evidence'"
# ---------------------------------------------------------------------------
# Tests: validate_plan mit gültigem Plan
# ---------------------------------------------------------------------------
def test_validate_plan_valid() -> None:
"""validate_plan soll einen gültigen Plan als gültig erkennen."""
plan = _valid_plan()
result = validate_plan(plan)
assert result["valid"] is True
assert result["errors"] == []
def test_validate_plan_invalid() -> None:
"""validate_plan soll einen ungültigen Plan als ungültig erkennen."""
plan = _invalid_plan()
result = validate_plan(plan)
assert result["valid"] is False
assert len(result["errors"]) > 0
# ---------------------------------------------------------------------------
# Tests: confidence im Bereich 0.0–1.0
# ---------------------------------------------------------------------------
def test_valid_plan_confidence_in_range() -> None:
"""confidence eines gültigen Plans muss im Bereich 0.0–1.0 sein."""
plan = _valid_plan()
assert 0.0 <= plan["confidence"] <= 1.0, (
f"confidence {plan['confidence']} außerhalb des Bereichs [0.0, 1.0]"
)
def test_validate_plan_confidence_out_of_range() -> None:
"""validate_plan soll confidence > 1.0 ablehnen."""
plan = _valid_plan()
plan["confidence"] = 1.5
result = validate_plan(plan)
assert result["valid"] is False
assert any("confidence" in err for err in result["errors"])
def test_validate_plan_confidence_negative() -> None:
"""validate_plan soll confidence < 0.0 ablehnen."""
plan = _valid_plan()
plan["confidence"] = -0.1
result = validate_plan(plan)
assert result["valid"] is False
assert any("confidence" in err for err in result["errors"])
# ---------------------------------------------------------------------------
# Tests: Search-Bias-Mitigation (politische Anfrage → beide Seiten)
# ---------------------------------------------------------------------------
def test_search_bias_both_sides_covered() -> None:
"""Politische Anfragen müssen beide Seiten abdecken."""
planner = _make_mock_planner()
plan = planner.plan(
research_question="Umweltpolitik der Bundesregierung 2024",
language="de",
)
# search_dimensions muss counter_evidence enthalten
assert "counter_evidence" in plan["search_dimensions"]
# Es muss eine counter_evidence-Query geben
has_counter = any(
q["category"] == "counter_evidence" for q in plan["queries"]
)
assert has_counter, "Keine counter_evidence-Query in der generierten Strategie"
# Es muss general/neutral Queries geben (andere Seite)
has_general = any(
q["category"] == "general" for q in plan["queries"]
)
assert has_general, "Keine general-Query in der generierten Strategie"
# Mindestens 6 queries für vollständige Bias-Mitigation
assert len(plan["queries"]) >= 6, (
f"Nur {len(plan['queries'])} queries — mindestens 6 für vollständige Bias-Mitigation"
)
# search_bias_mitigation muss nicht leer sein
assert len(plan["search_bias_mitigation"]) >= 2, (
"search_bias_mitigation sollte mindestens 2 Maßnahmen enthalten"
)
# counter_hypotheses muss nicht leer sein
assert len(plan["counter_hypotheses"]) >= 1, (
"counter_hypotheses sollte mindestens 1 Eintrag enthalten"
)
def test_search_bias_neutral_queries() -> None:
"""Suchanfragen müssen neutrale Formulierungen verwenden, nicht einseitig.»
Verifiziert, dass die generierten queries nicht nur einseitig
politische Begriffe enthalten.
"""
planner = _make_mock_planner()
plan = planner.plan(
research_question="Flüchtling政策 und Integration in Deutschland",
language="de",
)
# Alle Queries müssen purpose und category haben
for q in plan["queries"]:
assert q.get("query", "").strip(), "Query darf nicht leer sein"
assert q.get("purpose", "").strip(), "purpose darf nicht leer sein"
assert q.get("category"), "category darf nicht leer sein"
# Es muss sowohl supporting als auch counter_evidence geben
categories = [q["category"] for q in plan["queries"]]
assert "counter_evidence" in categories
assert "news" in categories or "general" in categories
# ---------------------------------------------------------------------------
# Tests: MockResearchPlanner generiert validen Plan
# ---------------------------------------------------------------------------
def test_mock_planner_generates_valid_plan() -> None:
"""Der Mock-Planner muss einen validen Plan generieren."""
planner = _make_mock_planner()
plan = planner.plan(
research_question="Test-Frage zu Klimapolitik",
language="de",
)
# Alle required fields vorhanden?
result = validate_plan(plan)
assert result["valid"] is True, f"Validation errors: {result['errors']}"
# topic nicht leer
assert plan["topic"].strip()
# search_dimensions muss korrekt sein
assert len(plan["search_dimensions"]) > 0
# Mindestens 6 queries
assert len(plan["queries"]) >= 6
def test_mock_planner_detects_depth_quick() -> None:
"""Depth-Erkennung für kurze Anfragen."""
planner = _make_mock_planner()
plan = planner.plan(
research_question="Was ist 2+2? Kurzantwort.",
language="de",
)
assert plan["estimated_depth"] == "quick"
def test_mock_planner_detects_depth_normal() -> None:
"""Depth-Erkennung für normale Anfragen."""
planner = _make_mock_planner()
plan = planner.plan(
research_question="Stand der Elektromobilität in Deutschland",
language="de",
)
assert plan["estimated_depth"] == "normal"
def test_mock_planner_detects_depth_deep() -> None:
"""Depth-Erkennung für tiefgehende Anfragen."""
planner = _make_mock_planner()
plan = planner.plan(
research_question="Tiefgehende Analyse der deutschen Energiewende und deren Auswirkungen",
language="de",
)
assert plan["estimated_depth"] == "deep"
# ---------------------------------------------------------------------------
# Tests: API-Endpoint (Mock, kein echtes LLM)
# ---------------------------------------------------------------------------
def test_planner_endpoint_returns_valid_plan() -> None:
"""POST /research/planner muss einen gültigen Plan zurückgeben."""
from nsct.api.main import create_app
from fastapi.testclient import TestClient
app = create_app()
with TestClient(app) as client:
resp = client.post(
"/research/planner",
json={"query": "Test-Frage", "language": "de"},
)
assert resp.status_code == 200
body = resp.json()
assert "plan" in body
assert "valid" in body
assert isinstance(body["plan"], dict)
assert isinstance(body["valid"], bool)
def test_planner_endpoint_minimal_query() -> None:
"""POST /research/planner mit minimaler Anfrage."""
from nsct.api.main import create_app
from fastapi.testclient import TestClient
app = create_app()
with TestClient(app) as client:
resp = client.post("/research/planner", json={"query": "Test"})
assert resp.status_code == 200
body = resp.json()
assert body["valid"] is True
def test_planner_endpoint_422_on_empty_query() -> None:
"""POST /research/planner mit leerer query muss 422 zurückgeben."""
from nsct.api.main import create_app
from fastapi.testclient import TestClient
app = create_app()
with TestClient(app) as client:
resp = client.post("/research/planner", json={"query": ""})
assert resp.status_code == 422
# ---------------------------------------------------------------------------
# Tests: ResearchPlan Pydantic-Schema (models/plan.py)
# ---------------------------------------------------------------------------
def test_research_plan_schema_valid() -> None:
"""ResearchPlan muss mit allen gültigen Feldern instanziert werden."""
from nsct.models.plan import (
QueryConfig,
PotentialSource,
ResearchPlan,
TimeRange,
)
plan = ResearchPlan(
topic="Test-Thema",
time_range=TimeRange(start=None, end=None, description="Test"),
entities=["Entität1"],
search_dimensions=["primary_sources", "counter_evidence"],
queries=[
QueryConfig(
query="Test query",
purpose="Test",
category="general",
language="de",
),
],
potential_sources=[
PotentialSource(type="primary_source", description="Test"),
],
counter_hypotheses=["Hypothese 1"],
search_bias_mitigation=["Mitigation 1"],
estimated_depth="normal",
confidence=0.7,
)
assert plan.topic == "Test-Thema"
assert plan.confidence == 0.7
def test_research_plan_schema_confidence_bounds() -> None:
"""ResearchPlan muss confidence 0.0 und 1.0 erlauben."""
from nsct.models.plan import (
QueryConfig,
PotentialSource,
ResearchPlan,
TimeRange,
)
# confidence = 0.0
plan_min = ResearchPlan(
topic="T",
time_range=TimeRange(start=None, end=None, description="T"),
entities=["E"],
search_dimensions=["primary_sources"],
queries=[QueryConfig(query="q", purpose="p", category="general", language="de")],
potential_sources=[PotentialSource(type="primary_source", description="d")],
confidence=0.0,
)
assert plan_min.confidence == 0.0
# confidence = 1.0
plan_max = ResearchPlan(
topic="T",
time_range=TimeRange(start=None, end=None, description="T"),
entities=["E"],
search_dimensions=["primary_sources"],
queries=[QueryConfig(query="q", purpose="p", category="general", language="de")],
potential_sources=[PotentialSource(type="primary_source", description="d")],
confidence=1.0,
)
assert plan_max.confidence == 1.0
# ---------------------------------------------------------------------------
# Tests: validator edge cases
# ---------------------------------------------------------------------------
def test_validate_missing_fields() -> None:
"""validate_plan soll fehlende Felder melden."""
result = validate_plan({})
assert result["valid"] is False
assert len(result["errors"]) > 0
def test_validate_missing_counter_evidence() -> None:
"""validate_plan soll fehlende counter_evidence-Query melden."""
plan = _valid_plan()
# Entferne die counter_evidence-Query
plan["queries"] = [
q for q in plan["queries"] if q["category"] != "counter_evidence"
]
result = validate_plan(plan)
assert result["valid"] is False
assert any("counter_evidence" in err for err in result["errors"])
def test_validate_missing_search_dimensions() -> None:
"""validate_plan soll fehlende search_dimensions melden."""
plan = _valid_plan()
plan["search_dimensions"] = []
result = validate_plan(plan)
assert result["valid"] is False
assert any("primary_sources" in err for err in result["errors"])
def test_validate_empty_topic() -> None:
"""validate_plan soll leeres topic melden."""
plan = _valid_plan()
plan["topic"] = ""
result = validate_plan(plan)
assert result["valid"] is False
# ---------------------------------------------------------------------------
# Tests: Mock Planner search_dimensions completeness
# ---------------------------------------------------------------------------
def test_mock_planner_search_dimensions() -> None:
"""Der Mock-Planner muss alle 4 search_dimensions setzen."""
planner = _make_mock_planner()
plan = planner.plan(research_question="Test", language="de")
required_dims = {"primary_sources", "independent_reporting", "counter_evidence", "scientific_sources"}
found = set(plan["search_dimensions"])
assert required_dims.issubset(found), (
f"search_dimensions fehlen: {required_dims - found}"
)