feat(agent): add index bootstrap and answer feedback

pv-agent/agent/README.md

@@ -62,11 +62,13 @@ oder Abrechnungsobjekte.
 ## Docker Compose auf `100.103.83.12`
 
 Der Stack in `compose.yaml` verbindet den Agenten mit dem bereits vorhandenen
-externen Docker-Netz `ollama-default`; Ollama wird nicht dupliziert. Setup und
+externen Docker-Netz `ollama_default`; Ollama wird nicht dupliziert. Setup und
 Smoke-Tests: `docs/DOCKER.md`.
 
 ```bash
-cp .env.example .env       # API-Key und Ollama-DNS-Alias setzen
+cp .env.example .env       # API-Key, UID/GID und Ollama-DNS-Alias setzen
+mkdir -p data
+sudo chown 1000:1000 data  # an PUID:PGID aus .env anpassen
 
 docker compose build
 docker compose up -d
@@ -74,7 +76,10 @@ docker compose up -d
 
 UI: `http://100.103.83.12:8080/`. `data/` wird schreibbar und
 `wissensbasis/` read-only eingebunden; weder Index noch lokale Quellkorpora
-landen im Image.
+landen im Image. Fehlt `data/index.db`, baut der Container ihn vor dem API-Start
+automatisch mit vollständigen Embeddings auf. Fragen, Antworten und Bewertungen
+werden im Compose-Profil standardmäßig nach `data/audit.db` und als
+strukturierte `AUDIT`-Zeilen in die Containerlogs geschrieben.
 
 ## Konfiguration (Umgebungsvariablen)
 
@@ -101,6 +106,11 @@ landen im Image.
 | `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 |
+| `PV_AUDIT_ENABLED` | `false` | Interaktions- und Bewertungsprotokoll aktivieren (Compose: `true`) |
+| `PV_AUDIT_DB_PATH` | `data/audit.db` | persistente SQLite-Datei für Fragen, Antworten und Bewertungen |
+| `PV_AUDIT_LOG_CONTENT` | `true` | Freitexte speichern; `false` = nur technische Metadaten und KB-IDs |
+| `PV_AUDIT_STDOUT` | `false` | strukturierte Audit-Ereignisse zusätzlich nach stdout (Compose: `true`) |
+| `PV_AUDIT_RETENTION_DAYS` | `30` | Aufbewahrung; `0` deaktiviert automatische Löschung |
 
 ## Deployment auf dem Host (Ollama-Maschine)
 

pv-agent/agent/api.py

@@ -20,6 +20,7 @@ from fastapi.responses import FileResponse
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
 from pydantic import BaseModel, ConfigDict, Field
 
+from .audit import AuditStore
 from .config import Config
 from .generate import CITE_RE, answer_question
 from .ingest import build_index
@@ -97,6 +98,7 @@ class AskResponse(StrictModel):
     model: str
     latency_ms: int
     regenerations: int = 0
+    ratings_enabled: bool = False
 
 
 class HealthResponse(StrictModel):
@@ -106,6 +108,20 @@ class HealthResponse(StrictModel):
     index: dict
     ollama_up: bool
     authentication_enabled: bool
+    ratings_enabled: bool
+
+
+class RatingRequest(StrictModel):
+    request_id: str = Field(pattern=r"^[A-Za-z0-9._:-]{1,128}$")
+    rating: Literal["up", "down"]
+    feedback: str | None = Field(default=None, max_length=1000)
+
+
+class RatingResponse(StrictModel):
+    api_version: Literal["v1"] = API_VERSION
+    request_id: str
+    rating: Literal["up", "down"]
+    accepted: Literal[True] = True
 
 
 class AppState:
@@ -113,6 +129,7 @@ class AppState:
         self.cfg: Config | None = None
         self.client: OllamaClient | None = None
         self.retriever: Retriever | None = None
+        self.audit: AuditStore | None = None
         self.maintenance_lock = threading.Lock()
         self.state_lock = threading.RLock()
 
@@ -139,6 +156,14 @@ class AppState:
                 self.retriever = Retriever(self.ensure())
             return self.retriever
 
+    def get_audit(self) -> AuditStore | None:
+        with self.state_lock:
+            if not self.ensure().audit_enabled:
+                return None
+            if self.audit is None:
+                self.audit = AuditStore(self.ensure())
+            return self.audit
+
     def reset_retriever(self) -> None:
         with self.state_lock:
             if self.retriever is not None:
@@ -152,6 +177,8 @@ async def lifespan(app: FastAPI):
     yield
     rag: AppState = app.state.rag
     rag.reset_retriever()
+    if rag.audit:
+        rag.audit.close()
     if rag.client:
         rag.client.close()
 
@@ -266,7 +293,9 @@ def _extract_clarification(answer: str, refused: bool) -> str | None:
     return question or None
 
 
-def _response_from_result(result: dict, request_id: str) -> AskResponse:
+def _response_from_result(
+    result: dict, request_id: str, ratings_enabled: bool = False
+) -> AskResponse:
     if not result["verified"]:
         status = "uncertain"
     elif result["refused"]:
@@ -304,6 +333,7 @@ def _response_from_result(result: dict, request_id: str) -> AskResponse:
         model=result["model"],
         latency_ms=result["latency_ms"],
         regenerations=result["regenerations"],
+        ratings_enabled=ratings_enabled,
     )
 
 
@@ -326,7 +356,17 @@ def _ask(req: AskRequest, request: Request) -> AskResponse:
             detail="Der Wissensdienst ist vorübergehend nicht verfügbar.",
         ) from exc
     result.pop("draft", None)
-    return _response_from_result(result, request_id)
+    response = _response_from_result(
+        result, request_id, ratings_enabled=cfg.audit_enabled
+    )
+    audit = rag.get_audit()
+    if audit is not None:
+        try:
+            audit.record_interaction(response.model_dump(mode="json"))
+        except Exception:
+            # Die Fachantwort darf bei einem reinen Audit-Fehler nicht verloren gehen.
+            logger.exception("Audit-Protokollierung fehlgeschlagen request_id=%s", request_id)
+    return response
 
 
 @app.post("/v1/ask", response_model=AskResponse, dependencies=[Depends(require_api_access)])
@@ -359,6 +399,7 @@ def _health(request: Request) -> HealthResponse:
         index=index,
         ollama_up=ollama_up,
         authentication_enabled=bool(cfg.api_key),
+        ratings_enabled=cfg.audit_enabled,
     )
 
 
@@ -372,6 +413,28 @@ def health_compat(request: Request) -> HealthResponse:
     return _health(request)
 
 
+@app.post(
+    "/v1/ratings",
+    response_model=RatingResponse,
+    dependencies=[Depends(require_api_access)],
+)
+def rate_answer(req: RatingRequest, request: Request) -> RatingResponse:
+    rag: AppState = request.app.state.rag
+    audit = rag.get_audit()
+    if audit is None:
+        raise HTTPException(status_code=503, detail="Bewertungen sind nicht aktiviert.")
+    try:
+        audit.record_rating(req.request_id, req.rating, req.feedback)
+    except KeyError as exc:
+        raise HTTPException(status_code=404, detail="Antwort nicht gefunden.") from exc
+    except Exception as exc:
+        logger.exception("Bewertung fehlgeschlagen request_id=%s", req.request_id)
+        raise HTTPException(
+            status_code=503, detail="Bewertung konnte nicht gespeichert werden."
+        ) from exc
+    return RatingResponse(request_id=req.request_id, rating=req.rating)
+
+
 def _reindex(request: Request) -> dict:
     rag: AppState = request.app.state.rag
     cfg = rag.ensure()

pv-agent/agent/audit.py

@@ -0,0 +1,179 @@
+"""Persistente, strukturierte Interaktions- und Bewertungsprotokolle."""
+from __future__ import annotations
+
+import json
+import sqlite3
+import threading
+import time
+from pathlib import Path
+
+from .config import Config
+
+SCHEMA = """
+CREATE TABLE IF NOT EXISTS interactions (
+    request_id TEXT PRIMARY KEY,
+    created_at INTEGER NOT NULL,
+    question TEXT,
+    answer TEXT,
+    status TEXT NOT NULL,
+    verified INTEGER NOT NULL,
+    refused INTEGER NOT NULL,
+    citations_json TEXT NOT NULL,
+    sources_json TEXT NOT NULL,
+    conflicts_json TEXT NOT NULL,
+    planned_queries_json TEXT NOT NULL,
+    model TEXT NOT NULL,
+    latency_ms INTEGER NOT NULL,
+    n_context INTEGER NOT NULL,
+    regenerations INTEGER NOT NULL
+);
+CREATE TABLE IF NOT EXISTS ratings (
+    request_id TEXT PRIMARY KEY,
+    created_at INTEGER NOT NULL,
+    updated_at INTEGER NOT NULL,
+    rating TEXT NOT NULL CHECK (rating IN ('up', 'down')),
+    feedback TEXT,
+    FOREIGN KEY (request_id) REFERENCES interactions(request_id) ON DELETE CASCADE
+);
+CREATE INDEX IF NOT EXISTS interactions_created_at_idx
+    ON interactions(created_at DESC);
+CREATE INDEX IF NOT EXISTS ratings_updated_at_idx
+    ON ratings(updated_at DESC);
+"""
+
+
+class AuditStore:
+    def __init__(self, cfg: Config):
+        self.cfg = cfg
+        path = Path(cfg.audit_db_path)
+        path.parent.mkdir(parents=True, exist_ok=True)
+        self._lock = threading.RLock()
+        self._con = sqlite3.connect(path, check_same_thread=False)
+        self._con.row_factory = sqlite3.Row
+        self._con.execute("PRAGMA foreign_keys = ON")
+        self._con.execute("PRAGMA journal_mode = WAL")
+        self._con.executescript(SCHEMA)
+        self._delete_expired()
+
+    def close(self) -> None:
+        with self._lock:
+            self._con.close()
+
+    def _delete_expired(self) -> None:
+        if self.cfg.audit_retention_days <= 0:
+            return
+        cutoff = int(time.time()) - self.cfg.audit_retention_days * 86_400
+        with self._lock:
+            self._con.execute("DELETE FROM interactions WHERE created_at < ?", (cutoff,))
+            self._con.commit()
+
+    def record_interaction(self, payload: dict) -> None:
+        now = int(time.time())
+        include_content = self.cfg.audit_log_content
+        question = payload.get("question") if include_content else None
+        answer = payload.get("answer") if include_content else None
+        sources = payload.get("sources", []) if include_content else []
+        conflicts = payload.get("conflicts", []) if include_content else []
+        planned_queries = payload.get("planned_queries", []) if include_content else []
+        values = (
+            payload["request_id"],
+            now,
+            question,
+            answer,
+            payload["status"],
+            int(bool(payload["verified"])),
+            int(bool(payload["refused"])),
+            json.dumps(payload.get("citations", []), ensure_ascii=False),
+            json.dumps(sources, ensure_ascii=False),
+            json.dumps(conflicts, ensure_ascii=False),
+            json.dumps(planned_queries, ensure_ascii=False),
+            payload.get("model", ""),
+            int(payload.get("latency_ms", 0)),
+            int(payload.get("n_context", 0)),
+            int(payload.get("regenerations", 0)),
+        )
+        with self._lock:
+            self._con.execute(
+                "INSERT OR REPLACE INTO interactions("
+                "request_id, created_at, question, answer, status, verified, refused, "
+                "citations_json, sources_json, conflicts_json, planned_queries_json, "
+                "model, latency_ms, n_context, regenerations"
+                ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+                values,
+            )
+            self._con.commit()
+        if self.cfg.audit_stdout:
+            if include_content:
+                event = {"event": "agent_interaction", "created_at": now, **payload}
+            else:
+                event = {
+                    "event": "agent_interaction",
+                    "created_at": now,
+                    "request_id": payload["request_id"],
+                    "status": payload["status"],
+                    "verified": payload["verified"],
+                    "refused": payload["refused"],
+                    "citations": payload.get("citations", []),
+                    "model": payload.get("model", ""),
+                    "latency_ms": payload.get("latency_ms", 0),
+                    "n_context": payload.get("n_context", 0),
+                    "regenerations": payload.get("regenerations", 0),
+                }
+            print("AUDIT " + json.dumps(event, ensure_ascii=False), flush=True)
+
+    def record_rating(self, request_id: str, rating: str, feedback: str | None) -> None:
+        now = int(time.time())
+        stored_feedback = feedback if self.cfg.audit_log_content else None
+        with self._lock:
+            exists = self._con.execute(
+                "SELECT 1 FROM interactions WHERE request_id = ?", (request_id,)
+            ).fetchone()
+            if exists is None:
+                raise KeyError(request_id)
+            self._con.execute(
+                "INSERT INTO ratings(request_id, created_at, updated_at, rating, feedback) "
+                "VALUES (?, ?, ?, ?, ?) "
+                "ON CONFLICT(request_id) DO UPDATE SET "
+                "updated_at=excluded.updated_at, rating=excluded.rating, "
+                "feedback=excluded.feedback",
+                (request_id, now, now, rating, stored_feedback),
+            )
+            self._con.commit()
+        if self.cfg.audit_stdout:
+            print(
+                "AUDIT "
+                + json.dumps(
+                    {
+                        "event": "agent_rating",
+                        "created_at": now,
+                        "request_id": request_id,
+                        "rating": rating,
+                        "feedback": stored_feedback,
+                    },
+                    ensure_ascii=False,
+                ),
+                flush=True,
+            )
+
+    def recent(self, limit: int = 20) -> list[dict]:
+        with self._lock:
+            rows = self._con.execute(
+                "SELECT i.*, r.rating, r.feedback, r.updated_at AS rating_updated_at "
+                "FROM interactions i LEFT JOIN ratings r USING(request_id) "
+                "ORDER BY i.created_at DESC LIMIT ?",
+                (limit,),
+            ).fetchall()
+        out = []
+        for row in rows:
+            item = dict(row)
+            for field in (
+                "citations_json",
+                "sources_json",
+                "conflicts_json",
+                "planned_queries_json",
+            ):
+                item[field.removesuffix("_json")] = json.loads(item.pop(field))
+            item["verified"] = bool(item["verified"])
+            item["refused"] = bool(item["refused"])
+            out.append(item)
+        return out

pv-agent/agent/bootstrap.py

@@ -0,0 +1,84 @@
+"""Container-Bootstrap: vollständigen Produktionsindex vor API-Start sicherstellen."""
+from __future__ import annotations
+
+import json
+import sqlite3
+from pathlib import Path
+
+from .config import Config
+from .ingest import build_index
+from .ollama_client import OllamaClient
+
+
+def index_is_ready(cfg: Config) -> bool:
+    path = Path(cfg.db_path)
+    if not path.is_file() or path.stat().st_size == 0:
+        return False
+    try:
+        con = sqlite3.connect(f"file:{path}?mode=ro", uri=True)
+        try:
+            n_chunks = con.execute("SELECT COUNT(*) FROM chunks").fetchone()[0]
+            if n_chunks <= 0:
+                return False
+            if cfg.embed_off:
+                return True
+            n_dense = con.execute(
+                "SELECT COUNT(*) FROM chunks c JOIN vectors v "
+                "ON v.content_hash = c.content_hash AND v.model = ?",
+                (cfg.embed_model,),
+            ).fetchone()[0]
+            return n_dense == n_chunks
+        finally:
+            con.close()
+    except (OSError, sqlite3.Error):
+        return False
+
+
+def ensure_index(cfg: Config) -> dict:
+    if index_is_ready(cfg):
+        result = {"event": "index_ready", "db_path": cfg.db_path}
+        print(json.dumps(result, ensure_ascii=False), flush=True)
+        return result
+
+    print(
+        json.dumps(
+            {
+                "event": "index_bootstrap_started",
+                "db_path": cfg.db_path,
+                "embed_model": None if cfg.embed_off else cfg.embed_model,
+            },
+            ensure_ascii=False,
+        ),
+        flush=True,
+    )
+    client = None
+    if not cfg.embed_off:
+        client = OllamaClient(
+            cfg.ollama_url,
+            embed_timeout_s=cfg.embed_timeout_s,
+            chat_timeout_s=cfg.chat_timeout_s,
+        )
+    try:
+        stats = build_index(cfg, client=client)
+    finally:
+        if client is not None:
+            client.close()
+    if stats.embed_error:
+        raise RuntimeError(
+            "Index-Bootstrap ohne vollständige Embeddings abgebrochen: "
+            + stats.embed_error
+        )
+    if not index_is_ready(cfg):
+        raise RuntimeError("Index-Bootstrap abgeschlossen, Index ist aber unvollständig")
+    result = {"event": "index_bootstrap_completed", **stats.as_dict()}
+    print(json.dumps(result, ensure_ascii=False), flush=True)
+    return result
+
+
+def main() -> int:
+    ensure_index(Config.from_env())
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

pv-agent/agent/cli.py

@@ -3,6 +3,7 @@
   python -m agent.cli ingest [--no-embed]   Index (neu) aufbauen
   python -m agent.cli ask "Frage?" [--top-k N] [--json]
   python -m agent.cli eval [--answers] [--limit N] [--k 8] [--json-out FILE]
+  python -m agent.cli audit [--limit N]
   python -m agent.cli serve [--host 0.0.0.0]
 """
 from __future__ import annotations
@@ -81,6 +82,21 @@ def _cmd_eval(args: argparse.Namespace, cfg: Config) -> int:
     return run_eval(cfg, args)
 
 
+def _cmd_audit(args: argparse.Namespace, cfg: Config) -> int:
+    from .audit import AuditStore
+
+    if not cfg.audit_enabled:
+        print("[Fehler] Audit ist nicht aktiviert (PV_AUDIT_ENABLED=true).", file=sys.stderr)
+        return 2
+    store = AuditStore(cfg)
+    try:
+        rows = store.recent(limit=args.limit)
+    finally:
+        store.close()
+    print(json.dumps(rows, indent=2, ensure_ascii=False))
+    return 0
+
+
 def _is_loopback_bind(host: str) -> bool:
     if host.casefold() == "localhost":
         return True
@@ -124,6 +140,9 @@ def main(argv: list[str] | None = None) -> int:
     p_eval.add_argument("--k", type=int, default=8, help="K für Recall@k")
     p_eval.add_argument("--json-out", default=None)
 
+    p_audit = sub.add_parser("audit", help="Letzte Fragen, Antworten und Bewertungen")
+    p_audit.add_argument("--limit", type=int, default=20)
+
     p_serve = sub.add_parser("serve", help="HTTP-API starten")
     p_serve.add_argument("--host", default="127.0.0.1")
 
@@ -132,6 +151,7 @@ def main(argv: list[str] | None = None) -> int:
         "ingest": _cmd_ingest,
         "ask": _cmd_ask,
         "eval": _cmd_eval,
+        "audit": _cmd_audit,
         "serve": _cmd_serve,
     }
     return handlers[args.cmd](args, cfg)

pv-agent/agent/config.py

@@ -92,6 +92,13 @@ class Config:
     api_key: str = ""          # leer = nur fuer lokale Entwicklung ohne Auth
     admin_api_key: str = ""    # leer = api_key auch fuer /reindex verwenden
 
+    # Audit / Feedback (Inhalte koennen personenbezogene Freitexte enthalten)
+    audit_enabled: bool = False
+    audit_db_path: str = "data/audit.db"
+    audit_log_content: bool = True
+    audit_stdout: bool = False
+    audit_retention_days: int = 30
+
     @classmethod
     def from_env(cls) -> Config:
         d = cls()
@@ -126,4 +133,11 @@ class Config:
             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),
+            audit_enabled=_env_bool("PV_AUDIT_ENABLED", d.audit_enabled),
+            audit_db_path=_env_str("PV_AUDIT_DB_PATH", d.audit_db_path),
+            audit_log_content=_env_bool("PV_AUDIT_LOG_CONTENT", d.audit_log_content),
+            audit_stdout=_env_bool("PV_AUDIT_STDOUT", d.audit_stdout),
+            audit_retention_days=_env_int(
+                "PV_AUDIT_RETENTION_DAYS", d.audit_retention_days
+            ),
         )