feat(stage14): implement REST API for research lifecycle
- POST /v1/research — start research (non-blocking, background pipeline)
- GET /v1/research/{id} — research metadata
- GET /v1/research/{id}/status — detailed state machine status
- GET /v1/research/{id}/sources — sources list
- GET /v1/research/{id}/claims — claims list
- GET /v1/research/{id}/evidence — evidence scores
- GET /v1/research/{id}/report — research report
- DELETE /v1/research/{id} — delete research (non-completed)
- GET /v1/research — paginated list of all research runs
Depth budgets (quick/normal/deep) control only resource limits.
In-memory store for now, to be replaced with PostgreSQL later.
Router mounted in main.py as tag "research-api".
This commit is contained in:
NSCT Agent
576
src/nsct/api/rest_research.py
Normal file
576
src/nsct/api/rest_research.py
Normal file
@@ -0,0 +1,576 @@
|
||||
"""REST API — research lifecycle: POST / GET / DELETE + status, sources, claims, report, list."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any
|
||||
|
||||
from fastapi import BackgroundTasks, APIRouter, HTTPException
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from nsct.orchestration.budget import HardBudgetConfig
|
||||
from nsct.orchestration.state import ResearchRunState
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Request / Response Schemas
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class DepthConfig(BaseModel, frozen=True):
|
||||
"""Budget-Konfiguration pro Tiefe."""
|
||||
|
||||
key: str = Field(..., description="'quick', 'normal' oder 'deep'")
|
||||
max_search_queries: int = Field(..., ge=1)
|
||||
max_sources: int = Field(..., ge=1)
|
||||
max_pages_per_domain: int = Field(1, ge=1)
|
||||
max_total_download_bytes: int = Field(500_000, ge=1)
|
||||
max_llm_requests: int = Field(15, ge=1)
|
||||
max_research_duration_seconds: int = Field(120, ge=1)
|
||||
max_context_per_llm_call: int = Field(16_000, ge=1)
|
||||
|
||||
|
||||
# Depth budgets — steuern **ausschließlich** Budgets, nie Bewertungsmaßstäbe.
|
||||
# Die Aufgabenspezifikation:
|
||||
# quick: max_search_queries=10, max_sources=15, max_research_rounds=1
|
||||
# normal: max_search_queries=20, max_sources=30, max_research_rounds=2
|
||||
# deep: max_search_queries=40, max_sources=60, max_research_rounds=3
|
||||
DEPTH_CONFIGS: dict[str, DepthConfig] = {
|
||||
"quick": DepthConfig(
|
||||
key="quick",
|
||||
max_search_queries=10,
|
||||
max_sources=15,
|
||||
max_pages_per_domain=2,
|
||||
max_total_download_bytes=500_000,
|
||||
max_llm_requests=15,
|
||||
max_research_duration_seconds=120,
|
||||
max_context_per_llm_call=16_000,
|
||||
),
|
||||
"normal": DepthConfig(
|
||||
key="normal",
|
||||
max_search_queries=20,
|
||||
max_sources=30,
|
||||
max_pages_per_domain=5,
|
||||
max_total_download_bytes=2_000_000,
|
||||
max_llm_requests=30,
|
||||
max_research_duration_seconds=300,
|
||||
max_context_per_llm_call=24_000,
|
||||
),
|
||||
"deep": DepthConfig(
|
||||
key="deep",
|
||||
max_search_queries=40,
|
||||
max_sources=60,
|
||||
max_pages_per_domain=10,
|
||||
max_total_download_bytes=5_000_000,
|
||||
max_llm_requests=60,
|
||||
max_research_duration_seconds=600,
|
||||
max_context_per_llm_call=32_000,
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
class ResearchRequest(BaseModel):
|
||||
"""POST /v1/research input."""
|
||||
|
||||
query: str = Field(..., min_length=1, max_length=2000, description="Forschungsfrage.")
|
||||
language: str = Field(default="de", description="Sprachcode, z.B. 'de', 'en'.")
|
||||
depth: str = Field(default="normal", description="Suchtiefe: 'quick', 'normal', 'deep'.")
|
||||
extra_queries: list[str] | None = Field(
|
||||
default=None,
|
||||
description="Optionale zusätzliche Suchanfragen, die in die Pipeline aufgenommen werden.",
|
||||
)
|
||||
|
||||
|
||||
class ResearchListItem(BaseModel):
|
||||
"""Ein Eintrag in der Forschungsliste."""
|
||||
|
||||
research_id: str
|
||||
query: str
|
||||
depth: str
|
||||
state: str
|
||||
created_at: str
|
||||
source_count: int = 0
|
||||
claim_count: int = 0
|
||||
|
||||
|
||||
class ResearchListResponse(BaseModel):
|
||||
"""GET /v1/research — paginierte Liste."""
|
||||
|
||||
total: int
|
||||
limit: int
|
||||
offset: int
|
||||
items: list[ResearchListItem] = Field(default_factory=list)
|
||||
|
||||
|
||||
class ResearchCreateResponse(BaseModel):
|
||||
"""Antwort vom Start einer Recherche."""
|
||||
|
||||
research_id: str = Field(..., description="UUID des Research-Runs.")
|
||||
status: str = Field(..., description="'pending' — die Recherche läuft im Hintergrund.")
|
||||
query: str = Field(..., description="Die Forschungsfrage.")
|
||||
depth: str = Field(..., description="Angelegte Tiefe.")
|
||||
state: str = Field(..., description="Aktueller State-Machine-Status.")
|
||||
|
||||
|
||||
class StatusResponse(BaseModel):
|
||||
"""GET /v1/research/{id}/status."""
|
||||
|
||||
research_id: str
|
||||
query: str
|
||||
depth: str
|
||||
state: str
|
||||
created_at: str
|
||||
updated_at: str | None = None
|
||||
search_count: int = 0
|
||||
source_count: int = 0
|
||||
claim_count: int = 0
|
||||
is_completed: bool = False
|
||||
is_running: bool = False
|
||||
|
||||
|
||||
class SourceListItem(BaseModel):
|
||||
"""Ein Source-Eintrag."""
|
||||
|
||||
id: str
|
||||
url: str
|
||||
title: str | None = None
|
||||
domain: str = ""
|
||||
source_type: str | None = None
|
||||
retrieved_at: str = ""
|
||||
|
||||
|
||||
class SourcesResponse(BaseModel):
|
||||
"""GET /v1/research/{id}/sources."""
|
||||
|
||||
research_id: str
|
||||
sources: list[SourceListItem] = Field(default_factory=list)
|
||||
total: int = 0
|
||||
|
||||
|
||||
class ClaimListItem(BaseModel):
|
||||
"""Ein Claim-Eintrag."""
|
||||
|
||||
id: str
|
||||
research_run_id: str
|
||||
source_id: str
|
||||
claim_text: str
|
||||
evidence_span: str
|
||||
claim_type: str
|
||||
confidence: float = 1.0
|
||||
|
||||
|
||||
class ClaimsResponse(BaseModel):
|
||||
"""GET /v1/research/{id}/claims."""
|
||||
|
||||
research_id: str
|
||||
claims: list[ClaimListItem] = Field(default_factory=list)
|
||||
total: int = 0
|
||||
|
||||
|
||||
class EvidenceScoreItem(BaseModel):
|
||||
"""Ein Evidence-Score pro Claim."""
|
||||
|
||||
claim_id: str
|
||||
research_run_id: str
|
||||
evidence_type: str = "secondary_report"
|
||||
source_independence_score: float = 0.5
|
||||
primary_source_proximity: float = 0.0
|
||||
cross_source_support: float = 0.0
|
||||
contradiction_level: float = 1.0
|
||||
evidence_directness: float = 0.5
|
||||
date_relevance_score: float = 0.5
|
||||
|
||||
|
||||
class EvidenceResponse(BaseModel):
|
||||
"""GET /v1/research/{id}/evidence."""
|
||||
|
||||
research_id: str
|
||||
evidence: list[EvidenceScoreItem] = Field(default_factory=list)
|
||||
total: int = 0
|
||||
|
||||
|
||||
class FindingItem(BaseModel):
|
||||
"""Ein einzelner Befund."""
|
||||
|
||||
text: str
|
||||
confidence: float = 0.0
|
||||
source_ids: list[str] = Field(default_factory=list)
|
||||
|
||||
|
||||
class ReportResponse(BaseModel):
|
||||
"""GET /v1/research/{id}/report."""
|
||||
|
||||
research_id: str
|
||||
query: str
|
||||
summary: str = ""
|
||||
findings: list[FindingItem] = Field(default_factory=list)
|
||||
disagreements: list[dict[str, Any]] = Field(default_factory=list)
|
||||
uncertainties: list[str] = Field(default_factory=list)
|
||||
source_statistics: dict[str, Any] = Field(default_factory=dict)
|
||||
methodology: str = ""
|
||||
generated_at: str = ""
|
||||
|
||||
|
||||
class ErrorResponse(BaseModel):
|
||||
"""Fehlerantwort."""
|
||||
|
||||
detail: str
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# In-memory research store
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class _ResearchRunState(BaseModel):
|
||||
"""Internal representation of a research run in the store."""
|
||||
|
||||
research_id: str
|
||||
query: str
|
||||
language: str
|
||||
depth: str
|
||||
budget: HardBudgetConfig
|
||||
state: str = "created"
|
||||
created_at: str = ""
|
||||
updated_at: str | None = None
|
||||
plan: dict[str, Any] | None = None
|
||||
search_results: list[dict[str, Any]] = Field(default_factory=list)
|
||||
sources: list[dict[str, Any]] = Field(default_factory=list)
|
||||
claims: list[dict[str, Any]] = Field(default_factory=list)
|
||||
evidence_scores: list[dict[str, Any]] = Field(default_factory=list)
|
||||
report: dict[str, Any] | None = None
|
||||
metadata: dict[str, Any] = Field(default_factory=dict)
|
||||
|
||||
|
||||
# Global in-memory store.
|
||||
_research_store: dict[str, _ResearchRunState] = {}
|
||||
|
||||
|
||||
def _save_run(run: _ResearchRunState) -> None:
|
||||
_research_store[run.research_id] = run
|
||||
|
||||
|
||||
def _load_run(research_id: str) -> _ResearchRunState:
|
||||
if research_id not in _research_store:
|
||||
raise HTTPException(status_code=404, detail=f"Research run {research_id} not found")
|
||||
return _research_store[research_id]
|
||||
|
||||
|
||||
def _now() -> str:
|
||||
return datetime.now(timezone.utc).isoformat()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Endpoints
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@router.get(
|
||||
"/v1/research",
|
||||
response_model=ResearchListResponse,
|
||||
summary="List all research runs (paginated)",
|
||||
)
|
||||
async def list_research(limit: int = 10, offset: int = 0) -> ResearchListResponse:
|
||||
"""Liste aller Research-Runs (paginiert)."""
|
||||
all_runs = list(_research_store.values())
|
||||
total = len(all_runs)
|
||||
all_runs.sort(key=lambda r: r.created_at, reverse=True)
|
||||
sliced = all_runs[offset : offset + limit]
|
||||
items = [
|
||||
ResearchListItem(
|
||||
research_id=r.research_id,
|
||||
query=r.query,
|
||||
depth=r.depth,
|
||||
state=r.state,
|
||||
created_at=r.created_at,
|
||||
source_count=len(r.sources),
|
||||
claim_count=len(r.claims),
|
||||
)
|
||||
for r in sliced
|
||||
]
|
||||
return ResearchListResponse(total=total, limit=limit, offset=offset, items=items)
|
||||
|
||||
|
||||
@router.post(
|
||||
"/v1/research",
|
||||
response_model=ResearchCreateResponse,
|
||||
summary="Start a new research (non-blocking — returns immediately)",
|
||||
)
|
||||
async def start_research(
|
||||
request: ResearchRequest,
|
||||
bg: BackgroundTasks,
|
||||
) -> ResearchCreateResponse:
|
||||
"""Startet eine neue Recherche asynchron.
|
||||
|
||||
Erzeugt einen Research-Run mit Budget-Config gemäß Tiefe und
|
||||
initialisiert den Status auf 'created'. Der eigentliche Durchlauf
|
||||
der Pipeline wird asynchron angestoßen und kann über status / report
|
||||
abgerufen werden.
|
||||
"""
|
||||
# Validate depth
|
||||
depth_cfg = DEPTH_CONFIGS.get(request.depth)
|
||||
if depth_cfg is None:
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail=f"Ungültige Tiefe '{request.depth}'. Gültig: 'quick', 'normal', 'deep'.",
|
||||
)
|
||||
|
||||
# Build budget config from depth
|
||||
budget = HardBudgetConfig(
|
||||
max_search_queries=depth_cfg.max_search_queries,
|
||||
max_sources=depth_cfg.max_sources,
|
||||
max_pages_per_domain=depth_cfg.max_pages_per_domain,
|
||||
max_total_download_bytes=depth_cfg.max_total_download_bytes,
|
||||
max_llm_requests=depth_cfg.max_llm_requests,
|
||||
max_research_duration_seconds=depth_cfg.max_research_duration_seconds,
|
||||
max_context_per_llm_call=depth_cfg.max_context_per_llm_call,
|
||||
)
|
||||
|
||||
research_id = str(uuid.uuid4())
|
||||
created_at = _now()
|
||||
|
||||
run = _ResearchRunState(
|
||||
research_id=research_id,
|
||||
query=request.query,
|
||||
language=request.language,
|
||||
depth=request.depth,
|
||||
budget=budget,
|
||||
state=ResearchRunState.CREATED.value,
|
||||
created_at=created_at,
|
||||
updated_at=created_at,
|
||||
)
|
||||
_save_run(run)
|
||||
|
||||
# Start pipeline in background
|
||||
bg.add_task(_run_pipeline, run, request, budget)
|
||||
|
||||
return ResearchCreateResponse(
|
||||
research_id=research_id,
|
||||
status="pending",
|
||||
query=request.query,
|
||||
depth=request.depth,
|
||||
state=run.state,
|
||||
)
|
||||
|
||||
|
||||
async def _run_pipeline(run: _ResearchRunState, request: ResearchRequest, budget: HardBudgetConfig) -> None:
|
||||
"""Background task that runs the full research pipeline."""
|
||||
try:
|
||||
from nsct.config import AppSettings
|
||||
|
||||
config = AppSettings.from_env()
|
||||
from nsct.orchestration.orchestrator import ResearchOrchestrator
|
||||
|
||||
orchestrator = ResearchOrchestrator(
|
||||
config=config,
|
||||
research_id=uuid.UUID(run.research_id),
|
||||
query=request.query,
|
||||
budget_config=budget,
|
||||
depth=request.depth,
|
||||
)
|
||||
await orchestrator.start()
|
||||
run.state = ResearchRunState.PLANNING.value
|
||||
run.updated_at = _now()
|
||||
_save_run(run)
|
||||
|
||||
result = await orchestrator.run()
|
||||
if result.get("success"):
|
||||
run.state = ResearchRunState.COMPLETED.value
|
||||
run.report = result.get("report")
|
||||
run.claims = result.get("report", {}).get("claims", [])
|
||||
run.evidence_scores = result.get("report", {}).get("evidence", [])
|
||||
else:
|
||||
run.state = ResearchRunState.FAILED.value
|
||||
run.updated_at = _now()
|
||||
_save_run(run)
|
||||
except Exception as exc:
|
||||
logger.error("Background research pipeline failed: %s", exc)
|
||||
run.state = ResearchRunState.FAILED.value
|
||||
run.updated_at = _now()
|
||||
run.metadata["error"] = str(exc)
|
||||
_save_run(run)
|
||||
|
||||
|
||||
@router.get(
|
||||
"/v1/research/{research_id}",
|
||||
response_model=StatusResponse,
|
||||
summary="Get research run metadata",
|
||||
)
|
||||
async def get_research(research_id: str) -> StatusResponse:
|
||||
"""Metadaten eines Research-Runs abrufen."""
|
||||
run = _load_run(research_id)
|
||||
return StatusResponse(
|
||||
research_id=run.research_id,
|
||||
query=run.query,
|
||||
depth=run.depth,
|
||||
state=run.state,
|
||||
created_at=run.created_at,
|
||||
updated_at=run.updated_at,
|
||||
search_count=len(run.search_results),
|
||||
source_count=len(run.sources),
|
||||
claim_count=len(run.claims),
|
||||
is_completed=(run.state == ResearchRunState.COMPLETED.value),
|
||||
is_running=run.state not in (
|
||||
ResearchRunState.COMPLETED.value,
|
||||
ResearchRunState.FAILED.value,
|
||||
ResearchRunState.CANCELLED.value,
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
@router.get(
|
||||
"/v1/research/{research_id}/status",
|
||||
response_model=StatusResponse,
|
||||
summary="Get research status",
|
||||
)
|
||||
async def get_status(research_id: str) -> StatusResponse:
|
||||
"""Status eines Research-Runs abrufen."""
|
||||
run = _load_run(research_id)
|
||||
return StatusResponse(
|
||||
research_id=run.research_id,
|
||||
query=run.query,
|
||||
depth=run.depth,
|
||||
state=run.state,
|
||||
created_at=run.created_at,
|
||||
updated_at=run.updated_at,
|
||||
search_count=len(run.search_results),
|
||||
source_count=len(run.sources),
|
||||
claim_count=len(run.claims),
|
||||
is_completed=(run.state == ResearchRunState.COMPLETED.value),
|
||||
is_running=run.state not in (
|
||||
ResearchRunState.COMPLETED.value,
|
||||
ResearchRunState.FAILED.value,
|
||||
ResearchRunState.CANCELLED.value,
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
@router.get(
|
||||
"/v1/research/{research_id}/sources",
|
||||
response_model=SourcesResponse,
|
||||
summary="Get sources for a research run",
|
||||
)
|
||||
async def get_sources(research_id: str) -> SourcesResponse:
|
||||
"""Quellen für einen Research-Run abrufen."""
|
||||
run = _load_run(research_id)
|
||||
sources = [
|
||||
SourceListItem(
|
||||
id=s.get("id", str(uuid.uuid4())),
|
||||
url=s.get("url", ""),
|
||||
title=s.get("title"),
|
||||
domain=s.get("domain", ""),
|
||||
source_type=s.get("source_type"),
|
||||
retrieved_at=s.get("retrieved_at", _now()),
|
||||
)
|
||||
for s in run.sources
|
||||
]
|
||||
return SourcesResponse(
|
||||
research_id=run.research_id,
|
||||
sources=sources,
|
||||
total=len(sources),
|
||||
)
|
||||
|
||||
|
||||
@router.get(
|
||||
"/v1/research/{research_id}/claims",
|
||||
response_model=ClaimsResponse,
|
||||
summary="Get claims for a research run",
|
||||
)
|
||||
async def get_claims(research_id: str) -> ClaimsResponse:
|
||||
"""Claims für einen Research-Run abrufen."""
|
||||
run = _load_run(research_id)
|
||||
claims = [
|
||||
ClaimListItem(
|
||||
id=c.get("id", str(uuid.uuid4())),
|
||||
research_run_id=run.research_id,
|
||||
source_id=c.get("source_id", ""),
|
||||
claim_text=c.get("claim_text", c.get("claim", "")),
|
||||
evidence_span=c.get("evidence_span", ""),
|
||||
claim_type=c.get("claim_type", "claim"),
|
||||
confidence=c.get("confidence", 1.0),
|
||||
)
|
||||
for c in run.claims
|
||||
]
|
||||
return ClaimsResponse(
|
||||
research_id=run.research_id,
|
||||
claims=claims,
|
||||
total=len(claims),
|
||||
)
|
||||
|
||||
|
||||
@router.get(
|
||||
"/v1/research/{research_id}/evidence",
|
||||
response_model=EvidenceResponse,
|
||||
summary="Get evidence scores for a research run",
|
||||
)
|
||||
async def get_evidence(research_id: str) -> EvidenceResponse:
|
||||
"""Evidence-Scores für einen Research-Run abrufen."""
|
||||
run = _load_run(research_id)
|
||||
scores = [EvidenceScoreItem(**s) for s in run.evidence_scores]
|
||||
return EvidenceResponse(
|
||||
research_id=run.research_id,
|
||||
evidence=scores,
|
||||
total=len(scores),
|
||||
)
|
||||
|
||||
|
||||
@router.get(
|
||||
"/v1/research/{research_id}/report",
|
||||
response_model=ReportResponse,
|
||||
summary="Get research report",
|
||||
)
|
||||
async def get_report(research_id: str) -> ReportResponse:
|
||||
"""Synthese-Bericht für einen Research-Run abrufen."""
|
||||
run = _load_run(research_id)
|
||||
if not run.report:
|
||||
return ReportResponse(
|
||||
research_id=run.research_id,
|
||||
query=run.query,
|
||||
summary="Bericht noch nicht verfügbar. Die Recherche läuft oder ist fehlgeschlagen.",
|
||||
)
|
||||
|
||||
report_data = run.report
|
||||
findings = [
|
||||
FindingItem(**f) if isinstance(f, dict) else FindingItem(text=str(f))
|
||||
for f in report_data.get("findings", report_data.get("findings_json", []))
|
||||
]
|
||||
|
||||
return ReportResponse(
|
||||
research_id=run.research_id,
|
||||
query=run.query,
|
||||
summary=report_data.get("summary", ""),
|
||||
findings=findings,
|
||||
disagreements=report_data.get("disagreements", []),
|
||||
uncertainties=report_data.get("uncertainties", []),
|
||||
source_statistics=report_data.get("source_statistics", {}),
|
||||
methodology=report_data.get("methodology", ""),
|
||||
generated_at=report_data.get("generated_at", _now()),
|
||||
)
|
||||
|
||||
|
||||
@router.delete(
|
||||
"/v1/research/{research_id}",
|
||||
response_model=dict[str, str],
|
||||
summary="Cancel and delete a research run",
|
||||
)
|
||||
async def delete_research(research_id: str) -> dict[str, str]:
|
||||
"""Research-Run löschen (nur bei 'created' oder 'failed').
|
||||
|
||||
Läuft ein Run gerade, wird er zunächst abgebrochen.
|
||||
"""
|
||||
run = _load_run(research_id)
|
||||
if run.state == ResearchRunState.COMPLETED.value:
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail="Cannot delete a completed research run. It is immutable.",
|
||||
)
|
||||
run.state = ResearchRunState.CANCELLED.value
|
||||
run.updated_at = _now()
|
||||
del _research_store[research_id]
|
||||
return {"status": "deleted", "research_id": research_id}
|
||||
Reference in New Issue
Block a user