81d4cf6fd8
Bisher wurde `language` zwar an STT/TTS übergeben, aber nie an den LLM.
Lokales Modell antwortete deshalb immer auf Englisch/Deutsch, egal ob
Französisch, Spanisch o.a. eingestellt war.
Lösung: `language`-Parameter durch die gesamte LLM-Schicht gezogen:
- base.py: `lang_instruction()` helper + Signatur erweitert
- local_openai_compatible.py: Sprachanweisung ("Respond in Français.")
wird als letzter System-Part in den Message-Stack eingefügt
- openrouter.py: Explizite Sprachanweisung ergänzt den bestehenden
"Answer in the same language as the user"-Prompt
- fallback.py: FallbackLLMProvider leitet `language` durch
- orchestrator.py: alle 3 LLM-Aufrufstellen übergeben `language`
- Tests: alle Stub-LLMs um `language=None` ergänzt
Co-Authored-By: Claude Sonnet 4.6 <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, language=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 == []
|