""" CONTEXT MANAGER - Chat-Historie und Kontext-Verständnis Speichert die letzten Nachrichten und nutzt diese für bessere Antworten """ import json import os from typing import Dict, List, Optional, Any from datetime import datetime from collections import deque import logging logger = logging.getLogger(__name__) class ContextManager: """Manages chat context and conversation history with ML learning""" def __init__(self, max_messages: int = 50, max_tokens: int = 10000000): """ Initialize context manager Args: max_messages: Maximum messages to keep in memory (default: 50) max_tokens: Maximum tokens per response (default: 10 million) """ self.max_messages = max_messages self.max_tokens = max_tokens self.message_history = deque(maxlen=max_messages) self.context_file = 'noahski_data/context_cache.json' self.token_usage = {'total': 0, 'session': 0, 'max': max_tokens} self.context_patterns = {} self.load_context() def add_message(self, session_id: str, user_msg: str, ai_response: str, intent: str = '', confidence: float = 0.0) -> Dict: """Add message to context history with metadata""" message_data = { 'timestamp': datetime.now().isoformat(), 'session_id': session_id, 'user_message': user_msg, 'ai_response': ai_response, 'intent': intent, 'confidence': confidence, 'tokens': self._estimate_tokens(ai_response), 'turn_number': len(self.message_history) + 1 } self.message_history.append(message_data) self.token_usage['session'] += message_data['tokens'] self.token_usage['total'] += message_data['tokens'] logger.info(f"Context: Added message (tokens: {message_data['tokens']}, total: {self.token_usage['session']})") # Learn patterns from this interaction self._learn_context_patterns(user_msg, ai_response, intent) return message_data def get_context(self, session_id: str, num_messages: int = 5) -> Dict: """ Get recent context for a specific session Returns the last N messages for context understanding """ session_messages = [ msg for msg in self.message_history if msg['session_id'] == session_id ][-num_messages:] context_text = self._build_context_string(session_messages) return { 'session_id': session_id, 'messages': session_messages, 'context_text': context_text, 'message_count': len(session_messages), 'total_tokens': sum(msg['tokens'] for msg in session_messages), 'patterns': self._get_relevant_patterns(session_messages) } def _build_context_string(self, messages: List[Dict]) -> str: """Build context summary from messages""" if not messages: return "No recent context available." context_parts = [] for msg in messages: context_parts.append(f"User: {msg['user_message'][:100]}") context_parts.append(f"AI: {msg['ai_response'][:100]}") if msg['intent']: context_parts.append(f"[Intent: {msg['intent']}]") context_parts.append("") return "\n".join(context_parts) def _estimate_tokens(self, text: str) -> int: """Estimate token count (rough estimate: 1 token ~ 4 characters)""" return max(1, len(text) // 4) def _learn_context_patterns(self, user_msg: str, response: str, intent: str): """Learn patterns from conversation""" # Extract key topics topics = self._extract_topics(user_msg) for topic in topics: if topic not in self.context_patterns: self.context_patterns[topic] = { 'count': 0, 'examples': [], 'response_patterns': [], 'confidence': 0.0 } self.context_patterns[topic]['count'] += 1 self.context_patterns[topic]['examples'].append(user_msg[:80]) self.context_patterns[topic]['response_patterns'].append(response[:80]) self.context_patterns[topic]['confidence'] = min( 1.0, self.context_patterns[topic]['count'] / 10.0 ) def _extract_topics(self, text: str) -> List[str]: """Extract topics/keywords from text""" # Simple keyword extraction keywords = { 'python': ['python', 'py', 'django', 'flask', 'code'], 'javascript': ['javascript', 'js', 'node', 'react'], 'math': ['calculate', 'math', 'algebra', 'equation', 'solve'], 'image': ['image', 'picture', 'generate', 'draw'], 'knowledge': ['what is', 'how', 'explain', 'tell me'], 'translation': ['translate', 'language', 'German', 'English'], } topics = [] text_lower = text.lower() for topic, keywords_list in keywords.items(): if any(kw in text_lower for kw in keywords_list): topics.append(topic) return topics if topics else ['general'] def _get_relevant_patterns(self, messages: List[Dict]) -> Dict: """Get relevant learned patterns for current context""" topics = set() for msg in messages: topics.update(self._extract_topics(msg['user_message'])) relevant = {} for topic in topics: if topic in self.context_patterns: relevant[topic] = { 'count': self.context_patterns[topic]['count'], 'confidence': self.context_patterns[topic]['confidence'], 'examples': self.context_patterns[topic]['examples'][:3] } return relevant def get_context_aware_prompt(self, message: str, session_id: str) -> str: """ Build context-aware prompt with recent conversation history """ context = self.get_context(session_id, num_messages=5) prompt = f"""Based on the recent conversation context: {context['context_text']} Current question: {message} Learned patterns from previous interactions: """ for topic, pattern in context['patterns'].items(): prompt += f"\n- {topic}: {pattern['count']} previous interactions (confidence: {pattern['confidence']:.2f})" return prompt def get_token_status(self) -> Dict: """Get current token usage status""" return { 'session_tokens': self.token_usage['session'], 'total_tokens': self.token_usage['total'], 'max_tokens': self.max_tokens, 'percentage': (self.token_usage['total'] / self.max_tokens) * 100, 'remaining': self.max_tokens - self.token_usage['total'], 'can_continue': self.token_usage['total'] < self.max_tokens } def should_optimize_response(self) -> bool: """Check if we should optimize responses (approaching token limit)""" usage = self.get_token_status() return usage['percentage'] > 80 # Optimize when > 80% token usage def save_context(self): """Save context to file for persistence""" try: os.makedirs('noahski_data', exist_ok=True) data = { 'messages': list(self.message_history), 'patterns': self.context_patterns, 'token_usage': self.token_usage, 'timestamp': datetime.now().isoformat() } with open(self.context_file, 'w', encoding='utf-8') as f: json.dump(data, f, indent=2, ensure_ascii=False) logger.info(f"Context saved: {len(self.message_history)} messages") except Exception as e: logger.error(f"Error saving context: {e}") def load_context(self): """Load context from file""" try: if os.path.exists(self.context_file): with open(self.context_file, 'r', encoding='utf-8') as f: data = json.load(f) self.message_history = deque(data.get('messages', []), maxlen=self.max_messages) self.context_patterns = data.get('patterns', {}) self.token_usage = data.get('token_usage', {'total': 0, 'session': 0, 'max': self.max_tokens}) logger.info(f"Context loaded: {len(self.message_history)} messages") except Exception as e: logger.error(f"Error loading context: {e}") def reset_session(self): """Reset session-only data while keeping long-term patterns""" self.token_usage['session'] = 0 logger.info("Session reset - long-term patterns preserved") def get_summary(self) -> Dict: """Get summary of context and learning""" total_interactions = len(self.message_history) total_patterns = len(self.context_patterns) top_topics = sorted( self.context_patterns.items(), key=lambda x: x[1]['count'], reverse=True )[:5] return { 'total_interactions': total_interactions, 'total_patterns_learned': total_patterns, 'top_topics': [(t, p['count']) for t, p in top_topics], 'token_status': self.get_token_status(), 'messages_stored': len(self.message_history), 'average_confidence': sum(p['confidence'] for p in self.context_patterns.values()) / max(1, total_patterns) } # Global instance _context_manager = None def get_context_manager(max_messages: int = 50, max_tokens: int = 10000000) -> ContextManager: """Get or create global context manager""" global _context_manager if _context_manager is None: _context_manager = ContextManager(max_messages, max_tokens) return _context_manager