Update evaluator_module.py

evaluator_module.py

@@ -14,6 +14,7 @@ import hashlib
 from datetime import datetime
 import concurrent.futures
 import random
+import gc
 
 class AetherScoreEvaluator:
     def __init__(self):
@@ -25,245 +26,395 @@ class AetherScoreEvaluator:
             spacy.cli.download("en_core_web_sm")
             self.nlp = spacy.load("en_core_web_sm")
 
-        # LLM Judge Model
-        self.judge_model = pipeline(
-            "text2text-generation", 
-            model="google/flan-t5-base", 
-            device=-1  # 0 for GPU 
-        )
-
-        # Sentence Transformer for Sentence ----> Embedding
-        self.sentence_model = SentenceTransformer('all-MiniLM-L6-v2')
-
-        # for hallucination
-        self.rouge = evaluate.load("rouge")
-        self.sacrebleu = evaluate.load("sacrebleu")
-        self.nli_tokenizer = AutoTokenizer.from_pretrained("prajjwal1/bert-mini-mnli")
-        self.nli_model = AutoModelForSequenceClassification.from_pretrained("prajjwal1/bert-mini-mnli")
-
-        # Scoring weights # Domain Specific weights can be added for better results
-        self.weights = {'instruction_following': 0.25, 'hallucination_score': 0.20,
-                        'assumption_control': 0.20, 'coherence': 0.20, 'accuracy': 0.15}
+        # Initialize models with error handling
+        self._initialize_models()
+        
+        # Scoring weights
+        self.weights = {
+            'instruction_following': 0.25, 
+            'hallucination_score': 0.20,
+            'assumption_control': 0.20, 
+            'coherence': 0.20, 
+            'accuracy': 0.15
+        }
         
         # In-memory cache
         self.cache = {}
 
+    def _initialize_models(self):
+        """Initialize all models with proper error handling"""
+        try:
+            # LLM Judge Model
+            self.judge_model = pipeline(
+                "text2text-generation", 
+                model="google/flan-t5-base", 
+                device=-1  # CPU only for stability
+            )
+            
+            # Sentence Transformer
+            self.sentence_model = SentenceTransformer('all-MiniLM-L6-v2')
+            
+            # Evaluation metrics
+            self.rouge = evaluate.load("rouge")
+            self.sacrebleu = evaluate.load("sacrebleu")
+            
+            # NLI models
+            self.nli_tokenizer = AutoTokenizer.from_pretrained("prajjwal1/bert-mini-mnli")
+            self.nli_model = AutoModelForSequenceClassification.from_pretrained("prajjwal1/bert-mini-mnli")
+            
+            print("All models initialized successfully")
+            
+        except Exception as e:
+            print(f"Error initializing models: {e}")
+            # Fallback to basic functionality
+            self._use_fallback_models()
+
+    def _use_fallback_models(self):
+        """Fallback to basic evaluation if model loading fails"""
+        print("Using fallback evaluation methods")
+        self.judge_model = None
+        self.sentence_model = None
+        self.rouge = None
+        self.sacrebleu = None
+        self.nli_tokenizer = None
+        self.nli_model = None
+
+    def _cleanup_models(self):
+        """Clean up model memory"""
+        if hasattr(self, 'nli_model') and self.nli_model is not None:
+            del self.nli_model
+        if hasattr(self, 'judge_model') and self.judge_model is not None:
+            del self.judge_model
+        torch.cuda.empty_cache() if torch.cuda.is_available() else None
+        gc.collect()
+
     def _evaluate_with_llm_judge(self, prompt: str, response: str) -> dict:
         """
-        Hallucination detection using:
-        - NLI (entailment, neutral, contradiction)
-        - Embedding similarity
-        - ROUGE-L
-        - SacreBLEU
-        Assumption control derived from NLI neutrality.
+        Hallucination detection with robust error handling
         """
-        # Step 1: Embedding similarity
-        emb_sim = self._semantic_similarity(prompt, response)
-    
-        # Step 2: NLI inference
-        inputs = self.nli_tokenizer.encode_plus(prompt, response, return_tensors="pt", truncation=True)
-        with torch.no_grad():
-            logits = self.nli_model(**inputs).logits
-            probs = torch.softmax(logits, dim=-1).cpu().numpy()[0]
-            entailment, neutral, contradiction = probs[2], probs[1], probs[0]
-    
-        # Step 3: ROUGE-L
-        rouge_l = self.rouge.compute(predictions=[response], references=[prompt])["rougeL"]
-    
-        # Step 4: SacreBLEU (normalized 0–1)
-        sacrebleu = self.sacrebleu.compute(predictions=[response], references=[[prompt]])["score"] / 100.0
-    
-        # Step 5: Weighted hallucination score
-        weights = {"entailment": 0.4, "embedding": 0.2, "rouge": 0.2, "sacrebleu": 0.2}
-        halluc_score = 1 - (
-            weights["entailment"] * entailment +
-            weights["embedding"] * emb_sim +
-            weights["rouge"] * rouge_l +
-            weights["sacrebleu"] * sacrebleu
-        )
-    
-        # Step 6: Assumption control from neutrality
-        assumption_score = 1 - neutral
-    
-        # Step 7: Explanations
-        halluc_expl = (
-            f"Entailment={entailment:.2f}, Embedding={emb_sim:.2f}, "
-            f"ROUGE-L={rouge_l:.2f}, SacreBLEU={sacrebleu:.2f}, Neutral={neutral:.2f}"
-        )
-        assumption_expl = (
-            f"Assumption control is derived from NLI neutrality={neutral:.2f}. "
-            "Lower neutrality → stronger confidence."
-        )
-    
-        return {
-            "hallucination_score": (float(halluc_score), halluc_expl),
-            "assumption_control": (float(assumption_score), assumption_expl),
-               }
+        try:
+            # Step 1: Embedding similarity (with fallback)
+            if self.sentence_model is not None:
+                emb_sim = self._semantic_similarity(prompt, response)
+            else:
+                emb_sim = 0.5  # neutral fallback
+
+            # Step 2: NLI check (with error handling)
+            if self.nli_tokenizer is not None and self.nli_model is not None:
+                try:
+                    inputs = self.nli_tokenizer.encode_plus(
+                        prompt, response, 
+                        return_tensors="pt", 
+                        truncation=True,
+                        max_length=512  # Limit token length
+                    )
+                    with torch.no_grad():
+                        logits = self.nli_model(**inputs).logits
+                        probs = torch.softmax(logits, dim=-1).cpu().numpy()[0]
+                        entailment, neutral, contradiction = probs[2], probs[1], probs[0]
+                except Exception as nli_error:
+                    print(f"NLI evaluation failed: {nli_error}")
+                    entailment, neutral, contradiction = 0.33, 0.33, 0.34
+            else:
+                entailment, neutral, contradiction = 0.33, 0.33, 0.34
+
+            # Step 3: ROUGE-L (with error handling)
+            if self.rouge is not None:
+                try:
+                    rouge_l = self.rouge.compute(predictions=[response], references=[prompt])["rougeL"]
+                except Exception as rouge_error:
+                    print(f"ROUGE evaluation failed: {rouge_error}")
+                    rouge_l = 0.5
+            else:
+                rouge_l = 0.5
+
+            # Step 4: SacreBLEU (with error handling)
+            if self.sacrebleu is not None:
+                try:
+                    sacrebleu = self.sacrebleu.compute(predictions=[response], references=[[prompt]])["score"] / 100.0
+                except Exception as bleu_error:
+                    print(f"BLEU evaluation failed: {bleu_error}")
+                    sacrebleu = 0.5
+            else:
+                sacrebleu = 0.5
+
+            # Step 5: Weighted hallucination score
+            weights = {"entailment": 0.4, "embedding": 0.2, "rouge": 0.2, "sacrebleu": 0.2}
+            halluc_score = 1 - (
+                weights["entailment"] * entailment +
+                weights["embedding"] * emb_sim +
+                weights["rouge"] * rouge_l +
+                weights["sacrebleu"] * sacrebleu
+            )
+
+            # Step 6: Assumption control from neutrality
+            assumption_score = 1 - neutral
+
+            # Ensure scores are in valid range
+            halluc_score = max(0.0, min(1.0, float(halluc_score)))
+            assumption_score = max(0.0, min(1.0, float(assumption_score)))
+
+            # Step 7: Explanations
+            halluc_expl = (
+                f"Entailment={entailment:.2f}, Embedding={emb_sim:.2f}, "
+                f"ROUGE-L={rouge_l:.2f}, SacreBLEU={sacrebleu:.2f}, Neutral={neutral:.2f}"
+            )
+            assumption_expl = (
+                f"Assumption control derived from NLI neutrality={neutral:.2f}. "
+                "Lower neutrality → stronger confidence."
+            )
+
+            return {
+                "hallucination_score": (halluc_score, halluc_expl),
+                "assumption_control": (assumption_score, assumption_expl),
+            }
+
+        except Exception as e:
+            print(f"Evaluation error: {e}")
+            # Return fallback scores
+            return {
+                "hallucination_score": (0.5, f"Evaluation failed: {str(e)}"),
+                "assumption_control": (0.5, f"Evaluation failed: {str(e)}"),
+            }
 
-    # Single Evaluation # Inputs-->> Prompt, Agent Response, Expected Answer(Optional), Agent Name and Task type( General, QA, Summarizaton)etc 
     def evaluate_single(self, prompt: str, response: str, expected_answer: Optional[str] = None, task_type: str = "general") -> Dict:
-        
-        # Generating Eval ID
-        eval_id = self._generate_eval_id(prompt, response)
-        
-        # If already stored in cache direclty we can return from there.
-        # if eval_id in self.cache:
-        #     return self.cache[eval_id]
+        """Single evaluation with enhanced error handling"""
+        try:
+            # Input validation
+            if not prompt or not response:
+                return {
+                    "scores": {"overall_score": 0.0},
+                    "reasons": {"error": "Empty prompt or response"}
+                }
 
-        scores, reasons = {}, {}
+            # Generating Eval ID
+            eval_id = self._generate_eval_id(prompt, response)
+            
+            scores, reasons = {}, {}
 
-        # Taking Back scores and reasons of hallucination and assumption control from LLM Judge
-        llm_judge_results = self._evaluate_with_llm_judge(prompt, response)
-        scores['hallucination_score'], reasons['hallucination_score'] = llm_judge_results['hallucination_score']
-        scores['assumption_control'], reasons['assumption_control'] = llm_judge_results['assumption_control']
+            # LLM Judge evaluation
+            llm_judge_results = self._evaluate_with_llm_judge(prompt, response)
+            scores['hallucination_score'], reasons['hallucination_score'] = llm_judge_results['hallucination_score']
+            scores['assumption_control'], reasons['assumption_control'] = llm_judge_results['assumption_control']
 
-        # Evaluating Instruction Following, Coherence and Accuracy
-        scores['instruction_following'], reasons['instruction_following'] = self._evaluate_instruction_following(prompt, response)
-        scores['coherence'], reasons['coherence'] = self._evaluate_coherence(response)
-        scores['accuracy'], reasons['accuracy'] = self._evaluate_accuracy(response, expected_answer, task_type) if expected_answer else (0.5, "No expected answer provided.")
+            # Other evaluations
+            scores['instruction_following'], reasons['instruction_following'] = self._evaluate_instruction_following(prompt, response)
+            scores['coherence'], reasons['coherence'] = self._evaluate_coherence(response)
+            
+            if expected_answer:
+                scores['accuracy'], reasons['accuracy'] = self._evaluate_accuracy(response, expected_answer, task_type)
+            else:
+                scores['accuracy'], reasons['accuracy'] = (0.5, "No expected answer provided.")
 
-        # Calculating Overall Score
-        scores['overall_score'] = self._calculate_overall_score(scores)
-        reasons['overall_score'] = f" Weighted Average Score based on component scores."
+            # Calculate overall score
+            scores['overall_score'] = self._calculate_overall_score(scores)
+            reasons['overall_score'] = "Weighted average of component scores."
 
-        # Updating Eval ID, Timestamp and Task Type in Scores
-        scores.update({'eval_id': eval_id, 'timestamp': datetime.now().isoformat(), 'task_type': task_type})
+            # Add metadata
+            scores.update({
+                'eval_id': eval_id, 
+                'timestamp': datetime.now().isoformat(), 
+                'task_type': task_type
+            })
 
-        # Updating scores(Eval ID, timestamp and task_type) and reasons(all scores) in result
-        result = {"scores": scores, "reasons": reasons}
+            return {"scores": scores, "reasons": reasons}
 
-        #Storing results with corresponding Eval ID in cache
-        # self.cache[eval_id] = result
-        
-        return result
+        except Exception as e:
+            print(f"Single evaluation error: {e}")
+            return {
+                "scores": {"overall_score": 0.0, "eval_id": "error"},
+                "reasons": {"error": str(e)}
+            }
 
-    # Batch Evaluation # Input of JSON/CSV file
     def evaluate_batch(self, data: List[Dict], mode: str = "comprehensive") -> List[Dict]:
-        """Process a batch of evaluations in parallel."""
-        
+        """Process batch with improved error handling and cleanup"""
+        if not data:
+            return []
+
         results = []
+        failed_count = 0
 
-        # Get Item function
         def process_item(item):
-            # Calling our Evalution function for Single prompt response pair
-            eval_result = self.evaluate_single(
-                prompt=item.get('prompt', ''),
-                response=item.get('response', ''),
-                expected_answer=item.get('expected_answer',''),
-                task_type=item.get('task_type', 'general')
-            )
-            # Combining with original metadata
-            eval_result.update({
-                'task_id': item.get('task_id', eval_result['scores']['eval_id']),
-                'agent_name': item.get('agent_name', 'Unknown'),
-            })
-            return eval_result
+            try:
+                return self.evaluate_single(
+                    prompt=item.get('prompt', ''),
+                    response=item.get('response', ''),
+                    expected_answer=item.get('expected_answer', ''),
+                    task_type=item.get('task_type', 'general')
+                )
+            except Exception as e:
+                print(f"Item processing failed: {e}")
+                return None
+
+        # Use smaller thread pool and add timeout
+        max_workers = min(4, len(data))  # Limit concurrent threads
         
-        with concurrent.futures.ThreadPoolExecutor() as executor:
-            future_to_item = {executor.submit(process_item, item): item for item in data}
-            for future in concurrent.futures.as_completed(future_to_item):
+        with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
+            # Submit all tasks with timeout
+            future_to_item = {
+                executor.submit(process_item, item): (i, item) 
+                for i, item in enumerate(data)
+            }
+            
+            for future in concurrent.futures.as_completed(future_to_item, timeout=300):  # 5 minute timeout
                 try:
-                    results.append(future.result())
+                    result = future.result(timeout=30)  # 30 second per item timeout
+                    if result:
+                        idx, item = future_to_item[future]
+                        result.update({
+                            'task_id': item.get('task_id', result['scores'].get('eval_id', f'task_{idx}')),
+                            'agent_name': item.get('agent_name', 'Unknown'),
+                        })
+                        results.append(result)
+                    else:
+                        failed_count += 1
                 except Exception as exc:
-                    print(f'An item generated an exception: {exc}')
+                    failed_count += 1
+                    print(f'Item generated exception: {exc}')
+
+        if failed_count > 0:
+            print(f"Warning: {failed_count} items failed to process")
+
+        # Cleanup after batch processing
+        if len(data) > 10:  # Only cleanup for larger batches
+            gc.collect()
 
         return results
- 
-    # Instruction Following Evaluation (Prompt, Response)
+
     def _evaluate_instruction_following(self, prompt: str, response: str) -> Tuple[float, str]:
-        score, checks, passed = 1.0, 0, 0
-        
-        # Check for negative constraints
-        negations = re.findall(r"(don't|do not|avoid|without) ([\w\s,]+)", prompt.lower())
-        for _, constraint_phrase in negations:
-            checks += 1
-            words_to_avoid = [w.strip() for w in constraint_phrase.split(',')]
-            if not any(word in response.lower() for word in words_to_avoid if len(word) > 2):
-                passed += 1
-        
-        # Fallback to semantic similarity if no specific instructions found
-        if checks == 0:
-            sim = self._semantic_similarity(prompt, response)
-            return sim, f"No specific constraints found. Score based on semantic similarity ({sim:.2f}) to prompt."
-
-        # Final Score calculation
-        score = passed / checks if checks > 0 else 1.0
-        reason = f"{passed}/{checks} specific constraints were followed."
-        
-        return score, reason
+        """Evaluate instruction following with better error handling"""
+        try:
+            score, checks, passed = 1.0, 0, 0
+            
+            # Check for negative constraints
+            negations = re.findall(r"(don't|do not|avoid|without) ([\w\s,]+)", prompt.lower())
+            for _, constraint_phrase in negations:
+                checks += 1
+                words_to_avoid = [w.strip() for w in constraint_phrase.split(',')]
+                if not any(word in response.lower() for word in words_to_avoid if len(word) > 2):
+                    passed += 1
+            
+            # Fallback to semantic similarity if no specific instructions found
+            if checks == 0:
+                sim = self._semantic_similarity(prompt, response)
+                return sim, f"No specific constraints found. Score based on semantic similarity ({sim:.2f}) to prompt."
+
+            score = passed / checks if checks > 0 else 1.0
+            reason = f"{passed}/{checks} specific constraints were followed."
+            
+            return score, reason
+            
+        except Exception as e:
+            return 0.5, f"Instruction evaluation failed: {str(e)}"
 
-    # Evaluating Coherence (response)
     def _evaluate_coherence(self, response: str) -> Tuple[float, str]:
+        """Evaluate coherence with error handling"""
+        try:
+            if not response.strip():
+                return 0.1, "Empty response"
 
-        # Extracting Sentences from Response
-        doc = self.nlp(response)
-        sentences = [sent.text for sent in doc.sents]
-        
-        # If only one Sentence then Coherence is Neutral
-        if len(sentences) < 2:
-            return 0.7, "Coherence is neutral for single-sentence responses."
+            doc = self.nlp(response)
+            sentences = [sent.text for sent in doc.sents if sent.text.strip()]
+            
+            if len(sentences) < 2:
+                return 0.7, "Coherence is neutral for single-sentence responses."
 
-        # Fetching Embeddings from our Sentence Model
-        embeddings = self.sentence_model.encode(sentences)
-        sims = [cosine_similarity([embeddings[i]], [embeddings[i+1]])[0][0] for i in range(len(sentences)-1)]
-        
-        score = np.mean(sims)
-        
-        reason = f"Average sentence-to-sentence similarity score is {score:.2f} across {len(sentences)} sentences."
-        return score, reason
+            if self.sentence_model is not None:
+                embeddings = self.sentence_model.encode(sentences)
+                sims = [cosine_similarity([embeddings[i]], [embeddings[i+1]])[0][0] for i in range(len(sentences)-1)]
+                score = np.mean(sims)
+            else:
+                score = 0.7  # fallback
+
+            reason = f"Average sentence-to-sentence similarity score is {score:.2f} across {len(sentences)} sentences."
+            return float(score), reason
+            
+        except Exception as e:
+            return 0.5, f"Coherence evaluation failed: {str(e)}"
 
-    # Evaluating Accuracy (Response, Expected, Task_type)
     def _evaluate_accuracy(self, response: str, expected: str, task_type: str) -> Tuple[float, str]:
-        sim = self._semantic_similarity(response, expected)
-        reason = f"Semantic similarity between response and expected answer is {sim:.2f}."
-        if sim > 0.95:
-            reason += " (High match)"
-        elif sim < 0.5:
-            reason += " (Low match)"
-        return sim, reason
-        
-    # Overall Score
+        """Evaluate accuracy with error handling"""
+        try:
+            sim = self._semantic_similarity(response, expected)
+            reason = f"Semantic similarity between response and expected answer is {sim:.2f}."
+            if sim > 0.95:
+                reason += " (High match)"
+            elif sim < 0.5:
+                reason += " (Low match)"
+            return sim, reason
+        except Exception as e:
+            return 0.5, f"Accuracy evaluation failed: {str(e)}"
+
     def _calculate_overall_score(self, scores: Dict) -> float:
-        total, weight_sum = 0.0, 0.0
-        for metric, weight in self.weights.items():
-            if metric in scores:
-                total += scores[metric] * weight
-                weight_sum += weight
-        return total / weight_sum #if weight_sum > 0 else 0.5
-
-    # Explanation Generator, work in progress
+        """Calculate overall score with error handling"""
+        try:
+            total, weight_sum = 0.0, 0.0
+            for metric, weight in self.weights.items():
+                if metric in scores and isinstance(scores[metric], (int, float)):
+                    total += float(scores[metric]) * weight
+                    weight_sum += weight
+            return total / weight_sum if weight_sum > 0 else 0.5
+        except Exception:
+            return 0.5
+
     def generate_explanation(self, scores: Dict) -> str:
-        explanation = []
-        overall = scores.get('overall_score', 0)
-        explanation.append(f"Overall Score: {overall:.2f}/1.00 - Reflects a weighted average of all dimensions.")
-
-        if scores.get('instruction_following', 0) < 0.6:
-            explanation.append("⚠️ Low Instruction Following: The response may have ignored key constraints or parts of the prompt.")
-        if scores.get('hallucination_score', 0) < 0.6:
-            explanation.append("⚠️ Potential Hallucination: The response might contain unverified or fabricated information.")
-        if scores.get('accuracy', 0) < 0.6 and scores.get('accuracy', 0.5) != 0.5:
-            explanation.append("⚠️ Low Accuracy: The response significantly differs from the provided expected answer.")
-        
-        if not explanation[1:]:
-             explanation.append("✅ Great Performance: The agent performed well across the primary evaluation dimensions.")
+        """Generate explanation with error handling"""
+        try:
+            explanation = []
+            overall = scores.get('overall_score', 0)
+            explanation.append(f"Overall Score: {overall:.2f}/1.00 - Reflects a weighted average of all dimensions.")
+
+            if scores.get('instruction_following', 0) < 0.6:
+                explanation.append("Low Instruction Following: The response may have ignored key constraints or parts of the prompt.")
+            if scores.get('hallucination_score', 0) < 0.6:
+                explanation.append("Potential Hallucination: The response might contain unverified or fabricated information.")
+            if scores.get('accuracy', 0) < 0.6 and scores.get('accuracy', 0.5) != 0.5:
+                explanation.append("Low Accuracy: The response significantly differs from the provided expected answer.")
+            
+            if len(explanation) == 1:
+                explanation.append("Great Performance: The agent performed well across the primary evaluation dimensions.")
 
-        return "\n".join(explanation)
+            return "\n".join(explanation)
+        except Exception as e:
+            return f"Explanation generation failed: {str(e)}"
 
-    # Agent Scores 
     def get_agent_scores_from_results(self, results: List[Dict]) -> Dict[str, List[float]]:
+        """Get agent scores with error handling"""
         agent_scores = defaultdict(list)
         for result in results:
-            agent_name = result.get('agent_name', 'Unknown')
-            overall_score = result.get('scores', {}).get('overall_score', 0)
-            agent_scores[agent_name].append(overall_score)
+            try:
+                agent_name = result.get('agent_name', 'Unknown')
+                overall_score = result.get('scores', {}).get('overall_score', 0)
+                if isinstance(overall_score, (int, float)) and not np.isnan(overall_score):
+                    agent_scores[agent_name].append(float(overall_score))
+            except Exception as e:
+                print(f"Error processing result: {e}")
+                continue
         return agent_scores
 
-    # Some Helper Functions
     def _generate_eval_id(self, prompt: str, response: str) -> str:
-        return hashlib.md5(f"{prompt}{response}".encode()).hexdigest()[:12]
+        """Generate evaluation ID"""
+        try:
+            return hashlib.md5(f"{prompt}{response}".encode()).hexdigest()[:12]
+        except Exception:
+            return hashlib.md5(f"fallback{datetime.now()}".encode()).hexdigest()[:12]
     
     def _semantic_similarity(self, text1: str, text2: str) -> float:
-        if not text1 or not text2: return 0.0
-        emb1 = self.sentence_model.encode([text1])
-        emb2 = self.sentence_model.encode([text2])
-        return cosine_similarity(emb1, emb2)[0][0]
+        """Calculate semantic similarity with error handling"""
+        try:
+            if not text1 or not text2 or self.sentence_model is None:
+                return 0.0
+            emb1 = self.sentence_model.encode([text1])
+            emb2 = self.sentence_model.encode([text2])
+            sim = cosine_similarity(emb1, emb2)[0][0]
+            return float(sim) if not np.isnan(sim) else 0.0
+        except Exception as e:
+            print(f"Similarity calculation failed: {e}")
+            return 0.0
+
+    def __del__(self):
+        """Cleanup when object is destroyed"""
+        try:
+            self._cleanup_models()
+        except Exception:
+            pass