NoahsKI / context_manager.py
noah33565's picture
Upload 221 files
8d3de43 verified
"""
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