mirror of
https://github.com/Heretek-AI/AetherTable.git
synced 2026-11-07 22:51:18 -10:00
John SmithandClaude
7c755caf4d
classify_with_llm() prompts the configured custom endpoint (LLM_API/LLM_KEY/LLM_MODEL) for JSON classification into the real IntentType enum, with layered honesty and safety: - SAFETY_INTERVENTION always wins — keyword safety short-circuits before any network call; an LLM safety label overrides keyword disagreement - any failure (kill-switch VTT_LLM_CLASSIFIER=0, no key, malformed or unknown output) falls back to the deterministic keyword classifier with provenance fields; never raises - RULE_OF_COOL has no enum counterpart and is deliberately not mapped complete_json() adds fence-tolerant JSON extraction + non-streaming chat-completions with response_format json_object; degraded-marker code preserved. 12 new tests, HTTP fully mocked. Co-Authored-By: Claude <[email protected]>
256 lines
10 KiB
Python
256 lines
10 KiB
Python
"""LLM-assisted intent classification (backlog item 4.8).
|
|
|
|
Contract under test:
|
|
- `IntentClassificationRouter.classify_with_llm()` prompts the configured
|
|
chat-completions endpoint (LLM_API/LLM_KEY/LLM_MODEL) and maps the reply onto
|
|
the SAME IntentType values the deterministic keyword router emits.
|
|
- SAFETY_TRIGGER always wins: if EITHER the keyword matcher or the LLM labels
|
|
the utterance a safety trigger, the final classification is
|
|
SAFETY_INTERVENTION regardless of confidence.
|
|
- Any LLM failure (malformed JSON, unknown label, HTTP error, kill switch
|
|
VTT_LLM_CLASSIFIER=0, or no API key configured) falls back to the keyword
|
|
classifier silently, tagged `"classifier": "keyword_fallback"` with lowered
|
|
confidence, never raising.
|
|
- No unit test touches the network: the HTTP layer (httpx.AsyncClient) is
|
|
monkeypatched with an in-process fake.
|
|
"""
|
|
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from vtt_orchestrator.routing import llm_client as llm_client_module
|
|
from vtt_orchestrator.routing.llm_client import LLMConfig, LLMStreamingGateway
|
|
from vtt_orchestrator.schemas.models import IntentType
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# HTTP-layer fakes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _chat_completion_payload(content: str) -> dict:
|
|
"""OpenAI-compatible chat-completions response body."""
|
|
return {
|
|
"id": "chatcmpl-test",
|
|
"object": "chat.completion",
|
|
"model": "test-model",
|
|
"choices": [{"index": 0, "message": {"role": "assistant", "content": content}, "finish_reason": "stop"}],
|
|
}
|
|
|
|
|
|
class _FakeResponse:
|
|
def __init__(self, body: dict, status_code: int = 200):
|
|
self._body = body
|
|
self.status_code = status_code
|
|
|
|
def raise_for_status(self) -> None:
|
|
if self.status_code >= 400:
|
|
raise RuntimeError(f"http {self.status_code}")
|
|
|
|
def json(self) -> dict:
|
|
return self._body
|
|
|
|
|
|
class _FakeAsyncClient:
|
|
"""httpx.AsyncClient stand-in serving canned completion bodies."""
|
|
|
|
bodies: list = []
|
|
calls: list = []
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
pass
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, *exc):
|
|
return False
|
|
|
|
async def post(self, url, headers=None, json=None, **kwargs):
|
|
type(self).calls.append({"url": url, "payload": json})
|
|
body = type(self).bodies.pop(0)
|
|
return _FakeResponse(body)
|
|
|
|
|
|
class _NeverCalledAsyncClient(_FakeAsyncClient):
|
|
async def post(self, *args, **kwargs):
|
|
raise AssertionError("network call attempted when it must not be")
|
|
|
|
|
|
def _install_upstream(monkeypatch, contents, client_cls=_FakeAsyncClient):
|
|
"""Point httpx.AsyncClient at a fake serving the given message contents."""
|
|
bodies = [_chat_completion_payload(c) if isinstance(c, str) else c for c in contents]
|
|
patched_cls = type("_PatchedClient", (client_cls,), {"bodies": bodies, "calls": []})
|
|
monkeypatch.setattr(llm_client_module.httpx, "AsyncClient", patched_cls)
|
|
return patched_cls
|
|
|
|
|
|
def _live_config(monkeypatch):
|
|
monkeypatch.setenv("LLM_API", "http://fake-llm.test/v1")
|
|
monkeypatch.setenv("LLM_KEY", "test-key")
|
|
monkeypatch.setenv("LLM_MODEL", "test-model")
|
|
# Ensure no ambient provider key leaks in from the developer environment.
|
|
for var in ("OPENAI_API_KEY", "ANTHROPIC_API_KEY", "GEMINI_API_KEY", "OLLAMA_BASE_URL"):
|
|
monkeypatch.delenv(var, raising=False)
|
|
|
|
|
|
def _make_router(monkeypatch, live: bool = True) -> "tuple":
|
|
if live:
|
|
_live_config(monkeypatch)
|
|
else:
|
|
for var in ("LLM_API", "LLM_KEY", "LLM_MODEL", "OPENAI_API_KEY", "ANTHROPIC_API_KEY", "GEMINI_API_KEY"):
|
|
monkeypatch.delenv(var, raising=False)
|
|
monkeypatch.delenv("OLLAMA_BASE_URL", raising=False)
|
|
gateway = LLMStreamingGateway(LLMConfig())
|
|
from vtt_orchestrator.routing.intent_router import IntentClassificationRouter
|
|
|
|
router = IntentClassificationRouter(llm_gateway=gateway)
|
|
return router, gateway
|
|
|
|
|
|
NEUTRAL_IC = "I lean on the tavern rail and describe the storm outside in character"
|
|
|
|
|
|
class TestLLMPath:
|
|
def test_valid_json_returns_llm_classified_intent(self, monkeypatch):
|
|
router, _ = _make_router(monkeypatch)
|
|
# Keywords alone would say IN_CHARACTER_DIALOGUE here; the LLM disagrees.
|
|
_install_upstream(monkeypatch, ['{"intent_type": "MECHANICAL_INVOCATION", "confidence": 0.91}'])
|
|
|
|
decision = asyncio_run(router.classify_with_llm(NEUTRAL_IC))
|
|
|
|
assert decision["classifier"] == "llm"
|
|
assert decision["classification"].intent_type is IntentType.MECHANICAL_INVOCATION
|
|
assert decision["confidence"] == pytest.approx(0.91)
|
|
assert decision["fallback_reason"] is None
|
|
|
|
def test_json_wrapped_in_code_fence_is_parsed(self, monkeypatch):
|
|
router, _ = _make_router(monkeypatch)
|
|
fenced = 'Sure! ```json\n{"intent_type": "OUT_OF_CHARACTER", "confidence": 0.9}\n```'
|
|
_install_upstream(monkeypatch, [fenced])
|
|
|
|
decision = asyncio_run(router.classify_with_llm("brb pizza"))
|
|
|
|
assert decision["classifier"] == "llm"
|
|
assert decision["classification"].intent_type is IntentType.OUT_OF_CHARACTER
|
|
|
|
|
|
class TestFallbacks:
|
|
def test_malformed_json_falls_back_to_keywords(self, monkeypatch):
|
|
router, _ = _make_router(monkeypatch)
|
|
_install_upstream(monkeypatch, ['this is not json {{{'])
|
|
|
|
decision = asyncio_run(router.classify_with_llm("brb pizza"))
|
|
keyword = router.classify_utterance("brb pizza")
|
|
|
|
assert decision["classifier"] == "keyword_fallback"
|
|
assert decision["classification"].intent_type is keyword.intent_type
|
|
assert decision["fallback_reason"]
|
|
# Degraded confidence must be strictly LOWER than the keyword result.
|
|
assert decision["confidence"] < keyword.confidence
|
|
|
|
def test_http_error_falls_back_to_keywords(self, monkeypatch):
|
|
router, _ = _make_router(monkeypatch)
|
|
|
|
class _FailingClient(_FakeAsyncClient):
|
|
async def post(self, *args, **kwargs):
|
|
raise RuntimeError("connection reset")
|
|
|
|
monkeypatch.setattr(
|
|
llm_client_module.httpx, "AsyncClient",
|
|
type("_Fail", (_FailingClient,), {"bodies": [], "calls": []}),
|
|
)
|
|
|
|
decision = asyncio_run(router.classify_with_llm("brb pizza"))
|
|
|
|
assert decision["classifier"] == "keyword_fallback"
|
|
assert decision["fallback_reason"]
|
|
|
|
def test_unknown_intent_label_falls_back_to_keywords(self, monkeypatch):
|
|
router, _ = _make_router(monkeypatch)
|
|
_install_upstream(monkeypatch, ['{"intent_type": "RULE_OF_COOL", "confidence": 0.99}'])
|
|
|
|
decision = asyncio_run(router.classify_with_llm(NEUTRAL_IC))
|
|
|
|
assert decision["classifier"] == "keyword_fallback"
|
|
assert decision["classification"].intent_type is router.classify_utterance(NEUTRAL_IC).intent_type
|
|
|
|
def test_no_key_environment_falls_back_silently(self, monkeypatch):
|
|
router, gateway = _make_router(monkeypatch, live=False)
|
|
assert gateway.config.is_mock is True
|
|
|
|
decision = asyncio_run(router.classify_with_llm("brb pizza"))
|
|
|
|
assert decision["classifier"] == "keyword_fallback"
|
|
assert decision["fallback_reason"]
|
|
assert decision["classification"].intent_type is IntentType.OUT_OF_CHARACTER
|
|
|
|
def test_kill_switch_forces_keyword_only_without_network(self, monkeypatch):
|
|
router, _ = _make_router(monkeypatch, live=False)
|
|
monkeypatch.setenv("VTT_LLM_CLASSIFIER", "0")
|
|
monkeypatch.setattr(llm_client_module.httpx, "AsyncClient", _NeverCalledAsyncClient)
|
|
|
|
decision = asyncio_run(router.classify_with_llm("brb pizza"))
|
|
|
|
assert decision["classifier"] == "keyword_fallback"
|
|
assert decision["fallback_reason"]
|
|
assert decision["classification"].intent_type is IntentType.OUT_OF_CHARACTER
|
|
|
|
def test_kill_switch_accepts_other_values_as_enabled(self, monkeypatch):
|
|
router, _ = _make_router(monkeypatch, live=False)
|
|
monkeypatch.setenv("VTT_LLM_CLASSIFIER", "true")
|
|
# No key configured anyway -> still silent fallback, never a crash.
|
|
decision = asyncio_run(router.classify_with_llm("brb pizza"))
|
|
assert decision["classification"].intent_type is IntentType.OUT_OF_CHARACTER
|
|
|
|
|
|
class TestSafetyPrecedence:
|
|
def test_keyword_safety_beats_llm_disagreement(self, monkeypatch):
|
|
router, _ = _make_router(monkeypatch)
|
|
# LLM confidently claims plain dialogue on an X-card utterance.
|
|
_install_upstream(monkeypatch, ['{"intent_type": "IN_CHARACTER_DIALOGUE", "confidence": 0.99}'])
|
|
|
|
decision = asyncio_run(router.classify_with_llm("we need an x-card please"))
|
|
|
|
assert decision["classification"].intent_type is IntentType.SAFETY_INTERVENTION
|
|
# Final label provenance is the deterministic path (safety precedence).
|
|
assert decision["classifier"] == "keyword_fallback"
|
|
assert "safety_precedence" in decision["fallback_reason"]
|
|
|
|
def test_llm_safety_beats_keyword_disagreement(self, monkeypatch):
|
|
router, _ = _make_router(monkeypatch)
|
|
# No safety keyword fires; the LLM flags it as a safety trigger anyway.
|
|
_install_upstream(monkeypatch, ['{"intent_type": "SAFETY_INTERVENTION", "confidence": 0.97}'])
|
|
|
|
decision = asyncio_run(router.classify_with_llm(NEUTRAL_IC))
|
|
|
|
assert decision["classification"].intent_type is IntentType.SAFETY_INTERVENTION
|
|
assert decision["classifier"] == "llm"
|
|
|
|
def test_llm_safety_beats_keyword_mechanical_match(self, monkeypatch):
|
|
router, _ = _make_router(monkeypatch)
|
|
_install_upstream(monkeypatch, ['{"intent_type": "SAFETY_INTERVENTION", "confidence": 0.8}'])
|
|
|
|
decision = asyncio_run(router.classify_with_llm("I attack the training dummy"))
|
|
|
|
assert decision["classification"].intent_type is IntentType.SAFETY_INTERVENTION
|
|
|
|
def test_keyword_safety_short_circuits_without_llm_call(self, monkeypatch):
|
|
router, _ = _make_router(monkeypatch)
|
|
patched = _install_upstream(monkeypatch, [])
|
|
|
|
decision = asyncio_run(router.classify_with_llm("lines and veils"))
|
|
|
|
assert decision["classification"].intent_type is IntentType.SAFETY_INTERVENTION
|
|
assert patched.calls == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def asyncio_run(coro):
|
|
import asyncio
|
|
|
|
return asyncio.run(coro)
|