from __future__ import annotations import os import re import logging from huggingface_hub import snapshot_download from dataclasses import dataclass from pathlib import Path from threading import Lock from typing import Any, Callable, Dict, Iterable, List, Literal, Optional, TypedDict from uuid import uuid4 # .env laden, bevor irgendein os.getenv() ausgewertet wird. # # Ohne diesen Schritt ist die gesamte .env tote Konfiguration: CHROMA_COLLECTIONS # erreicht specs_from_env() nie, die Registry fällt auf das einzelne # CHROMA_COLLECTION zurück und das zweite Korpus (SGB V) wird nie geöffnet. # # override=False ist Absicht: echte Prozess-Umgebungsvariablen (Render, Docker, # HF Spaces) haben Vorrang vor der lokalen Entwicklungsdatei. try: from dotenv import load_dotenv _ENV_FILE = Path(__file__).resolve().parent.parent / ".env" load_dotenv(_ENV_FILE if _ENV_FILE.exists() else None, override=False) except ImportError: # python-dotenv ist optional; Deployment setzt echte Env-Vars. _ENV_FILE = None from fastapi import FastAPI, HTTPException, Request, Response from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import FileResponse from fastapi.staticfiles import StaticFiles from pydantic import BaseModel, Field from retriever import LegalRetriever from corpus_amendments import ( append_amendment_note, describe as describe_amendments, reconcile_negative_answer, ) from corpus_boundary import append_boundary_note, detect_external_references import amrl_austauschbarkeit import amrl_biosimilars import amrl_lifestyle import amrl_otc import amrl_substitution import amrl_tabakentwoehnung import amrl_verordnungsausschluss import norm_anchors import norm_verweise from corpus_registry import CorpusRegistry, build_registry from corpus_router import explain_routing, mentioned_corpora, route_question from federated_retriever import FederatedRetriever from llm_client_groq import ( DEFAULT_SYSTEM_PROMPT, ConversationMemory, GroqClient, classify_question, is_meta_question, ) from answer_composer import AnswerComposer from orchestrator import NEGATIVE_ANSWER_RE, LegalAnswerOrchestrator, OrchestratorOptions try: from langgraph.graph import StateGraph, END except ImportError: # LangGraph ist nur für Workflow-Visualisierung erforderlich. StateGraph = None # type: ignore[assignment] END = "__end__" # type: ignore[assignment] # ----------------------------------------------------------------------------- # Konfiguration # ----------------------------------------------------------------------------- logger = logging.getLogger(__name__) APP_TITLE = os.getenv("APP_TITLE", "Juristischer RAG-Prototyp") APP_VERSION = os.getenv("APP_VERSION", "1.1") # Robuste Pfad-Ermittlung: # Funktioniert sowohl für: # repo/app.py # als auch für: # repo/src/app.py APP_DIR = Path(__file__).resolve().parent if (APP_DIR / "static").exists(): BASE_DIR = APP_DIR else: BASE_DIR = APP_DIR.parent # ---- HF Dataset -> lokale DB nach /data ziehen (Spaces) ---- HF_DATASET_REPO = os.getenv("HF_DATASET_REPO", "AlixJabda/bav-ki-db") DATA_DIR = Path(os.getenv("DATA_DIR", "/data")) CHROMA_DATA_DIR = DATA_DIR / "chroma_db" # Optional Token, wenn Dataset private ist HF_TOKEN = os.getenv("HF_TOKEN") or os.getenv("HUGGINGFACE_HUB_TOKEN") # Nur einmal laden, wenn noch nicht vorhanden try: if not CHROMA_DATA_DIR.exists(): DATA_DIR.mkdir(parents=True, exist_ok=True) snapshot_download( repo_id=HF_DATASET_REPO, repo_type="dataset", local_dir=str(DATA_DIR), token=HF_TOKEN, ) logger.warning("HF dataset geladen nach %s", CHROMA_DATA_DIR) except Exception as exc: logger.warning("HF dataset download fehlgeschlagen: %s", exc) def _resolve_project_path(env_name: str, default_relative: str | Path) -> str: raw = os.getenv(env_name) path = Path(raw) if raw else Path(default_relative) if not path.is_absolute(): path = BASE_DIR / path return str(path.resolve()) # Deine aktuelle Struktur: # # static/ # ├── index.html # ├── js/ # │ └── script.js # ├── styles/ # │ └── style.css # └── berliner_apotheker_verein_cover.jpg STATIC_DIR = _resolve_project_path("STATIC_DIR", "static") INDEX_FILE = _resolve_project_path("INDEX_FILE", Path("static") / "index.html") # --- PDF-Quelldokumente für die interaktive Fundstellen-Anzeige --- # # Suchreihenfolge: # 1. PDF_DIR (Env, explizit) # 2. DATA_DIR/pdfs -> HF-Dataset-Snapshot auf Spaces (dort pdfs/ ablegen) # 3. BASE_DIR/data/pdfs bzw. BASE_DIR/data -> im Docker-Image mitgelieferte PDFs # Kuratierte Änderungen zu Korpora, die nur als Grundfassung plus separate # Änderungstexte veröffentlicht werden (siehe src/corpus_amendments.py). AMENDMENTS_FILE = Path( os.getenv("AMENDMENTS_FILE", str(BASE_DIR / "data" / "amabrv_aenderungen.json")) ) # Teil B der AM-RL-Anlage VII als nachschlagbare Tabelle. Diese Liste wird nicht # indiziert: die entscheidende Auskunft ist oft die negative, und die kann ein # Vektor-Retrieval nicht belegen (siehe src/amrl_substitution.py). AMRL_SUBSTITUTION_FILE = Path( os.getenv( "AMRL_SUBSTITUTION_FILE", str(BASE_DIR / "data" / "amrl_anlage_vii_teil_b.json"), ) ) # Teil A derselben Anlage: welche Darreichungsformen desselben Wirkstoffs # gegeneinander austauschbar sind. Ebenfalls Lookup statt Index — die Aussage # hängt an der Gruppenzugehörigkeit, und die ist nur über die vollständige # Tabelle bestimmbar (siehe src/amrl_austauschbarkeit.py). AMRL_AUSTAUSCHBARKEIT_FILE = Path( os.getenv( "AMRL_AUSTAUSCHBARKEIT_FILE", str(BASE_DIR / "data" / "amrl_anlage_vii_teil_a.json"), ) ) # Anlage VIIa: welches Biosimilar zu welchem Referenzarzneimittel zugelassen ist. # Ebenfalls Lookup — die Zuordnung hängt an der Tabellenzeile, und ein Retrieval # zeigte Prolia und ein Xgeva-Biosimilar nebeneinander, weil beide unter # „Denosumab" stehen (siehe src/amrl_biosimilars.py). AMRL_BIOSIMILARS_FILE = Path( os.getenv( "AMRL_BIOSIMILARS_FILE", str(BASE_DIR / "data" / "amrl_anlage_viia.json"), ) ) # Anlage I: die OTC-Ausnahmen vom Verordnungsausschluss des § 34 Absatz 1 SGB V. # Lookup wie die übrigen — und hier trägt die Negativauskunft ausdrücklich, weil # § 12 Absatz 10 AM-RL abschließend regelt (siehe src/amrl_otc.py). AMRL_OTC_FILE = Path( os.getenv( "AMRL_OTC_FILE", str(BASE_DIR / "data" / "amrl_anlage_i.json"), ) ) # Anlage III: die Verordnungseinschränkungen und -ausschlüsse — das Gegenstück # zu Anlage I. Lookup wie die übrigen, aber die Negativauskunft trägt hier # *nicht*: die Anlage ist eine Übersicht und nennt sich nirgends abschließend # (siehe src/amrl_verordnungsausschluss.py). AMRL_VERORDNUNG_FILE = Path( os.getenv( "AMRL_VERORDNUNG_FILE", str(BASE_DIR / "data" / "amrl_anlage_iii.json"), ) ) # Anlage II: die Lifestyle-Arzneimittel. Von den drei Listen zur # Verordnungsfähigkeit ist ihr Ausschluss der härteste — er ruht auf § 34 # Absatz 1 Satz 7 SGB V, und der medizinisch begründete Einzelfall des § 16 # Absatz 5 AM-RL führt an ihm vorbei. Die Negativauskunft trägt hier wie bei # Anlage III nicht (siehe src/amrl_lifestyle.py). AMRL_LIFESTYLE_FILE = Path( os.getenv( "AMRL_LIFESTYLE_FILE", str(BASE_DIR / "data" / "amrl_anlage_ii.json"), ) ) # Anlage IIa: die einzige Ausnahme von Anlage II. Sie gehört mit ihr zusammen — # der Ausschluss der Raucherentwöhnung und der Anspruch auf Tabakentwöhnung nach # § 34 Absatz 2 SGB V sind dieselbe Frage von zwei Seiten. Anders als bei Anlage # II trägt die Negativauskunft hier: § 14a Absatz 3 Satz 1 AM-RL zählt die # Ausnahme abschließend auf (siehe src/amrl_tabakentwoehnung.py). AMRL_TABAKENTWOEHNUNG_FILE = Path( os.getenv( "AMRL_TABAKENTWOEHNUNG_FILE", str(BASE_DIR / "data" / "amrl_anlage_iia.json"), ) ) _PDF_ENV_DIR = os.getenv("PDF_DIR") PDF_SEARCH_DIRS: List[Path] = [ path for path in [ Path(_PDF_ENV_DIR).resolve() if _PDF_ENV_DIR else None, DATA_DIR / "pdfs", BASE_DIR / "data" / "pdfs", BASE_DIR / "data", ] if path is not None ] _PDF_NAME_RE = re.compile(r"^[\w.\- ]+\.pdf$", re.I) SESSION_COOKIE = os.getenv("SESSION_COOKIE", "session_id") SESSION_COOKIE_SECURE = os.getenv("SESSION_COOKIE_SECURE", "false").lower() == "true" SESSION_COOKIE_MAX_AGE = int(os.getenv("SESSION_COOKIE_MAX_AGE", str(60 * 60 * 8))) # Keine lokalen Windows-Pfade im Deployment. # Standard: /data/chroma_db für Spaces; per Env überschreibbar. # Wichtig: Die Chroma-DB liegt im Unterordner chroma_db (dorthin lädt der # HF-Snapshot, siehe CHROMA_DATA_DIR). DATA_DIR selbst enthält nur eine leere # chroma.sqlite3 -> "verfügbare Collections=[]". CHROMA_PERSIST_DIR = os.getenv("CHROMA_PERSIST_DIR", str(CHROMA_DATA_DIR)) if not Path(CHROMA_PERSIST_DIR).is_absolute(): CHROMA_PERSIST_DIR = str((BASE_DIR / CHROMA_PERSIST_DIR).resolve()) CHROMA_COLLECTION = os.getenv("CHROMA_COLLECTION", "rv129") # --- DEBUG: effektive Pfade prüfen, nur temporär --- try: print("DEBUG BASE_DIR:", BASE_DIR) print("DEBUG STATIC_DIR:", STATIC_DIR) print("DEBUG INDEX_FILE:", INDEX_FILE) print("DEBUG effective CHROMA_PERSIST_DIR:", CHROMA_PERSIST_DIR) if Path("/data").exists(): print("DEBUG /data listing:", os.listdir("/data")[:50]) print("DEBUG exists /data/chroma.sqlite3:", (Path("/data") / "chroma.sqlite3").exists()) print( "DEBUG exists CHROMA_PERSIST_DIR/chroma.sqlite3:", (Path(CHROMA_PERSIST_DIR) / "chroma.sqlite3").exists(), ) except Exception as e: print("DEBUG path check failed:", repr(e)) # --- /DEBUG --- DEFAULT_CONTAINER_ID = os.getenv("DEFAULT_CONTAINER_ID", "Vertrag") # "auto" liest das Embedding-Modell aus der Chroma-Collection-Metadata, die die # Ingest-Pipeline schreibt. Damit kann die Query-Seite nie mit einem anderen # Modell arbeiten als der Index (die Ursache stiller Ranking-Ausfälle bzw. # harter Dimensionsfehler nach einem Modellwechsel wie MiniLM -> e5-base). EMBEDDING_MODEL = os.getenv("EMBEDDING_MODEL", "auto") # Cross-Encoder-Reranking der fusionierten Kandidaten. Hob das Hybrid-Retrieval # im Golden-Eval (configs/eval_cases.json im Ingest-Repo) von 17/18 auf 18/18. ENABLE_RERANKER = os.getenv("ENABLE_RERANKER", "true").lower() == "true" RERANKER_MODEL = os.getenv("RERANKER_MODEL", "cross-encoder/mmarco-mMiniLMv2-L12-H384-v1") RERANKER_CANDIDATES = int(os.getenv("RERANKER_CANDIDATES", "20")) GROQ_MODEL = os.getenv("GROQ_MODEL", "openai/gpt-oss-120b") GROQ_TEMPERATURE = float(os.getenv("GROQ_TEMPERATURE", "0.05")) # Bei einem Reasoning-Modell (openai/gpt-oss-*) teilen sich Denk- und # Antworttokens dieses Budget. Mit 1400 lief jede Antwort in `finish_reason: # length`, und wenn das Denken es aufbrauchte, kam gar keine — gemessen am # 19.08.2026: 1046 von 1400 Tokens allein fürs Reasoning. Der Client meldet die # Kürzung jetzt im Log und behandelt eine leere Antwort als Fehler. # # Nach oben begrenzt das **Groq-Kontingent**, nicht das Modell: die # Organisation liegt auf dem on_demand-Tier bei 8.000 Tokens pro Minute, und # geprüft wird je Anfrage über Prompt **plus** max_tokens. Ein voller Kontext # (DEFAULT_RAG_CONTEXT_CHARS=12.000) sind gemessen rund 4.825 Prompt-Tokens; # 4.000 Antworttokens ergaben 8.825 und damit HTTP 413. 2.800 lässt gut 350 # Tokens Luft. Wer mehr braucht, braucht einen höheren Tier — oder einen # kleineren Kontext. GROQ_MAX_TOKENS = int(os.getenv("GROQ_MAX_TOKENS", "2800")) # Nachschlag bei leerer Antwort. Ohne höheres Kontingent ist er ausgeschaltet # (Decke = Budget): ein zweiter Aufruf mit mehr Tokens liefe in denselben 413. GROQ_MAX_TOKENS_CEILING = int(os.getenv("GROQ_MAX_TOKENS_CEILING", str(GROQ_MAX_TOKENS))) GROQ_DEBUG_PROMPTS = os.getenv("GROQ_DEBUG_PROMPTS", "false").lower() == "true" DEFAULT_TOP_K = int(os.getenv("DEFAULT_TOP_K", "6")) DEFAULT_FETCH_K = int(os.getenv("DEFAULT_FETCH_K", "18")) DEFAULT_MAX_FINAL_RESULTS = int(os.getenv("DEFAULT_MAX_FINAL_RESULTS", "10")) DEFAULT_MAX_SOURCES = int(os.getenv("DEFAULT_MAX_SOURCES", "5")) # Garantierte Trefferplätze je abgefragtem Korpus, bevor global gekürzt wird. # Ohne diese Untergrenze verliert ein kleines Korpus strukturell: die 106 Chunks # der Abrechnungsvereinbarung kamen gegen die 6.670 des SGB V nie an, auch wenn # die Regelung nur bei ihnen stand. Auf 0 gesetzt gilt wieder der reine # Score-Schnitt. Reserviert wird höchstens die Hälfte der Ergebnisliste. MIN_HITS_PER_CORPUS = int(os.getenv("MIN_HITS_PER_CORPUS", "2")) # Wie viele Normen ein Lauf nachladen darf, weil der gefundene Text auf sie # verweist. Auf 0 gesetzt ist die Verweisauflösung aus, und der Retriever # verhält sich wie vorher. # # **Die Vorgabe ist seit dem 21.08.2026 die 0, und zwar gemessen.** Die Idee war # richtig — ein einziger § des SGB V zitiert ein Dutzend andere, und was der Text # selbst in Bezug nimmt, ist ein Signal, das die Ähnlichkeitssuche nicht hat. # Nur ist das Budget fest: ein nachgeladener Verweis *ergänzt* den Kontext # nicht, er *ersetzt* etwas. # # 19.08.2026, VERWEIS_SCORE 0.97 — Kettenabdeckung 85 % -> 75 %. # 21.08.2026, VERWEIS_SCORE 0.50 — auf der damaligen Suite unauffällig (beide # Konfigurationen 100 %), weil die Suite gesättigt war und nichts mehr # trennen konnte. Auf der um fünf Fälle erweiterten Suite: 96,2 % ohne # Verweise, 93,6 % mit. # # Sichtbar wird es an `kette_off_label_abgabe`: nachgeladen wurden § 3 SGB V # (solidarische Finanzierung) und § 300 SGB V (Abrechnung), beide ohne Bezug zur # Frage und beide mit dem festen Verweis-Score 0.50 — verdrängt hat das # § 31 Abs. 1 SGB V mit Score 0.871. Kein Fall der Suite gewinnt umgekehrt durch # die Verweise ein Kettenglied. # # Modul und Messschalter (`--ohne-verweise`) bleiben: die Frage ist beantwortet, # nicht erledigt. Wer den Verweis wieder aufnimmt, braucht ein eigenes Budget # **neben** dem Top-k, keinen Platz darin — und muss ihn nach Einschlägigkeit # auswählen statt nach Häufigkeit im Text. MAX_VERWEIS_NORMEN = int(os.getenv("MAX_VERWEIS_NORMEN", "0")) DEFAULT_RAG_CONTEXT_CHARS = int(os.getenv("DEFAULT_RAG_CONTEXT_CHARS", "12000")) DEFAULT_MIN_SCORE = float(os.getenv("DEFAULT_MIN_SCORE", "0.0")) # Wenn True, werden nur Quellen zurückgegeben, die das Modell als [Quelle n] # in der Antwort erwähnt. Wenn das Modell keine Marker verwendet, fällt die App # auf die kuratierte Quellenliste zurück. SOURCE_SYNC_TO_ANSWER_MARKERS = os.getenv("SOURCE_SYNC_TO_ANSWER_MARKERS", "true").lower() == "true" # --- Ask-Graph-Orchestrierung --- # Organisiert die bestehende /ask-Funktionalität als expliziten Graphen. # Wichtig: Es werden keine neuen fachlichen Routen eingeführt. ASK_GRAPH_ENABLED = os.getenv("ASK_GRAPH_ENABLED", "true").lower() == "true" ORCHESTRATOR_ENABLED = os.getenv("ORCHESTRATOR_ENABLED", "true").lower() == "true" MAX_ASK_GRAPH_STEPS = int(os.getenv("MAX_ASK_GRAPH_STEPS", "10")) _raw_origins = os.getenv("CORS_ORIGINS", "*") CORS_ORIGINS = ( ["*"] if _raw_origins.strip() == "*" else [origin.strip() for origin in _raw_origins.split(",") if origin.strip()] ) # ----------------------------------------------------------------------------- # FastAPI-App # ----------------------------------------------------------------------------- app = FastAPI( title=APP_TITLE, description="Retriever + Groq LLM + juristische Dokumente", version=APP_VERSION, ) app.add_middleware( CORSMiddleware, allow_origins=CORS_ORIGINS, allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Static-Dateien ausliefern: # # /static/index.html # /static/styles/style.css # /static/js/script.js # /static/berliner_apotheker_verein_cover.jpg if os.path.isdir(STATIC_DIR): app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static") # ----------------------------------------------------------------------------- # Session State # ----------------------------------------------------------------------------- @dataclass class SessionState: """ Pro-Session-Zustand. ConversationMemory lebt bewusst in der Session, nicht im GroqClient. Der GroqClient wird pro Request neu instanziiert und erhält memory nur explizit. """ system_prompt: str = DEFAULT_SYSTEM_PROMPT memory: Optional[ConversationMemory] = None def get_memory(self) -> ConversationMemory: if self.memory is None: self.memory = ConversationMemory() return self.memory def reset(self) -> None: self.system_prompt = DEFAULT_SYSTEM_PROMPT self.get_memory().reset() sessions: Dict[str, SessionState] = {} sessions_lock = Lock() def get_or_create_session(session_id: Optional[str]) -> tuple[str, SessionState]: with sessions_lock: if not session_id or session_id not in sessions: session_id = str(uuid4()) sessions[session_id] = SessionState() return session_id, sessions[session_id] @app.middleware("http") async def session_middleware(request: Request, call_next): session_id, session = get_or_create_session(request.cookies.get(SESSION_COOKIE)) request.state.session_id = session_id request.state.session = session response: Response = await call_next(request) response.set_cookie( SESSION_COOKIE, session_id, httponly=True, samesite="lax", secure=SESSION_COOKIE_SECURE, max_age=SESSION_COOKIE_MAX_AGE, ) return response # ----------------------------------------------------------------------------- # RAG-Komponenten # ----------------------------------------------------------------------------- corpus_registry: Optional[CorpusRegistry] = None def get_corpus_registry() -> CorpusRegistry: global corpus_registry if corpus_registry is None: corpus_registry = build_registry( persist_dir=CHROMA_PERSIST_DIR, model_name=EMBEDDING_MODEL, enable_reranker=ENABLE_RERANKER, reranker_model=RERANKER_MODEL, reranker_candidates=RERANKER_CANDIDATES, ) return corpus_registry def _build_retriever() -> FederatedRetriever: """ Baut den Retriever über alle konfigurierten Korpora und gibt Startdiagnose aus. Bei genau einem Korpus verhält sich der föderierende Retriever wie der bisherige `LegalRetriever` — die Routing- und Fusionsschritte sind dann Durchreichen. """ registry = get_corpus_registry() instance = FederatedRetriever( registry, router=route_question, # Die maßgeblichen Normen werden abgerufen, nicht erhofft. Das Register # liefert nur Adressen; der Retriever bleibt frei von Rechtsinhalten. anchors=norm_anchors.pflichtabruf, # Und die Normen, die der gefundene Text selbst in Bezug nimmt. Das # Register kennt die Kette eines Fragetyps, der Text die Kette seines # eigenen Satzes — „nach Maßgabe der §§ 11 bis 14" löste bis dahin # nichts aus. Seit dem 21.08.2026 steht `MAX_VERWEIS_NORMEN` auf 0 und # damit ist der Abruf aus; die Begründung steht dort. verweise=norm_verweise.abrufziele, # Nennt die Frage ein Regelwerk, führt dessen Treffer die Liste an — # auch gegen den Pflichtabruf. Der Anker behält seinen Platz im # Kontext, aber nicht den ersten. named_corpora=mentioned_corpora, min_hits_per_corpus=MIN_HITS_PER_CORPUS, max_verweis_normen=MAX_VERWEIS_NORMEN, ) print("APP USING CHROMA PATH:", CHROMA_PERSIST_DIR) for entry in registry.diagnostics()["corpora"]: status = entry.get("count") if entry.get("ok") else entry.get("error") print(f"APP CORPUS {entry['corpus_id']}: collection={entry['collection']} count={status}") warning = registry.assert_consistent_embedding() if warning: print("APP WARNING:", warning) return instance retriever: Optional[FederatedRetriever] = None retriever_lock = Lock() def get_retriever() -> FederatedRetriever: """Initialisiert Chroma/SentenceTransformer erst beim ersten echten Zugriff. Vorteil im Deployment: FastAPI kann importieren und starten, auch wenn Chroma oder das Embedding-Modell beim Build/Cold-Start kurz zicken. Der konkrete Fehler kommt dann kontrolliert über /health, /debug/retriever oder /ask. """ global retriever if retriever is not None: return retriever with retriever_lock: if retriever is None: retriever = _build_retriever() return retriever # ----------------------------------------------------------------------------- # Data Models # ----------------------------------------------------------------------------- class Question(BaseModel): question: str = Field(..., min_length=1) top_k: Optional[int] = Field(default=None, ge=1, le=50) fetch_k: Optional[int] = Field(default=None, ge=1, le=100) max_final_results: Optional[int] = Field(default=None, ge=1, le=80) max_sources: Optional[int] = Field(default=None, ge=1, le=20) include_neighbors: bool = Field(default=True) include_explicit_sections: bool = Field(default=True) # False: Anlagen und Anhänge (z. B. Anlage 11, Dienstleistungs-Anhänge) # werden mitdurchsucht. Die Priorisierung des Hauptvertrags übernimmt das # Ranking/Reranking, nicht ein harter Container-Filter, der früher ganze # Fragenklassen (Beitritt, pharmazeutische Dienstleistungen) blind machte. restrict_to_default_container: bool = Field(default=False) verify_negative_answer: bool = Field(default=True) allow_clarification: bool = Field(default=True) enrich_citations: bool = Field(default=True) debug: bool = Field(default=False) class SystemPromptPayload(BaseModel): system_prompt: str = Field(..., min_length=1) # ----------------------------------------------------------------------------- # Hilfsfunktionen # ----------------------------------------------------------------------------- def _clean_question(text: str) -> str: return " ".join((text or "").strip().split()) def _build_llm(session: SessionState) -> GroqClient: return GroqClient( model=GROQ_MODEL, system_prompt=session.system_prompt, temperature=GROQ_TEMPERATURE, max_tokens=GROQ_MAX_TOKENS, max_tokens_ceiling=GROQ_MAX_TOKENS_CEILING, debug_prompts=GROQ_DEBUG_PROMPTS, ) def _build_composer(llm: GroqClient, payload: Question) -> AnswerComposer: """Baut den AnswerComposer im finalen Legal-RAG-Modus. Wichtig: Neue Composer-Versionen liefern nummernstabile, gruppierte Quellen mit Feldern wie display_title, source_numbers und canonical_refs. Diese Optionen werden hier bewusst explizit gesetzt, damit die API-/UI-Schicht nicht wieder auf die alte path/pages-Anzeige zurückfällt. """ allowed = [DEFAULT_CONTAINER_ID] if payload.restrict_to_default_container else None kwargs: Dict[str, Any] = { "max_context_chars": DEFAULT_RAG_CONTEXT_CHARS, "pass_memory_to_llm_for_documents": False, "allowed_container_ids": allowed, "max_hits_for_context": payload.max_final_results or DEFAULT_MAX_FINAL_RESULTS, "max_chunks_per_section": 4, "max_sources": payload.max_sources or DEFAULT_MAX_SOURCES, "min_score_for_context": DEFAULT_MIN_SCORE, "include_neighbor_hits_in_context": True, "include_pure_neighbors_as_sources": False, "append_sources_to_answer": False, # Neue Qualitätsoptionen des finalen Composer. "display_all_context_sources": True, "validate_source_markers": True, "prefer_direct_hits_over_neighbors": True, } try: return AnswerComposer(llm, **kwargs) except TypeError as exc: logger.warning( "AnswerComposer unterstützt nicht alle finalen Optionen; " "falle auf kompatible Minimalinitialisierung zurück: %s", exc, ) legacy_keys = { "max_context_chars", "pass_memory_to_llm_for_documents", "allowed_container_ids", "max_hits_for_context", "max_chunks_per_section", "max_sources", "min_score_for_context", "include_neighbor_hits_in_context", "include_pure_neighbors_as_sources", "append_sources_to_answer", } legacy_kwargs = {key: value for key, value in kwargs.items() if key in legacy_keys} try: return AnswerComposer(llm, **legacy_kwargs) except TypeError as exc2: logger.warning( "AnswerComposer unterstützt nur Minimalparameter; " "Quellenqualität kann eingeschränkt sein: %s", exc2, ) return AnswerComposer( llm, max_context_chars=DEFAULT_RAG_CONTEXT_CHARS, pass_memory_to_llm_for_documents=False, ) def _build_orchestrator( retriever_instance: LegalRetriever, composer: AnswerComposer, payload: Question, ) -> LegalAnswerOrchestrator: options = OrchestratorOptions( top_k=payload.top_k or DEFAULT_TOP_K, fetch_k=payload.fetch_k or DEFAULT_FETCH_K, max_final_results=payload.max_final_results or DEFAULT_MAX_FINAL_RESULTS, max_sources=payload.max_sources or DEFAULT_MAX_SOURCES, include_neighbors=payload.include_neighbors, include_explicit_sections=payload.include_explicit_sections, restrict_to_default_container=payload.restrict_to_default_container, default_container_id=DEFAULT_CONTAINER_ID, min_score=DEFAULT_MIN_SCORE, enable_negative_recheck=payload.verify_negative_answer, enable_clarification=payload.allow_clarification, enrich_citations_with_canonical_refs=payload.enrich_citations, debug=payload.debug, ) try: return LegalAnswerOrchestrator( retriever_instance, composer, options=options, question_classifier=classify_question, meta_detector=is_meta_question, ) except TypeError: return LegalAnswerOrchestrator(retriever_instance, composer, options=options) def _retrieve_hits(payload: Question) -> List[Dict[str, Any]]: question = _clean_question(payload.question) top_k = payload.top_k or DEFAULT_TOP_K fetch_k = payload.fetch_k or DEFAULT_FETCH_K max_final_results = payload.max_final_results or DEFAULT_MAX_FINAL_RESULTS where = {"container_id": DEFAULT_CONTAINER_ID} if payload.restrict_to_default_container else None retriever_instance = get_retriever() try: return retriever_instance.query( question=question, top_k=top_k, where=where, fetch_k=fetch_k, include_explicit_sections=payload.include_explicit_sections, include_neighbors=payload.include_neighbors, max_final_results=max_final_results, ) except TypeError: try: return retriever_instance.query(question=question, top_k=top_k, where=where) except TypeError: return retriever_instance.query(question=question, top_k=top_k) def _coalesce(*values: Any) -> Any: for value in values: if value is not None and value != "": return value return None def _page_value(hit: Dict[str, Any]) -> Optional[str]: """Normalisiert Seitenangaben aus Raw-Hits oder Composer-Sources.""" metadata = hit.get("metadata") or {} direct = _coalesce( hit.get("pages"), hit.get("page_range"), metadata.get("pages"), metadata.get("page_range"), ) if direct is not None: return str(direct) start = _coalesce(hit.get("page_start"), metadata.get("page_start")) end = _coalesce(hit.get("page_end"), metadata.get("page_end"), start) if start is not None and end is not None: return f"{start}–{end}" if start is not None: return str(start) return None def _list_value(*values: Any) -> List[Any]: """Gibt die erste nicht-leere Listen-/Skalarangabe als Liste zurück.""" for value in values: if value is None or value == "": continue if isinstance(value, (list, tuple, set)): return [item for item in value if item is not None and item != ""] return [value] return [] def _int_list(*values: Any) -> List[int]: out: List[int] = [] for value in _list_value(*values): try: number = int(value) except (TypeError, ValueError): continue if number not in out: out.append(number) return sorted(out) def _int_or_none(value: Any) -> Optional[int]: try: return int(value) except (TypeError, ValueError): return None def _highlight_text_from_hit(hit: Dict[str, Any], *, max_chars: int = 2000) -> str: """Chunk-Text für die PDF-Hervorhebung, ohne konstruierten Kontext-Header. Die Ingest-Pipeline stellt jedem Chunk eine Zeile "Container · § x Titel" voran, die im PDF nicht existiert und daher nicht gematcht werden kann. """ text = str(hit.get("text") or hit.get("document") or "").strip() if not text: return "" head, sep, rest = text.partition("\n\n") if sep and " · " in head and len(head) <= 200 and rest.strip(): text = rest.strip() return text[:max_chars] def _highlights_from_hit(hit: Dict[str, Any]) -> List[Dict[str, Any]]: """Normalisiert die Hervorhebungsliste einer Quelle bzw. eines Raw-Hits.""" metadata = hit.get("metadata") or {} existing = hit.get("highlights") if isinstance(existing, list) and existing: out = [] for entry in existing: if not isinstance(entry, dict): continue text = str(entry.get("text") or "").strip() if not text: continue out.append( { "page_start": _int_or_none(entry.get("page_start")), "page_end": _int_or_none(entry.get("page_end")) or _int_or_none(entry.get("page_start")), "text": text[:2000], } ) return out text = _highlight_text_from_hit(hit) if not text: return [] page_start = _int_or_none(_coalesce(hit.get("page_start"), metadata.get("page_start"))) page_end = _int_or_none(_coalesce(hit.get("page_end"), metadata.get("page_end"))) or page_start return [{"page_start": page_start, "page_end": page_end, "text": text}] def _canonical_refs_from_hit(hit: Dict[str, Any]) -> List[str]: metadata = hit.get("metadata") or {} refs = [ str(ref).strip() for ref in _list_value(hit.get("canonical_refs"), metadata.get("canonical_refs")) if str(ref).strip() ] single = _coalesce(hit.get("canonical_ref"), metadata.get("canonical_ref")) if single is not None and str(single).strip() and str(single).strip() not in refs: refs.append(str(single).strip()) return refs def _source_marker(numbers: List[int]) -> str: if not numbers: return "" if len(numbers) == 1: return f"[Quelle {numbers[0]}]" return "[Quellen " + ", ".join(str(n) for n in numbers) + "]" _SGB_BOOK_RE = re.compile(r"Sozialgesetzbuch.*?\(([IVXLC]+)\)", re.I) def _short_doc_title(title: str, *, max_chars: int = 32) -> str: """Kürzt einen Dokumenttitel auf eine zitierfähige Kurzform. Die vollen Titel aus den Chunk-Metadaten sind für eine Quellenzeile zu lang ("Sozialgesetzbuch (SGB) Fünftes Buch (V) - Gesetzliche Krankenversicherung"). Das LLM bekommt weiterhin den vollen Titel im Kontext; gekürzt wird nur die Anzeige. """ text = " ".join(str(title or "").split()) if not text: return "" match = _SGB_BOOK_RE.search(text) if match: return f"SGB {match.group(1).upper()}" # Untertitel nach Gedankenstrich abschneiden. text = re.split(r"\s+[–—-]\s+", text, maxsplit=1)[0].strip() if text.lower().startswith("rahmenvertrag"): return "Rahmenvertrag" if len(text) <= max_chars: return text return text[:max_chars].rsplit(" ", 1)[0].rstrip(" ,;:-–—") + "…" def _format_source_display_title(source: Dict[str, Any]) -> str: """Erzeugt eine UI-fertige, nummernstabile Quellenanzeige.""" marker = _coalesce(source.get("source_label"), source.get("source_marker")) or _source_marker( _int_list(source.get("source_numbers"), source.get("source_number")) ) container = _coalesce(source.get("container"), "Unbekannt") section = _coalesce(source.get("section"), "ohne Abschnitt") path = _coalesce(source.get("path"), source.get("section_path")) if not path: path = f"{container}::{section}" # Dokumentangabe voranstellen, sobald der Korpus mehr als ein Dokument # enthält. "§ 16" allein ist mehrdeutig: Rahmenvertrag und SGB V teilen sich # 34 §-Nummern mit unterschiedlichem Inhalt. doc = str(_coalesce(source.get("doc_title"), source.get("doc_id")) or "").strip() doc_part = f"{_short_doc_title(doc)} · " if doc else "" pages = _coalesce(source.get("page_range"), source.get("pages"), "?") canonical_refs = [ str(ref).strip() for ref in _list_value(source.get("canonical_refs"), source.get("canonical_ref")) if str(ref).strip() ] # Keine redundante Anzeige "§ 6 (§ 6)". canonical_refs = [ref for ref in dict.fromkeys(canonical_refs) if ref != section] ref_part = f" ({'; '.join(canonical_refs[:4])})" if canonical_refs else "" role = "" kinds = set(source.get("retrieval_kinds") or []) if kinds == {"neighbor"}: role = " · Kontext/Nachbar" return f"{marker} {doc_part}{path}{ref_part}, Seiten {pages}{role}".strip() def _source_from_hit(hit: Dict[str, Any], *, source_number: Optional[int] = None) -> Dict[str, Any]: """ Einheitliche Quellennormalisierung. Unterstützt Raw-Retriever-Hits und bereits kuratierte Composer-Quellen. Anders als die ältere Version bewahrt sie Composer-Felder wie display_title, source_numbers und canonical_refs, damit die UI nicht wieder auf bloße path/pages-Ausgaben zurückfällt. """ metadata = hit.get("metadata") or {} explicit_number = source_number if source_number is not None else _coalesce( hit.get("source_number"), metadata.get("source_number"), ) source_numbers = _int_list( hit.get("source_numbers"), metadata.get("source_numbers"), explicit_number, ) pages = _page_value(hit) container = _coalesce(hit.get("container"), hit.get("container_id"), metadata.get("container_id")) section = _coalesce(hit.get("section"), hit.get("section_id"), metadata.get("section_id")) canonical_refs = _canonical_refs_from_hit(hit) canonical_ref = _coalesce(hit.get("canonical_ref"), metadata.get("canonical_ref")) if canonical_ref is None and canonical_refs: canonical_ref = canonical_refs[0] path = _coalesce( hit.get("path"), hit.get("section_path"), metadata.get("path"), metadata.get("section_path"), ) if not path and container and section: path = f"{container}::{section}" source: Dict[str, Any] = { "source_number": explicit_number, "source_numbers": source_numbers, "source_marker": _coalesce(hit.get("source_marker"), metadata.get("source_marker"), _source_marker(source_numbers)), "source_label": _coalesce(hit.get("source_label"), metadata.get("source_label"), _source_marker(source_numbers)), "container": container, "section": section, "canonical_ref": canonical_ref, "canonical_refs": canonical_refs, "path": path, "pages": pages, "page_range": pages, "chunk_index": _coalesce( hit.get("chunk_index"), hit.get("chunk_index_in_section"), metadata.get("chunk_index"), metadata.get("chunk_index_in_section"), ), # PDF-Locator für die interaktive Fundstellen-Anzeige in der UI. "page_start": _int_or_none(_coalesce(hit.get("page_start"), metadata.get("page_start"))), "page_end": _int_or_none(_coalesce(hit.get("page_end"), metadata.get("page_end"), hit.get("page_start"), metadata.get("page_start"))), "source_file": _coalesce(hit.get("source_file"), metadata.get("source_file")), "doc_id": _coalesce(hit.get("doc_id"), metadata.get("doc_id")), "doc_title": _coalesce(hit.get("doc_title"), metadata.get("doc_title")), # Von FederatedRetriever je Treffer gesetzt. Ohne Durchreichen weiß die # API-Schicht nicht mehr, aus welchem Korpus eine Quelle stammt — der # Änderungshinweis in corpus_amendments.py könnte dann nie greifen. "corpus_id": _coalesce(hit.get("corpus_id"), metadata.get("corpus_id")), "highlights": _highlights_from_hit(hit), "legal_unit_id": _coalesce(hit.get("legal_unit_id"), metadata.get("legal_unit_id")), "parent_unit_id": _coalesce(hit.get("parent_unit_id"), metadata.get("parent_unit_id")), "chunk_kind": _coalesce(hit.get("chunk_kind"), metadata.get("chunk_kind")), "unit_type": _coalesce(hit.get("unit_type"), metadata.get("unit_type")), "score": hit.get("score"), "rank_score": hit.get("rank_score"), "retrieval_kinds": hit.get("retrieval_kinds", metadata.get("retrieval_kinds", [])), "display_label": _coalesce(hit.get("display_label"), metadata.get("display_label")), "display_title": _coalesce(hit.get("display_title"), metadata.get("display_title")), } if not source["display_title"]: source["display_title"] = _format_source_display_title(source) if not source["display_label"]: source["display_label"] = source["display_title"] return source def _extract_answer_source_numbers(answer: str) -> List[int]: """ Liest Quellenmarker aus der Modellantwort: [Quelle 1], [Quelle 2], ... """ nums: List[int] = [] for m in re.finditer(r"\[Quelle\s+(\d+)(?:[^\]]*)\]", answer or "", flags=re.I): try: nums.append(int(m.group(1))) except ValueError: continue return list(dict.fromkeys(nums)) def _answer_reports_nothing_found(answer: str) -> bool: """Die Antwort zitiert nichts, nennt keine Norm und sagt, es gebe nichts. Zwei Stellen hängen an dieser Frage: der Reichweiten-Hinweis, dessen Prämisse ist, dass Textstellen herangezogen *wurden*, und die zurückgegebene Quellenliste, die sonst fünf Fundstellen unter eine Aussage setzt, dass es keine gibt. Bewusst eng gefasst, damit der gewollte Fallback erhalten bleibt: vergisst das Modell bei einer inhaltlichen Antwort die Marker, sollen seine Quellen weiter erscheinen. Ein § irgendwo im Text heißt deshalb "etwas gefunden", unabhängig davon, was die Antwort darüber behauptet. """ text = answer or "" if _extract_answer_source_numbers(text): return False if "§" in text: return False return bool(NEGATIVE_ANSWER_RE.search(text)) def _merge_source(existing: Dict[str, Any], incoming: Dict[str, Any]) -> Dict[str, Any]: """Führt gruppierbare Quellen zusammen, ohne Nummern/Fundstellen zu verlieren.""" merged = dict(existing) numbers = _int_list(existing.get("source_numbers"), existing.get("source_number"), incoming.get("source_numbers"), incoming.get("source_number")) refs = [ str(ref).strip() for ref in _list_value(existing.get("canonical_refs"), existing.get("canonical_ref"), incoming.get("canonical_refs"), incoming.get("canonical_ref")) if str(ref).strip() ] refs = list(dict.fromkeys(refs)) kinds = list(dict.fromkeys( list(existing.get("retrieval_kinds") or []) + list(incoming.get("retrieval_kinds") or []) )) merged["source_numbers"] = numbers merged["source_number"] = numbers[0] if numbers else _coalesce(existing.get("source_number"), incoming.get("source_number")) merged["source_marker"] = _source_marker(numbers) or _coalesce(existing.get("source_marker"), incoming.get("source_marker")) merged["source_label"] = merged["source_marker"] merged["canonical_refs"] = refs merged["canonical_ref"] = _coalesce(existing.get("canonical_ref"), incoming.get("canonical_ref"), refs[0] if refs else None) merged["retrieval_kinds"] = kinds # Behalte den höchsten Score, ohne None/Strings zu erzwingen. for key in ("score", "rank_score"): try: old_score = float(existing.get(key) or 0.0) new_score = float(incoming.get(key) or 0.0) merged[key] = max(old_score, new_score) except (TypeError, ValueError): merged[key] = _coalesce(existing.get(key), incoming.get(key)) # Bei mehreren Chunks derselben Section ist chunk_index für die Anzeige nicht # mehr sinnvoll eindeutig. Nur behalten, wenn identisch. if existing.get("chunk_index") != incoming.get("chunk_index"): merged["chunk_index"] = None # PDF-Locator zusammenführen: Seitenbereich aufspannen, Datei behalten, # Hervorhebungen beider Quellen sammeln. starts = [p for p in (_int_or_none(existing.get("page_start")), _int_or_none(incoming.get("page_start"))) if p is not None] ends = [p for p in (_int_or_none(existing.get("page_end")), _int_or_none(incoming.get("page_end"))) if p is not None] merged["page_start"] = min(starts) if starts else None merged["page_end"] = max(ends) if ends else merged["page_start"] merged["source_file"] = _coalesce(existing.get("source_file"), incoming.get("source_file")) merged["doc_id"] = _coalesce(existing.get("doc_id"), incoming.get("doc_id")) merged["doc_title"] = _coalesce(existing.get("doc_title"), incoming.get("doc_title")) highlights: List[Dict[str, Any]] = [] seen_highlights: set = set() for entry in list(existing.get("highlights") or []) + list(incoming.get("highlights") or []): if not isinstance(entry, dict): continue text = str(entry.get("text") or "").strip() if not text: continue dedupe_key = (entry.get("page_start"), text[:120]) if dedupe_key in seen_highlights: continue seen_highlights.add(dedupe_key) highlights.append(entry) merged["highlights"] = highlights[:8] merged["display_title"] = _format_source_display_title(merged) merged["display_label"] = merged["display_title"] return merged def _source_dedupe_key(src: Dict[str, Any]) -> tuple[Any, ...]: """Dedupe-Key, der Composer-Gruppierungen respektiert.""" numbers = tuple(_int_list(src.get("source_numbers"), src.get("source_number"))) if numbers: return ("numbers", numbers) display_title = src.get("display_title") if display_title: return ("display", display_title) canonical_refs = tuple(src.get("canonical_refs") or []) return ( "location", # doc_id zuerst: gleiche §-Nummern in verschiedenen Dokumenten dürfen # nicht zu einer Quelle verschmelzen. src.get("doc_id"), src.get("container"), src.get("section"), src.get("pages") or src.get("page_range"), canonical_refs, ) def _dedupe_sources(sources: Iterable[Dict[str, Any]], *, max_sources: int) -> List[Dict[str, Any]]: """Dedupliziert Quellen, ohne Composer-Felder zu verlieren. Die alte Variante normalisierte jede Quelle auf path/pages/chunk_index zurück und zerstörte dadurch display_title, source_numbers und canonical_refs. """ grouped: Dict[tuple[Any, ...], Dict[str, Any]] = {} order: List[tuple[Any, ...]] = [] for raw in sources: src = _source_from_hit(raw) # Wenn keine Nummern vorhanden sind, gruppiere gleiche Norm/Seiten, nicht # einzelne Chunk-Indizes. Das verhindert doppelte "Vertrag::§ 6"-Zeilen. key = _source_dedupe_key(src) if key not in grouped: grouped[key] = src order.append(key) else: grouped[key] = _merge_source(grouped[key], src) out = [grouped[key] for key in order] out.sort(key=lambda s: (_int_list(s.get("source_numbers"), s.get("source_number")) or [10_000])[0]) return out[:max_sources] def _build_sources_from_composer_or_hits( composer: AnswerComposer, hits: List[Dict[str, Any]], payload: Question, ) -> List[Dict[str, Any]]: max_sources = payload.max_sources or DEFAULT_MAX_SOURCES allowed = [DEFAULT_CONTAINER_ID] if payload.restrict_to_default_container else None if hasattr(composer, "build_sources"): try: raw_sources = composer.build_sources( hits, max_sources=max(max_sources, min(len(hits), DEFAULT_MAX_FINAL_RESULTS)), include_pure_neighbors=False, allowed_container_ids=allowed, ) return _dedupe_sources(raw_sources, max_sources=max_sources) except TypeError: try: raw_sources = composer.build_sources(hits, max_sources=max_sources) return _dedupe_sources(raw_sources, max_sources=max_sources) except Exception as exc: logger.warning("Composer build_sources fallback failed: %s", exc) return _dedupe_sources(hits, max_sources=max_sources) def _sources_cover_cited_numbers(sources: List[Dict[str, Any]], cited_numbers: List[int]) -> bool: if not cited_numbers: return True covered = set() for source in sources: covered.update(_int_list(source.get("source_numbers"), source.get("source_number"))) return set(cited_numbers).issubset(covered) def _sources_for_cited_numbers_from_raw_hits( cited_numbers: List[int], hits: List[Dict[str, Any]], *, max_sources: int, ) -> List[Dict[str, Any]]: """Letzter Fallback: mappt [Quelle n] auf Treffer n. Dieser Pfad ist nur ein Sicherheitsnetz. Bevorzugt werden Composer-Sources, weil der Composer den Kontext sortiert/dedupliziert und die Nummern korrekt kennt. """ cited_sources: List[Dict[str, Any]] = [] for num in cited_numbers: idx = num - 1 if 0 <= idx < len(hits): cited_sources.append(_source_from_hit(hits[idx], source_number=num)) return _dedupe_sources(cited_sources, max_sources=max_sources) _FINE_REF_RE = re.compile( r"§\s*(\d{1,3}[a-z]?)\s+Abs\.\s*(\d+[a-z]?)\s+Buchst\.\s*([a-z])", flags=re.I, ) def _evidence_blob(sources: List[Dict[str, Any]], hits: List[Dict[str, Any]]) -> str: parts: List[str] = [] for item in list(sources or []) + list(hits or []): metadata = item.get("metadata") or {} parts.extend( str(value) for value in [ item.get("canonical_ref"), item.get("canonical_refs"), item.get("section"), item.get("section_id"), item.get("path"), item.get("text"), item.get("document"), metadata.get("canonical_ref"), metadata.get("section_id"), metadata.get("text"), ] if value ) return " ".join(parts).lower() def _sanitize_unsupported_fine_references( answer: str, *, sources: List[Dict[str, Any]], hits: List[Dict[str, Any]], ) -> str: """Entschärft erfundene Feinfundstellen wie '§ 6 Abs. 1 Buchst. a'. Wenn die genaue Buchstabenfundstelle nicht im Kontext/Metadaten belegt ist, wird auf die belastbarere Absatzfundstelle zurückgeführt. """ if not answer: return answer evidence = _evidence_blob(sources, hits) def repl(match: re.Match[str]) -> str: para, abs_no, letter = match.group(1), match.group(2), match.group(3).lower() exact_patterns = [ f"§ {para} abs. {abs_no} buchst. {letter}", f"§{para} abs. {abs_no} buchst. {letter}", f"§ {para} absatz {abs_no} buchstabe {letter}", f"§ {para} abs. {abs_no} lit. {letter}", ] if any(pattern in evidence for pattern in exact_patterns): return match.group(0) return f"§ {para} Abs. {abs_no}" cleaned = _FINE_REF_RE.sub(repl, answer) # Entfernt häufige Dopplung: "(§ 6 Abs. 1), § 6 Abs. 1" cleaned = re.sub( r"\((§\s*\d{1,3}[a-z]?\s+Abs\.\s*\d+[a-z]?)\),\s*\1", r"(\1)", cleaned, flags=re.I, ) return cleaned def _postprocess_answer( composer: AnswerComposer, answer: str, *, sources: List[Dict[str, Any]], hits: List[Dict[str, Any]], ) -> str: """Zentraler letzter Antwort-Postprocessor für Composer- und Orchestratorpfad.""" text = answer or "" if hasattr(composer, "_strip_model_generated_sources"): try: text = composer._strip_model_generated_sources(text) # type: ignore[attr-defined] except Exception as exc: logger.debug("Composer source-strip postprocessing skipped: %s", exc) if hasattr(composer, "_strip_invalid_source_markers"): try: # Erwartet meist used_hits/source_number-Strukturen; sources sind dafür # besser geeignet als Raw-Hits. text = composer._strip_invalid_source_markers(text, sources or hits) # type: ignore[attr-defined] except Exception as exc: logger.debug("Composer marker postprocessing skipped: %s", exc) text = _sanitize_unsupported_fine_references(text, sources=sources, hits=hits) # Zuletzt, damit die vorherigen Schritte erst entfernen können, was einen # Abschnitt nur scheinbar füllt (ungültige Marker, unbelegte Fundstellen). if hasattr(composer, "_strip_empty_section_headings"): try: text = composer._strip_empty_section_headings(text) # type: ignore[attr-defined] except Exception as exc: logger.debug("Composer section postprocessing skipped: %s", exc) return text.strip() def _normalize_returned_sources( *, answer: str, sources: List[Dict[str, Any]], hits: List[Dict[str, Any]], payload: Question, ) -> List[Dict[str, Any]]: """ Finale API-Quellenlogik. Wichtigste Regel: Composer-/Orchestrator-Sources sind der primäre Wahrheitsanker für [Quelle n]. Raw-Hits werden nur als Fallback verwendet. Dadurch bleiben display_title, source_numbers und canonical_refs erhalten. """ max_sources = payload.max_sources or DEFAULT_MAX_SOURCES cited_numbers = _extract_answer_source_numbers(answer) # Hat die Antwort nichts gefunden, gibt es auch nichts anzuzeigen. Die # kuratierte Liste wäre hier kein Fallback, sondern eine Behauptung. if _answer_reports_nothing_found(answer): return [] normalized = _dedupe_sources( sources or [], max_sources=max(max_sources, len(cited_numbers), DEFAULT_MAX_SOURCES), ) if normalized: if SOURCE_SYNC_TO_ANSWER_MARKERS and cited_numbers: # Zeige mindestens alle zitierten Quellen. Wenn Composer-Sources die # zitierten Nummern abdecken, keine Rückabbildung auf Raw-Hits. if _sources_cover_cited_numbers(normalized, cited_numbers): cited_set = set(cited_numbers) cited_first = [ src for src in normalized if cited_set.intersection(_int_list(src.get("source_numbers"), src.get("source_number"))) ] rest = [src for src in normalized if src not in cited_first] return (cited_first + rest)[:max(max_sources, len(cited_first))] fallback = _sources_for_cited_numbers_from_raw_hits(cited_numbers, hits, max_sources=max_sources) if fallback: return fallback return normalized[:max_sources] if SOURCE_SYNC_TO_ANSWER_MARKERS and cited_numbers and hits: fallback = _sources_for_cited_numbers_from_raw_hits(cited_numbers, hits, max_sources=max_sources) if fallback: return fallback return _dedupe_sources(hits, max_sources=max_sources) _INLINE_MARKER_RE = re.compile(r"\[Quellen?\s+\d+(?:[^\]]*)\]", flags=re.I) def _remap_inline_source_markers(answer: str, old_to_new: Dict[int, int]) -> str: """Schreibt Inline-Marker [Quelle n]/[Quellen n, m] auf die neuen Anzeigenummern um. Unbekannte Nummern (kein Mapping vorhanden) bleiben unverändert stehen; sie werden an anderer Stelle bereits als ungültige Marker entfernt. """ if not answer or not old_to_new: return answer def repl(match: re.Match[str]) -> str: mapped: List[int] = [] for raw in re.findall(r"\d+", match.group(0)): try: new = old_to_new.get(int(raw)) except (TypeError, ValueError): new = None if new is not None and new not in mapped: mapped.append(new) if not mapped: return match.group(0) mapped.sort() return _source_marker(mapped) return _INLINE_MARKER_RE.sub(repl, answer) def _renumber_sources_for_display( answer: str, sources: List[Dict[str, Any]], ) -> Tuple[str, List[Dict[str, Any]]]: """Vergibt stabile, fortlaufende Anzeigenummern (1..n) für die finalen Quellen. Hintergrund: Die ursprünglichen [Quelle n]-Nummern sind Retrieval-Ränge aus dem RAG-Kontext. Dadurch beginnt die Anzeige nicht bei 1 und enthält Lücken (z. B. "[Quellen 3, 8]"). Für die UI werden die tatsächlich angezeigten Quellen in ihrer Anzeige-Reihenfolge auf 1..n abgebildet und die Inline-Marker im Antworttext konsistent mitgezogen. Mehrere Alt-Nummern derselben Quelle (gruppierte Chunks) fallen dabei auf eine Anzeigenummer zusammen. """ if not sources: return answer, sources old_to_new: Dict[int, int] = {} next_number = 1 for source in sources: old_numbers = _int_list(source.get("source_numbers"), source.get("source_number")) if not old_numbers: # Quelle ohne Nummer: entsteht über den Rückfall auf die Roh-Treffer # (`_dedupe_sources(hits, …)`), denn ein Treffer trägt keine # Quellennummer — die vergibt erst der Composer. Übersprungen bekam # sie hier nie `display_title`/`display_label` und erschien in der # UI als **leere Zeile**. Sie bekommt deshalb ihre Anzeige, aber # keinen Marker: zitiert wurde sie nicht, und eine erfundene Nummer # stünde im Antworttext nirgends. source.setdefault("source_marker", "") source.setdefault("source_label", "") source["display_title"] = _format_source_display_title(source) source["display_label"] = source["display_title"] continue assigned: Optional[int] = None for old in old_numbers: if old in old_to_new: assigned = old_to_new[old] break if assigned is None: assigned = next_number next_number += 1 for old in old_numbers: old_to_new.setdefault(old, assigned) marker = f"[Quelle {assigned}]" source["source_number"] = assigned source["source_numbers"] = [assigned] source["source_marker"] = marker source["source_label"] = marker source["display_title"] = _format_source_display_title(source) source["display_label"] = source["display_title"] new_answer = _remap_inline_source_markers(answer, old_to_new) return new_answer, sources def _debug_hit(hit: Dict[str, Any]) -> Dict[str, Any]: text = (hit.get("text") or hit.get("document") or "").strip() return { **_source_from_hit(hit), "text_preview": text[:350], "text_length": len(text), } def _collection_count() -> Any: """Chunks über alle Korpora. Bei einem Korpus identisch zum bisherigen Wert.""" try: return get_corpus_registry().total_count() except Exception as exc: logger.warning("Chroma count unavailable: %s", exc) return "unknown" def _corpus_counts() -> Dict[str, Any]: try: return { entry["corpus_id"]: entry.get("count", entry.get("error")) for entry in get_corpus_registry().diagnostics()["corpora"] } except Exception as exc: # noqa: BLE001 logger.warning("Korpus-Diagnose nicht verfügbar: %s", exc) return {} # ----------------------------------------------------------------------------- # Ask Graph Workflow # ----------------------------------------------------------------------------- AskRoute = Literal["meta", "non_meta", "legacy", "clarification"] AskNodeName = Literal[ "__start__", "meta_decision", "retrieve_hits", "compose_answer", "orchestrate_answer", "normalize_sources", "update_memory", "build_response", "__end__", ] @dataclass class AskGraphStep: """Rückgabe eines Graph-Knotens.""" next_node: AskNodeName reason: str = "" @dataclass class AskGraphTraceEntry: node: str next_node: str reason: str = "" route: Optional[str] = None hit_count: int = 0 answer_type: str = "unknown" @dataclass class AskGraphContext: """ Gemeinsamer Zustand des /ask-Graphen. Dieser Graph kapselt ausschließlich die bereits vorhandene Funktionalität: Meta-Fragen werden ohne Retrieval beantwortet, alle anderen Fragen laufen durch Retrieval, AnswerComposer, Quellennormalisierung, Memory-Update und Response-Aufbau. """ payload: Question request: Request session: SessionState memory: ConversationMemory llm: GroqClient composer: AnswerComposer question: str orchestrator: Optional[LegalAnswerOrchestrator] = None route: Optional[AskRoute] = None hits: Optional[List[Dict[str, Any]]] = None # Die Teilmenge von `hits`, die die Antwort auch zitiert. Wird im # Normalisierungsschritt gesetzt, solange der Antworttext noch die # Retrieval-Ränge trägt (siehe `AnswerComposer.cited_hits`). cited_hits: Optional[List[Dict[str, Any]]] = None raw_sources: Optional[List[Dict[str, Any]]] = None sources: Optional[List[Dict[str, Any]]] = None answer: str = "" answer_type: str = "unknown" response_body: Optional[Dict[str, Any]] = None orchestrator_debug: Optional[Dict[str, Any]] = None needs_clarification: bool = False clarification_question: Optional[str] = None trace: Optional[List[AskGraphTraceEntry]] = None def __post_init__(self) -> None: if self.hits is None: self.hits = [] if self.raw_sources is None: self.raw_sources = [] if self.sources is None: self.sources = [] if self.cited_hits is None: self.cited_hits = [] if self.trace is None: self.trace = [] def _ask_graph_meta_decision(ctx: AskGraphContext) -> AskGraphStep: """Routet Meta-Fragen, Orchestrator-Fragen und Legacy-Fallback sauber.""" if is_meta_question(ctx.question): ctx.route = "meta" return AskGraphStep("compose_answer", "meta question without retrieval") if ORCHESTRATOR_ENABLED and ctx.orchestrator is not None: ctx.route = "non_meta" return AskGraphStep("orchestrate_answer", "regular question via legal orchestrator") ctx.route = "legacy" return AskGraphStep("retrieve_hits", "orchestrator unavailable or disabled; legacy retrieval") def _ask_graph_retrieve_hits(ctx: AskGraphContext) -> AskGraphStep: """Entspricht dem bisherigen Aufruf von _retrieve_hits(payload).""" ctx.hits = _retrieve_hits(ctx.payload) return AskGraphStep("compose_answer", f"retrieved {len(ctx.hits)} hits") def _ask_graph_orchestrate_answer(ctx: AskGraphContext) -> AskGraphStep: """Fachlicher Orchestrator-Knoten: Retrieval, Recheck, Antwort, Audit.""" if ctx.orchestrator is None: ctx.route = "legacy" return AskGraphStep("retrieve_hits", "orchestrator unavailable; falling back to legacy retrieval") result = ctx.orchestrator.run(ctx.question, memory=ctx.memory) ctx.answer = result.answer ctx.answer_type = result.answer_type ctx.hits = result.hits ctx.raw_sources = result.raw_sources ctx.sources = result.sources ctx.needs_clarification = result.needs_clarification ctx.clarification_question = result.clarification_question ctx.orchestrator_debug = result.to_debug_dict() if result.needs_clarification: ctx.route = "clarification" return AskGraphStep("normalize_sources", "answer orchestrated and audited") def _ask_graph_compose_answer(ctx: AskGraphContext) -> AskGraphStep: """ Entspricht der bisherigen Antwortgenerierung im /ask-Endpoint. Meta-Fragen werden mit leerem Kontext komponiert. Nicht-Meta-Fragen nutzen weiterhin compose_with_sources, sofern vorhanden, sonst compose plus Quellen-Fallback. """ if ctx.route == "meta": ctx.answer, ctx.answer_type = ctx.composer.compose(ctx.question, [], memory=ctx.memory) ctx.raw_sources = [] return AskGraphStep("normalize_sources", "meta answer composed") if hasattr(ctx.composer, "compose_with_sources"): try: ctx.answer, ctx.answer_type, raw_sources = ctx.composer.compose_with_sources( ctx.question, ctx.hits, memory=ctx.memory, ) except TypeError: ctx.answer, ctx.answer_type = ctx.composer.compose(ctx.question, ctx.hits, memory=ctx.memory) raw_sources = _build_sources_from_composer_or_hits(ctx.composer, ctx.hits, ctx.payload) else: ctx.answer, ctx.answer_type = ctx.composer.compose(ctx.question, ctx.hits, memory=ctx.memory) raw_sources = _build_sources_from_composer_or_hits(ctx.composer, ctx.hits, ctx.payload) ctx.raw_sources = raw_sources return AskGraphStep("normalize_sources", "answer composed") def _ask_graph_normalize_sources(ctx: AskGraphContext) -> AskGraphStep: """Normalisiert Quellen und führt finalen Antwort-Postprocessing-Schritt aus. Der Orchestratorpfad darf den Composer-Postprocessor nicht umgehen. Deshalb werden hier Quellen zuerst nummernstabil normalisiert, danach wird die Antwort gegen diese Quellen nachbearbeitet und anschließend erneut leicht synchronisiert. """ if ctx.answer_type == "document": raw_input_sources = ctx.raw_sources or ctx.sources or [] ctx.sources = _normalize_returned_sources( answer=ctx.answer, sources=raw_input_sources, hits=ctx.hits, payload=ctx.payload, ) ctx.answer = _postprocess_answer( ctx.composer, ctx.answer, sources=ctx.sources, hits=ctx.hits or [], ) ctx.sources = _normalize_returned_sources( answer=ctx.answer, sources=ctx.sources or raw_input_sources, hits=ctx.hits, payload=ctx.payload, ) # Welche Treffer die Antwort tatsächlich zitiert — noch mit den # Retrieval-Rängen, die gleich umnummeriert werden. Danach wäre die # Zuordnung verloren: der Text trüge Anzeigenummern, die Treffer weiter # ihre Ränge. try: ctx.cited_hits = AnswerComposer.cited_hits(ctx.answer, ctx.hits or []) except Exception as exc: # noqa: BLE001 - eine Nebenrechnung darf nie die Antwort verhindern. logger.warning("Zitatzuordnung übersprungen: %s", exc) ctx.cited_hits = [] # Anzeige-Nummern stabil auf 1..n abbilden (statt Retrieval-Ränge wie # "[Quellen 3, 8]"). Muss als letzter Schritt laufen, damit Antworttext # und Quellenliste dieselben Nummern zeigen. ctx.answer, ctx.sources = _renumber_sources_for_display(ctx.answer, ctx.sources) else: ctx.sources = [] ctx.cited_hits = [] return AskGraphStep("update_memory", "sources normalized and answer postprocessed") def _ask_graph_update_memory(ctx: AskGraphContext) -> AskGraphStep: """Entspricht dem bisherigen memory.add_turn(...).""" ctx.memory.add_turn( user_message=ctx.question, assistant_message=ctx.answer, question_kind=classify_question(ctx.question), ) return AskGraphStep("build_response", "conversation memory updated") _PAGE_RANGE_RE = re.compile(r"(\d+)\s*[–\-]\s*(\d+)|^(\d+)$") def _attach_pdf_locators(sources: List[Dict[str, Any]], hits: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """Sicherheitsnetz: fehlende PDF-Locator aus Hits bzw. page_range ableiten. Composer/Orchestrator liefern die Felder normalerweise bereits mit; ältere Pfade (Legacy-Fallbacks) können sie verlieren. Der Fallback wird pro Dokument gebildet, nicht global: bei mehreren Korpora würde "irgendein Hit" sonst das falsche PDF an die Quelle heften und der Viewer bei einer SGB-V-Fundstelle den Rahmenvertrag öffnen. """ by_doc: Dict[str, Dict[str, Any]] = {} for hit in hits or []: metadata = hit.get("metadata") or {} doc_id = str(_coalesce(hit.get("doc_id"), metadata.get("doc_id")) or "") entry = by_doc.setdefault(doc_id, {}) entry.setdefault("source_file", _coalesce(hit.get("source_file"), metadata.get("source_file"))) entry.setdefault("doc_title", _coalesce(hit.get("doc_title"), metadata.get("doc_title"))) # Nur wenn der Trefferpool eindeutig aus einem Dokument stammt, darf eine # Quelle ohne doc_id daraus ergänzt werden. single_doc = by_doc.get(next(iter(by_doc))) if len(by_doc) == 1 else None for source in sources or []: doc_id = str(source.get("doc_id") or "") entry = by_doc.get(doc_id) if doc_id else single_doc if entry: if not source.get("source_file") and entry.get("source_file"): source["source_file"] = entry["source_file"] if not source.get("doc_title") and entry.get("doc_title"): source["doc_title"] = entry["doc_title"] if not source.get("doc_id") and single_doc is not None: source["doc_id"] = next(iter(by_doc)) if source.get("page_start") is None: match = _PAGE_RANGE_RE.search(str(source.get("page_range") or source.get("pages") or "")) if match: start = match.group(1) or match.group(3) end = match.group(2) or start source["page_start"] = _int_or_none(start) source["page_end"] = _int_or_none(end) if source.get("page_end") is None: source["page_end"] = source.get("page_start") if not source.get("highlights"): source["highlights"] = [] return sources def _listenverdikt( befund: Any, austausch: Any, biosimilar: Any = None, otc: Any = None, verordnung: Any = None, lifestyle: Any = None, tabak: Any = None, ) -> str: """Welcher der Listenbefunde die Kurzantwort setzt. Die fünf Listen beantworten teils dieselbe Frageformulierung, aber verschiedene Fragen — „dasselbe Präparat eines anderen Herstellers?" (Anlage VII Teil B), „eine andere Darreichungsform?" (Teil A), „ein Biosimilar statt des Originals?" (Anlage VIIa), „überhaupt zu Lasten der GKV?" (Anlagen II, III und I). Die Reihenfolge folgt daraus, wie einschneidend die Auskunft ist: 1. Ein Substitutionsausschluss aus Teil B ist eine harte Schranke und steht über allem anderen. 2. Dann Anlage VIIa, wenn sie entschieden hat: ihre Wirkstoffe sind biotechnologisch hergestellt und kommen in keiner der anderen Listen vor (geprüft: die Wirkstoffmengen sind disjunkt). Wer nach Humira und Amgevita fragt, bekommt sonst „Adalimumab steht nicht auf der Substitutionsausschlussliste" — richtig, aber nicht die Frage. 3. Sonst Teil A, wenn die Frage zwei Formen benannt hat. 4. Dann Anlage IIa, dann Anlage II, danach Anlage III und erst danach Anlage I. Alle vier beantworten dieselbe Frage — „darf das überhaupt zu Lasten der GKV verordnet werden?" —, aber sie tragen verschieden weit. Anlage IIa steht vor Anlage II, obwohl sie deren Ausnahme ist und nicht ihre Regel: ihr Verdikt trägt **beide** Hälften („grundsätzlich ausgeschlossen, ausnahmsweise verordnungsfähig, wenn …"), das der Anlage II nur eine. Bei Champix stünde sonst „ist ausgeschlossen" als Kurzantwort über einem Block, der zwei Absätze tiefer den Anspruch nach § 34 Absatz 2 SGB V nennt — genau der Widerspruch, den `antwort_mit_befunden` auflösen soll. Anlage II steht vor III und I, weil ihr Ausschluss der härteste ist: er beruht auf § 34 Absatz 1 Satz 7 SGB V, und der medizinisch begründete Einzelfall des § 16 Absatz 5 AM-RL führt an ihm vorbei. Anlage III entscheidet auch für verschreibungspflichtige Arzneimittel, hält den Einzelfall bei ihren Markern 3 bis 6 aber offen; Anlage I urteilt nur über nicht verschreibungspflichtige. Bei Sildenafil steht so „nach Anlage II ausgeschlossen" statt des blasseren „steht nicht in Anlage III". 5. Alle drei stehen hinter den Austauschlisten, weil ihre Signale („erstattungsfähig", „Kassenrezept") auch in einer Austauschfrage vorkommen, umgekehrt aber nicht. 6. Erst danach das „nicht gelistet" aus Teil B — die schwächste Aussage von allen. Ohne diese Reihenfolge würde eine reine Darreichungsformfrage mit „Ambroxol steht nicht auf der Substitutionsausschlussliste" beantwortet: richtig, aber am Thema vorbei. """ if befund is not None and befund.status in { "ausschluss", "ausschluss_bedingt", "ausschluss_zwischen_varianten", }: return amrl_substitution.verdikt(befund) for modul, listenbefund in ( (amrl_biosimilars, biosimilar), (amrl_austauschbarkeit, austausch), (amrl_tabakentwoehnung, tabak), (amrl_lifestyle, lifestyle), (amrl_verordnungsausschluss, verordnung), (amrl_otc, otc), ): if listenbefund is not None: satz = modul.verdikt(listenbefund) if satz: return satz return amrl_substitution.verdikt(befund) if befund is not None else "" def _ask_graph_build_response(ctx: AskGraphContext) -> AskGraphStep: """Baut exakt die bisherige API-Response-Struktur.""" _attach_pdf_locators(ctx.sources or [], ctx.hits or []) # Korpusgrenze offenlegen: verweist der herangezogene Kontext auf ein # Regelwerk, das nicht indiziert ist (AM-RL, AMPreisV, …), wird das benannt. # Ohne diesen Hinweis liest sich eine aus § 31 SGB V gebaute Antwort # vollständig, obwohl die Verordnungsausschlüsse der AM-RL fehlen. # # Gelesen werden die *zitierten* Treffer, nicht der ganze Pool. Der Hinweis # spricht von den „herangezogenen Textstellen"; im Ausdruck vom 20.08.2026 # stand unter einer Frage zur Austauschbarkeit von Darreichungsformen der # Hinweis auf die BtMVV — sie kommt weder in § 9 Abs. 3 Rahmenvertrag noch # in § 40c AM-RL vor, wohl aber in § 6 Rahmenvertrag, den das Modell # verworfen hatte. Ein Hinweis auf ein Regelwerk, das die Antwort gar nicht # berührt, entwertet die Hinweise, die tragen. # # Ist keine Zuordnung möglich (Antwort ohne Marker), bleibt es beim Pool: # lieber ein Hinweis zu viel als eine verschwiegene Lücke. boundary: List[Dict[str, str]] = [] if ctx.route != "meta": try: boundary = detect_external_references( ctx.cited_hits or ctx.hits or [], available_corpora=get_corpus_registry().corpus_ids, ) except Exception as exc: # noqa: BLE001 - ein Hinweis darf nie die Antwort verhindern. logger.warning("Korpusgrenzen-Erkennung übersprungen: %s", exc) boundary = [] # Stand der zitierten Vorschriften: nennt konkret, welche spätere # Änderungsvereinbarung eine zitierte Norm geändert hat und was jetzt gilt. # Betrifft Korpora, die nur als Grundfassung plus separate Änderungstexte # veröffentlicht werden — dort wäre der indizierte Text sonst still veraltet. # Beide Hinweise unten hängen an derselben Frage. Sie muss vor dem Abgleich # beantwortet werden: reconcile_negative_answer setzt die Fundstelle der # Änderungsvereinbarung in den Text und kippte die Prüfung sonst ins # Gegenteil. nothing_found = _answer_reports_nothing_found(ctx.answer) amendments: List[Dict[str, Any]] = [] amendment_note = "" if ctx.route != "meta": try: # Bewusst die ungefilterten Quellen: ob eine geänderte Vorschrift # abgerufen wurde, entscheidet der Trefferbestand, nicht die für die # Anzeige gekürzte Liste — die bei einer Verneinung leer ist. amendment_note, amendments = describe_amendments( ctx.raw_sources or ctx.sources or [], path=AMENDMENTS_FILE, question=ctx.question, ) except Exception as exc: # noqa: BLE001 - ein Hinweis darf nie die Antwort verhindern. logger.warning("Änderungshinweis übersprungen: %s", exc) amendment_note, amendments = "", [] # Das "nicht geregelt" des Modells gilt nur für den indizierten Stand. Folgt # darunter der Änderungshinweis mit der geltenden Regelung, widersprechen # sich beide Aussagen im Text, obwohl jede für sich zutrifft. Deshalb wird # die Verneinung vor dem Anhängen auf ihren Geltungsbereich zurückgeführt. answer = ctx.answer if amendments: try: answer = reconcile_negative_answer(answer, amendments, path=AMENDMENTS_FILE) except Exception as exc: # noqa: BLE001 - siehe oben: nie die Antwort verhindern. logger.warning("Abgleich mit dem Änderungshinweis übersprungen: %s", exc) # Der Reichweiten-Hinweis spricht von den "herangezogenen Textstellen". Hat # die Antwort nichts gefunden, wurde nichts herangezogen: die Verweise # stammen dann aus Treffern, die das Modell verworfen hat, und benennen ein # fremdes Thema. Beim Beispiel Chargendokumentation waren das AM-RL und BtMVV. if boundary and nothing_found: logger.info("Reichweiten-Hinweis unterdrückt: Antwort zitiert keine Quelle.") boundary = [] # Substitutionsausschluss: deterministische Prüfung gegen AM-RL Anlage VII # Teil B. Die Liste ist bewusst nicht indiziert — über Retrieval wäre die # praktisch wichtigste Auskunft ("steht nicht drauf, also austauschbar") # nicht belegbar, weil top-k immer Treffer liefert. befund = None austausch = None biosimilar = None otc = None verordnung = None lifestyle = None tabak = None if ctx.route != "meta": # Teil A zuerst: seine Wirkstoffliste ist rund achtmal so lang wie die # von Teil B und liefert dessen Negativauskunft den Namen, den die # Endungsheuristik allein nicht sicher erkennt. try: austausch = amrl_austauschbarkeit.pruefe(ctx.question, path=AMRL_AUSTAUSCHBARKEIT_FILE) except Exception as exc: # noqa: BLE001 - eine Prüfung darf nie die Antwort verhindern. logger.warning("Austauschbarkeitsprüfung übersprungen: %s", exc) austausch = None try: befund = amrl_substitution.pruefe( ctx.question, path=AMRL_SUBSTITUTION_FILE, kandidat=(austausch.wirkstoff if austausch is not None else None), ) except Exception as exc: # noqa: BLE001 - siehe oben. logger.warning("Substitutionsprüfung übersprungen: %s", exc) befund = None try: biosimilar = amrl_biosimilars.pruefe(ctx.question, path=AMRL_BIOSIMILARS_FILE) except Exception as exc: # noqa: BLE001 - siehe oben. logger.warning("Biosimilar-Prüfung übersprungen: %s", exc) biosimilar = None try: otc = amrl_otc.pruefe(ctx.question, path=AMRL_OTC_FILE) except Exception as exc: # noqa: BLE001 - siehe oben. logger.warning("OTC-Prüfung übersprungen: %s", exc) otc = None try: verordnung = amrl_verordnungsausschluss.pruefe( ctx.question, path=AMRL_VERORDNUNG_FILE ) except Exception as exc: # noqa: BLE001 - siehe oben. logger.warning("Verordnungsausschluss-Prüfung übersprungen: %s", exc) verordnung = None try: lifestyle = amrl_lifestyle.pruefe(ctx.question, path=AMRL_LIFESTYLE_FILE) except Exception as exc: # noqa: BLE001 - siehe oben. logger.warning("Lifestyle-Prüfung übersprungen: %s", exc) lifestyle = None try: # Der Wirkstoff wird aus Anlage II übergeben, nicht neu erkannt: # „Champix" steht dort in der Fertigarzneimittelspalte und wird zu # „Vareniclin" aufgelöst, während Anlage IIa gar keine Produkte führt # („alle marktverfügbaren Arzneimittel"). Übergeben wird nur, wenn # der Treffer die Nikotinabhängigkeit betrifft — sonst beantwortete # eine Frage nach Sildenafil eine Frage nach der Tabakentwöhnung. tabak = amrl_tabakentwoehnung.pruefe( ctx.question, path=AMRL_TABAKENTWOEHNUNG_FILE, stoff=( lifestyle.treffer[0].wirkstoff if lifestyle is not None and lifestyle.tabakentwoehnung and lifestyle.treffer else None ), ) except Exception as exc: # noqa: BLE001 - siehe oben. logger.warning("Tabakentwöhnungs-Prüfung übersprungen: %s", exc) tabak = None geprueft = [ b for b in (befund, austausch, biosimilar, otc, verordnung, lifestyle, tabak) if b is not None and b.ist_belastbar ] # Trägt einer der Befunde eine Auskunft, wurde die Liste geprüft — dann darf # der Reichweiten-Hinweis nicht weiter behaupten, sie sei es nicht. Die # übrigen Anlagen (Medizinprodukte, Verordnungsausschlüsse …) bleiben # ungeprüft, deshalb wird nur der eine Eintrag entfernt und nicht der ganze # Hinweis. # Dasselbe gilt, wenn das Normregister eine AM-RL-Anlage als Fundstelle # benennt: der Hinweis meldete die Anlagen als ungeprüft, während die Antwort # sich zwei Absätze darüber auf Anlage VII Teil A berief. anlage_benannt = False if ctx.route != "meta": try: anlage_benannt = norm_anchors.anlagen_geprueft(ctx.question) except Exception as exc: # noqa: BLE001 - siehe oben. logger.warning("Anlagenprüfung des Normregisters übersprungen: %s", exc) if (geprueft or anlage_benannt) and boundary: boundary = [ref for ref in boundary if ref.get("key") != "am_rl_anlagen"] # Die Listenbefunde stehen ganz oben: sie sind die Auskunft, das Modell # darunter ordnet ein, was daraus für Abgabe und Abrechnung folgt. Die # Kurzantwort wird dabei angeglichen — ohne die Listen kommt das Modell zur # allgemeinen Ersetzungsregel und widerspricht dem Befund zwei Zeilen darüber. if geprueft: answer = amrl_substitution.antwort_mit_befunden( answer, bloecke=[ amrl_substitution.befund_block(befund) if befund is not None else "", amrl_austauschbarkeit.befund_block(austausch) if austausch is not None else "", amrl_biosimilars.befund_block(biosimilar) if biosimilar is not None else "", amrl_lifestyle.befund_block(lifestyle) if lifestyle is not None else "", amrl_tabakentwoehnung.befund_block(tabak) if tabak is not None else "", amrl_verordnungsausschluss.befund_block(verordnung) if verordnung is not None else "", amrl_otc.befund_block(otc) if otc is not None else "", ], verdikt=_listenverdikt( befund, austausch, biosimilar, otc, verordnung, lifestyle, tabak ), ) # Normebene statt Stoffebene: die vier Listen prüfen, ob ein Wirkstoff auf # einer Anlage steht, und melden bei einer reinen Normfrage planmäßig # `nicht_pruefbar`. Genau dort lagen die beiden schwersten Fehler der # Auswertung — eine OTC-Frage mit § 31 statt § 34 beantwortet, das # Zuweisungsverbot aus § 360 hergeleitet statt aus § 31 Abs. 1 Satz 6. normbefund = None if ctx.route != "meta": try: normbefund = norm_anchors.pruefe(ctx.question, answer) except Exception as exc: # noqa: BLE001 - ein Abgleich darf nie die Antwort verhindern. logger.warning("Normabgleich übersprungen: %s", exc) normbefund = None if normbefund is not None: # Steht die Antwort auf der falschen Vorschrift, werden „Kurzantwort" und # „Maßgebliche Norm" neu gesetzt — ein bloßer Nachtrag stünde neben der # falschen Antwort, und nichts im Text sagte, welcher gilt. Fehlt nur ein # Glied der Kette, bleibt die Antwort stehen und bekommt den Nachtrag. answer = norm_anchors.in_antwort_einsetzen(answer, normbefund) # Beide Hinweise erst jetzt: sie werden ans Ende gehängt, und dort steht # ohne eigene Überschrift noch der Abschnitt „Kurzantwort". Setzt ein # Listenbefund diese Kurzantwort, ersetzt `antwort_mit_befunden` den ganzen # Abschnitt — ein vorher angehängter Hinweis steckte darin und wäre spurlos # verschwunden. Die Reihenfolge der beiden zueinander bleibt: der # Änderungshinweis trägt die eigentliche Auskunft und steht deshalb vor dem # Reichweiten-Hinweis, der nur eine Einschränkung benennt. answer = append_amendment_note(answer, amendment_note) if boundary: answer = append_boundary_note(answer, boundary) ctx.response_body = { "question": ctx.question, "answer": answer, "answer_type": ctx.answer_type, "session_id": ctx.request.state.session_id, "factual_question_index": ctx.memory.factual_question_count, "total_turns": ctx.memory.total_turns, "sources": ctx.sources or [], "needs_clarification": ctx.needs_clarification, "clarification_question": ctx.clarification_question, # Maschinenlesbar für UI/Monitoring: welche Regelwerke fehlten. "corpus_boundary": boundary, # Zitierte Vorschriften, deren indizierter Stand überholt ist. "corpus_amendments": [ {"locator": p.get("locator"), "status": p.get("status"), "current_rule": p.get("current_rule")} for p in amendments ], # Welche Norm das Register für maßgeblich hält und ob die Antwort sie nennt. "norm_anchor": normbefund.to_dict() if normbefund is not None else None, # Maschinenlesbare Listenbefunde samt Stand — für UI und Monitoring. "substitutionsausschluss": befund.to_dict() if befund is not None else None, "austauschbarkeit": austausch.to_dict() if austausch is not None else None, "biosimilars": biosimilar.to_dict() if biosimilar is not None else None, "otc_verordnungsfaehigkeit": otc.to_dict() if otc is not None else None, "verordnungsausschluss": verordnung.to_dict() if verordnung is not None else None, "lifestyle_ausschluss": lifestyle.to_dict() if lifestyle is not None else None, "tabakentwoehnung": tabak.to_dict() if tabak is not None else None, } return AskGraphStep("__end__", "response built") ASK_GRAPH_NODES: Dict[AskNodeName, Callable[[AskGraphContext], AskGraphStep]] = { "meta_decision": _ask_graph_meta_decision, "retrieve_hits": _ask_graph_retrieve_hits, "compose_answer": _ask_graph_compose_answer, "orchestrate_answer": _ask_graph_orchestrate_answer, "normalize_sources": _ask_graph_normalize_sources, "update_memory": _ask_graph_update_memory, "build_response": _ask_graph_build_response, } ASK_GRAPH: Dict[str, List[Dict[str, str]]] = { "__start__": [{"to": "meta_decision", "label": ""}], "meta_decision": [ {"to": "compose_answer", "label": "meta"}, {"to": "orchestrate_answer", "label": "non_meta"}, {"to": "retrieve_hits", "label": "legacy"}, ], "retrieve_hits": [{"to": "compose_answer", "label": ""}], "orchestrate_answer": [{"to": "normalize_sources", "label": ""}], "compose_answer": [{"to": "normalize_sources", "label": ""}], "normalize_sources": [{"to": "update_memory", "label": ""}], "update_memory": [{"to": "build_response", "label": ""}], "build_response": [{"to": "__end__", "label": ""}], } class AskGraphVizState(TypedDict, total=False): """Minimaler LangGraph-State nur für Visualisierung/Rendering.""" route: str def _ask_langgraph_passthrough(state: AskGraphVizState) -> AskGraphVizState: """Dummy-Node für LangGraph-Rendering; die echte Logik bleibt in run_ask_graph.""" return state def _make_ask_langgraph_router(source: str, labels: List[str]): """ Erzeugt einen Router für LangGraph-Visualisierung. Für das Rendering ist nur die Mapping-Struktur wichtig. Falls der Graph doch testweise ausgeführt wird, kann pro Node über '_route' geroutet werden; sonst wird der erste Label-Zweig genutzt. """ default_label = labels[0] def _router(state: AskGraphVizState) -> str: return state.get(f"{source}_route", state.get("route", default_label)) return _router def build_ask_workflow(): """ Baut den LangGraph-Workflow aus ASK_GRAPH. Diese Funktion ist die stabile Schnittstelle für visualize_graph.py: from app import build_ask_workflow workflow = build_ask_workflow() workflow.get_graph(xray=True).draw_mermaid_png() Wenn ASK_GRAPH in app.py geändert wird, übernimmt die Visualisierung diese Änderung automatisch, ohne dass visualize_graph.py angepasst werden muss. """ if StateGraph is None: raise RuntimeError( "LangGraph ist nicht installiert. Bitte ausführen: pip install -U langgraph langchain-core" ) workflow = StateGraph(AskGraphVizState) node_names = set() for source, edges in ASK_GRAPH.items(): if source not in {"__start__", "__end__"}: node_names.add(source) for edge in edges: target = edge["to"] if target not in {"__start__", "__end__"}: node_names.add(target) for node_name in sorted(node_names): workflow.add_node(node_name, _ask_langgraph_passthrough) start_edges = ASK_GRAPH.get("__start__") or [] if not start_edges: raise RuntimeError("ASK_GRAPH benötigt eine __start__-Kante.") workflow.set_entry_point(start_edges[0]["to"]) for source, edges in ASK_GRAPH.items(): if source in {"__start__", "__end__"}: continue labelled_edges = [edge for edge in edges if edge.get("label")] plain_edges = [edge for edge in edges if not edge.get("label")] if labelled_edges: labels = [edge["label"] for edge in labelled_edges] workflow.add_conditional_edges( source, _make_ask_langgraph_router(source, labels), {edge["label"]: (END if edge["to"] == "__end__" else edge["to"]) for edge in labelled_edges}, ) for edge in plain_edges: target = END if edge["to"] == "__end__" else edge["to"] workflow.add_edge(source, target) return workflow.compile() def get_ask_workflow(): """Alias für Visualisierungsskripte, die eine Getter-Funktion bevorzugen.""" return build_ask_workflow() # Backwards-/Convenience-Aliase für Visualisierungsskripte. # Wichtig: FastAPI bleibt weiterhin in der Variable `app`; der LangGraph darf # deshalb nicht ebenfalls `app` heißen. try: ASK_WORKFLOW = build_ask_workflow() except Exception: ASK_WORKFLOW = None AGENT_GRAPH = ASK_GRAPH ASK_LANGGRAPH_APP = ASK_WORKFLOW def _ask_graph_mermaid() -> str: lines = ["flowchart TD"] for source, edges in ASK_GRAPH.items(): for edge in edges: target = edge["to"] label = edge.get("label") or "" if label: lines.append(f' {source}["{source}"] -- "{label}" --> {target}["{target}"]') else: lines.append(f' {source}["{source}"] --> {target}["{target}"]') return "\n".join(lines) def run_ask_graph(ctx: AskGraphContext) -> AskGraphContext: """Führt den /ask-Graphen aus, ohne fachliche Zusatzpfade einzuführen.""" current: AskNodeName = "meta_decision" for _ in range(MAX_ASK_GRAPH_STEPS): node = ASK_GRAPH_NODES.get(current) if node is None: raise HTTPException(status_code=500, detail=f"Unbekannter Ask-Graph-Knoten: {current}") step = node(ctx) ctx.trace.append( AskGraphTraceEntry( node=current, next_node=step.next_node, reason=step.reason, route=ctx.route, hit_count=len(ctx.hits or []), answer_type=ctx.answer_type, ) ) if step.next_node == "__end__": return ctx current = step.next_node raise HTTPException( status_code=500, detail=f"Ask-Graph nach {MAX_ASK_GRAPH_STEPS} Schritten abgebrochen.", ) def _legacy_answer_flow(ctx: AskGraphContext) -> AskGraphContext: """ Fallback auf die bisherige monolithische /ask-Logik. Dieser Pfad bleibt absichtlich funktionsgleich zum Graphen und dient nur als schneller Rollback über ASK_GRAPH_ENABLED=false. """ if is_meta_question(ctx.question): ctx.route = "meta" ctx.hits = [] ctx.answer, ctx.answer_type = ctx.composer.compose(ctx.question, [], memory=ctx.memory) ctx.sources = [] else: ctx.route = "non_meta" ctx.hits = _retrieve_hits(ctx.payload) if hasattr(ctx.composer, "compose_with_sources"): try: ctx.answer, ctx.answer_type, raw_sources = ctx.composer.compose_with_sources( ctx.question, ctx.hits, memory=ctx.memory, ) except TypeError: ctx.answer, ctx.answer_type = ctx.composer.compose(ctx.question, ctx.hits, memory=ctx.memory) raw_sources = _build_sources_from_composer_or_hits(ctx.composer, ctx.hits, ctx.payload) else: ctx.answer, ctx.answer_type = ctx.composer.compose(ctx.question, ctx.hits, memory=ctx.memory) raw_sources = _build_sources_from_composer_or_hits(ctx.composer, ctx.hits, ctx.payload) ctx.raw_sources = raw_sources if ctx.answer_type == "document": ctx.sources = _normalize_returned_sources( answer=ctx.answer, sources=raw_sources, hits=ctx.hits, payload=ctx.payload, ) ctx.answer = _postprocess_answer( ctx.composer, ctx.answer, sources=ctx.sources, hits=ctx.hits or [], ) ctx.sources = _normalize_returned_sources( answer=ctx.answer, sources=ctx.sources or raw_sources, hits=ctx.hits, payload=ctx.payload, ) else: ctx.sources = [] ctx.memory.add_turn( user_message=ctx.question, assistant_message=ctx.answer, question_kind=classify_question(ctx.question), ) _ask_graph_build_response(ctx) return ctx def _ask_graph_debug_payload(ctx: AskGraphContext) -> Dict[str, Any]: return { "ask_graph_enabled": ASK_GRAPH_ENABLED, "orchestrator_enabled": ORCHESTRATOR_ENABLED, "orchestrator": ctx.orchestrator_debug, "workflow_trace": [entry.__dict__ for entry in (ctx.trace or [])], "workflow_graph": ASK_GRAPH, "workflow_mermaid": _ask_graph_mermaid(), "route": ctx.route, "is_meta_question": is_meta_question(ctx.question), "question_kind": classify_question(ctx.question), "retrieved_hit_count": len(ctx.hits or []), "answer_source_numbers": _extract_answer_source_numbers(ctx.answer), "raw_sources": ctx.raw_sources or [], "normalized_sources": ctx.sources or [], "hits": [_debug_hit(hit) for hit in (ctx.hits or [])], "top_k": ctx.payload.top_k or DEFAULT_TOP_K, "fetch_k": ctx.payload.fetch_k or DEFAULT_FETCH_K, "max_final_results": ctx.payload.max_final_results or DEFAULT_MAX_FINAL_RESULTS, "max_sources": ctx.payload.max_sources or DEFAULT_MAX_SOURCES, "include_neighbors": ctx.payload.include_neighbors, "include_explicit_sections": ctx.payload.include_explicit_sections, "restrict_to_default_container": ctx.payload.restrict_to_default_container, "source_sync_to_answer_markers": SOURCE_SYNC_TO_ANSWER_MARKERS, "chroma_persist_dir": CHROMA_PERSIST_DIR, "chroma_collection": CHROMA_COLLECTION, "chroma_count": _collection_count(), "corpus_counts": _corpus_counts(), "routed_corpora": sorted({ str(hit.get("corpus_id")) for hit in (ctx.hits or []) if hit.get("corpus_id") }), } # ----------------------------------------------------------------------------- # API Endpoints # ----------------------------------------------------------------------------- @app.post("/ask") def ask(payload: Question, request: Request): question = _clean_question(payload.question) if not question: raise HTTPException(status_code=422, detail="question darf nicht leer sein.") session: SessionState = request.state.session memory = session.get_memory() llm = _build_llm(session) composer = _build_composer(llm, payload) orchestrator = _build_orchestrator(get_retriever(), composer, payload) if ORCHESTRATOR_ENABLED else None ctx = AskGraphContext( payload=payload, request=request, session=session, memory=memory, llm=llm, composer=composer, question=question, orchestrator=orchestrator, ) ctx = run_ask_graph(ctx) if ASK_GRAPH_ENABLED else _legacy_answer_flow(ctx) response_body: Dict[str, Any] = ctx.response_body or { "question": question, "answer": ctx.answer, "answer_type": ctx.answer_type, "session_id": request.state.session_id, "factual_question_index": memory.factual_question_count, "total_turns": memory.total_turns, "sources": ctx.sources or [], "needs_clarification": ctx.needs_clarification, "clarification_question": ctx.clarification_question, "corpus_boundary": [], # Der Fallback greift nur, wenn der Graph keinen Body gebaut hat; die # Schlüssel bleiben trotzdem stabil, damit die UI nicht zwei Formen kennen muss. "substitutionsausschluss": None, "austauschbarkeit": None, "biosimilars": None, "otc_verordnungsfaehigkeit": None, "verordnungsausschluss": None, "lifestyle_ausschluss": None, "tabakentwoehnung": None, } if payload.debug: response_body["debug"] = _ask_graph_debug_payload(ctx) return response_body @app.get("/pdf/{filename}") def get_pdf(filename: str): """Liefert ein Quell-PDF für den eingebetteten Fundstellen-Viewer aus. Es werden nur einfache PDF-Dateinamen akzeptiert (kein Pfadanteil), und die Datei muss in einem der konfigurierten PDF-Verzeichnisse liegen. """ name = os.path.basename(filename or "").strip() if not name or not _PDF_NAME_RE.match(name): raise HTTPException(status_code=400, detail="Ungültiger PDF-Dateiname.") for directory in PDF_SEARCH_DIRS: candidate = (directory / name).resolve() try: candidate.relative_to(directory.resolve()) except (ValueError, OSError): continue if candidate.is_file(): return FileResponse( str(candidate), media_type="application/pdf", headers={ "Content-Disposition": f'inline; filename="{name}"', "Cache-Control": "public, max-age=3600", }, ) raise HTTPException(status_code=404, detail=f"PDF nicht gefunden: {name}") @app.get("/debug/workflow") def debug_workflow(): return { "ok": True, "ask_graph_enabled": ASK_GRAPH_ENABLED, "max_ask_graph_steps": MAX_ASK_GRAPH_STEPS, "langgraph_available": StateGraph is not None, "graph": ASK_GRAPH, "mermaid": _ask_graph_mermaid(), } @app.get("/debug/workflow/mermaid") def debug_workflow_mermaid(): return Response(_ask_graph_mermaid(), media_type="text/plain") @app.get("/health") def health(): return { "ok": True, "app": APP_TITLE, "version": APP_VERSION, "collection": CHROMA_COLLECTION, "corpora": _corpus_counts(), "chroma_persist_dir": CHROMA_PERSIST_DIR, "chroma_count": _collection_count(), "default_container_id": DEFAULT_CONTAINER_ID, "embedding_model": EMBEDDING_MODEL, "groq_model": GROQ_MODEL, "source_sync_to_answer_markers": SOURCE_SYNC_TO_ANSWER_MARKERS, "ask_graph_enabled": ASK_GRAPH_ENABLED, "orchestrator_enabled": ORCHESTRATOR_ENABLED, } @app.get("/debug/retriever") def debug_retriever(): """Diagnose pro Korpus statt für eine einzelne Collection.""" try: registry = get_corpus_registry() diagnostics = registry.diagnostics() except Exception as exc: return { "ok": False, "error": f"{type(exc).__name__}: {exc}", "chroma_persist_dir": CHROMA_PERSIST_DIR, } for entry in diagnostics["corpora"]: if not entry.get("ok"): continue try: res = registry.retriever(entry["corpus_id"]).col.get(limit=3, include=["metadatas", "documents"]) docs = res.get("documents") or [] metas = res.get("metadatas") or [] ids = res.get("ids") or [] entry["sample"] = [ { "id": ids[idx] if idx < len(ids) else None, "metadata": meta, "normalized_source": _source_from_hit({"metadata": meta}), "text_preview": (docs[idx] if idx < len(docs) else "")[:250], } for idx, meta in enumerate(metas) ] except Exception as exc: # noqa: BLE001 entry["sample_error"] = f"{type(exc).__name__}: {exc}" return { "ok": True, "chroma_persist_dir": CHROMA_PERSIST_DIR, "count": _collection_count(), "embedding_consistency": registry.assert_consistent_embedding() or "ok", **diagnostics, } @app.get("/debug/routing") def debug_routing(question: str = ""): """Zeigt, welche Korpora eine Frage erreichen würde — und warum.""" try: available = get_corpus_registry().available() except Exception as exc: # noqa: BLE001 return {"ok": False, "error": f"{type(exc).__name__}: {exc}"} out: Dict[str, Any] = {"ok": True, **explain_routing(question, available)} # Der Router sieht nur den Wortlaut. Ob ein Korpus zusätzlich zwingend # abgefragt wird, entscheidet das Normregister — und genau dieser Fall ist # der, den man debuggen will ("Sonderkennzeichen" nennt kein Korpus). try: out["federation"] = get_retriever().explain_selection(question) except Exception as exc: # noqa: BLE001 out["federation_error"] = f"{type(exc).__name__}: {exc}" return out # ----------------------------------------------------------------------------- # Session System-Prompt # ----------------------------------------------------------------------------- @app.get("/system-prompt") def get_system_prompt(request: Request): session: SessionState = request.state.session return { "system_prompt": session.system_prompt, "default_system_prompt": DEFAULT_SYSTEM_PROMPT, } @app.post("/system-prompt") def set_system_prompt(payload: SystemPromptPayload, request: Request): prompt = payload.system_prompt.strip() if not prompt: raise HTTPException(status_code=422, detail="system_prompt darf nicht leer sein.") request.state.session.system_prompt = prompt return { "ok": True, "system_prompt": request.state.session.system_prompt, } @app.post("/system-prompt/reset") def reset_system_prompt(request: Request): request.state.session.system_prompt = DEFAULT_SYSTEM_PROMPT return { "ok": True, "system_prompt": request.state.session.system_prompt, } # ----------------------------------------------------------------------------- # Session zurücksetzen und Verlauf abrufen # ----------------------------------------------------------------------------- @app.post("/session/reset") def reset_session(request: Request): session: SessionState = request.state.session session.reset() return {"ok": True} @app.get("/session/history") def get_history(request: Request): session: SessionState = request.state.session memory = session.get_memory() return { "factual_questions": memory.get_factual_qa(), "total_turns": memory.total_turns, "factual_question_count": memory.factual_question_count, } @app.get("/session") def get_session_info(request: Request): session: SessionState = request.state.session memory = session.get_memory() return { "session_id": request.state.session_id, "total_turns": memory.total_turns, "factual_question_count": memory.factual_question_count, } # ----------------------------------------------------------------------------- # UI # ----------------------------------------------------------------------------- @app.get("/") def index(): if not os.path.exists(INDEX_FILE): raise HTTPException( status_code=404, detail=f"UI-Datei nicht gefunden: {INDEX_FILE}", ) return FileResponse(INDEX_FILE)