# -*- coding: utf-8 -*- """ NLP Classifier Plugin — Intent-Erkennung via trainiertem ONNX-Modell. 3-stufige Pipeline: 1. KeywordMatcher (< 5ms) — Pattern-basiert, hoechste Praezision 2. ONNX-Classifier (< 100ms) — Trainiertes Sentence-Embedding-Modell 3. LLM-Fallback — Nur wenn nlp_llm Plugin aktiv und Confidence zu niedrig Hoert auf speech_recognized mit HIGHEST Prioritaet. Bei ausreichender Confidence wird das Event gecancelt, damit das nlp_llm Plugin nicht zusaetzlich aufgerufen wird. Event-Flow: speech_recognized (HIGHEST Prioritaet) ↓ [Classifier Plugin] KeywordMatcher → ONNX → Fallback ↓ intent_received (mit confidence, slots, tones) ↓ [Intent-Handler Plugin] verarbeitet Intent """ from __future__ import annotations import time from pathlib import Path from typing import Any from trixy_core.events.decorators import TrixyEvent from trixy_core.events.enums import EventPriority from trixy_core.events.event_data.basic import IntentReceived, SpeechRecognized from trixy_core.nlp.intent_classifier import IntentClassifier from trixy_core.nlp.tone import get_analyzer, analyze_tone, emit_tone_events from trixy_core.plugins.trixy_plugin import TrixyPlugin from trixy_core.utils.debug import pdebug, perror, pinfo, pwarn class NLPClassifierPlugin(TrixyPlugin): """ NLP-Backend via trainiertem Intent-Classifier. Nutzt ein ONNX-Modell fuer schnelle Intent-Erkennung (<100ms auf Pi). Faellt auf das LLM-Plugin zurueck wenn Confidence zu niedrig. """ NAME = "nlp_classifier" VERSION = "1.0.0" DESCRIPTION = "Intent-Erkennung via trainiertem ONNX-Classifier" AUTHOR = "Trixy" def __init__(self, application, plugin_path, config=None) -> None: super().__init__(application, plugin_path, config) self._classifier: IntentClassifier | None = None self._keyword_matcher = None self._entity_registry = None async def on_load(self) -> None: """Plugin laden: Classifier, KeywordMatcher, ToneAnalyzer initialisieren.""" pinfo(f"[Classifier] Lade Plugin v{self.VERSION}...") # 1. Intent-Classifier laden model_dir = self.get_config_value("model_dir", "models/intent") self._classifier = IntentClassifier() ok = await self._classifier.load(model_dir) if ok: pinfo(f"[Classifier] ONNX-Modell geladen: {self._classifier.intent_count} Intents") else: pwarn("[Classifier] ONNX-Modell nicht verfuegbar — nur KeywordMatcher aktiv") # 2. KeywordMatcher initialisieren (Pattern-basiert, schnellster Pfad) try: from trixy_core.nlp.keyword_matcher import KeywordIntentMatcher from trixy_core.nlp.entities import EntityRegistry from trixy_core.nlp.intent_registry import IntentRegistry matcher_config = {"enabled": True} self._keyword_matcher = KeywordIntentMatcher(matcher_config) # Entity-Registry laden project_root = Path(__file__).resolve().parent.parent.parent self._entity_registry = EntityRegistry() self._entity_registry.load(project_root / "entities", "de") self._entity_registry.load_plugin_entities(project_root / "plugins", "de") self._keyword_matcher.set_entity_registry(self._entity_registry) # Intents aus der Registry laden und kompilieren registry = IntentRegistry.get_instance() all_intents = registry.get_all_as_dict() # System-Intents hinzufuegen try: from trixy_core.nlp.system_intents import get_system_intents_for_context system_intents = get_system_intents_for_context() all_intents = system_intents + all_intents except ImportError: pass self._keyword_matcher.compile_intents(all_intents) pinfo(f"[Classifier] KeywordMatcher: {len(all_intents)} Intents kompiliert") except Exception as e: pwarn(f"[Classifier] KeywordMatcher nicht verfuegbar: {e}") # 3. ToneAnalyzer laden tone_file = self.get_config_value("tone_keywords_file", "config/tone_keywords.json") if Path(tone_file).is_file(): get_analyzer().load_keywords(tone_file) pdebug(f"[Classifier] ToneAnalyzer geladen") pinfo(f"[Classifier] Plugin geladen — Pipeline: KeywordMatcher → ONNX → LLM-Fallback") async def on_unload(self) -> None: """Plugin entladen.""" self._classifier = None self._keyword_matcher = None self._entity_registry = None pdebug("[Classifier] Plugin entladen") # === KeywordMatcher aktualisieren bei Plugin-Aenderungen === @TrixyEvent(["plugin_loaded", "plugin_unloaded"]) async def on_plugin_change(self, event_name: str, event_data) -> None: """Aktualisiert den KeywordMatcher wenn Plugins sich aendern.""" plugin_name = getattr(event_data, "plugin_name", "") if plugin_name == self.NAME: return if self._keyword_matcher: try: from trixy_core.nlp.intent_registry import IntentRegistry registry = IntentRegistry.get_instance() all_intents = registry.get_all_as_dict() self._keyword_matcher.compile_intents(all_intents) pdebug(f"[Classifier] KeywordMatcher aktualisiert: {len(all_intents)} Intents") except Exception: pass # === Intent-Erkennung === @TrixyEvent(["speech_recognized"], priority=EventPriority.HIGHEST) async def on_speech_recognized(self, event_name: str, event_data: SpeechRecognized) -> None: """ Erkennt Intent aus Spracheingabe. Prioritaet HIGHEST — laeuft VOR dem nlp_llm Plugin. Bei ausreichender Confidence wird das Event gecancelt. """ text = getattr(event_data, "text", "") if not text or not text.strip(): return satellite_id = getattr(event_data, "satellite_id", "") min_confidence = self.get_config_value("min_confidence", 0.7) log_predictions = self.get_config_value("log_predictions", True) t_start = time.monotonic() # --- Stufe 1: KeywordMatcher (< 5ms) --- intent = "" confidence = 0.0 slots: dict[str, Any] = {} matched_by = "" if self._keyword_matcher: kw_match = self._keyword_matcher.match(text) if kw_match and kw_match.confidence >= min_confidence: intent = kw_match.intent confidence = kw_match.confidence slots = kw_match.slots matched_by = "keyword" if log_predictions: t_kw = (time.monotonic() - t_start) * 1000 pinfo( f"[Classifier] KeywordMatch: '{intent}' " f"(conf={confidence:.2f}, {t_kw:.1f}ms)" ) # --- Stufe 2: ONNX-Classifier (< 100ms) --- if not intent and self._classifier and self._classifier.is_loaded: prediction = await self._classifier.classify(text, min_confidence=min_confidence) if prediction.confidence >= min_confidence: intent = prediction.intent confidence = prediction.confidence matched_by = "classifier" # Slot-Extraktion: KeywordMatcher gegen den erkannten Intent matchen slots = self._extract_slots_for_intent(text, intent) if log_predictions: pinfo( f"[Classifier] ONNX: '{intent}' " f"(conf={confidence:.2f}, {prediction.inference_ms:.1f}ms, " f"slots={slots})" ) # --- Stufe 3: LLM-Fallback --- if not intent: fallback = self.get_config_value("fallback_to_llm", True) if fallback: t_elapsed = (time.monotonic() - t_start) * 1000 if log_predictions: pdebug( f"[Classifier] Keine Erkennung ({t_elapsed:.1f}ms) " f"— Fallback an LLM" ) # Event NICHT canceln → nlp_llm Plugin uebernimmt return # Kein Fallback — unknown emittieren intent = "unknown" confidence = 0.0 matched_by = "none" # --- Tone-Analyse --- tone_results = analyze_tone(text) tones = [r.tag for r in tone_results] # Tone-Events emittieren events = getattr(self._application, "events", None) if events and tone_results: await emit_tone_events(tone_results, text, events) # --- intent_received emittieren --- t_total = (time.monotonic() - t_start) * 1000 # Session-Info sammeln session_info = self._get_session_info(satellite_id) intent_event = IntentReceived( satellite_id=satellite_id, intent=intent, confidence=confidence, slots=slots, original_text=text, session_id=session_info.get("session_id", ""), room_id=session_info.get("room_id", ""), wakeword_type=session_info.get("wakeword_type", "custom"), is_authenticated=self._is_authenticated(satellite_id), language=getattr(event_data, "language", "de") or "de", metadata={ "matched_by": matched_by, "tones": [{"tag": r.tag, "category": r.category} for r in tone_results], "inference_ms": t_total, }, ) pinfo( f"[Classifier] Intent: '{intent}' (conf={confidence:.2f}, " f"by={matched_by}, {t_total:.1f}ms, " f"tones={tones[:3]})" ) await self._application.events.trigger("intent_received", intent_event) # Event canceln — verhindert dass nlp_llm nochmal laeuft if hasattr(event_data, "cancel"): event_data.cancel() # === Hilfsmethoden === def _extract_slots_for_intent(self, text: str, intent_name: str) -> dict[str, Any]: """ Extrahiert Slots fuer einen bestimmten Intent. Nutzt den KeywordMatcher um die Patterns des erkannten Intents gezielt gegen den Text zu matchen. Fallback auf einfache Heuristik. Args: text: Eingabetext intent_name: Vom Classifier erkannter Intent Returns: Dict mit extrahierten Slots """ if not self._keyword_matcher: return {} # Strategie 1: KeywordMatcher Patterns fuer diesen Intent direkt matchen compiled_patterns = self._keyword_matcher._compiled.get(intent_name, []) if compiled_patterns: text_lower = text.lower() # Tokenisieren (wie der KeywordMatcher es macht) import re tokens = re.findall(r"[\w]+", text_lower) best_slots: dict[str, Any] = {} best_conf = 0.0 for pattern in compiled_patterns: result = self._keyword_matcher._match_pattern(tokens, pattern) if result and result.get("confidence", 0) > best_conf: best_conf = result["confidence"] best_slots = result.get("slots", {}) if best_slots: pdebug(f"[Classifier] Slots extrahiert: {best_slots}") return best_slots # Strategie 2: Generischen Match versuchen (beliebiger Intent) kw_result = self._keyword_matcher.match(text) if kw_result and kw_result.slots: pdebug(f"[Classifier] Slots via generischem Match ({kw_result.intent}): {kw_result.slots}") return kw_result.slots # Strategie 3: Heuristik — Stoppwoerter entfernen, Rest als Query # Fuer Intents mit freiem {query} Slot (play_music, search_notes, etc.) query_intents = { "play_music", "search_notes", "search_appointment", } if intent_name in query_intents: query = self._extract_query_heuristic(text) if query: pdebug(f"[Classifier] Query via Heuristik: '{query}'") return {"query": query} return {} @staticmethod def _extract_query_heuristic(text: str) -> str: """ Extrahiert einen Query aus dem Text durch Entfernen von Stoppwoertern. "koenntest du bitte musik von rammstein spielen" → entferne: koenntest, du, bitte, musik, von, spielen → uebrig: "rammstein" """ stop_words = { # Hoeflichkeit "koenntest", "kannst", "wuerdest", "waerst", "bitte", "mal", "du", "mir", "doch", "gerne", "gern", # Musik-spezifisch "musik", "spiel", "spiele", "spielen", "abspielen", "abspiel", "mach", "mache", "hoeren", "hoer", "an", "von", "etwas", "lied", "song", "track", # Allgemein "ich", "moechte", "will", "haette", "lass", "wir", "den", "die", "das", "der", "dem", "ein", "eine", "einen", "und", "oder", "fuer", "mit", "auf", "in", "im", "sei", "so", "lieb", "nett", "es", # Suche "suche", "such", "suchen", "finde", "finden", "nach", } words = text.lower().split() # Stoppwoerter entfernen query_words = [w for w in words if w not in stop_words] # Satzzeichen entfernen query = " ".join(query_words).strip("?!.,") return query if query else "" def _get_session_info(self, satellite_id: str) -> dict[str, Any]: """Holt Session-Info fuer den Satellite.""" try: if hasattr(self._application, "satellites"): satellite = self._application.satellites.get(satellite_id) if satellite: conv = getattr(satellite, "conversation", None) if conv: return { "session_id": getattr(conv, "session_id", ""), "room_id": getattr(satellite, "room", ""), "wakeword_type": getattr(conv, "wakeword_type", "custom"), } return {"room_id": getattr(satellite, "room", "")} except Exception: pass return {} def _is_authenticated(self, satellite_id: str) -> bool: """Prueft ob der Satellite authentifiziert ist (Admin-Befehle).""" # TODO: Admin-Session-Verwaltung (wie in nlp_llm) return False