Files
NSCT Agent b8181deb05 Stage 4: Research Planner — LLM-basierte Recherchestrategie-Generierung
- 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
2026-08-23 13:28:23 +00:00

532 lines
18 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""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.01.0
# ---------------------------------------------------------------------------
def test_valid_plan_confidence_in_range() -> None:
"""confidence eines gültigen Plans muss im Bereich 0.01.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}"
)