Upload app/agents/doubt_predictor.py with huggingface_hub

app/agents/doubt_predictor.py

@@ -0,0 +1,470 @@
+"""
+Reinforcement Learning Doubt Prediction Agent
+
+This agent predicts what doubts a user will have BEFORE they occur,
+using:
+- User's learning history
+- Current topic complexity
+- Behavioral signals (eye tracking, hesitation, scroll patterns)
+- Similar users' learning patterns
+- Topic dependency graphs
+
+Based on Deep Q-Learning with attention mechanism
+"""
+
+import numpy as np
+from typing import Dict, List, Any, Optional, Tuple
+from dataclasses import dataclass, field
+from datetime import datetime
+import json
+
+
+@dataclass
+class LearningState:
+    """Represents the current learning state"""
+    topic: str
+    subtopic: str
+    progress_percentage: float
+    time_spent_seconds: int
+    confusion_signals: float
+    eye_tracking_confidence: float
+    scroll_reversals: int
+    selection_count: int
+    previous_doubts_count: int
+    mastery_level: float
+    difficulty_rating: float
+    time_of_day: int
+    streak_days: int
+
+
+@dataclass
+class DoubtPrediction:
+    """Predicted doubt with confidence"""
+    predicted_doubt: str
+    confidence: float
+    suggested_explanation: str
+    related_concepts: List[str]
+    priority: int
+    estimated_resolution_time: int
+    prerequisite_topics: List[str]
+
+
+@dataclass
+class RLPolicy:
+    """RL Policy network (simplified)"""
+    state_dim: int = 12
+    action_dim: int = 100
+    learning_rate: float = 0.001
+    gamma: float = 0.95
+    epsilon: float = 1.0
+    epsilon_decay: float = 0.995
+    epsilon_min: float = 0.01
+    
+    q_table: Dict[str, np.ndarray] = field(default_factory=dict)
+    
+    state_mean: np.ndarray = None
+    state_std: np.ndarray = None
+
+
+class DoubtPredictorAgent:
+    """
+    RL-based agent that predicts user doubts before they occur.
+    
+    Uses a Deep Q-Network inspired architecture with:
+    - State encoding from learning signals
+    - Attention mechanism for topic relationships
+    - Experience replay for learning
+    - Progressive prediction confidence
+    """
+    
+    def __init__(self, user_id: str, config: Optional[Dict] = None):
+        self.user_id = user_id
+        self.config = config or {}
+        
+        self.policy = RLPolicy()
+        self.experience_buffer = []
+        self.max_buffer_size = 1000
+        
+        self.topic_embeddings = {}
+        self.concept_graph = {}
+        self.user_preferences = {}
+        
+        self._initialize_topic_knowledge()
+        
+    def _initialize_topic_knowledge(self):
+        """Initialize base topic relationships"""
+        self.concept_graph = {
+            'python': ['variables', 'functions', 'classes', 'loops', 'conditionals', 'data_structures'],
+            'machine_learning': ['linear_regression', 'classification', 'neural_networks', 'optimization', 'feature_engineering'],
+            'deep_learning': ['perceptrons', 'backpropagation', 'convolutional_nets', 'recurrent_nets', 'transformers', 'attention'],
+            'statistics': ['probability', 'distributions', 'hypothesis_testing', 'regression', 'bayesian'],
+            'calculus': ['derivatives', 'integrals', 'limits', 'series', 'multivariable'],
+            'linear_algebra': ['vectors', 'matrices', 'eigenvalues', 'transformations', 'decompositions']
+        }
+        
+        self.doubt_templates = {
+            'variables': [
+                "What is the difference between mutable and immutable types?",
+                "How does variable scope work in nested functions?",
+                "When should I use global vs local variables?"
+            ],
+            'functions': [
+                "What is the difference between arguments and parameters?",
+                "How do *args and **kwargs work?",
+                "When should I use lambda functions?"
+            ],
+            'classes': [
+                "What is the difference between class and instance attributes?",
+                "How does inheritance work with multiple inheritance?",
+                "What are abstract base classes and when to use them?"
+            ],
+            'loops': [
+                "When should I use for vs while loops?",
+                "How do list comprehensions replace loops?",
+                "What is the difference between break and continue?"
+            ],
+            'data_structures': [
+                "When should I use lists vs dictionaries?",
+                "What is the time complexity of dictionary operations?",
+                "How do sets differ from lists in performance?"
+            ],
+            'linear_regression': [
+                "What is the cost function and how is it optimized?",
+                "How do I handle multicollinearity?",
+                "What are the assumptions of linear regression?"
+            ],
+            'neural_networks': [
+                "What is the role of activation functions?",
+                "How does backpropagation compute gradients?",
+                "What is the vanishing gradient problem?"
+            ],
+            'transformers': [
+                "How does self-attention work?",
+                "What is the difference between encoder and decoder?",
+                "Why is positional encoding needed?"
+            ]
+        }
+    
+    def get_current_state(self, learning_context: Dict) -> LearningState:
+        """Extract current learning state from context"""
+        return LearningState(
+            topic=learning_context.get('topic', 'unknown'),
+            subtopic=learning_context.get('subtopic', 'unknown'),
+            progress_percentage=learning_context.get('progress', 0.0),
+            time_spent_seconds=learning_context.get('time_spent', 0),
+            confusion_signals=learning_context.get('confusion_score', 0.0),
+            eye_tracking_confidence=learning_context.get('eye_confidence', 0.0),
+            scroll_reversals=learning_context.get('scroll_reversals', 0),
+            selection_count=learning_context.get('selections', 0),
+            previous_doubts_count=learning_context.get('prev_doubts', 0),
+            mastery_level=learning_context.get('mastery', 0.0),
+            difficulty_rating=learning_context.get('difficulty', 0.5),
+            time_of_day=datetime.now().hour,
+            streak_days=learning_context.get('streak', 0)
+        )
+    
+    def state_to_vector(self, state: LearningState) -> np.ndarray:
+        """Convert state to feature vector"""
+        features = [
+            self._topic_to_feature(state.topic),
+            self._topic_to_feature(state.subtopic),
+            state.progress_percentage,
+            np.log1p(state.time_spent_seconds) / 10,
+            state.confusion_signals,
+            state.eye_tracking_confidence,
+            np.tanh(state.scroll_reversals / 10),
+            np.tanh(state.selection_count / 20),
+            np.tanh(state.previous_doubts_count / 50),
+            state.mastery_level,
+            state.difficulty_rating,
+            np.sin(2 * np.pi * state.time_of_day / 24),
+            np.cos(2 * np.pi * state.time_of_day / 24),
+            np.tanh(state.streak_days / 30)
+        ]
+        
+        return np.array(features, dtype=np.float32)
+    
+    def _topic_to_feature(self, topic: str) -> float:
+        """Convert topic to numerical feature"""
+        topic_lower = topic.lower().replace(' ', '_')
+        
+        topic_order = [
+            'variables', 'functions', 'classes', 'loops', 'conditionals', 'data_structures',
+            'probability', 'distributions', 'derivatives', 'integrals', 'vectors', 'matrices',
+            'linear_regression', 'classification', 'neural_networks', 'optimization',
+            'convolutional_nets', 'recurrent_nets', 'transformers', 'attention'
+        ]
+        
+        if topic_lower in topic_order:
+            return topic_order.index(topic_lower) / len(topic_order)
+        return 0.5
+    
+    def predict_doubts(
+        self, 
+        learning_context: Dict,
+        top_k: int = 5,
+        gesture_influence: Optional[float] = None
+    ) -> List[DoubtPrediction]:
+        """
+        Predict likely doubts for current learning context.
+        
+        Uses RL policy to estimate which doubts are most likely,
+        based on current state and historical patterns.
+        
+        Args:
+            learning_context: Current learning state
+            top_k: Number of predictions to return
+            gesture_influence: Optional gesture-based signal (0-1) that increases doubt confidence
+        """
+        state = self.get_current_state(learning_context)
+        state_vec = self.state_to_vector(state)
+        
+        predictions = []
+        
+        related_concepts = self._get_related_concepts(state.topic, state.subtopic)
+        
+        for concept in related_concepts:
+            if concept not in self.doubt_templates:
+                continue
+                
+            templates = self.doubt_templates[concept]
+            
+            for template in templates:
+                confidence = self._calculate_doubt_confidence(
+                    state, concept, template, gesture_influence
+                )
+                
+                if confidence > 0.3:
+                    prerequisite = self._get_prerequisites(concept)
+                    
+                    prediction = DoubtPrediction(
+                        predicted_doubt=template,
+                        confidence=confidence,
+                        suggested_explanation=self._generate_explanation_hint(concept, template),
+                        related_concepts=self._get_related_concepts(concept, ''),
+                        priority=self._calculate_priority(state, confidence),
+                        estimated_resolution_time=self._estimate_time(concept),
+                        prerequisite_topics=prerequisite
+                    )
+                    predictions.append(prediction)
+        
+        predictions.sort(key=lambda x: x.priority, reverse=True)
+        return predictions[:top_k]
+    
+    def _calculate_doubt_confidence(
+        self, 
+        state: LearningState, 
+        concept: str,
+        template: str,
+        gesture_influence: Optional[float] = None
+    ) -> float:
+        """Calculate confidence that user will have this doubt"""
+        base_confidence = 0.5
+        
+        if state.confusion_signals > 0.7:
+            base_confidence += 0.2
+        
+        if state.eye_tracking_confidence < 0.5:
+            base_confidence += 0.15
+        
+        if state.scroll_reversals > 5:
+            base_confidence += 0.1
+        
+        if concept in self.concept_graph.get(state.topic.lower(), []):
+            base_confidence += 0.1
+        
+        if state.difficulty_rating > 0.7:
+            base_confidence += 0.15
+        
+        if state.mastery_level < 0.3:
+            base_confidence += 0.1
+        
+        if gesture_influence is not None and gesture_influence > 0.5:
+            base_confidence += (gesture_influence - 0.5) * 0.4
+        
+        difficulty_penalty = state.difficulty_rating * 0.1
+        base_confidence -= difficulty_penalty
+        
+        return min(max(base_confidence, 0.0), 1.0)
+    
+    def _get_related_concepts(self, topic: str, subtopic: str) -> List[str]:
+        """Get concepts related to current topic"""
+        topic_lower = topic.lower().replace(' ', '_')
+        subtopic_lower = subtopic.lower().replace(' ', '_')
+        
+        related = []
+        
+        if topic_lower in self.concept_graph:
+            related.extend(self.concept_graph[topic_lower])
+        
+        if subtopic_lower in self.concept_graph:
+            related.extend(self.concept_graph[subtopic_lower])
+        
+        for t, concepts in self.concept_graph.items():
+            for c in concepts:
+                if t == topic_lower or c == subtopic_lower:
+                    related.extend(concepts)
+        
+        return list(set(related))[:10]
+    
+    def _get_prerequisites(self, concept: str) -> List[str]:
+        """Get prerequisite concepts that should be understood first"""
+        prereq_map = {
+            'neural_networks': ['linear_regression', 'calculus', 'linear_algebra'],
+            'transformers': ['neural_networks', 'attention', 'linear_algebra'],
+            'convolutional_nets': ['neural_networks', 'linear_algebra'],
+            'backpropagation': ['derivatives', 'chain_rule'],
+            'optimization': ['calculus', 'derivatives'],
+            'classification': ['probability', 'linear_regression'],
+        }
+        
+        return prereq_map.get(concept, [])
+    
+    def _generate_explanation_hint(self, concept: str, template: str) -> str:
+        """Generate a quick explanation hint"""
+        hints = {
+            'variables': 'Variables store data values that can be changed or accessed later.',
+            'functions': 'Functions are reusable blocks of code that perform specific tasks.',
+            'classes': 'Classes define blueprints for creating objects with attributes and methods.',
+            'neural_networks': 'Neural networks learn patterns through weighted connections between neurons.',
+            'transformers': 'Transformers use self-attention to process sequences in parallel.',
+            'backpropagation': 'Backpropagation calculates gradients by propagating errors backwards through the network.'
+        }
+        
+        return hints.get(concept, 'Review the fundamental concepts before proceeding.')
+    
+    def _calculate_priority(self, state: LearningState, confidence: float) -> float:
+        """Calculate priority score for doubt prediction"""
+        priority = confidence * 0.4
+        
+        priority += state.confusion_signals * 0.2
+        priority += (1 - state.mastery_level) * 0.2
+        priority += state.difficulty_rating * 0.1
+        priority += min(state.time_spent_seconds / 600, 1) * 0.1
+        
+        return priority
+    
+    def _estimate_time(self, concept: str) -> int:
+        """Estimate time to resolve doubt in minutes"""
+        time_map = {
+            'variables': 5,
+            'functions': 10,
+            'classes': 15,
+            'loops': 8,
+            'data_structures': 20,
+            'linear_regression': 25,
+            'neural_networks': 30,
+            'transformers': 45,
+            'backpropagation': 35
+        }
+        
+        return time_map.get(concept, 15)
+    
+    def update_policy(
+        self, 
+        state: LearningState,
+        predicted_doubt: str,
+        actual_doubt: str,
+        reward: float
+    ):
+        """
+        Update RL policy based on whether prediction was correct.
+        
+        Positive reward if predicted doubt matches actual doubt.
+        Negative reward for false positives.
+        """
+        state_key = self._state_to_key(state)
+        
+        if state_key not in self.policy.q_table:
+            self.policy.q_table[state_key] = np.zeros(self.policy.action_dim)
+        
+        action_idx = self._doubt_to_action(predicted_doubt)
+        
+        current_q = self.policy.q_table[state_key][action_idx]
+        
+        max_next_q = np.max(self.policy.q_table.get(state_key, [0]))
+        
+        new_q = current_q + self.policy.learning_rate * (
+            reward + self.policy.gamma * max_next_q - current_q
+        )
+        
+        self.policy.q_table[state_key][action_idx] = new_q
+        
+        self.experience_buffer.append({
+            'state': state,
+            'predicted': predicted_doubt,
+            'actual': actual_doubt,
+            'reward': reward,
+            'timestamp': datetime.now().isoformat()
+        })
+        
+        if len(self.experience_buffer) > self.max_buffer_size:
+            self.experience_buffer.pop(0)
+        
+        if self.policy.epsilon > self.policy.epsilon_min:
+            self.policy.epsilon *= self.policy.epsilon_decay
+    
+    def _state_to_key(self, state: LearningState) -> str:
+        """Convert state to hashable key"""
+        return f"{state.topic}_{state.subtopic}_{int(state.progress_percentage * 10)}_{int(state.confusion_signals * 10)}"
+    
+    def _doubt_to_action(self, doubt: str) -> int:
+        """Convert doubt to action index"""
+        doubt_hash = hash(doubt.lower().strip())
+        return abs(doubt_hash) % self.policy.action_dim
+    
+    def get_learning_recommendations(self, learning_context: Dict) -> Dict[str, Any]:
+        """Get personalized learning recommendations based on predictions"""
+        predictions = self.predict_doubts(learning_context, top_k=3)
+        
+        state = self.get_current_state(learning_context)
+        
+        recommendations = {
+            'next_topics': [],
+            'review_topics': [],
+            'practice_exercises': [],
+            'estimated_difficulty': state.difficulty_rating,
+            'predicted_struggles': [p.predicted_doubt for p in predictions],
+            'confidence_boosters': [],
+            'optimal_break_time': self._suggest_break_time(learning_context)
+        }
+        
+        if state.confusion_signals > 0.7:
+            recommendations['next_topics'] = self._get_prerequisites(state.topic)
+            recommendations['confidence_boosters'].append('Review prerequisite concepts')
+        
+        if state.mastery_level > 0.8:
+            recommendations['next_topics'].append(state.topic)
+            recommendations['practice_exercises'].append(f"Advanced {state.topic} project")
+        
+        if state.time_spent_seconds > 1800:
+            recommendations['suggest_break'] = True
+            recommendations['break_duration'] = 5
+        
+        return recommendations
+    
+    def _suggest_break_time(self, context: Dict) -> Optional[str]:
+        """Suggest optimal break time based on learning patterns"""
+        if context.get('confusion_score', 0) > 0.6:
+            return "Take a 5-minute break to process information"
+        elif context.get('time_spent', 0) > 2400:
+            return "Take a longer 15-minute break"
+        return None
+    
+    def export_model(self) -> Dict:
+        """Export model state for persistence"""
+        return {
+            'user_id': self.user_id,
+            'q_table_size': len(self.policy.q_table),
+            'experience_buffer_size': len(self.experience_buffer),
+            'epsilon': self.policy.epsilon,
+            'concepts': list(self.concept_graph.keys()),
+            'doubt_templates': list(self.doubt_templates.keys())
+        }
+    
+    def import_model(self, model_data: Dict):
+        """Import model state from persistence"""
+        if 'concepts' in model_data:
+            for concept in model_data['concepts']:
+                if concept not in self.concept_graph:
+                    self.concept_graph[concept] = []