from __future__ import annotations import os import time import random import re import logging from typing import Optional, List, Dict, Any from groq import Groq logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Frageklassifikation: Single Source of Truth # --------------------------------------------------------------------------- QUESTION_KIND_META = "meta" QUESTION_KIND_LEGAL = "legal" QUESTION_KIND_TECHNICAL = "technical" QUESTION_KIND_OTHER = "other" _ORDINAL_WORDS = ( "erste", "zweite", "dritte", "vierte", "fünfte", "sechste", "siebte", "achte", "neunte", "zehnte", "elfte", "zwölfte", "letzte", "vorletzte", "vorherige", ) _ORDINAL_PATTERN = "|".join(_ORDINAL_WORDS) _META_SIGNALS = [ "was war meine", "wie lautete", "was habe ich gefragt", "was war die frage", "wie war die antwort", "was hast du geantwortet", "welche frage", "erinnere dich", "wiederhole meine", "was war deine antwort", "nochmal die antwort", "antwort auf meine", "antwort auf die", "frage davor", "vorherige frage", "letzte frage", "deine erste antwort", "deine zweite antwort", "deine dritte antwort", "deine vierte antwort", "deine fünfte antwort", ] _META_PATTERNS = [ re.compile( rf"\b(meine|die|deine)\s+({_ORDINAL_PATTERN})\s+(frage|antwort)\b", re.I, ), re.compile(r"\bwas\s+war\s+meine\b", re.I), re.compile(r"\bwas\s+habe\s+ich\s+(zuletzt|vorher|davor)\s+gefragt\b", re.I), re.compile(r"\b(letzte|vorletzte|vorherige)\s+(frage|antwort)\b", re.I), re.compile(r"\b(\d+)\.?\s+(frage|antwort)\b", re.I), re.compile(r"\bantwort\s+(auf|zu)\s+(meine|die|der|deine|frage)\b", re.I), re.compile(r"\bwie\s+hast\s+du\s+(auf|zu)\b", re.I), re.compile(r"\bwas\s+hast\s+du\s+geantwortet\b", re.I), ] _META_ANSWER_REQUEST_PATTERNS = [ re.compile( rf"\b(meine|die|deine)\s+({_ORDINAL_PATTERN}|\d+\.?)\s+antwort\b", re.I, ), re.compile(r"\bantwort\s+(auf|zu)\b", re.I), re.compile(r"\bwie\s+(hast\s+du|lautete\s+die\s+antwort)\b", re.I), re.compile(r"\bwas\s+hast\s+du\s+geantwortet\b", re.I), re.compile(r"\bwas\s+war\s+deine\s+antwort\b", re.I), re.compile(r"\bgib\s+mir\s+die\s+antwort\b", re.I), ] _TECHNICAL_COMMAND_PATTERNS = [ re.compile( r"^\s*(uvicorn|python|python3|pip|pip3|poetry|conda|npm|npx|pnpm|yarn|node|" r"git|docker|docker-compose|kubectl|curl|wget|streamlit|fastapi|pytest|" r"ruff|black|mypy)\b", re.I, ), re.compile(r"\b(app:app|--reload|localhost|127\.0\.0\.1|0\.0\.0\.0)\b", re.I), re.compile(r"^\s*(import\s+\w+|from\s+\w+(\.\w+)*\s+import\s+)", re.I), re.compile(r"^\s*(def|class)\s+\w+\s*[\(:]", re.I), re.compile(r"^\s*@\w+", re.I), re.compile( r"\b(traceback|exception|syntaxerror|typeerror|valueerror|runtimeerror|" r"modulenotfounderror|notfounderror)\b", re.I, ), ] _TECHNICAL_TERMS = { "app", "api", "endpoint", "fastapi", "uvicorn", "reload", "server", "client", "retriever", "composer", "skript", "script", "code", "python", "chroma", "chromadb", "embedding", "vector", "vektor", "vektordatenbank", "datenbank", "session", "cookie", "cors", "groq", "llm", "prompt", "debug", "docker", "git", "npm", } _LEGAL_TERMS = { "§", "§§", "sgb", "bgb", "stgb", "zpo", "vvg", "amvv", "btmvv", "rv", "rahmenvertrag", "vertrag", "anlage", "anhang", "norm", "normen", "paragraph", "absatz", "satz", "gesetz", "verordnung", "rechtsfolge", "rechtsfolgen", "rechtlich", "juristisch", "anspruch", "vergütungsanspruch", "anspruchsgrundlage", "anspruchstatbestand", "retaxation", "retax", "retaxationsvorbehalt", "heilung", "ausnahme", "ausnahmen", "verstoß", "verstöße", "krankenkasse", "apotheke", "apotheker", "apothekerin", "arzneimittel", "belieferung", "abgabe", "auswahlvorgaben", "vertragsärztlich", "vertragsärztliche", "e-rezept", "erezept", "rezept", "verschreibung", "leistungspflicht", "öffentlich-rechtlich", "ordnungsgemäß", "vertragskonform", "versicherter", "versicherten", } _LEGAL_PATTERNS = [ re.compile(r"§{1,2}\s*\d+[a-zA-Z]?", re.I), re.compile(r"\b(sgb|bgb|stgb|zpo|vvg)\s*[ivx\d]*\b", re.I), re.compile(r"\b(amvv|btmvv)\b", re.I), re.compile(r"\b(rahmenvertrag|vergütungsanspruch|retax\w*|vertragsärzt\w*)\b", re.I), ] def _normalize(text: str) -> str: return ( " ".join((text or "").lower().strip().split()) .replace("?", "") .replace("!", "") .replace(".", "") .replace(":", "") ) def is_meta_question(text: str) -> bool: """Erkennt Meta-Fragen zum bisherigen Gespräch.""" if not text: return False lower = text.lower() if any(signal in lower for signal in _META_SIGNALS): return True normalized = _normalize(text) return any(pattern.search(normalized) for pattern in _META_PATTERNS) def is_meta_answer_request(text: str) -> bool: """True, wenn eine Meta-Frage die Antwort und nicht die Frage meint.""" if not text: return False normalized = _normalize(text) return any(pattern.search(normalized) for pattern in _META_ANSWER_REQUEST_PATTERNS) def is_technical_or_nonlegal_question(text: str) -> bool: """Erkennt technische Eingaben, die nicht als juristische Fachfragen zählen.""" if not text or not text.strip(): return False normalized = _normalize(text) if any(pattern.search(text) for pattern in _TECHNICAL_COMMAND_PATTERNS): return True tokens = set(re.findall(r"[a-zA-ZäöüÄÖÜß0-9_\-]+", normalized)) technical_hits = tokens.intersection(_TECHNICAL_TERMS) legal_hits = tokens.intersection(_LEGAL_TERMS) return bool(technical_hits and not legal_hits) def is_legal_question(text: str) -> bool: """Heuristische Klassifikation für den Fachfragen-Index.""" if not text or not text.strip(): return False if is_meta_question(text): return False if is_technical_or_nonlegal_question(text): return False normalized = _normalize(text) if any(pattern.search(text) for pattern in _LEGAL_PATTERNS): return True tokens = set(re.findall(r"[a-zA-ZäöüÄÖÜß0-9_\-]+", normalized)) return bool(tokens.intersection(_LEGAL_TERMS)) def classify_question(text: str) -> str: """Klassifiziert Nutzer-Eingaben für die Gedächtnislogik.""" if is_meta_question(text): return QUESTION_KIND_META if is_legal_question(text): return QUESTION_KIND_LEGAL if is_technical_or_nonlegal_question(text): return QUESTION_KIND_TECHNICAL return QUESTION_KIND_OTHER # --------------------------------------------------------------------------- # Prompts # --------------------------------------------------------------------------- DEFAULT_SYSTEM_PROMPT = """ Du bist ein juristischer Antwortassistent für die Auswertung bereitgestellter juristischer Textstellen. Grundregeln: - Antworte bei juristischen Sachfragen ausschließlich anhand der bereitgestellten Textstellen. - Verwende kein allgemeines Rechtswissen als eigene Quelle. - Erfinde keine Paragraphen, Anlagen, Fundstellen, Seitenangaben oder Inhalte. - Verwende nur Quellenmarker, die im bereitgestellten Kontext vorkommen, z. B. [Quelle 1]. - Wenn eine Aussage nicht aus den Textstellen folgt, sage das ausdrücklich. - Wenn Textstellen widersprüchlich oder unzureichend sind, benenne die Grenze. Juristische Methode: - Prüfe Wortlaut, Systematik, Normzusammenhang und erkennbare Regelungsstruktur. - Unterscheide ausdrücklich geregelte Ergebnisse von systematisch ableitbaren Ergebnissen. - Bei Anspruchsfragen unterscheide Anspruchstatbestand, ordnungsgemäße Leistung, Verstöße, Heilungsmöglichkeiten, Ausnahmen und Rechtsfolgen. - Bleibe präzise und vermeide unnötige Breite. """.strip() LEGAL_TASK_INSTRUCTIONS = """ === Bearbeitungsauftrag für juristische Sachfragen === Beantworte die Nutzerfrage nach diesem Schema: 1. Kurzantwort: - Gib das Ergebnis in 1–3 Sätzen an. 2. Einordnung der Textgrundlage: - Nenne, ob das Ergebnis ausdrücklich geregelt, systematisch ableitbar oder nicht belastbar ableitbar ist. - Verweise im Text auf die relevanten Quellenmarker, z. B. [Quelle 1]. 3. Begründung/Subsumtion: - Verknüpfe die relevanten Textstellen. - Wiederhole nicht bloß die Fundstellen. - Nutze nur Informationen aus den bereitgestellten Textstellen. 4. Ausnahmen und Einschränkungen: - Nenne Heilungs-, Ergänzungs-, Ausnahme- oder Retaxationsregeln nur, soweit sie in den Textstellen enthalten sind. - Stelle klar, wenn eine Rechtsfolge nicht aus den Textstellen folgt. 5. Ergebnisformel: - Schließe mit einem prägnanten Satz. Quellenregeln: - Verwende im Fließtext nur Quellenmarker aus dem Kontext: [Quelle 1], [Quelle 2] usw. - Erfinde keine weiteren Quellen. - Erstelle am Ende keine eigene Quellenliste, wenn die Anwendung Quellen separat anzeigt. === Ende des Bearbeitungsauftrags === """.strip() META_TASK_INSTRUCTIONS = """ === Bearbeitungsauftrag für Meta-Fragen === Beantworte ausschließlich anhand des bereitgestellten Konversationsverlaufs. Regeln: - Juristische Textstellen ignorieren. - Nur juristische Sachfragen als Fachfragen zählen. - Meta-Fragen, technische Kommandos, Code-Fragen und Bedienfragen nicht mitzählen. - Wenn nach einer Antwort auf eine Fachfrage gefragt wird, gib die damals gespeicherte Antwort wieder oder fasse sie klar erkennbar zusammen. === Ende des Bearbeitungsauftrags === """.strip() # --------------------------------------------------------------------------- # ConversationMemory # --------------------------------------------------------------------------- class ConversationMemory: """ Strukturiertes Gesprächsgedächtnis. Speichert: - vollständige Turn-History, - separat nur juristische Fachfragen und Antworten für Meta-Fragen. """ def __init__(self): self._messages: List[Dict[str, str]] = [] self._factual_qa: List[Dict[str, str]] = [] def is_meta_question(self, text: str) -> bool: return is_meta_question(text) def is_legal_question(self, text: str) -> bool: return is_legal_question(text) def classify_question(self, text: str) -> str: return classify_question(text) def add_turn( self, user_message: str, assistant_message: str, *, question_kind: Optional[str] = None, force_factual: Optional[bool] = None, ) -> None: kind = question_kind or classify_question(user_message) self._messages.append({ "role": "user", "content": user_message, "kind": kind, }) self._messages.append({ "role": "assistant", "content": assistant_message, "kind": kind, }) if force_factual is None: should_store_as_factual = kind == QUESTION_KIND_LEGAL else: should_store_as_factual = bool(force_factual) if should_store_as_factual: idx = len(self._factual_qa) + 1 self._factual_qa.append({ "index": str(idx), "question": user_message, "answer": assistant_message, }) def get_messages(self) -> List[Dict[str, str]]: return list(self._messages) def get_factual_qa(self) -> List[Dict[str, str]]: return list(self._factual_qa) def build_context_block( self, *, include_answers: bool = True, max_chars: Optional[int] = None, ) -> str: if not self._factual_qa: return "" entries: List[str] = [] for entry in self._factual_qa: if include_answers: block = ( f"[Fachfrage {entry['index']}]\n" f"Frage: {entry['question']}\n" f"Antwort: {entry['answer']}" ) else: block = ( f"[Fachfrage {entry['index']}]\n" f"Frage: {entry['question']}" ) entries.append(block) if max_chars is not None and max_chars > 0: selected: List[str] = [] total = 0 for block in reversed(entries): block_len = len(block) + 2 if selected and total + block_len > max_chars: break selected.append(block) total += block_len entries = list(reversed(selected)) return "\n\n".join( ["=== Bisherige juristische Fachfragen und Antworten ==="] + entries + ["=== Ende des Verlaufs ==="] ) def reset(self) -> None: self._messages.clear() self._factual_qa.clear() @property def factual_question_count(self) -> int: return len(self._factual_qa) @property def total_turns(self) -> int: return len(self._messages) // 2 # --------------------------------------------------------------------------- # GroqClient # --------------------------------------------------------------------------- def _reasoning_tokens(response: Any) -> Optional[int]: """Wie viele der verbrauchten Tokens ins Denken gingen, wenn das Modell es meldet. Nur fürs Log, und deshalb bewusst zahnlos: ein Modell ohne `completion_tokens_details` ist keine Störung, sondern der Normalfall bei den nicht-reasoning-Modellen. """ try: details = getattr(getattr(response, "usage", None), "completion_tokens_details", None) return getattr(details, "reasoning_tokens", None) except Exception: # noqa: BLE001 - eine Logzeile darf nie die Antwort kosten. return None class GroqClient: """ Robuster Client-Wrapper um Groq Chat Completions. Kompatibel mit deiner bisherigen App: - generate(prompt) - chat(prompt, memory=None, rag_context="") - generate_messages(messages) Verbesserungen: - strengere Quellenregeln, - keine Quellenfantasie, - klare Trennung zwischen Meta- und juristischen Fragen, - optionales Debugging des tatsächlich an Groq gesendeten Prompts, - stabilere Fehlerdiagnose. """ def __init__( self, model: str = "openai/gpt-oss-120b", system_prompt: str = DEFAULT_SYSTEM_PROMPT, temperature: float = 0.05, max_tokens: int = 4000, max_retries: int = 2, # Obergrenze für den Nachschlag bei leerer Antwort. Siehe `_call`. max_tokens_ceiling: int = 8000, retry_backoff_base: float = 0.8, history_for_factual_questions: bool = False, max_factual_history_chars: int = 8000, debug_prompts: bool = False, ): api_key = os.environ.get("GROQ_API_KEY") if not api_key: raise RuntimeError( "GROQ_API_KEY ist nicht gesetzt. " "Setze die Umgebungsvariable, bevor du den Server startest." ) self.client = Groq(api_key=api_key) self.model = model self.system_prompt = (system_prompt or "").strip() self.temperature = temperature self.max_tokens = max_tokens self.max_tokens_ceiling = max(max_tokens, max_tokens_ceiling) self.max_retries = max_retries self.retry_backoff_base = retry_backoff_base self.history_for_factual_questions = history_for_factual_questions self.max_factual_history_chars = max_factual_history_chars self.debug_prompts = debug_prompts def set_system_prompt(self, system_prompt: str) -> None: self.system_prompt = (system_prompt or "").strip() @staticmethod def _looks_like_augmented_document_prompt(prompt: str) -> bool: """Erkennt, ob AnswerComposer bereits fachliche Instruktionen enthält. Wichtig: Wenn diese Erkennung fehlschlägt, hängt der Client seine eigenen LEGAL_TASK_INSTRUCTIONS zusätzlich an und erzeugt zwei widersprüchliche Anweisungssets im selben Prompt (z. B. "Ergebnisformel" vs. "Keine Ergebnisformel"). Der Composer-Prompt nutzt Block-Header in Großbuchstaben, daher wird case-insensitiv geprüft. """ p = (prompt or "") if "Nutzerfrage:" not in p: return False lowered = p.lower() markers = ( "du beantwortest eine juristische sachfrage", "antwortschema", "verbindliche regeln:", "evidence first", ) return any(marker in lowered for marker in markers) @staticmethod def _source_marker_count(rag_context: str) -> int: return len(set(re.findall(r"\[Quelle\s+\d+\]", rag_context or ""))) def _call_chat_completion(self, messages: List[Dict[str, str]], **kwargs: Any) -> str: """Call Groq with a locked model. Production safety rule: - The billed Groq model is always `self.model`, i.e. the model configured when `GroqClient` is instantiated, typically from `GROQ_MODEL` in app.py. - Per-call model overrides through kwargs are rejected instead of being silently accepted. This prevents accidental billing of other models such as `openai/gpt-oss-120b` or `qwen/qwen3-32b`. - `service_tier` overrides are also rejected here. The app should not switch Groq billing tiers from arbitrary downstream calls. """ blocked_model = kwargs.pop("model", None) if blocked_model is not None and str(blocked_model) != str(self.model): raise RuntimeError( "Blocked Groq model override: " f"requested={blocked_model!r}, configured={self.model!r}. " "Only the configured model may be used for billing." ) blocked_service_tier = kwargs.pop("service_tier", None) if blocked_service_tier is not None: raise RuntimeError( "Blocked Groq service_tier override: " f"requested={blocked_service_tier!r}. " "Billing-tier selection must not be changed per request." ) actual_model = self.model temperature = kwargs.pop("temperature", self.temperature) max_tokens = kwargs.pop("max_tokens", self.max_tokens) if kwargs: logger.warning( "Ignoring unsupported Groq call kwargs", extra={"ignored_kwargs": sorted(kwargs.keys())}, ) logger.info( "calling Groq chat completion", extra={ "groq_model": actual_model, "temperature": temperature, "max_tokens": max_tokens, }, ) last_err: Optional[Exception] = None # Ein Reasoning-Modell teilt sich `max_tokens` zwischen Denk- und # Antworttokens. Reicht das Budget nicht, kommt eine **leere** Antwort # zurück — mit HTTP 200 und `finish_reason: "length"`. # # Gemessen am 19.08.2026 mit `openai/gpt-oss-120b` und dem damaligen # Budget von 1400: dieselbe Frage nach der Austauschbarkeit von # Darreichungsformen lieferte einmal 0 Zeichen (reasoning_tokens 1400 # von 1400) und einmal 1143 (reasoning_tokens 1046). Die alte Zeile # `return content or ""` machte daraus eine erfolgreiche Antwort; beim # Nutzer kam der Reichweiten-Hinweis an und sonst nichts, und nichts im # System hat es bemerkt. Ein leerer Inhalt ist kein Ergebnis. versuchsbudget = max_tokens for attempt in range(self.max_retries + 1): try: response = self.client.chat.completions.create( model=actual_model, messages=messages, temperature=temperature, max_tokens=versuchsbudget, ) choice = response.choices[0] content = (choice.message.content or "").strip() finish_reason = getattr(choice, "finish_reason", None) if finish_reason == "length": # Auch mit Inhalt: die Antwort endet dann mitten im Satz. # Das gehört ins Log, weil es sonst als fachliche # Unvollständigkeit gelesen wird. logger.warning( "Groq response hit the token limit", extra={ "groq_model": actual_model, "max_tokens": versuchsbudget, "content_chars": len(content), "reasoning_tokens": _reasoning_tokens(response), }, ) if content: return content if versuchsbudget < self.max_tokens_ceiling and attempt < self.max_retries: versuchsbudget = min(versuchsbudget * 2, self.max_tokens_ceiling) logger.warning( "Groq returned no content; retrying with a larger budget", extra={"groq_model": actual_model, "next_max_tokens": versuchsbudget}, ) continue last_err = RuntimeError( f"Groq lieferte keinen Inhalt (finish_reason={finish_reason!r}, " f"max_tokens={versuchsbudget})" ) break except Exception as exc: last_err = exc if attempt < self.max_retries: sleep_s = ( self.retry_backoff_base * (2 ** attempt) + random.uniform(0, 0.25) ) time.sleep(sleep_s) continue raise RuntimeError( f"Groq API call failed after {self.max_retries + 1} attempts: {last_err}" ) from last_err def _build_augmented_prompt( self, *, prompt: str, memory: Optional[ConversationMemory], rag_context: str, include_history_for_factual: Optional[bool] = None, ) -> str: meta = is_meta_question(prompt) parts: List[str] = [] if meta: if memory: history_block = memory.build_context_block( include_answers=True, max_chars=None, ) if history_block: parts.append(history_block) parts.append(META_TASK_INSTRUCTIONS) parts.append(f"Meta-Frage: {prompt}") return "\n\n".join(parts) use_history = ( self.history_for_factual_questions if include_history_for_factual is None else include_history_for_factual ) if use_history and memory: history_block = memory.build_context_block( include_answers=False, max_chars=self.max_factual_history_chars, ) if history_block: parts.append( "=== Gesprächskontext bisheriger juristischer Fachfragen " "(keine Rechtsquelle, nur Kontext) ===\n" f"{history_block}\n" "=== Ende des Gesprächskontexts ===" ) if rag_context and rag_context.strip(): marker_count = self._source_marker_count(rag_context) parts.append( "=== Juristische Textstellen ===\n" f"{rag_context.strip()}\n" "=== Ende der Textstellen ===\n" f"Hinweis: Es gibt {marker_count} zitierbare Quellenmarker im Kontext. " "Zitiere nur diese Marker." ) else: parts.append( "=== Juristische Textstellen ===\n" "[Keine juristischen Textstellen bereitgestellt.]\n" "=== Ende der Textstellen ===" ) # Wenn der AnswerComposer bereits einen vollständigen Dokument-Prompt liefert, # nicht nochmals dieselbe lange juristische Instruktion verdoppeln. if self._looks_like_augmented_document_prompt(prompt): parts.append(prompt) else: parts.append(LEGAL_TASK_INSTRUCTIONS) parts.append(f"Juristische Frage: {prompt}") return "\n\n".join(parts) def chat( self, prompt: str, memory: Optional[ConversationMemory] = None, rag_context: str = "", include_history_for_factual: Optional[bool] = None, **kwargs: Any, ) -> str: augmented = self._build_augmented_prompt( prompt=prompt, memory=memory, rag_context=rag_context, include_history_for_factual=include_history_for_factual, ) if self.debug_prompts: print("=== GROQ DEBUG PROMPT START ===") print(augmented[:6000]) if len(augmented) > 6000: print(f"... [truncated, total chars={len(augmented)}]") print("=== GROQ DEBUG PROMPT END ===") messages = [ {"role": "system", "content": self.system_prompt}, {"role": "user", "content": augmented}, ] return self._call_chat_completion(messages, **kwargs) def generate(self, prompt: str, system_prompt: Optional[str] = None, **kwargs: Any) -> str: sys_prompt = (system_prompt if system_prompt is not None else self.system_prompt).strip() messages = [ {"role": "system", "content": sys_prompt}, {"role": "user", "content": prompt}, ] return self._call_chat_completion(messages, **kwargs) def generate_messages( self, messages: List[Dict[str, str]], system_prompt: Optional[str] = None, **kwargs: Any, ) -> str: sys_prompt = (system_prompt if system_prompt is not None else self.system_prompt).strip() full = [{"role": "system", "content": sys_prompt}] + messages return self._call_chat_completion(full, **kwargs)