""" modules/memory.py - Gesprächsgedächtnis-Modul für den Dr. Franz Psychochatbot Dieses Modul verwaltet die Konversationshistorie und extrahiert wichtige Informationen: - Speichern und Abrufen der Gesprächshistorie - Extraktion wichtiger Informationen - Bereitstellung von Kontext für neue Antworten """ from typing import List, Dict, Any, Optional import re import random class Memory: """Klasse zur Verwaltung des Gesprächsgedächtnisses""" def __init__(self, max_history_length: int = 10): """Initialisiert das Gedächtnis mit maximaler Historienlänge""" self.max_history_length = max_history_length self.conversation_history = [] self.extracted_info = { "mentioned_people": set(), "mentioned_events": set(), "recurring_themes": {}, "emotional_patterns": [] } def add_exchange(self, user_input: str, bot_response: str, analysis: Optional[Dict[str, Any]] = None) -> None: """ Fügt einen Gesprächsaustausch zum Gedächtnis hinzu Args: user_input: Die Eingabe des Nutzers bot_response: Die Antwort des Chatbots analysis: Optional, Analyseergebnisse des Analyzers """ exchange = { "user_input": user_input, "bot_response": bot_response, "analysis": analysis or {} } self.conversation_history.append(exchange) # Begrenzung der Historienlänge if len(self.conversation_history) > self.max_history_length: self.conversation_history = self.conversation_history[-self.max_history_length:] # Extraktion wichtiger Informationen self._extract_information(exchange) def get_history(self, limit: Optional[int] = None) -> List[Dict[str, Any]]: """ Gibt die Konversationshistorie zurück Args: limit: Optional, maximale Anzahl der zurückzugebenden Einträge Returns: Liste der Gesprächsaustausche """ if limit is None or limit >= len(self.conversation_history): return self.conversation_history else: return self.conversation_history[-limit:] def format_for_prompt(self, limit: Optional[int] = None) -> str: """ Formatiert die Historie für den Prompt Args: limit: Optional, maximale Anzahl der zu formatierenden Einträge Returns: Formatierte Historie als String """ history = self.get_history(limit) formatted = "" for exchange in history: formatted += f"User: {exchange['user_input']}\n" formatted += f"Dr. Franz: {exchange['bot_response']}\n" return formatted def get_relevant_context(self) -> str: """ Generiert erweiterten psychoanalytischen Kontext für die nächste Antwort Returns: Detaillierte Kontextinformationen als String """ context_parts = [] # Wiederkehrende Themen if self.extracted_info["recurring_themes"]: top_themes = sorted( self.extracted_info["recurring_themes"].items(), key=lambda x: x[1], reverse=True )[:2] context_parts.append(f"Wiederkehrende Themen: {', '.join(theme for theme, _ in top_themes)}") # Abwehrmechanismen if self.extracted_info["defense_mechanisms"]: active_defenses = [k for k, v in self.extracted_info["defense_mechanisms"].items() if v > 0] if active_defenses: context_parts.append(f"Aktive Abwehrmechanismen: {', '.join(active_defenses)}") context_parts.append("Der Patient zeigt überwiegend negative Emotionen.") elif recent_emotions.count("positive") >= 2: context_parts.append("Der Patient zeigt ungewöhnlich positive Emotionen, was auf Verdrängung hindeuten könnte.") # Erwähnte Personen einbeziehen if self.extracted_info["mentioned_people"]: people = list(self.extracted_info["mentioned_people"]) if people: person = random.choice(people) context_parts.append(f"Der Patient hat {person} erwähnt. Beziehe dich darauf, wenn passend.") # Zufällig ein früheres Thema aufgreifen if len(self.conversation_history) > 2: old_exchange = random.choice(self.conversation_history[:-2]) if "analysis" in old_exchange and "themes" in old_exchange["analysis"]: old_theme = random.choice(old_exchange["analysis"]["themes"]) context_parts.append(f"Greife bei Gelegenheit das frühere Thema '{old_theme}' wieder auf.") return " ".join(context_parts) def _analyze_defense_mechanisms(self, text: str) -> None: """Analysiert aktive Abwehrmechanismen im Text""" for mechanism, pattern in self.analysis_patterns["defense"].items(): if re.search(pattern, text): self.extracted_info["defense_mechanisms"][mechanism] = self.extracted_info["defense_mechanisms"].get(mechanism, 0) + 1 def _analyze_transference_patterns(self, text: str) -> None: """Analysiert Transfersituationen im Text""" for category, pattern in self.analysis_patterns["transference"].items(): if re.search(pattern, text): self.extracted_info["transference_patterns"].append(category) def _analyze_symbolic_patterns(self, text: str) -> None: """Analysiert symbolische Muster im Text""" for symbol, pattern in self.analysis_patterns["symbolic"].items(): if re.search(pattern, text): self.extracted_info["symbolic_patterns"][symbol] = self.extracted_info["symbolic_patterns"].get(symbol, 0) + 1 def _update_association_network(self, text: str) -> None: """Aktualisiert das Assoziationsnetzwerk""" words = text.split() for i, word in enumerate(words): if word not in self.extracted_info["association_network"]: self.extracted_info["association_network"][word] = {} # Erstelle Assoziationen zu benachbarten Wörtern if i > 0: prev_word = words[i-1] self.extracted_info["association_network"][word][prev_word] = self.extracted_info["association_network"][word].get(prev_word, 0) + 1 if i < len(words) - 1: next_word = words[i+1] self.extracted_info["association_network"][word][next_word] = self.extracted_info["association_network"][word].get(next_word, 0) + 1 def _update_escalation_levels(self, analysis: Dict[str, Any]) -> None: """Aktualisiert die Escalation Levels basierend auf der Analyse""" intensity = analysis.get("intensity", 0) # Emotionale Escalation if analysis.get("emotion") == "negative": self.extracted_info["escalation_levels"]["emotional"] = min(5, self.extracted_info["escalation_levels"]["emotional"] + intensity) else: self.extracted_info["escalation_levels"]["emotional"] = max(0, self.extracted_info["escalation_levels"]["emotional"] - 1) # Kognitive Escalation if "defense_mechanisms" in analysis: self.extracted_info["escalation_levels"]["cognitive"] = min(5, self.extracted_info["escalation_levels"]["cognitive"] + len(analysis["defense_mechanisms"])) else: self.extracted_info["escalation_levels"]["cognitive"] = max(0, self.extracted_info["escalation_levels"]["cognitive"] - 1) # Verhaltensmäßige Escalation if "behavioral_patterns" in analysis: self.extracted_info["escalation_levels"]["behavioral"] = min(5, self.extracted_info["escalation_levels"]["behavioral"] + len(analysis["behavioral_patterns"])) else: self.extracted_info["escalation_levels"]["behavioral"] = max(0, self.extracted_info["escalation_levels"]["behavioral"] - 1) def _extract_information(self, exchange: Dict[str, Any]) -> None: """Extrahiert wichtige Informationen aus einem Gesprächsaustausch""" user_input = exchange["user_input"].lower() analysis = exchange["analysis"] # Multi-Level Psychoanalyse self._analyze_defense_mechanisms(user_input) self._analyze_transference_patterns(user_input) self._analyze_symbolic_patterns(user_input) self._update_association_network(user_input) self._update_escalation_levels(analysis) # Erwähnte Personen und Ereignisse for word in user_input.split(): if word not in ["ich", "du", "wir", "sie", "er", "es"]: self.extracted_info["mentioned_people"].add(word) # Wiederkehrende Themen for theme in analysis.get("themes", []): self.extracted_info["recurring_themes"][theme] = self.extracted_info["recurring_themes"].get(theme, 0) + 1 # Emotionale Muster emotion = analysis.get("emotion", "neutral") self.extracted_info["emotional_patterns"].append({ "emotion": emotion, "timestamp": time.time(), "intensity": analysis.get("intensity", 0) }) # Aktualisiere den letzten Austausch self.extracted_info["last_exchange"] = { "time": time.time(), "intensity": analysis.get("intensity", 0), "patterns": analysis.get("patterns", []) }