""" ╔══════════════════════════════════════════════════════════════════════════════╗ ║ ║ ║ 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)}")