Upload agents.py with huggingface_hub

agents.py

@@ -0,0 +1,445 @@
+"""
+ContextFlow Multi-Agent Integration
+
+Brings together the core ContextFlow agents for the OpenEnv environment:
+- DoubtPredictorAgent: RL-based confusion prediction
+- BehavioralAgent: Behavior signal analysis  
+- HandGestureAgent: Gesture-based learning signals
+"""
+
+import numpy as np
+from typing import Dict, List, Any, Optional, Tuple
+from dataclasses import dataclass, field
+from datetime import datetime
+from enum import Enum
+import json
+
+
+class ConfusionLevel(str, Enum):
+    LOW = "low"
+    MEDIUM = "medium"
+    HIGH = "high"
+    CRITICAL = "critical"
+
+
+class InterventionType(str, Enum):
+    HINT = "hint"
+    SIMPLIFY = "simplify"
+    BREAKDOWN = "breakdown"
+    EXAMPLE = "example"
+    SCAFFOLD = "scaffold"
+    PEER_CONNECT = "peer_connect"
+    BREAK = "break"
+    ENCOURAGE = "encourage"
+
+
+@dataclass
+class LearningState:
+    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 BehavioralSignal:
+    signal_type: str
+    value: float
+    timestamp: datetime
+    source: str
+    metadata: Dict = field(default_factory=dict)
+
+
+@dataclass
+class GestureTemplate:
+    gesture_id: str
+    name: str
+    description: str
+    samples: List[List[float]] = field(default_factory=list)
+    centroid: Optional[List[float]] = None
+    threshold: float = 0.3
+    trained: bool = False
+
+
+@dataclass
+class AgentPrediction:
+    confusion_probability: float
+    confusion_level: ConfusionLevel
+    confidence: float
+    recommended_intervention: InterventionType
+    intervention_intensity: float
+    reasoning: str
+    supporting_signals: Dict[str, float]
+
+
+class MultiModalFusion:
+    """Fuses signals from multiple modalities"""
+    
+    def __init__(self):
+        self.weights = {
+            "behavioral": 0.25,
+            "gesture": 0.25,
+            "biometric": 0.20,
+            "temporal": 0.15,
+            "content": 0.15,
+        }
+    
+    def fuse(
+        self,
+        behavioral: float,
+        gesture: float,
+        biometric: float,
+        temporal: float,
+        content: float,
+    ) -> float:
+        return (
+            self.weights["behavioral"] * behavioral +
+            self.weights["gesture"] * gesture +
+            self.weights["biometric"] * biometric +
+            self.weights["temporal"] * temporal +
+            self.weights["content"] * content
+        )
+
+
+class ContextFlowAgent:
+    """
+    Integrated ContextFlow agent combining:
+    - RL-based doubt prediction
+    - Behavioral signal analysis
+    - Gesture recognition
+    - Multi-modal fusion
+    """
+    
+    def __init__(self, config: Optional[Dict] = None):
+        self.config = config or {}
+        
+        self.doubt_predictor = RLBasedPredictor()
+        self.behavioral_analyzer = BehavioralAnalyzer()
+        self.gesture_recognizer = GestureRecognizer()
+        self.fusion = MultiModalFusion()
+        
+        self.history: List[Dict] = []
+        self.episode_rewards: List[float] = []
+        
+        self.epsilon = 1.0
+        self.epsilon_decay = 0.995
+        self.epsilon_min = 0.01
+    
+    def predict(
+        self,
+        observation: Dict[str, Any],
+        use_exploration: bool = True,
+    ) -> AgentPrediction:
+        behavioral_signal = self.behavioral_analyzer.analyze(observation)
+        gesture_signal = self.gesture_recognizer.recognize(observation)
+        biometric_signal = self._extract_biometric_signal(observation)
+        temporal_signal = self._extract_temporal_signal(observation)
+        content_signal = self._extract_content_signal(observation)
+        
+        fused_signal = self.fusion.fuse(
+            behavioral=behavioral_signal,
+            gesture=gesture_signal,
+            biometric=biometric_signal,
+            temporal=temporal_signal,
+            content=content_signal,
+        )
+        
+        if use_exploration and np.random.random() < self.epsilon:
+            confusion_prob = np.random.uniform(0.3, 0.8)
+        else:
+            confusion_prob = self.doubt_predictor.predict(fused_signal)
+        
+        confusion_level = self._get_confusion_level(confusion_prob)
+        intervention, intensity = self._get_recommendation(confusion_prob, confusion_level)
+        
+        return AgentPrediction(
+            confusion_probability=confusion_prob,
+            confusion_level=confusion_level,
+            confidence=0.85,
+            recommended_intervention=intervention,
+            intervention_intensity=intensity,
+            reasoning=self._generate_reasoning(behavioral_signal, gesture_signal, biometric_signal),
+            supporting_signals={
+                "behavioral": behavioral_signal,
+                "gesture": gesture_signal,
+                "biometric": biometric_signal,
+                "temporal": temporal_signal,
+                "content": content_signal,
+                "fused": fused_signal,
+            }
+        )
+    
+    def update(self, reward: float, observation: Dict[str, Any]):
+        self.episode_rewards.append(reward)
+        self.epsilon = max(self.epsilon_min, self.epsilon * self.epsilon_decay)
+        
+        if len(self.episode_rewards) > 100:
+            recent_avg = np.mean(self.episode_rewards[-100:])
+            self.doubt_predictor.update_q_value(recent_avg)
+    
+    def _extract_biometric_signal(self, obs: Dict) -> float:
+        biometric = obs.get("biometric_features", [])
+        if not biometric:
+            return 0.5
+        
+        hr = biometric[0] if len(biometric) > 0 else 70.0
+        gsr = biometric[1] if len(biometric) > 1 else 0.5
+        
+        hr_signal = min(1.0, max(0.0, (hr - 60) / 40))
+        gsr_signal = min(1.0, max(0.0, gsr * 2))
+        
+        return (hr_signal + gsr_signal) / 2
+    
+    def _extract_temporal_signal(self, obs: Dict) -> float:
+        time_spent = obs.get("learning_context", {}).get("time_spent", 0)
+        
+        if time_spent < 300:
+            return 0.2
+        elif time_spent < 900:
+            return 0.4
+        elif time_spent < 1800:
+            return 0.6
+        else:
+            return 0.8 + min(0.2, (time_spent - 1800) / 3600)
+    
+    def _extract_content_signal(self, obs: Dict) -> float:
+        difficulty = obs.get("learning_context", {}).get("difficulty", "medium")
+        difficulty_map = {"easy": 0.2, "medium": 0.5, "hard": 0.8}
+        return difficulty_map.get(difficulty, 0.5)
+    
+    def _get_confusion_level(self, prob: float) -> ConfusionLevel:
+        if prob < 0.25:
+            return ConfusionLevel.LOW
+        elif prob < 0.5:
+            return ConfusionLevel.MEDIUM
+        elif prob < 0.75:
+            return ConfusionLevel.HIGH
+        else:
+            return ConfusionLevel.CRITICAL
+    
+    def _get_recommendation(self, prob: float, level: ConfusionLevel) -> Tuple[InterventionType, float]:
+        recommendations = {
+            ConfusionLevel.LOW: (InterventionType.ENCOURAGE, 0.3),
+            ConfusionLevel.MEDIUM: (InterventionType.HINT, 0.5),
+            ConfusionLevel.HIGH: (InterventionType.SIMPLIFY, 0.7),
+            ConfusionLevel.CRITICAL: (InterventionType.SCAFFOLD, 0.9),
+        }
+        return recommendations[level]
+    
+    def _generate_reasoning(self, behavioral: float, gesture: float, biometric: float) -> str:
+        reasons = []
+        
+        if behavioral > 0.6:
+            reasons.append("High scroll reversals and hesitation detected")
+        if gesture > 0.6:
+            reasons.append("Confusion-related gestures identified")
+        if biometric > 0.6:
+            reasons.append("Elevated physiological stress indicators")
+        
+        if not reasons:
+            reasons.append("All signals within normal range")
+        
+        return "; ".join(reasons)
+
+
+class RLBasedPredictor:
+    """Q-learning based confusion predictor"""
+    
+    def __init__(self):
+        self.q_values: Dict[float, float] = {}
+        self.gamma = 0.95
+        self.learning_rate = 0.1
+    
+    def predict(self, state: float) -> float:
+        if state not in self.q_values:
+            self.q_values[state] = 0.5
+        
+        base = self.q_values[state]
+        noise = np.random.normal(0, 0.05)
+        return np.clip(base + noise, 0.0, 1.0)
+    
+    def update_q_value(self, reward: float, state: float = 0.5):
+        if state not in self.q_values:
+            self.q_values[state] = 0.5
+        
+        self.q_values[state] += self.learning_rate * (
+            reward - self.q_values[state]
+        )
+
+
+class BehavioralAnalyzer:
+    """Analyzes behavioral signals for confusion indicators"""
+    
+    def __init__(self):
+        self.baseline_scroll_speed = 1.0
+        self.baseline_click_rate = 1.0
+    
+    def analyze(self, observation: Dict[str, Any]) -> float:
+        behavioral = observation.get("behavioral_features", [])
+        
+        if not behavioral or len(behavioral) < 4:
+            return 0.5
+        
+        scroll_reversal = behavioral[0]
+        hesitation = behavioral[1]
+        click_pattern = behavioral[2]
+        time_on_task = behavioral[3]
+        
+        signals = [
+            min(1.0, scroll_reversal * 2),
+            min(1.0, hesitation * 2),
+            min(1.0, click_pattern * 2),
+            min(1.0, time_on_task / 1800),
+        ]
+        
+        return np.mean(signals)
+
+
+class GestureRecognizer:
+    """Recognizes confusion-related gestures"""
+    
+    CONFUCSION_GESTURES = {
+        "head_scratch": {"pattern": [0.7, 0.8, 0.9], "confidence": 0.85},
+        "brow_furrow": {"pattern": [0.6, 0.7, 0.8], "confidence": 0.75},
+        "hand_wave": {"pattern": [0.5, 0.6, 0.7], "confidence": 0.70},
+        "thinking": {"pattern": [0.4, 0.5, 0.6], "confidence": 0.65},
+    }
+    
+    def __init__(self):
+        self.last_gesture = None
+        self.gesture_duration = 0
+    
+    def recognize(self, observation: Dict[str, Any]) -> float:
+        gesture_features = observation.get("gesture_features", [])
+        
+        if not gesture_features or len(gesture_features) < 21:
+            return 0.3
+        
+        hand_variance = np.var(gesture_features[:21])
+        movement_intensity = np.mean(np.abs(np.diff(gesture_features[:21])))
+        
+        confusion_score = min(1.0, (hand_variance * 5 + movement_intensity * 3))
+        
+        return confusion_score
+
+
+class KnowledgeGraphAgent:
+    """Tracks concept relationships and prerequisite chains"""
+    
+    def __init__(self):
+        self.concepts: Dict[str, Dict] = {}
+        self.prerequisites: Dict[str, List[str]] = {}
+    
+    def add_concept(self, concept: str, mastery: float, prerequisites: List[str]):
+        self.concepts[concept] = {
+            "mastery": mastery,
+            "last_accessed": datetime.now(),
+        }
+        self.prerequisites[concept] = prerequisites
+    
+    def get_prerequisite_mastery(self, concept: str) -> float:
+        prereqs = self.prerequisites.get(concept, [])
+        if not prereqs:
+            return 1.0
+        
+        masteries = [self.concepts.get(p, {}).get("mastery", 0.0) for p in prereqs]
+        return min(masteries) if masteries else 1.0
+    
+    def predict_confusion_risk(self, concept: str) -> float:
+        mastery = self.concepts.get(concept, {}).get("mastery", 0.0)
+        prereq_mastery = self.get_prerequisite_mastery(concept)
+        
+        risk = (1 - mastery) * 0.6 + (1 - prereq_mastery) * 0.4
+        return risk
+
+
+class PeerLearningAgent:
+    """Connects learners with similar struggles"""
+    
+    def __init__(self):
+        self.learners: Dict[str, Dict] = {}
+        self.doubt_patterns: Dict[str, List[str]] = {}
+    
+    def register_doubt(self, user_id: str, doubt: str):
+        if user_id not in self.doubt_patterns:
+            self.doubt_patterns[user_id] = []
+        self.doubt_patterns[user_id].append(doubt)
+    
+    def find_similar_learners(self, doubt: str, top_k: int = 3) -> List[Dict]:
+        matches = []
+        for user_id, doubts in self.doubt_patterns.items():
+            overlap = len(set(doubts) & {doubt})
+            if overlap > 0:
+                matches.append({
+                    "user_id": user_id,
+                    "overlap": overlap,
+                    "solutions_shared": len(doubts) - overlap,
+                })
+        
+        matches.sort(key=lambda x: x["overlap"], reverse=True)
+        return matches[:top_k]
+
+
+class RecallAgent:
+    """Spaced repetition for concept reinforcement"""
+    
+    def __init__(self):
+        self.cards: Dict[str, Dict] = {}
+    
+    def add_card(self, concept: str, quality: int = 0):
+        self.cards[concept] = {
+            "interval": 1,
+            "ease_factor": 2.5,
+            "repetitions": 0,
+            "next_review": datetime.now(),
+            "quality": quality,
+        }
+    
+    def process_review(self, concept: str, quality: int) -> Dict:
+        if concept not in self.cards:
+            self.add_card(concept, quality)
+            return {"interval": 1, "message": "New card added"}
+        
+        card = self.cards[concept]
+        
+        if quality < 3:
+            card["repetitions"] = 0
+            card["interval"] = 1
+        else:
+            if card["repetitions"] == 0:
+                card["interval"] = 1
+            elif card["repetitions"] == 1:
+                card["interval"] = 6
+            else:
+                card["interval"] = int(card["interval"] * card["ease_factor"])
+            
+            card["repetitions"] += 1
+        
+        card["ease_factor"] = max(1.3, card["ease_factor"] + (0.1 - (5 - quality) * (0.08 + (5 - quality) * 0.02)))
+        card["next_review"] = datetime.now()
+        
+        return {"interval": card["interval"], "ease_factor": card["ease_factor"]}
+
+
+__all__ = [
+    "ContextFlowAgent",
+    "RLBasedPredictor",
+    "BehavioralAnalyzer",
+    "GestureRecognizer",
+    "KnowledgeGraphAgent",
+    "PeerLearningAgent",
+    "RecallAgent",
+    "MultiModalFusion",
+    "ConfusionLevel",
+    "InterventionType",
+    "AgentPrediction",
+]