# -*- coding: utf-8 -*- """ Intent Registry. Globale Registry für alle verfügbaren Intents im System. Wird vom NLP-Provider verwendet, um verfügbare Intents zu ermitteln. """ from dataclasses import dataclass, field from typing import Any, Callable, TYPE_CHECKING import threading if TYPE_CHECKING: from trixy_core.plugins.trixy_plugin import TrixyPlugin @dataclass class IntentSlot: """ Definition eines Intent-Slots. Attributes: name: Slot-Name slot_type: Datentyp des Slots required: Ist der Slot erforderlich? description: Beschreibung des Slots examples: Beispielwerte """ name: str slot_type: type = str required: bool = False description: str = "" examples: list[str] = field(default_factory=list) def to_dict(self) -> dict[str, Any]: """Konvertiert den Slot in ein Dictionary.""" return { "name": self.name, "type": self.slot_type.__name__, "required": self.required, "description": self.description, "examples": self.examples, } @dataclass class IntentDefinition: """ Definition eines Intents. Attributes: name: Eindeutiger Intent-Name handler: Handler-Funktion plugin_name: Name des registrierenden Plugins slots: Slot-Definitionen examples: Beispiel-Phrasen für den Intent description: Beschreibung des Intents priority: Priorität (höher = wird bevorzugt bei Konflikten) enabled: Ist der Intent aktiv? """ name: str handler: Callable[..., Any] plugin_name: str = "" slots: dict[str, IntentSlot] = field(default_factory=dict) examples: list[str] = field(default_factory=list) description: str = "" priority: int = 0 enabled: bool = True patterns: list[str] = field(default_factory=list) def to_dict(self) -> dict[str, Any]: """ Konvertiert den Intent in ein Dictionary für NLP-Provider. Returns: Dictionary mit Intent-Informationen """ result = { "name": self.name, "description": self.description, "slots": {name: slot.to_dict() for name, slot in self.slots.items()}, "examples": self.examples, "plugin": self.plugin_name, } if self.patterns: result["patterns"] = self.patterns return result class IntentRegistry: """ Singleton Registry für alle Intents im System. Plugins registrieren ihre Intents hier, und NLP-Provider fragen die Registry ab, um verfügbare Intents zu ermitteln. Verwendung: # Intent registrieren registry = IntentRegistry.get_instance() registry.register(IntentDefinition( name="turn_on_light", handler=my_handler, plugin_name="smarthome", slots={"device": IntentSlot("device", str, required=True)}, examples=["Licht an", "Lampe einschalten"] )) # Intents für NLP abfragen intents = registry.get_all_as_dict() # Handler aufrufen handler = registry.get_handler("turn_on_light") if handler: await handler(intent_data) """ _instance: "IntentRegistry | None" = None _lock = threading.Lock() def __new__(cls) -> "IntentRegistry": """Singleton-Pattern.""" if cls._instance is None: with cls._lock: if cls._instance is None: cls._instance = super().__new__(cls) cls._instance._initialized = False return cls._instance def __init__(self) -> None: """Initialisiert die Registry.""" if self._initialized: return self._intents: dict[str, IntentDefinition] = {} self._plugin_intents: dict[str, list[str]] = {} # Plugin -> Intent-Namen self._initialized = True @classmethod def get_instance(cls) -> "IntentRegistry": """Gibt die Singleton-Instanz zurück.""" return cls() def register(self, intent: IntentDefinition) -> bool: """ Registriert einen Intent. Args: intent: Intent-Definition Returns: True bei Erfolg, False wenn Intent bereits existiert """ if intent.name in self._intents: existing = self._intents[intent.name] # Bei gleichem Plugin ersetzen if existing.plugin_name == intent.plugin_name: self._intents[intent.name] = intent return True # Bei höherer Priorität ersetzen if intent.priority > existing.priority: self._intents[intent.name] = intent self._update_plugin_intents(existing.plugin_name, intent.plugin_name, intent.name) return True return False self._intents[intent.name] = intent # Plugin-Zuordnung aktualisieren if intent.plugin_name: if intent.plugin_name not in self._plugin_intents: self._plugin_intents[intent.plugin_name] = [] self._plugin_intents[intent.plugin_name].append(intent.name) return True def _update_plugin_intents( self, old_plugin: str, new_plugin: str, intent_name: str ) -> None: """Aktualisiert die Plugin-Intent-Zuordnung.""" if old_plugin and old_plugin in self._plugin_intents: if intent_name in self._plugin_intents[old_plugin]: self._plugin_intents[old_plugin].remove(intent_name) if new_plugin: if new_plugin not in self._plugin_intents: self._plugin_intents[new_plugin] = [] if intent_name not in self._plugin_intents[new_plugin]: self._plugin_intents[new_plugin].append(intent_name) def unregister(self, intent_name: str) -> bool: """ Entfernt einen Intent aus der Registry. Args: intent_name: Name des Intents Returns: True bei Erfolg """ if intent_name not in self._intents: return False intent = self._intents.pop(intent_name) # Plugin-Zuordnung entfernen if intent.plugin_name and intent.plugin_name in self._plugin_intents: if intent_name in self._plugin_intents[intent.plugin_name]: self._plugin_intents[intent.plugin_name].remove(intent_name) return True def unregister_plugin(self, plugin_name: str) -> int: """ Entfernt alle Intents eines Plugins. Args: plugin_name: Plugin-Name Returns: Anzahl entfernter Intents """ if plugin_name not in self._plugin_intents: return 0 intent_names = self._plugin_intents.pop(plugin_name, []).copy() count = 0 for name in intent_names: if name in self._intents: del self._intents[name] count += 1 return count def get(self, intent_name: str) -> IntentDefinition | None: """ Gibt eine Intent-Definition zurück. Args: intent_name: Name des Intents Returns: IntentDefinition oder None """ return self._intents.get(intent_name) def resolve_intent(self, intent_name: str) -> IntentDefinition | None: """ Loest einen Intent-Namen auf (exakt oder Suffix-Match). Sucht zuerst nach exaktem Match. Falls nicht gefunden, wird ein Suffix-Match versucht (z.B. "weather" findet "current_weather"). Nur bei genau einem Treffer wird der Suffix-Match akzeptiert. Args: intent_name: Name des Intents (ggf. verkuerzt vom LLM) Returns: IntentDefinition oder None """ # Exakter Match intent = self._intents.get(intent_name) if intent and intent.enabled: return intent # Suffix-Match: LLM gibt manchmal verkuerzte Namen zurueck candidates = [ i for i in self._intents.values() if i.enabled and i.name.endswith(f"_{intent_name}") ] if len(candidates) == 1: return candidates[0] return None def get_handler(self, intent_name: str) -> Callable[..., Any] | None: """ Gibt den Handler für einen Intent zurück. Nutzt resolve_intent() fuer exakten und Suffix-Match. Args: intent_name: Name des Intents Returns: Handler-Funktion oder None """ intent = self.resolve_intent(intent_name) if intent: return intent.handler return None def get_all(self) -> list[IntentDefinition]: """ Gibt alle registrierten Intents zurück. Returns: Liste aller IntentDefinitions """ return list(self._intents.values()) def get_enabled(self) -> list[IntentDefinition]: """ Gibt alle aktivierten Intents zurück. Returns: Liste aktivierter IntentDefinitions """ return [i for i in self._intents.values() if i.enabled] def get_all_as_dict(self) -> list[dict[str, Any]]: """ Gibt alle Intents als Dictionary-Liste zurück. Wird vom NLP-Provider verwendet, um verfügbare Intents in den System-Prompt aufzunehmen. Returns: Liste von Intent-Dictionaries """ return [i.to_dict() for i in self._intents.values() if i.enabled] def get_plugin_intents(self, plugin_name: str) -> list[IntentDefinition]: """ Gibt alle Intents eines Plugins zurück. Args: plugin_name: Plugin-Name Returns: Liste der Plugin-Intents """ intent_names = self._plugin_intents.get(plugin_name, []) return [self._intents[name] for name in intent_names if name in self._intents] def has_intent(self, intent_name: str) -> bool: """Prüft ob ein Intent registriert ist.""" return intent_name in self._intents def enable(self, intent_name: str) -> bool: """ Aktiviert einen Intent. Args: intent_name: Name des Intents Returns: True bei Erfolg """ if intent_name in self._intents: self._intents[intent_name].enabled = True return True return False def disable(self, intent_name: str) -> bool: """ Deaktiviert einen Intent. Args: intent_name: Name des Intents Returns: True bei Erfolg """ if intent_name in self._intents: self._intents[intent_name].enabled = False return True return False def clear(self) -> None: """Löscht alle registrierten Intents.""" self._intents.clear() self._plugin_intents.clear() def count(self) -> int: """Anzahl registrierter Intents.""" return len(self._intents) def __len__(self) -> int: """Anzahl registrierter Intents.""" return len(self._intents) def __contains__(self, intent_name: str) -> bool: """Prüft ob ein Intent registriert ist.""" return intent_name in self._intents def __iter__(self): """Iterator über alle Intent-Definitionen.""" return iter(self._intents.values())