| """
|
| 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']})")
|
|
|
|
|
| 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"""
|
|
|
|
|
| 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"""
|
|
|
| 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
|
|
|
| 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)
|
| }
|
|
|
|
|
|
|
| _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
|
|
|