96 lines
3.4 KiB
Python
96 lines
3.4 KiB
Python
|
"""LLM-Notfall-Klassifikation (zweite Stufe der Notfall-Erkennung).
|
||
|
|
|
||
|
|
Ergaenzt die schnelle Stichwort-Heuristik (`app.safety.emergency.detect`) um einen
|
||
|
|
LLM-Klassifikator, der Formulierungen erkennt, die keine Stichwoerter treffen.
|
||
|
|
|
||
|
|
Bewusst **best-effort** und mit Konfidenz-Schwelle (sensibler Senioren-Kontext):
|
||
|
|
Ein LLM-Fehler oder kaputtes JSON fuehrt nie zu einem Alarm und nie zu einem Fehler
|
||
|
|
im Antwort-Turn.
|
||
|
|
"""
|
||
|
|
|
||
|
|
import json
|
||
|
|
import logging
|
||
|
|
import re
|
||
|
|
|
||
|
|
from app.config import Settings, settings
|
||
|
|
|
||
|
|
logger = logging.getLogger(__name__)
|
||
|
|
|
||
|
|
# Gueltige Notfall-Kategorien (deckungsgleich mit der Stichwort-Heuristik).
|
||
|
|
VALID_CATEGORIES = {"medical", "self_harm", "help"}
|
||
|
|
|
||
|
|
_SYSTEM_PROMPT = (
|
||
|
|
"Du bist ein Sicherheits-Klassifikator fuer einen Senioren-Sprachassistenten. "
|
||
|
|
"Beurteile, ob die Nutzeraeusserung einen akuten Notfall beschreibt. Kategorien: "
|
||
|
|
"'medical' (akute medizinische Notlage, z. B. Brustschmerz, Atemnot, Sturz, "
|
||
|
|
"Schlaganfall), 'self_harm' (Suizidalitaet/Selbstgefaehrdung), 'help' (akuter "
|
||
|
|
"Hilferuf, z. B. Feuer, Notruf), 'none' (kein Notfall). Antworte AUSSCHLIESSLICH "
|
||
|
|
"mit JSON: {\"category\": \"medical|self_harm|help|none\", \"confidence\": 0.0-1.0, "
|
||
|
|
"\"reason\": \"kurze Begruendung\"}. Sei zurueckhaltend: nur echte, akute Notlagen "
|
||
|
|
"sind ein Notfall, keine beilaeufigen Erwaehnungen oder Vergangenes."
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def _build_classifier_llm(cfg: Settings):
|
||
|
|
"""Baut eine eigene LLM-Instanz fuer die Klassifikation (eigener JSON-Prompt)."""
|
||
|
|
provider = cfg.emergency_llm_provider or cfg.default_llm_provider
|
||
|
|
if provider == "local-openai-compatible":
|
||
|
|
from app.providers.llm.local_openai_compatible import LocalOpenAICompatibleLLM
|
||
|
|
|
||
|
|
return LocalOpenAICompatibleLLM(
|
||
|
|
cfg.local_llm_base_url,
|
||
|
|
cfg.local_llm_api_key,
|
||
|
|
cfg.local_llm_model,
|
||
|
|
system_prompt=_SYSTEM_PROMPT,
|
||
|
|
disable_reasoning=True,
|
||
|
|
max_tokens=128,
|
||
|
|
temperature=0.0,
|
||
|
|
)
|
||
|
|
|
||
|
|
from app.dependencies import get_llm_provider
|
||
|
|
|
||
|
|
return get_llm_provider(provider, cfg)
|
||
|
|
|
||
|
|
|
||
|
|
def parse_classification(raw: str) -> dict | None:
|
||
|
|
"""Liest {category, confidence, reason} aus der (evtl. verrauschten) LLM-Antwort."""
|
||
|
|
if not raw:
|
||
|
|
return None
|
||
|
|
match = re.search(r"\{.*\}", raw, re.DOTALL)
|
||
|
|
if not match:
|
||
|
|
return None
|
||
|
|
try:
|
||
|
|
data = json.loads(match.group(0))
|
||
|
|
except ValueError:
|
||
|
|
return None
|
||
|
|
if not isinstance(data, dict):
|
||
|
|
return None
|
||
|
|
category = data.get("category")
|
||
|
|
if category not in VALID_CATEGORIES:
|
||
|
|
return None
|
||
|
|
try:
|
||
|
|
confidence = float(data.get("confidence", 0.0))
|
||
|
|
except (TypeError, ValueError):
|
||
|
|
confidence = 0.0
|
||
|
|
return {
|
||
|
|
"category": category,
|
||
|
|
"confidence": confidence,
|
||
|
|
"reason": str(data.get("reason", "")),
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
async def classify_emergency(text: str, cfg: Settings = settings) -> dict | None:
|
||
|
|
"""Klassifiziert eine Aeusserung. Liefert {category, confidence, reason} oder None.
|
||
|
|
|
||
|
|
None bedeutet: kein Notfall (bzw. unter der Konfidenz-Schwelle / nicht parsebar).
|
||
|
|
"""
|
||
|
|
if not text or not text.strip():
|
||
|
|
return None
|
||
|
|
llm = _build_classifier_llm(cfg)
|
||
|
|
raw = await llm.complete(text)
|
||
|
|
result = parse_classification(raw)
|
||
|
|
if result is None:
|
||
|
|
return None
|
||
|
|
if result["confidence"] < cfg.emergency_llm_min_confidence:
|
||
|
|
return None
|
||
|
|
return result
|