Trixy7/trixy_core/nlp/provider.py

# -*- coding: utf-8 -*-
"""
NLP Provider Interfaces.

Definiert die Basisklassen und Datenstrukturen für NLP-Provider.
"""

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from enum import Enum, auto
from typing import Any, TYPE_CHECKING

if TYPE_CHECKING:
    from trixy_core.conversation.session import ConversationSession


class NLPState(Enum):
    """Status eines NLP-Providers."""
    UNINITIALIZED = auto()
    INITIALIZING = auto()
    READY = auto()
    PROCESSING = auto()
    ERROR = auto()
    SHUTDOWN = auto()


@dataclass
class NLPConfig:
    """
    Konfiguration für einen NLP-Provider.

    Attributes:
        backend: Name des Backends (z.B. "llama_cpp", "ollama")
        model_name: Name des Modells
        model_path: Optionaler Pfad zum Modell
        use_gpu: GPU-Beschleunigung verwenden
        num_threads: Anzahl CPU-Threads
        temperature: Sampling-Temperatur (0.0 = deterministisch)
        max_tokens: Maximale Antwortlänge
        context_window: Größe des Kontextfensters
        extra: Zusätzliche backend-spezifische Optionen
    """
    backend: str = "llama_cpp"
    model_name: str = ""
    model_path: str | None = None
    use_gpu: bool = False
    num_threads: int = 4
    temperature: float = 0.1
    max_tokens: int = 256
    context_window: int = 2048
    extra: dict[str, Any] = field(default_factory=dict)


@dataclass
class NLPContext:
    """
    Kontext für eine NLP-Anfrage.

    Enthält alle relevanten Informationen für die Intent-Erkennung.

    Attributes:
        text: Der zu verarbeitende Text (STT-Ergebnis)
        satellite_id: ID des Satellites
        room_id: Raum-ID
        session_id: Konversations-Session-ID
        session: Optionale Konversations-Session für Verlauf
        available_intents: Liste verfügbarer Intents mit Metadaten
        user_context: Zusätzlicher Benutzerkontext
        language: Sprachcode (z.B. "de", "en")
    """
    text: str
    satellite_id: str = ""
    room_id: str = ""
    session_id: str = ""
    session: "ConversationSession | None" = None
    available_intents: list[dict[str, Any]] = field(default_factory=list)
    user_context: dict[str, Any] = field(default_factory=dict)
    language: str = "de"

    def get_conversation_history(self, max_turns: int = 5) -> list[dict[str, str]]:
        """
        Gibt die letzten N Konversations-Turns zurück.

        Args:
            max_turns: Maximale Anzahl Turns

        Returns:
            Liste von {"role": "user"|"assistant", "content": "..."}
        """
        if self.session is None:
            return []

        history = []
        turns = list(self.session.turns)[-max_turns:]

        for turn in turns:
            if turn.turn_type.name.startswith("USER"):
                history.append({"role": "user", "content": turn.content})
            elif turn.turn_type.name.startswith("ASSISTANT"):
                history.append({"role": "assistant", "content": turn.content})

        return history


@dataclass
class NLPResult:
    """
    Ergebnis einer NLP-Verarbeitung.

    Attributes:
        intent: Erkannter Intent-Name
        confidence: Konfidenz der Erkennung (0.0 - 1.0)
        slots: Extrahierte Slot-Werte
        response_text: Generierte Antwort (optional, für LLM-basierte NLP)
        raw_output: Rohausgabe des NLP-Modells
        processing_time: Verarbeitungszeit in Sekunden
        success: Ob die Verarbeitung erfolgreich war
        error: Fehlermeldung falls nicht erfolgreich
    """
    intent: str = ""
    confidence: float = 0.0
    slots: dict[str, Any] = field(default_factory=dict)
    response_text: str = ""
    raw_output: str = ""
    processing_time: float = 0.0
    success: bool = True
    error: str = ""

    @classmethod
    def failure(cls, error: str) -> "NLPResult":
        """Erstellt ein Fehler-Ergebnis."""
        return cls(success=False, error=error)

    def has_response(self) -> bool:
        """Prüft ob eine Antwort generiert wurde."""
        return bool(self.response_text)


class NLPProvider(ABC):
    """
    Abstrakte Basisklasse für NLP-Provider.

    Ein NLP-Provider verarbeitet Text und erkennt Intents mit Slots.
    Optional kann er auch direkt Antworten generieren (LLM-basiert).

    Beispiel:
        class LLMNLPProvider(NLPProvider):
            async def initialize(self, config: NLPConfig) -> bool:
                self._model = load_model(config.model_path)
                return True

            async def process(self, context: NLPContext) -> NLPResult:
                response = await self._model.generate(context.text)
                return NLPResult(
                    intent=response.intent,
                    confidence=response.confidence,
                    slots=response.slots,
                    response_text=response.text
                )
    """

    def __init__(self) -> None:
        """Initialisiert den Provider."""
        self._state = NLPState.UNINITIALIZED
        self._config: NLPConfig | None = None

    @property
    def state(self) -> NLPState:
        """Aktueller Status des Providers."""
        return self._state

    @property
    def is_ready(self) -> bool:
        """Ist der Provider einsatzbereit?"""
        return self._state == NLPState.READY

    @property
    def config(self) -> NLPConfig | None:
        """Aktuelle Konfiguration."""
        return self._config

    @abstractmethod
    async def initialize(self, config: NLPConfig) -> bool:
        """
        Initialisiert den Provider mit der gegebenen Konfiguration.

        Args:
            config: Provider-Konfiguration

        Returns:
            True bei erfolgreicher Initialisierung
        """
        pass

    @abstractmethod
    async def process(self, context: NLPContext) -> NLPResult:
        """
        Verarbeitet Text und erkennt Intents.

        Args:
            context: NLP-Kontext mit Text und Metadaten

        Returns:
            NLP-Ergebnis mit Intent, Slots und optionaler Antwort
        """
        pass

    @abstractmethod
    async def shutdown(self) -> None:
        """Fährt den Provider herunter und gibt Ressourcen frei."""
        pass

    async def health_check(self) -> bool:
        """
        Führt einen Health-Check durch.

        Returns:
            True wenn der Provider funktionsfähig ist
        """
        return self._state == NLPState.READY

    def get_capabilities(self) -> list[str]:
        """
        Gibt die Fähigkeiten des Providers zurück.

        Returns:
            Liste von Fähigkeiten (z.B. ["intent", "response", "entity"])
        """
        return ["intent"]

    def supports_streaming(self) -> bool:
        """
        Unterstützt der Provider Streaming-Antworten?

        Returns:
            True wenn Streaming unterstützt wird
        """
        return False