From c8b55fbd38c7bf566319020d26ada52dddae5770 Mon Sep 17 00:00:00 2001 From: Florian Egger Date: Wed, 16 Sep 2026 21:12:47 +0200 Subject: [PATCH] feat(agent): harden API-first service --- pv-agent/.agents/MEMORY.md | 34 ++- pv-agent/.gitignore | 3 +- pv-agent/agent/README.md | 11 +- pv-agent/agent/api.py | 379 +++++++++++++++++++++++++------- pv-agent/agent/cli.py | 17 ++ pv-agent/agent/config.py | 6 +- pv-agent/agent/generate.py | 8 +- pv-agent/agent/retrieve.py | 163 +++++++------- pv-agent/docs/API.md | 168 ++++++++++++++ pv-agent/planung.md | 15 ++ pv-agent/tests/test_api.py | 187 ++++++++++++++++ pv-agent/tests/test_retrieve.py | 10 + 12 files changed, 839 insertions(+), 162 deletions(-) create mode 100644 pv-agent/docs/API.md create mode 100644 pv-agent/tests/test_api.py diff --git a/pv-agent/.agents/MEMORY.md b/pv-agent/.agents/MEMORY.md index 201c4cd..51f504c 100644 --- a/pv-agent/.agents/MEMORY.md +++ b/pv-agent/.agents/MEMORY.md @@ -5,11 +5,10 @@ Rollender Übergabe-Log für agent-Threads. Workflow: `.agents/SKILL.md` ## Current focus -Decision-Support-Tuning (D16) ist stabilisiert und im 53-Fragen- -Bestätigungslauf zitierseitig validiert. Verbleibend ist die bekannte -stochastische Survey-Fehlverweigerung q-029; danach folgen Rechtsprechungs- -Publikations-/Lizenzfreigabe und M4 (API-first Odoo-Integration inkl. -Privacy-Grenze für Lohndaten). +API-first-Festigung (D19): Der eigenständige Wissensdienst erhält einen +versionierten, authentisierbaren v1-Vertrag und eine explizite Privacy-Grenze. +Als Nächstes folgt der dünne Odoo-Client; Lohndaten bleiben bis zum separaten +tenant-autorisierten Datenvertrag außerhalb der Agenten-API. ## Completed (2026-09-14) @@ -255,6 +254,15 @@ Privacy-Grenze für Lohndaten). bisherigen Expected-ID. Optional später Goldset-Expected-IDs fachlich rekalibrieren, aber kein Release-Blocker. +- **D18 (Rechtsprechungsnutzung, 2026-09-16, User-Freigabe):** Auf + gerichtliche Entscheidungen darf in Antworten verwiesen werden. Ihr Inhalt + darf in eigenen Worten wiedergegeben und stellenweise zitiert werden. Die + Agentenantwort muss weiterhin die retrieved `rj-*`-/RIS-KB-ID nennen, + nichtamtliche Wiedergaben ausdrücklich kennzeichnen und darf keine nicht + retrieved Entscheidung ergänzen. Diese Freigabe umfasst Referenz, + Zusammenfassung und auszugsweise Zitate; sie ist keine pauschale Freigabe + zur Veröffentlichung vollständiger quellentreuer Entscheidungstexte. + ## Open issues / blockers - **Push**: Remote localhost:3003 aus der Sandbox nicht erreichbar — User @@ -414,6 +422,20 @@ Privacy-Grenze für Lohndaten). erwartete Quelle 92,7 %, Latenz mean 39,6 s / p95 78 s (+5 s: vollstaendige statt abgeschnittener Antworten). q-024: 4/4 stabil. Tests 59 → 60. Report `data/eval-qwen38-lengthfix.json`. +- **D19 (API-first-Vertrag, 2026-09-16):** `/v1/ask`, `/v1/health` und + `/v1/reindex` sind die stabilen Endpunkte; alte Pfade bleiben deprecated. + Der Ask-Response enthält Status, Quellen, belegte Konflikte, extrahierte + Rückfrage, Planungs- und Grounding-Metadaten sowie eine Request-ID. Optionaler + Bearer-Schutz über `PV_API_KEY`; Reindex kann mit `PV_ADMIN_API_KEY` getrennt + werden. Leere Keys sind nur lokaler Entwicklungsmodus. Health und öffentliche + Fehler nennen keine internen Hosts/Exceptions. v1 ist zustandslos, + `knowledge_base_only`, lehnt unbekannte JSON-Felder ab und akzeptiert keine + Payroll-/Mitarbeiterobjekte. Odoo-Lohndaten erfordern später einen separaten + tenant-autorisierten Vertrag mit Datensatzregeln, Datenminimierung, Audit und + Cross-Tenant-Tests. Vertrag: `docs/API.md`. Der gemeinsame SQLite-Retriever + ist durch `check_same_thread=False` plus `RLock` worker-thread-sicher. Neue + Nicht-lokaler CLI-Bind ohne `PV_API_KEY` wird fail-closed abgelehnt. Neue + API-/Concurrency-Tests erhöhen die Offline-Suite von 75 auf **82 grüne Tests**. - **Bake-off-Protokoll** (Skill): Modellwechsel nur über dokumentierten Goldset-Vergleich; Kriterium: Zitier-Präzision > Verweigerungs- korrektheit > Latenz. @@ -423,6 +445,8 @@ Privacy-Grenze für Lohndaten). - `planung.md` — Plan + Entscheidungspunkte (Abschnitt 12) + Stand. - `.agents/skills/pv-rag-agent/SKILL.md` — verbindliche Regeln. - `agent/README.md` — Betrieb, Konfiguration, Host-Schritte. +- `docs/API.md` — v1-Vertrag, Authentisierung, Privacy-Grenze und Odoo-Aufruf. +- `agent/api.py` — versionierte Service-Oberfläche und Sicherheitsgrenzen. - `agent/eval/goldset.yaml` — Goldset (IDs gegen kb.json verifiziert). - `agent/generate.py` — Grounding-Kern (Prompt, Post-Validierung). - `wissensbasis/README.md` — Layer-2-Schema + Rechtsprechungs-Provenance. diff --git a/pv-agent/.gitignore b/pv-agent/.gitignore index dbef656..a80622f 100644 --- a/pv-agent/.gitignore +++ b/pv-agent/.gitignore @@ -11,4 +11,5 @@ data/ __pycache__/ *.pyc .venv/ -.pytest_cache/ \ No newline at end of file +.pytest_cache/ +.env diff --git a/pv-agent/agent/README.md b/pv-agent/agent/README.md index d609f91..58d781c 100644 --- a/pv-agent/agent/README.md +++ b/pv-agent/agent/README.md @@ -43,9 +43,14 @@ python -m agent.cli eval python -m agent.cli eval --answers --json-out data/eval-report.json # 4) HTTP-API + Test-Chat -python -m agent.cli serve # http://127.0.0.1:8080 (/ask, /health, /reindex) +python -m agent.cli serve # http://127.0.0.1:8080 (/v1/ask, /v1/health, /v1/reindex) ``` +Der stabile v1-Vertrag, Bearer-Authentisierung, Fehlersemantik und die +Datenschutzgrenze für die spätere Odoo-Anbindung sind in `docs/API.md` +dokumentiert. v1 ist zustandslos und akzeptiert keine Mandanten-, Mitarbeiter- +oder Abrechnungsobjekte. + ## Konfiguration (Umgebungsvariablen) | Variable | Default | Bedeutung | @@ -69,6 +74,8 @@ python -m agent.cli serve # http://127.0.0.1:8080 (/ask, /health, /r | `PV_MAX_CONTEXT_CHARS` | `90000` | User-Content-Budget; niedrig gerankte Blöcke werden ganz weggelassen (`trim_results`) | | `PV_CONTEXT_BLOCKS` | `8` | Kontextblöcke im Prompt | | `PV_PORT` | `8080` | API-Port | +| `PV_API_KEY` | leer | Bearer-Key für `/v1/ask`; leer nur für lokale Entwicklung ohne Auth | +| `PV_ADMIN_API_KEY` | leer | separater Bearer-Key für `/v1/reindex`; leer = `PV_API_KEY` verwenden | ## Deployment auf dem Host (Ollama-Maschine) @@ -206,7 +213,7 @@ agent/ retrieve.py Hybrid-Retrieval (BM25 + Dense, RRF, cross_refs) ollama_client.py Ollama-HTTP (embed + chat, think-Fallback) generate.py Systemprompt, Post-Validierung, Verweigerung - api.py FastAPI (/ask, /health, /reindex) + api.py FastAPI v1 (/v1/ask, /v1/health, /v1/reindex), Auth + Request-IDs cli.py ingest | ask | eval | serve eval/ goldset.yaml + evaluate.py web/index.html Minimaler Test-Chat diff --git a/pv-agent/agent/api.py b/pv-agent/agent/api.py index 5346792..b1f1a4e 100644 --- a/pv-agent/agent/api.py +++ b/pv-agent/agent/api.py @@ -1,31 +1,51 @@ -"""FastAPI-Oberfläche des PV RAG Agent. +"""Versionierte FastAPI-Oberfläche des PV RAG Agent. -Endpunkte: - POST /ask — Frage -> belegte Antwort (oder Verweigerung) - GET /health — Index- und Ollama-Status - POST /reindex — Index-Neuaufbau (nach neuem Wissensbasis-Batch) +Die API bleibt ein reiner Wissensdienst: Requests enthalten eine Frage, aber +keinen Mandanten- oder Payroll-Datenkontext. Eine spätere Lohndaten-Anbindung +benötigt einen getrennten, mandantenautorisierten Vertrag. """ from __future__ import annotations +import logging +import re +import secrets import threading +import uuid from contextlib import asynccontextmanager +from typing import Literal -from fastapi import FastAPI, HTTPException -from pydantic import BaseModel, Field +from fastapi import Depends, FastAPI, HTTPException, Request +from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer +from pydantic import BaseModel, ConfigDict, Field from .config import Config -from .generate import answer_question +from .generate import CITE_RE, answer_question from .ingest import build_index from .ollama_client import OllamaClient from .retrieve import Retriever +API_VERSION = "v1" +DATA_SCOPE = "knowledge_base_only" +_REQUEST_ID_RE = re.compile(r"^[A-Za-z0-9._:-]{1,128}$") +logger = logging.getLogger(__name__) +bearer = HTTPBearer(auto_error=False) +bearer_credentials = Depends(bearer) -class AskRequest(BaseModel): + +class StrictModel(BaseModel): + model_config = ConfigDict(extra="forbid") + + +class AskRequest(StrictModel): question: str = Field(min_length=3, max_length=2000) top_k: int | None = Field(default=None, ge=1, le=20) + mode: Literal["knowledge"] = Field( + default="knowledge", + description="Derzeit ausschließlich KB-Wissen; kein Payroll-Datenkontext.", + ) -class SourceOut(BaseModel): +class SourceOut(StrictModel): id: str title: str section: str | None = None @@ -33,51 +53,93 @@ class SourceOut(BaseModel): work: str | None = None -class AskResponse(BaseModel): +class ConflictOut(StrictModel): + summary: str + source_ids: list[str] + + +class PlannedQueryOut(StrictModel): + text: str + stand_year: str | None = None + scope: str | None = None + + +class GroundingOut(StrictModel): + data_scope: Literal["knowledge_base_only"] = DATA_SCOPE + citations_verified: bool + context_count: int + regenerations: int + + +class AskResponse(StrictModel): + api_version: Literal["v1"] = API_VERSION + request_id: str + status: Literal["answered", "refused", "uncertain"] question: str answer: str refused: bool verified: bool citations: list[str] sources: list[SourceOut] + conflicts: list[ConflictOut] = Field(default_factory=list) + assumptions: list[str] = Field(default_factory=list) + clarification_question: str | None = None + alternatives: list[str] = Field(default_factory=list) + answer_type: Literal["specific", "survey"] = "specific" + planned: bool = False + planned_queries: list[PlannedQueryOut] = Field(default_factory=list) + grounding: GroundingOut n_context: int model: str latency_ms: int regenerations: int = 0 +class HealthResponse(StrictModel): + api_version: Literal["v1"] = API_VERSION + service: Literal["pv-rag-agent"] = "pv-rag-agent" + status: Literal["ok", "degraded"] + index: dict + ollama_up: bool + authentication_enabled: bool + + class AppState: def __init__(self) -> None: self.cfg: Config | None = None self.client: OllamaClient | None = None self.retriever: Retriever | None = None - self.lock = threading.Lock() + self.maintenance_lock = threading.Lock() + self.state_lock = threading.RLock() def ensure(self) -> Config: - if self.cfg is None: - self.cfg = Config.from_env() - return self.cfg + with self.state_lock: + if self.cfg is None: + self.cfg = Config.from_env() + return self.cfg def get_client(self) -> OllamaClient: - if self.client is None: - cfg = self.ensure() - self.client = OllamaClient( - cfg.ollama_url, - embed_timeout_s=cfg.embed_timeout_s, - chat_timeout_s=cfg.chat_timeout_s, - ) - return self.client + with self.state_lock: + if self.client is None: + cfg = self.ensure() + self.client = OllamaClient( + cfg.ollama_url, + embed_timeout_s=cfg.embed_timeout_s, + chat_timeout_s=cfg.chat_timeout_s, + ) + return self.client def get_retriever(self) -> Retriever: - if self.retriever is None: - cfg = self.ensure() - self.retriever = Retriever(cfg) - return self.retriever + with self.state_lock: + if self.retriever is None: + self.retriever = Retriever(self.ensure()) + return self.retriever def reset_retriever(self) -> None: - if self.retriever is not None: - self.retriever.close() - self.retriever = None + with self.state_lock: + if self.retriever is not None: + self.retriever.close() + self.retriever = None @asynccontextmanager @@ -85,72 +147,233 @@ async def lifespan(app: FastAPI): app.state.rag = AppState() yield rag: AppState = app.state.rag - if rag.retriever: - rag.retriever.close() + rag.reset_retriever() if rag.client: rag.client.close() -app = FastAPI(title="PV RAG Agent", version="0.1.0", lifespan=lifespan) +app = FastAPI( + title="PV RAG Agent", + version="1.0.0", + description=( + "Eigenständiger, KB-gebundener Wissensdienst. Der v1-Vertrag nimmt " + "keine Mandanten- oder Mitarbeiterdaten entgegen." + ), + lifespan=lifespan, +) -@app.post("/ask", response_model=AskResponse) -def ask(req: AskRequest) -> AskResponse: - rag: AppState = app.state.rag +@app.middleware("http") +async def add_request_id(request: Request, call_next): + supplied = request.headers.get("X-Request-ID", "") + request_id = supplied if _REQUEST_ID_RE.fullmatch(supplied) else uuid.uuid4().hex + request.state.request_id = request_id + response = await call_next(request) + response.headers["X-Request-ID"] = request_id + return response + + +def _request_id(request: Request) -> str: + return getattr(request.state, "request_id", uuid.uuid4().hex) + + +def _check_bearer( + expected: str, + credentials: HTTPAuthorizationCredentials | None, +) -> None: + # Leere Keys erhalten den bisherigen lokalen Entwicklungsmodus. Für ein + # exponiertes Deployment muss PV_API_KEY gesetzt sein. + if not expected: + return + supplied = credentials.credentials if credentials and credentials.scheme.lower() == "bearer" else "" + if not supplied or not secrets.compare_digest(supplied, expected): + raise HTTPException( + status_code=401, + detail="Authentisierung erforderlich.", + headers={"WWW-Authenticate": "Bearer"}, + ) + + +def require_api_access( + request: Request, + credentials: HTTPAuthorizationCredentials | None = bearer_credentials, +) -> None: + rag: AppState = request.app.state.rag + _check_bearer(rag.ensure().api_key, credentials) + + +def require_admin_access( + request: Request, + credentials: HTTPAuthorizationCredentials | None = bearer_credentials, +) -> None: + rag: AppState = request.app.state.rag cfg = rag.ensure() + _check_bearer(cfg.admin_api_key or cfg.api_key, credentials) + + +def _extract_conflicts(answer: str, citations: list[str]) -> list[ConflictOut]: + allowed = set(citations) + conflicts: list[ConflictOut] = [] + for paragraph in re.split(r"\n\s*\n", answer): + text = " ".join(paragraph.split()) + if "⚠" not in text: + continue + ids = sorted(set(CITE_RE.findall(text)) & allowed) + if ids: + conflicts.append(ConflictOut(summary=text, source_ids=ids)) + return conflicts + + +def _extract_clarification(answer: str, refused: bool) -> str | None: + if refused: + return None + match = re.search(r"([^.!?\n]*\?)\s*$", answer.strip()) + if not match: + return None + question = match.group(1).strip().lstrip("-• ") + return question or None + + +def _response_from_result(result: dict, request_id: str) -> AskResponse: + if not result["verified"]: + status = "uncertain" + elif result["refused"]: + status = "refused" + else: + status = "answered" + citations = list(result["citations"]) + return AskResponse( + request_id=request_id, + status=status, + question=result["question"], + answer=result["answer"], + refused=result["refused"], + verified=result["verified"], + citations=citations, + sources=result["sources"], + conflicts=_extract_conflicts(result["answer"], citations), + # Keine Annahmen oder Alternativen aus Freitext erraten. Diese Felder + # sind Teil des stabilen Vertrags und werden erst befüllt, wenn die + # Generierung sie selbst belegbar strukturiert liefert. + assumptions=[], + clarification_question=_extract_clarification( + result["answer"], result["refused"] + ), + alternatives=[], + answer_type=result.get("answer_type", "specific"), + planned=result.get("planned", False), + planned_queries=result.get("planned_queries", []), + grounding=GroundingOut( + citations_verified=result["verified"], + context_count=result["n_context"], + regenerations=result["regenerations"], + ), + n_context=result["n_context"], + model=result["model"], + latency_ms=result["latency_ms"], + regenerations=result["regenerations"], + ) + + +def _ask(req: AskRequest, request: Request) -> AskResponse: + rag: AppState = request.app.state.rag + cfg = rag.ensure() + request_id = _request_id(request) try: result = answer_question( - req.question, cfg, + req.question, + cfg, client=rag.get_client(), retriever=rag.get_retriever(), top_k=req.top_k, ) - except RuntimeError as e: # Index fehlt - raise HTTPException(status_code=503, detail=str(e)) from e - except Exception as e: # Ollama nicht erreichbar o. Ä. + except Exception as exc: + logger.exception("Antwortgenerierung fehlgeschlagen request_id=%s", request_id) raise HTTPException( status_code=503, - detail=f"Antwortgenerierung fehlgeschlagen: {type(e).__name__}: {e}", - ) from e + detail="Der Wissensdienst ist vorübergehend nicht verfügbar.", + ) from exc result.pop("draft", None) - return AskResponse(**result) + return _response_from_result(result, request_id) -@app.get("/health") -def health() -> dict: - rag: AppState = app.state.rag +@app.post("/v1/ask", response_model=AskResponse, dependencies=[Depends(require_api_access)]) +def ask_v1(req: AskRequest, request: Request) -> AskResponse: + return _ask(req, request) + + +@app.post( + "/ask", + response_model=AskResponse, + dependencies=[Depends(require_api_access)], + deprecated=True, +) +def ask_compat(req: AskRequest, request: Request) -> AskResponse: + return _ask(req, request) + + +def _health(request: Request) -> HealthResponse: + rag: AppState = request.app.state.rag cfg = rag.ensure() - out: dict = {"service": "pv-rag-agent", "config": { - "ollama_url": cfg.ollama_url, - "answer_model": cfg.answer_model, - "embed_model": cfg.embed_model, - }} try: - retriever = rag.get_retriever() - out["index"] = retriever.stats() - except RuntimeError as e: - out["index"] = {"error": str(e)} - client = rag.get_client() - out["ollama_up"] = client.is_up() - if out["ollama_up"]: - try: - out["ollama_models"] = client.list_models() - except Exception: - out["ollama_models"] = None - return out - - -@app.post("/reindex") -def reindex() -> dict: - rag: AppState = app.state.rag - cfg = rag.ensure() - with rag.lock: - stats = build_index(cfg, client=rag.get_client()) - rag.reset_retriever() - result = stats.as_dict() - result["warning"] = ( - "Index ohne Dense-Vektoren aufgebaut (Ollama-Embedding nicht verfügbar) — " - "BM25-only. 'ollama pull " + cfg.embed_model + "' prüfen und erneut reindexen." - if stats.embed_error else None + index = rag.get_retriever().stats() + index_ok = True + except RuntimeError: + index = {"available": False} + index_ok = False + ollama_up = rag.get_client().is_up() + return HealthResponse( + status="ok" if index_ok and ollama_up else "degraded", + index=index, + ollama_up=ollama_up, + authentication_enabled=bool(cfg.api_key), ) - return result \ No newline at end of file + + +@app.get("/v1/health", response_model=HealthResponse) +def health_v1(request: Request) -> HealthResponse: + return _health(request) + + +@app.get("/health", response_model=HealthResponse, deprecated=True) +def health_compat(request: Request) -> HealthResponse: + return _health(request) + + +def _reindex(request: Request) -> dict: + rag: AppState = request.app.state.rag + cfg = rag.ensure() + with rag.maintenance_lock: + try: + stats = build_index(cfg, client=rag.get_client()) + rag.reset_retriever() + except Exception as exc: + request_id = _request_id(request) + logger.exception("Reindex fehlgeschlagen request_id=%s", request_id) + raise HTTPException( + status_code=503, + detail="Der Index konnte nicht neu aufgebaut werden.", + ) from exc + result = stats.as_dict() + result["api_version"] = API_VERSION + result["request_id"] = _request_id(request) + result["warning"] = ( + "Index ohne Dense-Vektoren aufgebaut; Embedding-Dienst prüfen und erneut reindexen." + if stats.embed_error + else None + ) + return result + + +@app.post("/v1/reindex", dependencies=[Depends(require_admin_access)]) +def reindex_v1(request: Request) -> dict: + return _reindex(request) + + +@app.post( + "/reindex", + dependencies=[Depends(require_admin_access)], + deprecated=True, +) +def reindex_compat(request: Request) -> dict: + return _reindex(request) diff --git a/pv-agent/agent/cli.py b/pv-agent/agent/cli.py index 361c79b..49bf3a6 100644 --- a/pv-agent/agent/cli.py +++ b/pv-agent/agent/cli.py @@ -8,6 +8,7 @@ from __future__ import annotations import argparse +import ipaddress import json import sys @@ -80,7 +81,23 @@ def _cmd_eval(args: argparse.Namespace, cfg: Config) -> int: return run_eval(cfg, args) +def _is_loopback_bind(host: str) -> bool: + if host.casefold() == "localhost": + return True + try: + return ipaddress.ip_address(host).is_loopback + except ValueError: + return False + + def _cmd_serve(args: argparse.Namespace, cfg: Config) -> int: + if not _is_loopback_bind(args.host) and not cfg.api_key: + print( + "[Fehler] Nicht-lokaler API-Bind ohne PV_API_KEY abgelehnt. " + "Service-Key setzen oder nur an 127.0.0.1 binden.", + file=sys.stderr, + ) + return 2 import uvicorn uvicorn.run("agent.api:app", host=args.host, port=cfg.port, log_level="info") return 0 diff --git a/pv-agent/agent/config.py b/pv-agent/agent/config.py index 443ac74..0b1357c 100644 --- a/pv-agent/agent/config.py +++ b/pv-agent/agent/config.py @@ -89,9 +89,11 @@ class Config: # Service port: int = 8080 + api_key: str = "" # leer = nur fuer lokale Entwicklung ohne Auth + admin_api_key: str = "" # leer = api_key auch fuer /reindex verwenden @classmethod - def from_env(cls) -> "Config": + def from_env(cls) -> Config: d = cls() return cls( kb_dir=_env_str("PV_KB_DIR", d.kb_dir), @@ -122,4 +124,6 @@ class Config: survey_blocks=_env_int("PV_SURVEY_BLOCKS", d.survey_blocks), temporal_boost=_env_float("PV_TEMPORAL_BOOST", d.temporal_boost), port=_env_int("PV_PORT", d.port), + api_key=_env_str("PV_API_KEY", d.api_key), + admin_api_key=_env_str("PV_ADMIN_API_KEY", d.admin_api_key), ) \ No newline at end of file diff --git a/pv-agent/agent/generate.py b/pv-agent/agent/generate.py index c99189c..aeae2c8 100644 --- a/pv-agent/agent/generate.py +++ b/pv-agent/agent/generate.py @@ -296,8 +296,14 @@ def answer_question( "regenerations": regenerations, "latency_ms": round((time.perf_counter() - t0) * 1000), "draft": draft, + "answer_type": qtype, + "planned": planned, "planned_queries": [ - {"text": sq.text, "stand_year": sq.stand_year} + { + "text": sq.text, + "stand_year": sq.stand_year, + "scope": sq.scope, + } for sq in sub_queries ], } diff --git a/pv-agent/agent/retrieve.py b/pv-agent/agent/retrieve.py index 38e9f02..dcac660 100644 --- a/pv-agent/agent/retrieve.py +++ b/pv-agent/agent/retrieve.py @@ -5,6 +5,7 @@ from __future__ import annotations import json import sqlite3 +import threading from dataclasses import dataclass, field from pathlib import Path @@ -66,11 +67,16 @@ class Retriever: raise RuntimeError( f"index fehlt ({self.db_path}) — zuerst 'python -m agent.cli ingest' ausführen" ) - self._con = sqlite3.connect(self.db_path) + # FastAPI fuehrt synchrone Endpunkte in Worker-Threads aus. Die + # Verbindung wird deshalb thread-uebergreifend verwendet; ein RLock + # serialisiert die kurzen SQLite-/Matrix-Abschnitte je Retriever. + self._lock = threading.RLock() + self._con = sqlite3.connect(self.db_path, check_same_thread=False) self._con.row_factory = sqlite3.Row self._client = client self._owns_client = client is None self._mat: np.ndarray | None = None + self._mat_norm: np.ndarray | None = None self._mat_chunk_ids: list[int] | None = None self._scope_cache: dict[str, set[int]] = {} row = self._con.execute("SELECT MIN(stand), MAX(stand) FROM chunks").fetchone() @@ -78,31 +84,33 @@ class Retriever: self._stand_max = int((row[1] or "2026-01").replace("-", "")) def close(self) -> None: - self._con.close() - if self._owns_client and self._client is not None: - self._client.close() + with self._lock: + self._con.close() + if self._owns_client and self._client is not None: + self._client.close() # -- Index-Kennzahlen --------------------------------------------------- def stats(self) -> dict: - n_chunks = self._con.execute("SELECT COUNT(*) FROM chunks").fetchone()[0] - n_entries = self._con.execute( - "SELECT COUNT(DISTINCT entry_id) FROM chunks" - ).fetchone()[0] - n_vec = self._con.execute( - "SELECT COUNT(*) FROM vectors WHERE model = ?", - (self.cfg.embed_model,), - ).fetchone()[0] - meta = dict(self._con.execute("SELECT key, value FROM meta").fetchall()) - return { - "n_entries": n_entries, - "n_chunks": n_chunks, - "n_vectors": n_vec, - "dense_available": n_vec > 0 and not self.cfg.embed_off, - "stand_min": str(self._stand_min), - "stand_max": str(self._stand_max), - "built_at": meta.get("built_at"), - } + with self._lock: + n_chunks = self._con.execute("SELECT COUNT(*) FROM chunks").fetchone()[0] + n_entries = self._con.execute( + "SELECT COUNT(DISTINCT entry_id) FROM chunks" + ).fetchone()[0] + n_vec = self._con.execute( + "SELECT COUNT(*) FROM vectors WHERE model = ?", + (self.cfg.embed_model,), + ).fetchone()[0] + meta = dict(self._con.execute("SELECT key, value FROM meta").fetchall()) + return { + "n_entries": n_entries, + "n_chunks": n_chunks, + "n_vectors": n_vec, + "dense_available": n_vec > 0 and not self.cfg.embed_off, + "stand_min": str(self._stand_min), + "stand_max": str(self._stand_max), + "built_at": meta.get("built_at"), + } # -- Einzelverfahren ---------------------------------------------------- @@ -141,7 +149,12 @@ class Retriever: if self.cfg.embed_off: return {} self._ensure_matrix() - if self._mat is None or len(self._mat) == 0: + if ( + self._mat is None + or self._mat_norm is None + or self._mat_chunk_ids is None + or len(self._mat) == 0 + ): return {} if self._client is None: self._client = OllamaClient( @@ -166,7 +179,7 @@ class Retriever: for i in order: if len(out) >= limit: break - cid = self._mat_chunk_ids[i] + cid = self._mat_chunk_ids[int(i)] if scope_ids is not None and cid not in scope_ids: continue out[cid] = float(sims[i]) @@ -392,16 +405,17 @@ class Retriever: deterministisch in Direktzahlung und Alternativen zerlegt. Das gilt auch für den Offline-Retrieval-Eval, der keinen LLM-Planer aufruft. """ - n = n_entries or self.cfg.context_blocks - latest_year = str(self._stand_max)[:4] - deterministic = decision_support_plan(question, default_year=latest_year) - if deterministic: - sub_queries, _qtype = deterministic - return self.search_multi(sub_queries, n_entries=n) - fused, bm_all, dn_all = self._fuse_queries( - [SubQuery(text=question)], self.cfg.candidate_pool - ) - return self._context_from_fused(fused, n, bm_all, dn_all) + with self._lock: + n = n_entries or self.cfg.context_blocks + latest_year = str(self._stand_max)[:4] + deterministic = decision_support_plan(question, default_year=latest_year) + if deterministic: + sub_queries, _qtype = deterministic + return self.search_multi(sub_queries, n_entries=n) + fused, bm_all, dn_all = self._fuse_queries( + [SubQuery(text=question)], self.cfg.candidate_pool + ) + return self._context_from_fused(fused, n, bm_all, dn_all) def search_multi( self, sub_queries: list, n_entries: int | None = None @@ -412,44 +426,45 @@ class Retriever: seiner besten Quelle hinein — sonst dominieren Einträge, die in mehreren Sub-Queries mittelgut matchen). Temporal-Intent: kv- Einträge im gefragten Geltungsjahr erhalten temporal_boost.""" - n = n_entries or self.cfg.context_blocks - if ( - len(sub_queries) == 1 - and sub_queries[0].scope is None - and sub_queries[0].stand_year is None - ): - return self.search(sub_queries[0].text, n_entries=n) - pool = self.cfg.candidate_pool - fused_total: dict[int, float] = {} - bm_all: set[int] = set() - dn_all: set[int] = set() - per_query_entries: list[list[str]] = [] - for sq in sub_queries: - bm, dn, fused_q = self._search_one(sq, pool) - bm_all.update(bm) - dn_all.update(dn) - for cid, score in fused_q.items(): - fused_total[cid] = fused_total.get(cid, 0.0) + score - rows_q = self._fetch_chunks(list(fused_q)) - entry_best: dict[str, float] = {} - for cid, score in fused_q.items(): - row = rows_q.get(cid) - if row is not None: - eid = row["entry_id"] - entry_best[eid] = max(entry_best.get(eid, 0.0), score) - per_query_entries.append( - [e for e, _ in sorted(entry_best.items(), key=lambda kv: -kv[1])] + with self._lock: + n = n_entries or self.cfg.context_blocks + if ( + len(sub_queries) == 1 + and sub_queries[0].scope is None + and sub_queries[0].stand_year is None + ): + return self.search(sub_queries[0].text, n_entries=n) + pool = self.cfg.candidate_pool + fused_total: dict[int, float] = {} + bm_all: set[int] = set() + dn_all: set[int] = set() + per_query_entries: list[list[str]] = [] + for sq in sub_queries: + bm, dn, fused_q = self._search_one(sq, pool) + bm_all.update(bm) + dn_all.update(dn) + for cid, score in fused_q.items(): + fused_total[cid] = fused_total.get(cid, 0.0) + score + rows_q = self._fetch_chunks(list(fused_q)) + entry_best: dict[str, float] = {} + for cid, score in fused_q.items(): + row = rows_q.get(cid) + if row is not None: + eid = row["entry_id"] + entry_best[eid] = max(entry_best.get(eid, 0.0), score) + per_query_entries.append( + [e for e, _ in sorted(entry_best.items(), key=lambda kv: -kv[1])] + ) + reserved: list[str] = [] + for entries in per_query_entries: + taken = 0 + for eid in entries: + if taken >= self.cfg.per_query_slots: + break + if eid not in reserved: + reserved.append(eid) + taken += 1 + years = {sq.stand_year for sq in sub_queries if sq.stand_year} + return self._context_from_fused( + fused_total, n, bm_all, dn_all, stand_years=years, reserved=reserved ) - reserved: list[str] = [] - for entries in per_query_entries: - taken = 0 - for eid in entries: - if taken >= self.cfg.per_query_slots: - break - if eid not in reserved: - reserved.append(eid) - taken += 1 - years = {sq.stand_year for sq in sub_queries if sq.stand_year} - return self._context_from_fused( - fused_total, n, bm_all, dn_all, stand_years=years, reserved=reserved - ) \ No newline at end of file diff --git a/pv-agent/docs/API.md b/pv-agent/docs/API.md new file mode 100644 index 0000000..307c57e --- /dev/null +++ b/pv-agent/docs/API.md @@ -0,0 +1,168 @@ +# PV RAG Agent — HTTP API v1 + +## Zweck und Grenze + +Die API ist der eigenständige Zugriffspunkt für CLI-, Web- und spätere +Odoo-Clients. Version 1 ist ein **zustandsloser Wissensdienst**: + +- verarbeitet ausschließlich `question`, optionale Retrieval-Tiefe `top_k` + und den festen Modus `knowledge`; +- antwortet ausschließlich aus der Layer-2-Wissensbasis; +- nimmt keine Mandanten-, Mitarbeiter- oder Abrechnungsobjekte entgegen; +- speichert keinen Gesprächsverlauf; +- nutzt keine Websuche und keine externen Tools. + +Personenbezogene Lohndaten dürfen nicht in den Fragetext eingebettet werden. +Die spätere Odoo-Lohndatenintegration bekommt einen getrennten Vertrag mit +Mandantenbindung, Rollenprüfung, Datenminimierung, Auditierung und definierter +Aufbewahrung. Ein allgemeiner API-Key allein reicht dafür nicht aus. + +## Endpunkte + +| Methode | Pfad | Auth | Zweck | +|---|---|---|---| +| `POST` | `/v1/ask` | Service-Key | belegte Wissensantwort | +| `GET` | `/v1/health` | öffentlich | Readiness ohne interne Hostdetails | +| `POST` | `/v1/reindex` | Admin-Key | Index nach KB-Änderung neu aufbauen | + +Die bisherigen Pfade `/ask`, `/health` und `/reindex` bleiben vorläufig als +`deprecated` Kompatibilitätsrouten erhalten und liefern denselben Vertrag. + +## Authentisierung + +```bash +export PV_API_KEY='service-key-aus-secret-store' +export PV_ADMIN_API_KEY='separater-admin-key-aus-secret-store' +python -m agent.cli serve --host 127.0.0.1 +``` + +Requests senden den Key als Bearer-Token: + +```http +Authorization: Bearer +``` + +- Ist `PV_ADMIN_API_KEY` leer, verwendet `/v1/reindex` den normalen + `PV_API_KEY`. +- Sind beide Variablen leer, läuft der Dienst zur lokalen Entwicklung ohne + Authentisierung. Der CLI-Start lehnt in diesem Zustand nicht-lokale Bind- + Adressen wie `0.0.0.0` ab. Ein exponiertes Deployment muss mindestens + `PV_API_KEY` setzen. +- Keys gehören in einen Secret Store bzw. eine nicht versionierte + Umgebungsdatei; sie dürfen nicht in Odoo-Quellcode oder Git liegen. +- TLS wird am Reverse Proxy bzw. Service Mesh terminiert. Bearer-Tokens dürfen + nicht unverschlüsselt über fremde Netze übertragen werden. + +## `POST /v1/ask` + +Request: + +```json +{ + "question": "Wie hoch ist der steuerfreie Tagesgeldsatz?", + "top_k": 8, + "mode": "knowledge" +} +``` + +Unbekannte Felder werden mit HTTP `422` abgewiesen. Das ist insbesondere die +technische Vertragsgrenze gegen ad-hoc-Felder wie `employee_data` oder +`payroll_context`. + +Response (gekürzt): + +```json +{ + "api_version": "v1", + "request_id": "odoo-request-123", + "status": "answered", + "question": "Wie hoch ist der steuerfreie Tagesgeldsatz?", + "answer": "... [lb-rei-09] ...", + "refused": false, + "verified": true, + "citations": ["lb-rei-09"], + "sources": [ + { + "id": "lb-rei-09", + "title": "...", + "section": "Kernwerte & Fristen", + "stand": "2026-01", + "work": "..." + } + ], + "conflicts": [], + "assumptions": [], + "clarification_question": null, + "alternatives": [], + "answer_type": "specific", + "planned": false, + "planned_queries": [], + "grounding": { + "data_scope": "knowledge_base_only", + "citations_verified": true, + "context_count": 8, + "regenerations": 0 + }, + "n_context": 8, + "model": "qwen3.8:27b", + "latency_ms": 38800, + "regenerations": 0 +} +``` + +### Semantik + +- `status=answered`: beantwortet und zitierseitig verifiziert. +- `status=refused`: Wissensbasis deckt die Frage nicht; sichere Verweigerung. +- `status=uncertain`: Regenerierung konnte die Grounding-Regeln nicht erfüllen. +- `sources` enthält nur tatsächlich zitierte Quellen, nicht sämtliche + Retrieval-Treffer. +- `conflicts` enthält mit `⚠` markierte, bereits in der belegten Antwort + vorkommende Konfliktpassagen und deren Quellen-IDs. +- `clarification_question` wird aus einer abschließenden Rückfrage übernommen. +- `assumptions` und `alternatives` sind bereits stabile Vertragsfelder, bleiben + in v1 aber leer. Der Dienst errät diese Strukturen nicht aus Freitext; ihre + spätere Befüllung benötigt einen eigenen belegbaren Generierungsvertrag. +- `grounding.data_scope=knowledge_base_only` ist in v1 unveränderlich. + +Der Client darf `verified=false` nicht als normale Fachantwort darstellen. +Empfohlen ist ein sichtbarer Warnzustand ohne automatische Folgeverarbeitung. + +## Request-Korrelation + +Ein Client kann einen technisch neutralen Header mitsenden: + +```http +X-Request-ID: odoo-request-123 +``` + +Erlaubt sind 1–128 Zeichen aus `A-Z`, `a-z`, `0-9`, `.`, `_`, `:`, `-`. +Ungültige oder fehlende IDs werden durch eine zufällige ID ersetzt. Die ID +wird im Response-Header und Response-Body zurückgegeben. Keine Namen, +Personalnummern oder sonstigen personenbezogenen Daten als Request-ID nutzen. + +## Fehlerverhalten + +- `401`: fehlender oder falscher Bearer-Key; +- `422`: ungültiger Request bzw. nicht erlaubte Felder; +- `503`: Index, Modellserver oder Reindex vorübergehend nicht verfügbar. + +Öffentliche Fehlerantworten enthalten keine internen Hosts, Dateipfade oder +Exception-Texte. Der Server loggt technische Fehler mit `request_id`. + +## Odoo-Anbindung der Wissens-API + +Odoo soll `/v1/ask` serverseitig aufrufen, nicht direkt aus dem Browser: + +1. Service-URL und Key verschlüsselt bzw. als Deployment-Secret verwalten; +2. Benutzerzugriff in Odoo über eine eigene Gruppe steuern; +3. nur die Frage und eine nicht personenbezogene `X-Request-ID` senden; +4. `answer`, `sources`, `conflicts` und `clarification_question` rendern; +5. `status` und `verified` zwingend auswerten; +6. keine eigene Retrieval- oder Grounding-Logik in Odoo duplizieren. + +Die Payroll-Datenintegration wird nicht durch zusätzliche freie JSON-Felder an +`/v1/ask` umgesetzt. Sie benötigt mindestens: Odoo-seitige Datensatzregeln, +service-seitige Tenant-Bindung, erlaubte Datenprojektionen statt Rohobjekten, +Zweckbindung, Audit-Events, kurze Aufbewahrung und Tests gegen Cross-Tenant- +Datenabfluss. diff --git a/pv-agent/planung.md b/pv-agent/planung.md index 5eaa9a7..3d10b62 100644 --- a/pv-agent/planung.md +++ b/pv-agent/planung.md @@ -342,4 +342,19 @@ tests/ # pytest: Ingest-, Retrieval-, Grounding-Unit-Tests quellentreuen Volltexte ist die Publikations-/Lizenzfreigabe bewusst zu bestätigen; sie folgt nicht automatisch aus der Regel zu amtlichen Gesetzestexten. +- **API-first-Festigung (2026-09-16, D19):** Der eigenständige Service hat + einen versionierten, strikt validierten Vertrag unter `/v1/ask`, + `/v1/health` und `/v1/reindex`; die alten Pfade bleiben deprecated. Antworten + liefern Status, Quellen, extrahierte belegte Konflikte, Rückfrage, + Planungsmetadaten und einen expliziten Grounding-Block. Request-IDs erlauben + technische Korrelation ohne Personenkennzeichen. Optionaler Bearer-Schutz + trennt Service- und Admin-Key; ein nicht-lokaler Bind ohne Service-Key wird + fail-closed abgelehnt. Health und Fehlerantworten geben keine interne + Ollama-URL bzw. Exception-Details mehr aus. Der gemeinsam genutzte SQLite- + Retriever ist für FastAPI-Worker-Threads serialisiert. **Privacy-Grenze:** v1 + akzeptiert ausschließlich `mode=knowledge` und keine freien Payroll-/ + Mitarbeiterdatenfelder. Die spätere Odoo-Lohndatenintegration erhält einen + getrennten tenant-autorisierten Vertrag; sie wird nicht durch Anhängen von + Rohdaten an `/v1/ask` umgesetzt. Vertrag und Odoo-Clientregeln: + `docs/API.md`. - Betrieb: `agent/README.md`. diff --git a/pv-agent/tests/test_api.py b/pv-agent/tests/test_api.py new file mode 100644 index 0000000..9ca39ea --- /dev/null +++ b/pv-agent/tests/test_api.py @@ -0,0 +1,187 @@ +"""Offline-Tests für den versionierten API-Vertrag und seine Schutzgrenzen.""" +from __future__ import annotations + +from dataclasses import dataclass + +from fastapi.testclient import TestClient + +from agent.api import app +from agent.cli import _cmd_serve, _is_loopback_bind +from agent.config import Config + + +class FakeClient: + def is_up(self): + return True + + def close(self): + pass + + +class FakeRetriever: + def stats(self): + return { + "n_entries": 3, + "n_chunks": 9, + "n_vectors": 9, + "dense_available": True, + "stand_min": "202601", + "stand_max": "202612", + "built_at": "2026-09-16T00:00:00Z", + } + + def close(self): + pass + + +def answer_result(answer: str = "Belegte Antwort [lb-min-01].") -> dict: + return { + "question": "Was gilt?", + "answer": answer, + "refused": False, + "verified": True, + "citations": ["lb-min-01"], + "sources": [ + { + "id": "lb-min-01", + "title": "Testquelle", + "section": "Zusammenfassung", + "stand": "2026-01", + "work": "Testwerk", + } + ], + "n_context": 1, + "model": "test-model", + "regenerations": 0, + "latency_ms": 12, + "draft": None, + "answer_type": "specific", + "planned": False, + "planned_queries": [ + {"text": "Was gilt?", "stand_year": None, "scope": None} + ], + } + + +def configure_state(*, api_key: str = "", admin_key: str = "") -> None: + app.state.rag.cfg = Config(api_key=api_key, admin_api_key=admin_key) + app.state.rag.client = FakeClient() + app.state.rag.retriever = FakeRetriever() + + +def test_nonlocal_bind_without_api_key_is_rejected(capsys): + class Args: + host = "0.0.0.0" + + assert _is_loopback_bind("127.0.0.1") is True + assert _is_loopback_bind("::1") is True + assert _is_loopback_bind("0.0.0.0") is False + assert _cmd_serve(Args(), Config()) == 2 + assert "PV_API_KEY" in capsys.readouterr().err + + +def test_v1_ask_requires_configured_bearer(monkeypatch): + monkeypatch.setattr("agent.api.answer_question", lambda *args, **kwargs: answer_result()) + with TestClient(app) as client: + configure_state(api_key="service-secret") + unauthorized = client.post("/v1/ask", json={"question": "Was gilt?"}) + assert unauthorized.status_code == 401 + assert unauthorized.headers["www-authenticate"] == "Bearer" + + response = client.post( + "/v1/ask", + json={"question": "Was gilt?"}, + headers={ + "Authorization": "Bearer service-secret", + "X-Request-ID": "odoo-42", + }, + ) + assert response.status_code == 200 + body = response.json() + assert body["api_version"] == "v1" + assert body["request_id"] == "odoo-42" + assert response.headers["x-request-id"] == "odoo-42" + assert body["status"] == "answered" + assert body["grounding"] == { + "data_scope": "knowledge_base_only", + "citations_verified": True, + "context_count": 1, + "regenerations": 0, + } + assert body["planned_queries"][0]["text"] == "Was gilt?" + + +def test_v1_contract_rejects_payroll_context(monkeypatch): + monkeypatch.setattr("agent.api.answer_question", lambda *args, **kwargs: answer_result()) + with TestClient(app) as client: + configure_state() + response = client.post( + "/v1/ask", + json={ + "question": "Was gilt?", + "employee_data": {"name": "Max", "salary": 5000}, + }, + ) + assert response.status_code == 422 + + +def test_v1_extracts_grounded_conflict_and_clarification(monkeypatch): + answer = ( + "⚠ Quelle A und Quelle B widersprechen einander [lb-min-01].\n\n" + "Für welche Branche soll die Aussage geprüft werden?" + ) + monkeypatch.setattr( + "agent.api.answer_question", lambda *args, **kwargs: answer_result(answer) + ) + with TestClient(app) as client: + configure_state() + response = client.post("/v1/ask", json={"question": "Was gilt?"}) + assert response.status_code == 200 + body = response.json() + assert body["conflicts"][0]["source_ids"] == ["lb-min-01"] + assert body["clarification_question"] == ( + "Für welche Branche soll die Aussage geprüft werden?" + ) + assert body["assumptions"] == [] + assert body["alternatives"] == [] + + +def test_health_does_not_expose_internal_ollama_url(): + with TestClient(app) as client: + configure_state(api_key="service-secret") + response = client.get("/v1/health") + assert response.status_code == 200 + body = response.json() + assert body["status"] == "ok" + assert body["authentication_enabled"] is True + assert "ollama_url" not in response.text + assert "100.103.83.12" not in response.text + + +@dataclass +class FakeStats: + embed_error: str | None = None + + def as_dict(self): + return {"n_entries": 3, "n_chunks": 9} + + +def test_reindex_uses_separate_admin_key(monkeypatch): + monkeypatch.setattr( + "agent.api.build_index", lambda cfg, client=None: FakeStats() + ) + with TestClient(app) as client: + configure_state(api_key="service-secret", admin_key="admin-secret") + denied = client.post( + "/v1/reindex", + headers={"Authorization": "Bearer service-secret"}, + ) + assert denied.status_code == 401 + + response = client.post( + "/v1/reindex", + headers={"Authorization": "Bearer admin-secret"}, + ) + assert response.status_code == 200 + assert response.json()["api_version"] == "v1" + assert response.json()["n_entries"] == 3 diff --git a/pv-agent/tests/test_retrieve.py b/pv-agent/tests/test_retrieve.py index b59f7fa..4bd379a 100644 --- a/pv-agent/tests/test_retrieve.py +++ b/pv-agent/tests/test_retrieve.py @@ -1,5 +1,7 @@ """Tests: Hybrid-Retrieval (BM25-only offline): Fusion, Entry-Dedup, cross_ref-Erweiterung, leeres Retrieval.""" +from concurrent.futures import ThreadPoolExecutor + import pytest from agent.retrieve import Retriever @@ -62,6 +64,14 @@ def test_recency_boost_prefers_newer_stand(retriever): assert main[0].entry_id in {"lb-min-01", "lb-min-02"} +def test_shared_retriever_is_safe_across_worker_threads(retriever): + """FastAPI darf dieselbe Retriever-Instanz in mehreren Threads nutzen.""" + questions = ["Altersteilzeit Lohnausgleich", "Urlaubsanspruch Werktage"] * 4 + with ThreadPoolExecutor(max_workers=4) as pool: + result_sets = list(pool.map(retriever.search, questions)) + assert all(results for results in result_sets) + + def test_stats_report(mini_index): r = Retriever(mini_index) try: