Spaces:
Sleeping
Sleeping
| """ | |
| Adaptive Feedback 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, GlobalPrompt, LocalPrompt, Feedback, FeedbackPattern | |
| class AdaptiveFeedbackSystem: | |
| """Adaptive feedback system for processing and learning from feedback""" | |
| def __init__(self): | |
| self.global_prompt_template = self._get_global_prompt_template() | |
| def _get_global_prompt_template(self) -> str: | |
| """Get the base global prompt template""" | |
| return """ | |
| 🎯 GOAL: Generate Smart Hiring Criteria to evaluate candidates for any role. | |
| ## Base Scoring Categories: | |
| - Education (Weight: 20%) | |
| - Career Trajectory (Weight: 25%) | |
| - Company Relevance (Weight: 20%) | |
| - Tenure Stability (Weight: 15%) | |
| - Skills Match (Weight: 20%) | |
| ## Scoring Rules: | |
| 1. **Education Score (0-10)**: | |
| - PhD: 10 points | |
| - Masters: 9 points | |
| - Bachelors: 7 points | |
| - Associate: 5 points | |
| - High School: 3 points | |
| 2. **Career Trajectory (0-10)**: | |
| - Progressive roles: +2 points | |
| - Industry relevance: +2 points | |
| - Leadership experience: +2 points | |
| - Technical depth: +2 points | |
| - Innovation/achievements: +2 points | |
| 3. **Company Relevance (0-10)**: | |
| - Fortune 500: 8-10 points | |
| - Mid-size (100-1000): 6-8 points | |
| - Startup: 4-6 points | |
| - Industry alignment: +2 points | |
| 4. **Tenure Stability (0-10)**: | |
| - 5+ years average: 10 points | |
| - 3-5 years average: 8 points | |
| - 2-3 years average: 6 points | |
| - 1-2 years average: 4 points | |
| - <1 year average: 2 points | |
| 5. **Skills Match (0-10)**: | |
| - 90%+ match: 10 points | |
| - 70-89% match: 8 points | |
| - 50-69% match: 6 points | |
| - 30-49% match: 4 points | |
| - <30% match: 2 points | |
| ## Final Score Calculation: | |
| Total Score = (Education × 0.20) + (Career Trajectory × 0.25) + (Company Relevance × 0.20) + (Tenure Stability × 0.15) + (Skills Match × 0.20) | |
| ## Acceptance Criteria: | |
| - Score ≥ 7.0: Strong candidate | |
| - Score 5.0-6.9: Consider with reservations | |
| - Score < 5.0: Reject | |
| """ | |
| def create_initial_global_prompt(self): | |
| """Create initial global prompt in database""" | |
| try: | |
| db = next(get_db()) | |
| # Check if global prompt already exists | |
| existing_prompt = db.query(GlobalPrompt).filter(GlobalPrompt.is_active == True).first() | |
| if existing_prompt: | |
| return existing_prompt | |
| # Create new global prompt | |
| global_prompt = GlobalPrompt( | |
| version="v1.0", | |
| prompt_content=self.global_prompt_template, | |
| category_weights={ | |
| "education": 0.20, | |
| "career_trajectory": 0.25, | |
| "company_relevance": 0.20, | |
| "tenure_stability": 0.15, | |
| "skills_match": 0.20 | |
| }, | |
| scoring_rules={ | |
| "education": {"PhD": 10, "Masters": 9, "Bachelors": 7, "Associate": 5, "High School": 3}, | |
| "career_trajectory": {"progressive": 2, "industry_relevance": 2, "leadership": 2, "technical_depth": 2, "innovation": 2}, | |
| "company_relevance": {"fortune_500": 10, "mid_size": 8, "startup": 6, "industry_alignment": 2}, | |
| "tenure_stability": {"5_plus": 10, "3_5": 8, "2_3": 6, "1_2": 4, "less_1": 2}, | |
| "skills_match": {"90_plus": 10, "70_89": 8, "50_69": 6, "30_49": 4, "less_30": 2} | |
| }, | |
| is_active=True | |
| ) | |
| db.add(global_prompt) | |
| db.commit() | |
| db.refresh(global_prompt) | |
| print("✅ Initial global prompt created successfully!") | |
| return global_prompt | |
| except Exception as e: | |
| print(f"❌ Error creating initial global prompt: {e}") | |
| return None | |
| def add_feedback(self, job_id: str, company_id: str, analysis_id: str, | |
| feedback_type: str, feedback_text: str, feedback_category: str, | |
| confidence_score: float, email: Optional[str] = None, | |
| linkedin_url: Optional[str] = None): | |
| """Add feedback to the system""" | |
| try: | |
| db = next(get_db()) | |
| feedback = Feedback( | |
| feedback_id=str(uuid.uuid4()), | |
| job_id=job_id, | |
| company_id=company_id, | |
| analysis_id=analysis_id, | |
| feedback_type=feedback_type, | |
| feedback_text=feedback_text, | |
| feedback_category=feedback_category, | |
| confidence_score=confidence_score, | |
| email=email, | |
| linkedin_url=linkedin_url | |
| ) | |
| db.add(feedback) | |
| db.commit() | |
| db.refresh(feedback) | |
| # Process feedback for patterns | |
| self._process_feedback_patterns(feedback, db) | |
| return feedback | |
| except Exception as e: | |
| print(f"❌ Error adding feedback: {e}") | |
| raise | |
| def _process_feedback_patterns(self, feedback: Feedback, db: Session): | |
| """Process feedback to identify patterns""" | |
| try: | |
| # Simple pattern detection based on feedback text | |
| feedback_lower = feedback.feedback_text.lower() | |
| # Check for common patterns | |
| patterns = { | |
| "technical_skills": ["technical", "skills", "programming", "coding", "development"], | |
| "experience": ["experience", "years", "background", "history"], | |
| "communication": ["communication", "soft skills", "interpersonal", "teamwork"], | |
| "culture_fit": ["culture", "fit", "values", "personality", "attitude"] | |
| } | |
| for pattern_name, keywords in patterns.items(): | |
| if any(keyword in feedback_lower for keyword in keywords): | |
| # Check if pattern already exists | |
| existing_pattern = db.query(FeedbackPattern).filter( | |
| FeedbackPattern.pattern_text == pattern_name, | |
| FeedbackPattern.category == feedback.feedback_category | |
| ).first() | |
| if existing_pattern: | |
| existing_pattern.frequency += 1 | |
| existing_pattern.last_seen = datetime.utcnow() | |
| else: | |
| new_pattern = FeedbackPattern( | |
| pattern_text=pattern_name, | |
| category=feedback.feedback_category, | |
| frequency=1, | |
| affected_jobs=[feedback.job_id], | |
| affected_companies=[feedback.company_id], | |
| is_global=False | |
| ) | |
| db.add(new_pattern) | |
| db.commit() | |
| except Exception as e: | |
| print(f"❌ Error processing feedback patterns: {e}") | |
| def get_feedback_analytics(self) -> Dict[str, Any]: | |
| """Get feedback analytics""" | |
| try: | |
| db = next(get_db()) | |
| total_feedback = db.query(Feedback).count() | |
| # Feedback by type | |
| feedback_by_type = {} | |
| for feedback_type in ['hired', 'accepted', 'rejected', 'interviewed']: | |
| count = db.query(Feedback).filter(Feedback.feedback_type == feedback_type).count() | |
| feedback_by_type[feedback_type] = count | |
| # Feedback by category | |
| feedback_by_category = {} | |
| for category in ['skills', 'experience', 'location', 'education', 'other']: | |
| count = db.query(Feedback).filter(Feedback.feedback_category == category).count() | |
| feedback_by_category[category] = count | |
| return { | |
| "total_feedback": total_feedback, | |
| "feedback_by_type": feedback_by_type, | |
| "feedback_by_category": feedback_by_category, | |
| "patterns_count": db.query(FeedbackPattern).count() | |
| } | |
| except Exception as e: | |
| print(f"❌ Error getting feedback analytics: {e}") | |
| return { | |
| "total_feedback": 0, | |
| "feedback_by_type": {}, | |
| "feedback_by_category": {}, | |
| "patterns_count": 0 | |
| } |