# -*- coding: utf-8 -*- """ Conversation Context - Kontext-Verwaltung für Sessions. """ from dataclasses import dataclass, field from datetime import datetime from enum import Enum from typing import Any class ContextScope(Enum): """Gültigkeitsbereich von Kontext-Variablen.""" TURN = "turn" # Nur für aktuellen Turn SESSION = "session" # Für gesamte Session USER = "user" # Persistent für User GLOBAL = "global" # Global über alle Sessions @dataclass class ContextVariable: """Eine Kontext-Variable.""" name: str value: Any scope: ContextScope = ContextScope.SESSION created_at: datetime = field(default_factory=datetime.now) updated_at: datetime = field(default_factory=datetime.now) metadata: dict[str, Any] = field(default_factory=dict) def update(self, value: Any) -> None: """Aktualisiert den Wert.""" self.value = value self.updated_at = datetime.now() def to_dict(self) -> dict: """Konvertiert zu Dictionary.""" return { "name": self.name, "value": self.value, "scope": self.scope.value, "created_at": self.created_at.isoformat(), "updated_at": self.updated_at.isoformat(), } class ConversationContext: """ Kontext-Container für Conversation-Sessions. Verwaltet Variablen verschiedener Gültigkeitsbereiche: - TURN: Nur für aktuellen Gesprächsbeitrag - SESSION: Für gesamte Conversation-Session - USER: Persistent für einen User (über Sessions hinweg) - GLOBAL: Global verfügbar """ def __init__(self): """Initialisiert den Kontext.""" self._variables: dict[str, ContextVariable] = {} self._turn_variables: dict[str, ContextVariable] = {} def set( self, name: str, value: Any, scope: ContextScope = ContextScope.SESSION, metadata: dict | None = None, ) -> None: """ Setzt eine Kontext-Variable. Args: name: Variablen-Name value: Wert scope: Gültigkeitsbereich metadata: Zusätzliche Metadaten """ var = ContextVariable( name=name, value=value, scope=scope, metadata=metadata or {}, ) if scope == ContextScope.TURN: self._turn_variables[name] = var else: self._variables[name] = var def get(self, name: str, default: Any = None) -> Any: """ Gibt Kontext-Variable zurück. Args: name: Variablen-Name default: Standard-Wert wenn nicht gefunden Returns: Wert oder Default """ # Erst Turn-Variables prüfen if name in self._turn_variables: return self._turn_variables[name].value if name in self._variables: return self._variables[name].value return default def get_variable(self, name: str) -> ContextVariable | None: """ Gibt Kontext-Variable mit Metadaten zurück. Args: name: Variablen-Name Returns: Variable oder None """ if name in self._turn_variables: return self._turn_variables[name] return self._variables.get(name) def has(self, name: str) -> bool: """ Prüft ob Variable existiert. Args: name: Variablen-Name Returns: True wenn vorhanden """ return name in self._turn_variables or name in self._variables def remove(self, name: str) -> bool: """ Entfernt Variable. Args: name: Variablen-Name Returns: True wenn entfernt """ if name in self._turn_variables: del self._turn_variables[name] return True if name in self._variables: del self._variables[name] return True return False def clear_turn(self) -> None: """Löscht alle Turn-Variablen.""" self._turn_variables.clear() def clear_scope(self, scope: ContextScope) -> None: """ Löscht alle Variablen eines Scopes. Args: scope: Zu löschender Scope """ if scope == ContextScope.TURN: self._turn_variables.clear() else: to_remove = [ name for name, var in self._variables.items() if var.scope == scope ] for name in to_remove: del self._variables[name] def clear(self) -> None: """Löscht alle Variablen.""" self._variables.clear() self._turn_variables.clear() def get_by_scope(self, scope: ContextScope) -> dict[str, Any]: """ Gibt alle Variablen eines Scopes zurück. Args: scope: Scope Returns: Dictionary mit Variablen """ if scope == ContextScope.TURN: return {name: var.value for name, var in self._turn_variables.items()} return { name: var.value for name, var in self._variables.items() if var.scope == scope } def get_all(self) -> dict[str, Any]: """ Gibt alle Variablen zurück. Returns: Dictionary mit allen Variablen """ result = {} # Session/User/Global zuerst for name, var in self._variables.items(): result[name] = var.value # Turn überschreibt for name, var in self._turn_variables.items(): result[name] = var.value return result def merge(self, other: "ConversationContext") -> None: """ Merged anderen Kontext in diesen. Args: other: Anderer Kontext """ for name, var in other._variables.items(): self._variables[name] = var for name, var in other._turn_variables.items(): self._turn_variables[name] = var def to_dict(self) -> dict: """Konvertiert zu Dictionary.""" return { "variables": { name: var.to_dict() for name, var in self._variables.items() }, "turn_variables": { name: var.to_dict() for name, var in self._turn_variables.items() }, } # === Spezielle Kontext-Methoden für Conversations === def set_intent(self, intent: str, confidence: float = 1.0) -> None: """Setzt erkannten Intent.""" self.set("_intent", intent, ContextScope.TURN) self.set("_intent_confidence", confidence, ContextScope.TURN) def get_intent(self) -> str | None: """Gibt aktuellen Intent zurück.""" return self.get("_intent") def set_entities(self, entities: dict[str, Any]) -> None: """Setzt extrahierte Entities.""" self.set("_entities", entities, ContextScope.TURN) def get_entities(self) -> dict[str, Any]: """Gibt Entities zurück.""" return self.get("_entities", {}) def set_slot(self, name: str, value: Any) -> None: """Setzt Slot-Wert (für Dialoge).""" self.set(f"slot_{name}", value, ContextScope.SESSION) def get_slot(self, name: str, default: Any = None) -> Any: """Gibt Slot-Wert zurück.""" return self.get(f"slot_{name}", default) def clear_slots(self) -> None: """Löscht alle Slots.""" to_remove = [ name for name in self._variables if name.startswith("slot_") ] for name in to_remove: del self._variables[name] @property def slots(self) -> dict[str, Any]: """Gibt alle Slots zurück.""" return { name[5:]: var.value # Entferne "slot_" Prefix for name, var in self._variables.items() if name.startswith("slot_") }