""" Advanced Learning System for FitScore Feedback Agent """ import uuid from datetime import datetime from typing import Dict, Any, Optional from sqlalchemy.orm import Session from .database import get_db, LearningEvent, ContrastivePair, FewShotExample, CurriculumStage, BayesianUpdate, ActiveLearningRequest class AdvancedLearningSystem: """Advanced learning system with multiple AI approaches""" def __init__(self): self.learning_methods = [ "rlhf", "contrastive", "few_shot", "curriculum", "online", "bayesian", "active" ] def process_learning_event(self, event_data: Dict[str, Any]) -> Dict[str, Any]: """Process a learning event using multiple AI approaches""" try: db = next(get_db()) event_id = str(uuid.uuid4()) # Create learning event learning_event = LearningEvent( event_id=event_id, event_type=event_data["event_type"], candidate_id=event_data["candidate_id"], job_id=event_data["job_id"], submission_id=event_data.get("submission_id"), input_data=event_data["input_data"], outcome_data=event_data["outcome_data"], confidence_score=event_data.get("confidence_score", 0.0), model_version=event_data.get("model_version", "v1.0"), learning_signal=0.0, # Will be calculated based on event type processed=False ) # Process based on event type if event_data["event_type"] == "rlhf": learning_event.learning_signal = self._process_rlhf_event(event_data) elif event_data["event_type"] == "contrastive": learning_event.learning_signal = self._process_contrastive_event(event_data) elif event_data["event_type"] == "few_shot": learning_event.learning_signal = self._process_few_shot_event(event_data) elif event_data["event_type"] == "curriculum": learning_event.learning_signal = self._process_curriculum_event(event_data) elif event_data["event_type"] == "online": learning_event.learning_signal = self._process_online_event(event_data) elif event_data["event_type"] == "bayesian": learning_event.learning_signal = self._process_bayesian_event(event_data) elif event_data["event_type"] == "active": learning_event.learning_signal = self._process_active_event(event_data) learning_event.processed = True db.add(learning_event) db.commit() return { "success": True, "event_id": event_id, "learning_signal": learning_event.learning_signal, "message": f"Learning event processed using {event_data['event_type']} method" } except Exception as e: print(f"❌ Error processing learning event: {e}") raise def get_learning_analytics(self) -> Dict[str, Any]: """Get comprehensive learning analytics""" try: db = next(get_db()) total_events = db.query(LearningEvent).count() processed_events = db.query(LearningEvent).filter(LearningEvent.processed == True).count() # Events by type events_by_type = {} for event_type in self.learning_methods: count = db.query(LearningEvent).filter(LearningEvent.event_type == event_type).count() events_by_type[event_type] = count # Average learning signals avg_learning_signal = db.query(LearningEvent.learning_signal).filter( LearningEvent.processed == True ).all() avg_signal = sum(signal[0] for signal in avg_learning_signal) / len(avg_learning_signal) if avg_learning_signal else 0 return { "total_events": total_events, "processed_events": processed_events, "processing_rate": (processed_events / total_events * 100) if total_events > 0 else 0, "events_by_type": events_by_type, "average_learning_signal": avg_signal, "learning_methods": self.learning_methods } except Exception as e: print(f"❌ Error getting learning analytics: {e}") return { "total_events": 0, "processed_events": 0, "processing_rate": 0, "events_by_type": {}, "average_learning_signal": 0, "learning_methods": self.learning_methods } def _process_rlhf_event(self, event_data: Dict[str, Any]) -> float: """Process RLHF (Reinforcement Learning from Human Feedback) event""" # Simplified RLHF processing outcome_data = event_data.get("outcome_data", {}) # Calculate reward based on outcome if outcome_data.get("outcome") == "hired": return 10.0 elif outcome_data.get("outcome") == "accepted": return 5.0 elif outcome_data.get("outcome") == "interviewed": return 1.0 elif outcome_data.get("outcome") == "rejected": return -1.0 return 0.0 def _process_contrastive_event(self, event_data: Dict[str, Any]) -> float: """Process contrastive learning event""" # Simplified contrastive learning input_data = event_data.get("input_data", {}) # Calculate separation score between positive and negative examples positive_score = input_data.get("positive_score", 0.0) negative_score = input_data.get("negative_score", 0.0) separation_score = positive_score - negative_score return max(0.0, separation_score) def _process_few_shot_event(self, event_data: Dict[str, Any]) -> float: """Process few-shot learning event""" # Simplified few-shot learning confidence_score = event_data.get("confidence_score", 0.0) # Higher confidence in few-shot examples indicates better learning return confidence_score * 2.0 def _process_curriculum_event(self, event_data: Dict[str, Any]) -> float: """Process curriculum learning event""" # Simplified curriculum learning input_data = event_data.get("input_data", {}) # Calculate progress through curriculum stages current_stage = input_data.get("current_stage", 1) total_stages = input_data.get("total_stages", 5) progress = current_stage / total_stages return progress * 5.0 def _process_online_event(self, event_data: Dict[str, Any]) -> float: """Process online learning event""" # Simplified online learning confidence_score = event_data.get("confidence_score", 0.0) # Online learning benefits from continuous updates return confidence_score * 1.5 def _process_bayesian_event(self, event_data: Dict[str, Any]) -> float: """Process Bayesian learning event""" # Simplified Bayesian learning input_data = event_data.get("input_data", {}) # Calculate uncertainty reduction prior_uncertainty = input_data.get("prior_uncertainty", 1.0) posterior_uncertainty = input_data.get("posterior_uncertainty", 0.5) uncertainty_reduction = prior_uncertainty - posterior_uncertainty return max(0.0, uncertainty_reduction * 3.0) def _process_active_event(self, event_data: Dict[str, Any]) -> float: """Process active learning event""" # Simplified active learning input_data = event_data.get("input_data", {}) # Calculate information gain uncertainty_score = input_data.get("uncertainty_score", 0.0) # Higher uncertainty means more potential for learning return uncertainty_score * 2.0