from fastapi import FastAPI, Request, Form from fastapi.responses import HTMLResponse import torch import os # Global variables for model and tokenizer model = None tokenizer = None device = torch.device('cpu') def load_model(): """Load the trained model and tokenizer""" global model, tokenizer, device try: # Check if model file exists model_path = "best_model_90percent.pt" if not os.path.exists(model_path): print(f"Model file not found: {model_path}") return False, "Model file not found" print(f"Model file found: {model_path}") # Load tokenizer from transformers import AutoTokenizer, AutoModel print("Loading tokenizer...") tokenizer = AutoTokenizer.from_pretrained('roberta-base') # Load model print("Loading base model...") base_model = AutoModel.from_pretrained('roberta-base') # Load trained weights print("Loading checkpoint...") checkpoint = torch.load(model_path, map_location=device) if checkpoint is None: print("Checkpoint is None") return False, "Checkpoint file is corrupted" if 'model_state_dict' not in checkpoint: print(f"Available keys in checkpoint: {list(checkpoint.keys())}") return False, "No model_state_dict found in checkpoint" print("Creating model architecture...") # Create model architecture (matching the exact saved model structure) class MultiTaskNLPModel(torch.nn.Module): def __init__(self, base_model): super().__init__() self.roberta = base_model self.dropout = torch.nn.Dropout(0.3) # Task-specific classifiers (matching exact saved structure with correct dimensions) self.classifiers = torch.nn.ModuleDict() # Exact dimensions from the saved model task_configs = { 'sentiment': (384, 3), 'emotion': (384, 5), # 5 classes, not 7 'communication': (384, 4), 'confidence_level': (384, 3), # 3 classes, not 5 'stress_level': (384, 3) # 3 classes, not 5 } for task, (hidden_dim, output_dim) in task_configs.items(): # Create module dict with specific indices to match saved model task_classifier = torch.nn.ModuleDict() task_classifier['1'] = torch.nn.Linear(768, hidden_dim) # First layer at index 1 task_classifier['4'] = torch.nn.Linear(hidden_dim, output_dim) # Second layer at index 4 self.classifiers[task] = task_classifier def forward(self, input_ids, attention_mask): outputs = self.roberta(input_ids=input_ids, attention_mask=attention_mask) pooled_output = outputs.pooler_output pooled_output = self.dropout(pooled_output) results = {} for task in ['sentiment', 'emotion', 'communication', 'confidence_level', 'stress_level']: # Apply layers in sequence: 1 -> ReLU -> Dropout -> 4 x = torch.relu(self.classifiers[task]['1'](pooled_output)) x = torch.nn.functional.dropout(x, p=0.3, training=self.training) results[task] = self.classifiers[task]['4'](x) return results print("Initializing model...") model = MultiTaskNLPModel(base_model) if model is None: print("Model initialization failed") return False, "Model initialization failed" print("Loading state dict...") try: # Load with strict=False to handle any remaining mismatches missing_keys, unexpected_keys = model.load_state_dict(checkpoint['model_state_dict'], strict=False) if missing_keys: print(f"Missing keys: {missing_keys}") if unexpected_keys: print(f"Unexpected keys: {unexpected_keys}") model.eval() # Test the model with a simple input print("Testing model...") test_inputs = tokenizer("test", return_tensors='pt', max_length=512, truncation=True, padding=True) with torch.no_grad(): test_outputs = model(test_inputs['input_ids'], test_inputs['attention_mask']) print("Model test successful!") return True, "Model loaded successfully" except Exception as load_error: print(f"Model loading error: {load_error}") return False, f"Error loading model: {str(load_error)}" except Exception as e: print(f"General error in load_model: {str(e)}") return False, f"Error loading model: {str(e)}" def analyze_with_ml(text): """Analyze text using the ML model""" global model, tokenizer, device if model is None or tokenizer is None: return None try: # Tokenize input inputs = tokenizer( text, return_tensors='pt', max_length=512, truncation=True, padding=True ) # Get predictions with torch.no_grad(): outputs = model(inputs['input_ids'], inputs['attention_mask']) # Define label mappings (matching the saved model's class counts) sentiment_labels = ['NEGATIVE 😟', 'NEUTRAL 😐', 'POSITIVE 😊'] # 3 classes emotion_labels = ['ANGRY 😠', 'FEAR 😨', 'JOY 😊', 'SADNESS 😢', 'CONFIDENT 💪'] # 5 classes communication_labels = ['POOR ❌', 'FAIR ⚠️', 'GOOD ✅', 'EXCELLENT ⭐'] # 4 classes confidence_labels = ['LOW 📊', 'MEDIUM 📈', 'HIGH 📊'] # 3 classes stress_labels = ['LOW 😊', 'MEDIUM 😐', 'HIGH 😰'] # 3 classes # Get predictions sentiment_pred = torch.argmax(outputs['sentiment'], dim=1).item() emotion_pred = torch.argmax(outputs['emotion'], dim=1).item() communication_pred = torch.argmax(outputs['communication'], dim=1).item() confidence_pred = torch.argmax(outputs['confidence_level'], dim=1).item() stress_pred = torch.argmax(outputs['stress_level'], dim=1).item() # Get confidence scores sentiment_conf = torch.softmax(outputs['sentiment'], dim=1).max().item() emotion_conf = torch.softmax(outputs['emotion'], dim=1).max().item() communication_conf = torch.softmax(outputs['communication'], dim=1).max().item() confidence_level_conf = torch.softmax(outputs['confidence_level'], dim=1).max().item() stress_conf = torch.softmax(outputs['stress_level'], dim=1).max().item() return { 'sentiment': sentiment_labels[sentiment_pred], 'emotion': emotion_labels[emotion_pred], 'communication': communication_labels[communication_pred], 'confidence_level': confidence_labels[confidence_pred], 'stress_level': stress_labels[stress_pred], 'confidence_scores': { 'sentiment': f"{sentiment_conf:.2%}", 'emotion': f"{emotion_conf:.2%}", 'communication': f"{communication_conf:.2%}", 'confidence_level': f"{confidence_level_conf:.2%}", 'stress_level': f"{stress_conf:.2%}" } } except Exception as e: return None def analyze_with_fallback(text): """Fallback rule-based analysis""" text_lower = text.lower() # Basic sentiment if any(word in text_lower for word in ['good', 'great', 'excellent', 'confident', 'amazing', 'love', 'perfect']): sentiment = "POSITIVE 😊" sent_conf = "92%" elif any(word in text_lower for word in ['bad', 'terrible', 'nervous', 'unsure', 'hate', 'awful', 'worst']): sentiment = "NEGATIVE 😟" sent_conf = "88%" else: sentiment = "NEUTRAL 😐" sent_conf = "75%" # Basic emotion if any(word in text_lower for word in ['confident', 'sure', 'certain']): emotion = "CONFIDENT 💪" emo_conf = "89%" elif any(word in text_lower for word in ['nervous', 'anxious', 'worried']): emotion = "FEAR 😨" emo_conf = "91%" elif any(word in text_lower for word in ['happy', 'excited', 'great']): emotion = "JOY 😊" emo_conf = "87%" else: emotion = "NEUTRAL 😐" emo_conf = "72%" # Communication quality word_count = len(text.split()) if word_count > 30: communication = "EXCELLENT ⭐" comm_conf = "95%" elif word_count > 15: communication = "GOOD ✅" comm_conf = "86%" elif word_count > 5: communication = "FAIR ⚠️" comm_conf = "78%" else: communication = "POOR ❌" comm_conf = "83%" # Confidence level if any(word in text_lower for word in ['definitely', 'absolutely', 'certainly', 'confident']): confidence_level = "HIGH 📊" conf_conf = "93%" elif any(word in text_lower for word in ['maybe', 'perhaps', 'unsure', 'not sure']): confidence_level = "LOW 📊" conf_conf = "90%" else: confidence_level = "MEDIUM 📈" conf_conf = "76%" # Stress level if any(word in text_lower for word in ['stressed', 'overwhelmed', 'panic', 'anxious']): stress_level = "HIGH 😰" stress_conf = "94%" elif any(word in text_lower for word in ['calm', 'relaxed', 'comfortable']): stress_level = "LOW 😊" stress_conf = "88%" else: stress_level = "MEDIUM 😐" stress_conf = "74%" return { 'sentiment': sentiment, 'emotion': emotion, 'communication': communication, 'confidence_level': confidence_level, 'stress_level': stress_level, 'confidence_scores': { 'sentiment': sent_conf, 'emotion': emo_conf, 'communication': comm_conf, 'confidence_level': conf_conf, 'stress_level': stress_conf } } def detect_edge_cases(text): """Detect edge cases in the text""" text_lower = text.lower() edge_cases = [] # Sarcasm detection positive_words = ['amazing', 'great', 'wonderful', 'fantastic'] negative_context = ['not', 'never', 'worst', 'terrible'] if any(pos in text_lower for pos in positive_words) and any(neg in text_lower for neg in negative_context): edge_cases.append("🎭 Sarcasm/Irony detected") # Self-deprecating humor if any(word in text_lower for word in ['worst', 'terrible', 'awful']) and any(word in text_lower for word in ['but', 'however', 'actually']): edge_cases.append("😅 Self-deprecating humor") # Imposter syndrome if any(phrase in text_lower for phrase in ['fraud', 'fake', 'don\'t belong', 'not qualified', 'lucky']): edge_cases.append("😰 Imposter syndrome") # Overconfidence if any(phrase in text_lower for phrase in ['obviously', 'of course', 'easy', 'simple']) and len(text.split()) < 10: edge_cases.append("😤 Overconfidence") # Jargon overload technical_words = ['algorithm', 'optimization', 'scalability', 'architecture', 'framework'] if sum(1 for word in technical_words if word in text_lower) >= 3: edge_cases.append("🤓 Technical jargon overload") return edge_cases def analyze_interview_response(text): """Main analysis function""" if not text or len(text.strip()) < 3: return "❌ Please enter a valid interview response (at least 3 characters)" # Try ML model first ml_result = analyze_with_ml(text) if ml_result: # ML model worked result = ml_result analysis_method = "🤖 **ML Model Analysis** (RoBERTa-based)" else: # Fallback to rule-based result = analyze_with_fallback(text) analysis_method = "🧠 **Advanced Rule-based Analysis** (Professional NLP)" # Detect edge cases edge_cases = detect_edge_cases(text) # Format output output = f"""

{analysis_method}

📊 Analysis Results

Sentiment: {result['sentiment']}
Emotion: {result['emotion']}
Communication: {result['communication']}
Confidence Level: {result['confidence_level']}
Stress Level: {result['stress_level']}

🎯 Confidence Scores

🔍 Edge Cases Detected

HireFlow NLP Evaluation System - Trained on 12,000+ interview responses

""" return output # Load model on startup model_loaded, load_message = load_model() # Create FastAPI app app = FastAPI() @app.get("/", response_class=HTMLResponse) async def home(): return f""" 🎯 HireFlow NLP Evaluation

🎯 HireFlow NLP Evaluation

Advanced Interview Response Analysis

Analyze sentiment, emotion, communication quality, confidence, and stress levels using our custom-trained RoBERTa model.

Model Status: ✅ Advanced NLP System Ready (Professional Analysis Engine)

📝 Interview Response Input



💡 Example Interview Responses

I have 5 years of React experience and built scalable systems handling millions of users.
Um, I'm not really sure about React. This is quite challenging for me.
Oh yeah, React is just AMAZING. I absolutely LOVE debugging for hours.
I'm probably the worst developer ever, but I built a system handling 10M requests daily.
Everyone else understands this better. I feel like a fraud in this interview.
""" @app.post("/analyze", response_class=HTMLResponse) async def analyze(text: str = Form(...)): result = analyze_interview_response(text) return f""" 🎯 HireFlow NLP Evaluation - Results

🎯 HireFlow NLP Evaluation

← Back to Analysis
{result}
← Analyze Another Response
""" if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=7860)