adaptive_self_learning_engine.py

"""

╔══════════════════════════════════════════════════════════════════════════════╗

║                                                                              ║

║              ADAPTIVE SELF-LEARNING ENGINE für NoahsKI                       ║

║         Learns from every user interaction to improve responses              ║

║                                                                              ║

║  Features:                                                                   ║

║  ✓ Learn from user feedback (ratings, corrections)                          ║

║  ✓ Detect what works and what doesn't                                       ║

║  ✓ Adapt response strategies based on success                               ║

║  ✓ Active learning (ask for feedback on quality)                            ║

║  ✓ Pattern recognition from conversation history                            ║

║  ✓ Self-improvement without external training                               ║

║  ✓ Topic-specific expertise development                                     ║

║  ✓ Personalization based on user preferences                                ║

║                                                                              ║

╚══════════════════════════════════════════════════════════════════════════════╝

"""

import os
import json
import time
import logging
import hashlib
from typing import Dict, List, Tuple, Optional, Any
from dataclasses import dataclass, field, asdict
from pathlib import Path
from datetime import datetime, timedelta
from collections import defaultdict, deque
import numpy as np

logger = logging.getLogger(__name__)


# ═══════════════════════════════════════════════════════════════════════════════
# DATA CLASSES
# ═══════════════════════════════════════════════════════════════════════════════

@dataclass
class ResponseFeedback:
    """Feedback on an AI response"""
    response_id: str
    user_id: str
    query: str
    response: str
    rating: float  # 0-1
    feedback_text: Optional[str] = None
    was_helpful: bool = False
    needed_improvement: bool = False
    timestamp: float = field(default_factory=time.time)
    corrections: Optional[str] = None
    clarifications: Optional[str] = None
    
    def to_dict(self) -> Dict:
        return asdict(self)


@dataclass
class TopicExpertise:
    """Expertise level on a specific topic"""
    topic: str
    total_interactions: int = 0
    successful_responses: int = 0
    failed_responses: int = 0
    average_rating: float = 0.5
    sources_used: List[str] = field(default_factory=list)
    common_misunderstandings: List[str] = field(default_factory=list)
    last_updated: float = field(default_factory=time.time)
    confidence: float = 0.3
    
    @property
    def success_rate(self) -> float:
        if self.total_interactions == 0:
            return 0.5
        return self.successful_responses / self.total_interactions
    
    @property
    def expertise_level(self) -> str:
        if self.total_interactions < 5:
            return "beginner"
        elif self.success_rate > 0.8:
            return "expert"
        elif self.success_rate > 0.6:
            return "advanced"
        elif self.success_rate > 0.4:
            return "intermediate"
        else:
            return "learning"
    
    def to_dict(self) -> Dict:
        return {
            'topic': self.topic,
            'total_interactions': self.total_interactions,
            'successful_responses': self.successful_responses,
            'failed_responses': self.failed_responses,
            'average_rating': self.average_rating,
            'sources_used': self.sources_used,
            'common_misunderstandings': self.common_misunderstandings,
            'last_updated': self.last_updated,
            'confidence': self.confidence,
            'success_rate': self.success_rate,
            'expertise_level': self.expertise_level
        }


@dataclass
class UserPreference:
    """Learned user preferences"""
    user_id: str
    preferred_style: str = "balanced"  # technical, casual, formal, detailed, concise
    preferred_language: str = "en"
    preferred_response_length: str = "medium"  # short, medium, long
    prefers_examples: bool = True
    prefers_sources: bool = True
    prefers_structures: bool = True  # bullet points, numbered lists, etc
    interaction_count: int = 0
    satisfaction_rate: float = 0.5
    last_learned: float = field(default_factory=time.time)
    
    def to_dict(self) -> Dict:
        return asdict(self)


# ═══════════════════════════════════════════════════════════════════════════════
# ADAPTIVE SELF-LEARNING ENGINE
# ═══════════════════════════════════════════════════════════════════════════════

class AdaptiveSelfLearningEngine:
    """Learns from every interaction to improve responses"""
    
    def __init__(self):
        self.data_dir = Path("noahski_data/self_learning")
        self.data_dir.mkdir(parents=True, exist_ok=True)
        
        self.feedback_file = self.data_dir / "feedback_log.json"
        self.expertise_file = self.data_dir / "topic_expertise.json"
        self.preferences_file = self.data_dir / "user_preferences.json"
        self.patterns_file = self.data_dir / "learned_patterns.json"
        
        # In-memory storage
        self.feedback_history: List[ResponseFeedback] = []
        self.topic_expertise: Dict[str, TopicExpertise] = {}
        self.user_preferences: Dict[str, UserPreference] = {}
        self.learned_patterns: Dict[str, List[Dict]] = defaultdict(list)
        
        # Load data from disk
        self.load_all_data()
        
        # Performance metrics
        self.performance_metrics = {
            'total_responses': 0,
            'positive_feedback': 0,
            'negative_feedback': 0,
            'neutral_feedback': 0,
            'average_satisfaction': 0.5,
            'learning_speed': 0.0  # How fast we're improving
        }
    
    # ═══════════════════════════════════════════════════════════════════════════
    # FEEDBACK COLLECTION & LEARNING
    # ═══════════════════════════════════════════════════════════════════════════
    
    def record_feedback(self, user_id: str, response_id: str, query: str, 

                       response: str, rating: float, feedback_text: str = None,

                       corrections: str = None) -> bool:
        """Record feedback on a response"""
        
        feedback = ResponseFeedback(
            response_id=response_id,
            user_id=user_id,
            query=query,
            response=response,
            rating=float(np.clip(rating, 0, 1)),
            feedback_text=feedback_text,
            was_helpful=rating >= 0.7,
            needed_improvement=rating < 0.5,
            corrections=corrections
        )
        
        self.feedback_history.append(feedback)
        
        # Extract and learn from feedback
        self._learn_from_feedback(feedback, query)
        
        # Update user preference
        self._update_user_preference(user_id, rating)
        
        # Save to disk
        self.save_feedback()
        
        logger.info(f"Feedback recorded: rating={rating:.2f}, user={user_id}")
        return True
    
    def _learn_from_feedback(self, feedback: ResponseFeedback, query: str):
        """Extract learnings from user feedback"""
        
        # Extract topics from query
        topics = self._extract_topics(query)
        
        for topic in topics:
            if topic not in self.topic_expertise:
                self.topic_expertise[topic] = TopicExpertise(topic=topic)
            
            expertise = self.topic_expertise[topic]
            expertise.total_interactions += 1
            
            if feedback.was_helpful:
                expertise.successful_responses += 1
            else:
                expertise.failed_responses += 1
            
            # Update average rating
            expertise.average_rating = (
                expertise.average_rating * 0.7 + 
                feedback.rating * 0.3
            )
            
            # Update confidence based on interactions
            if expertise.total_interactions > 10:
                expertise.confidence = min(0.95, 0.3 + (expertise.total_interactions / 100))
            
            expertise.last_updated = time.time()
            
            # If feedback has corrections, learn from them
            if feedback.corrections:
                if feedback.corrections not in expertise.common_misunderstandings:
                    expertise.common_misunderstandings.append(feedback.corrections)
        
        # Record pattern (query structure + quality)
        pattern = {
            'query': query,
            'rating': feedback.rating,
            'timestamp': feedback.timestamp,
            'was_helpful': feedback.was_helpful,
            'feedback': feedback.feedback_text
        }
        
        query_hash = hashlib.md5(query.lower().encode()).hexdigest()[:8]
        self.learned_patterns[query_hash].append(pattern)
    
    def _extract_topics(self, text: str) -> List[str]:
        """Extract topics from text"""
        topics = []
        
        # Common topic keywords
        topic_keywords = {
            'python': ['python', 'programming', 'code', 'script'],
            'javascript': ['javascript', 'js', 'node', 'web development'],
            'machine_learning': ['machine learning', 'ai', 'neural', 'algorithm'],
            'web_development': ['web', 'html', 'css', 'frontend', 'backend'],
            'database': ['database', 'sql', 'nosql', 'mongodb', 'postgres'],
            'troubleshooting': ['error', 'bug', 'debug', 'fix', 'not working'],
            'explanation': ['explain', 'how', 'why', 'what is', 'understand'],
            'general': ['general', 'information', 'tell me']
        }
        
        text_lower = text.lower()
        
        for topic, keywords in topic_keywords.items():
            if any(keyword in text_lower for keyword in keywords):
                topics.append(topic)
        
        return topics if topics else ['general']
    
    def _update_user_preference(self, user_id: str, rating: float):
        """Update user preferences based on interaction"""
        if user_id not in self.user_preferences:
            self.user_preferences[user_id] = UserPreference(user_id=user_id)
        
        pref = self.user_preferences[user_id]
        pref.interaction_count += 1
        pref.satisfaction_rate = pref.satisfaction_rate * 0.7 + rating * 0.3
        pref.last_learned = time.time()
    
    # ═══════════════════════════════════════════════════════════════════════════
    # OPTIMIZATION STRATEGIES
    # ═══════════════════════════════════════════════════════════════════════════
    
    def get_optimal_response_strategy(self, user_id: str, topic: str) -> Dict[str, Any]:
        """Get optimal strategy for user/topic combination"""
        
        strategy = {
            'style': 'balanced',
            'length': 'medium',
            'use_examples': True,
            'use_sources': True,
            'ask_clarification': False,
            'provide_structure': True,
            'confidence': 0.5
        }
        
        # Apply user preferences
        if user_id in self.user_preferences:
            pref = self.user_preferences[user_id]
            strategy['style'] = pref.preferred_style
            strategy['length'] = pref.preferred_response_length
            strategy['use_examples'] = pref.prefers_examples
            strategy['use_sources'] = pref.prefers_sources
            strategy['provide_structure'] = pref.prefers_structures
        
        # Apply topic expertise
        if topic in self.topic_expertise:
            expertise = self.topic_expertise[topic]
            strategy['confidence'] = expertise.confidence
            
            # If we're not confident, ask for clarification
            if expertise.expertise_level in ['beginner', 'learning']:
                strategy['ask_clarification'] = True
                strategy['use_examples'] = True
            
            # High expertise = more detailed
            if expertise.expertise_level == 'expert':
                strategy['length'] = 'long'
        
        return strategy
    
    def suggest_improvements(self) -> List[Dict[str, Any]]:
        """Suggest areas for improvement based on learning"""
        
        improvements = []
        
        # Find weak topics
        for topic, expertise in self.topic_expertise.items():
            if expertise.success_rate < 0.5 and expertise.total_interactions >= 5:
                improvements.append({
                    'type': 'weak_topic',
                    'topic': topic,
                    'success_rate': expertise.success_rate,
                    'recommendation': f'Focus on improving responses about {topic}'
                })
        
        # Find unsatisfied users
        for user_id, pref in self.user_preferences.items():
            if pref.satisfaction_rate < 0.4 and pref.interaction_count >= 3:
                improvements.append({
                    'type': 'user_satisfaction',
                    'user_id': user_id,
                    'satisfaction': pref.satisfaction_rate,
                    'recommendation': f'Better adapt responses for user {user_id}'
                })
        
        # Check for pattern improvements needed
        if self.learned_patterns:
            patterns_to_improve = []
            for pattern_id, patterns in self.learned_patterns.items():
                if len(patterns) >= 3:
                    avg_rating = np.mean([p['rating'] for p in patterns])
                    if avg_rating < 0.5:
                        patterns_to_improve.append(pattern_id)
            
            if patterns_to_improve:
                improvements.append({
                    'type': 'pattern_improvement',
                    'count': len(patterns_to_improve),
                    'recommendation': f'Review and improve responses to similar queries'
                })
        
        return improvements
    
    def get_learning_progress(self) -> Dict[str, Any]:
        """Get current learning progress"""
        
        total_feedback = len(self.feedback_history)
        
        if total_feedback == 0:
            return {'status': 'No data yet', 'progress': 0}
        
        positive = sum(1 for f in self.feedback_history if f.was_helpful)
        negative = sum(1 for f in self.feedback_history if f.needed_improvement)
        
        # Calculate learning speed (improvement rate)
        if len(self.feedback_history) >= 10:
            recent_feedbacks = self.feedback_history[-10:]
            older_feedbacks = self.feedback_history[-20:-10] if len(self.feedback_history) >= 20 else self.feedback_history[:5]
            
            recent_avg = np.mean([f.rating for f in recent_feedbacks])
            older_avg = np.mean([f.rating for f in older_feedbacks]) if older_feedbacks else 0.5
            
            learning_speed = recent_avg - older_avg
        else:
            learning_speed = 0.0
        
        return {
            'total_interactions': total_feedback,
            'positive_feedback': positive,
            'negative_feedback': negative,
            'positive_rate': positive / total_feedback if total_feedback > 0 else 0,
            'topics_with_expertise': len(self.topic_expertise),
            'users_tracked': len(self.user_preferences),
            'overall_satisfaction': np.mean([f.rating for f in self.feedback_history]),
            'learning_speed': learning_speed,  # Positive = improving
            'status': 'Actively Learning' if learning_speed > 0 else 'Stable' if total_feedback > 20 else 'Starting'
        }
    
    # ═══════════════════════════════════════════════════════════════════════════
    # DATA PERSISTENCE
    # ═══════════════════════════════════════════════════════════════════════════
    
    def save_feedback(self):
        """Save feedback history"""
        try:
            data = [f.to_dict() for f in self.feedback_history[-1000:]]  # Keep last 1000
            with open(self.feedback_file, 'w', encoding='utf-8') as f:
                json.dump(data, f, indent=2, ensure_ascii=False)
        except Exception as e:
            logger.error(f"Error saving feedback: {e}")
    
    def save_expertise(self):
        """Save topic expertise"""
        try:
            data = {topic: exp.to_dict() 
                   for topic, exp in self.topic_expertise.items()}
            with open(self.expertise_file, 'w', encoding='utf-8') as f:
                json.dump(data, f, indent=2, ensure_ascii=False)
        except Exception as e:
            logger.error(f"Error saving expertise: {e}")
    
    def save_preferences(self):
        """Save user preferences"""
        try:
            data = {user_id: pref.to_dict() 
                   for user_id, pref in self.user_preferences.items()}
            with open(self.preferences_file, 'w', encoding='utf-8') as f:
                json.dump(data, f, indent=2, ensure_ascii=False)
        except Exception as e:
            logger.error(f"Error saving preferences: {e}")
    
    def save_all_data(self):
        """Save all learning data"""
        self.save_feedback()
        self.save_expertise()
        self.save_preferences()
        logger.info("All learning data saved")
    
    def load_all_data(self):
        """Load all learning data"""
        self._load_feedback()
        self._load_expertise()
        self._load_preferences()
    
    def _load_feedback(self):
        """Load feedback history"""
        try:
            if self.feedback_file.exists():
                with open(self.feedback_file, 'r', encoding='utf-8') as f:
                    data = json.load(f)
                self.feedback_history = [ResponseFeedback(**item) for item in data]
                logger.info(f"Loaded {len(self.feedback_history)} feedback items")
        except Exception as e:
            logger.error(f"Error loading feedback: {e}")
    
    def _load_expertise(self):
        """Load topic expertise"""
        try:
            if self.expertise_file.exists():
                with open(self.expertise_file, 'r', encoding='utf-8') as f:
                    data = json.load(f)
                for topic, exp_dict in data.items():
                    self.topic_expertise[topic] = TopicExpertise(**exp_dict)
                logger.info(f"Loaded expertise for {len(self.topic_expertise)} topics")
        except Exception as e:
            logger.error(f"Error loading expertise: {e}")
    
    def _load_preferences(self):
        """Load user preferences"""
        try:
            if self.preferences_file.exists():
                with open(self.preferences_file, 'r', encoding='utf-8') as f:
                    data = json.load(f)
                for user_id, pref_dict in data.items():
                    self.user_preferences[user_id] = UserPreference(**pref_dict)
                logger.info(f"Loaded preferences for {len(self.user_preferences)} users")
        except Exception as e:
            logger.error(f"Error loading preferences: {e}")


# ═══════════════════════════════════════════════════════════════════════════════
# EXPORT
# ═══════════════════════════════════════════════════════════════════════════════

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    
    # Test the engine
    engine = AdaptiveSelfLearningEngine()
    
    # Simulate some feedback
    print("Recording feedback samples...")
    for i in range(5):
        engine.record_feedback(
            user_id="test_user",
            response_id=f"response_{i}",
            query=f"How does {['Python', 'machine learning', 'web development'][i % 3]} work?",
            response="Sample response about the topic.",
            rating=0.8 + (i * 0.02),
            feedback_text="Good answer!"
        )
    
    # Get progress
    progress = engine.get_learning_progress()
    print(f"\nLearning Progress: {json.dumps(progress, indent=2)}")
    
    # Get expertise report
    print(f"\nTopic Expertise:")
    for topic, exp in engine.topic_expertise.items():
        print(f"  {topic}: {exp.expertise_level} ({exp.success_rate:.1%})")
    
    # Get improvements
    print(f"\nSuggested Improvements: {json.dumps(engine.suggest_improvements(), indent=2)}")