855fc71a1e
- app/safety/llm_classifier.py: LLM klassifiziert Aeusserungen (medical/self_harm/ help/none) mit Konfidenz-Schwelle, eigener JSON-Prompt, Reasoning aus, temp 0 - emergency.py: Eskalation in _escalate() gebuendelt (Log/Metrik/Webhook, source-Label); schedule_llm_emergency_check() faehrt Stufe 2 als Hintergrund-Task NUR wenn die Stichwort-Heuristik nichts fand -> faengt verpasste Formulierungen ohne Mehrlatenz - /ws/voice schickt bei LLM-Treffer ein nachgelagertes emergency-Event (source=llm) - Konfig: EMERGENCY_LLM_ENABLED/_PROVIDER/_MIN_CONFIDENCE - Tests: Parsing, Konfidenz-Schwelle, Skip-bei-Keyword-Treffer, Eskalation+Callback, disabled, kaputtes JSON - Doku: README + Architektur-Roadmap (Punkt 6) Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
128 lines
4.1 KiB
Python
128 lines
4.1 KiB
Python
import asyncio
|
|
|
|
import pytest
|
|
|
|
from app.config import settings
|
|
from app.safety import emergency as em
|
|
from app.safety import llm_classifier as lc
|
|
|
|
|
|
class StubLLM:
|
|
def __init__(self, raw):
|
|
self.raw = raw
|
|
self.calls = 0
|
|
|
|
async def complete(self, text, history=None, session_id=None):
|
|
self.calls += 1
|
|
return self.raw
|
|
|
|
|
|
class FakeUser:
|
|
id = "anonymous"
|
|
display_name = "Test"
|
|
|
|
|
|
class FakeStore:
|
|
def __init__(self):
|
|
self.logged = []
|
|
|
|
def log_emergency(self, user_id, category, snippet):
|
|
self.logged.append((user_id, category, snippet))
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _clear_pending():
|
|
em._pending.clear()
|
|
yield
|
|
em._pending.clear()
|
|
|
|
|
|
def test_parse_classification_variants():
|
|
assert lc.parse_classification('{"category":"medical","confidence":0.9}')["category"] == "medical"
|
|
# umschlossen von Text
|
|
got = lc.parse_classification('Antwort: {"category":"help","confidence":0.7,"reason":"Feuer"}')
|
|
assert got["category"] == "help" and got["confidence"] == 0.7
|
|
# 'none' ist kein Notfall
|
|
assert lc.parse_classification('{"category":"none","confidence":0.99}') is None
|
|
# unbekannte Kategorie
|
|
assert lc.parse_classification('{"category":"foo","confidence":0.99}') is None
|
|
assert lc.parse_classification("kein json") is None
|
|
assert lc.parse_classification("") is None
|
|
|
|
|
|
def test_classify_respects_confidence_threshold(monkeypatch):
|
|
monkeypatch.setattr(settings, "emergency_llm_min_confidence", 0.6)
|
|
monkeypatch.setattr(lc, "_build_classifier_llm",
|
|
lambda cfg: StubLLM('{"category":"medical","confidence":0.4}'))
|
|
# unter der Schwelle -> kein Notfall
|
|
assert asyncio.run(lc.classify_emergency("mir ist schwindelig")) is None
|
|
|
|
monkeypatch.setattr(lc, "_build_classifier_llm",
|
|
lambda cfg: StubLLM('{"category":"medical","confidence":0.85}'))
|
|
res = asyncio.run(lc.classify_emergency("mir wird ganz schwarz vor augen"))
|
|
assert res["category"] == "medical"
|
|
|
|
|
|
def test_schedule_skips_when_keyword_already_hit(monkeypatch):
|
|
store = FakeStore()
|
|
called = StubLLM('{"category":"medical","confidence":0.9}')
|
|
monkeypatch.setattr(lc, "_build_classifier_llm", lambda cfg: called)
|
|
|
|
async def run():
|
|
# keyword_hit ist gesetzt -> Stufe 2 wird uebersprungen
|
|
task = em.schedule_llm_emergency_check(
|
|
FakeUser(), "egal", store, {"category": "medical", "matched": "x"}
|
|
)
|
|
return task
|
|
|
|
assert asyncio.run(run()) is None
|
|
assert called.calls == 0
|
|
assert store.logged == []
|
|
|
|
|
|
def test_schedule_disabled_is_noop(monkeypatch):
|
|
monkeypatch.setattr(settings, "emergency_llm_enabled", False)
|
|
store = FakeStore()
|
|
|
|
async def run():
|
|
return em.schedule_llm_emergency_check(FakeUser(), "hilfe", store, None)
|
|
|
|
assert asyncio.run(run()) is None
|
|
assert store.logged == []
|
|
|
|
|
|
def test_schedule_escalates_and_calls_callback(monkeypatch):
|
|
monkeypatch.setattr(settings, "emergency_llm_enabled", True)
|
|
monkeypatch.setattr(settings, "emergency_llm_min_confidence", 0.6)
|
|
monkeypatch.setattr(lc, "_build_classifier_llm",
|
|
lambda cfg: StubLLM('{"category":"self_harm","confidence":0.95}'))
|
|
store = FakeStore()
|
|
events = []
|
|
|
|
async def on_emergency(category):
|
|
events.append(category)
|
|
|
|
async def run():
|
|
task = em.schedule_llm_emergency_check(
|
|
FakeUser(), "ich will nicht mehr weiterleben",
|
|
store, None, on_emergency=on_emergency,
|
|
)
|
|
await task
|
|
|
|
asyncio.run(run())
|
|
assert store.logged == [("anonymous", "self_harm", "ich will nicht mehr weiterleben")]
|
|
assert events == ["self_harm"]
|
|
|
|
|
|
def test_schedule_malformed_output_no_escalation(monkeypatch):
|
|
monkeypatch.setattr(settings, "emergency_llm_enabled", True)
|
|
monkeypatch.setattr(lc, "_build_classifier_llm",
|
|
lambda cfg: StubLLM("Tut mir leid, kein JSON."))
|
|
store = FakeStore()
|
|
|
|
async def run():
|
|
task = em.schedule_llm_emergency_check(FakeUser(), "irgendwas", store, None)
|
|
await task
|
|
|
|
asyncio.run(run())
|
|
assert store.logged == []
|