Update evaluator_module.py (#1)

- Update evaluator_module.py (114f82fe7eca7efe34d222f4681d4ccba6b1f895)


Co-authored-by: Manay Porwal <manayporwal07@users.noreply.huggingface.co>

evaluator_module.py

@@ -10,7 +10,7 @@ from sklearn.metrics.pairwise import cosine_similarity
 import hashlib
 from datetime import datetime
 import concurrent.futures
-import random # For LLM simulation
+import random
 
 class AetherScoreEvaluator:
     def __init__(self):
@@ -21,69 +21,118 @@ class AetherScoreEvaluator:
             print("Downloading 'en_core_web_sm' spacy model...")
             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-small", 
+            device=-1  # 0 for GPU 
+        )
+
+        # Sentence Transformer for Sentence ----> Embedding
         self.sentence_model = SentenceTransformer('all-MiniLM-L6-v2')
         
-        # Scoring weights
+        # 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}
         
         # In-memory cache
         self.cache = {}
 
-    # --- LLM Judge (Simulated) ---
+    # LLM Judge
     def _evaluate_with_llm_judge(self, prompt: str, response: str) -> Dict:
-        """
-        Simulated LLM Judge for nuanced evaluation.
-        Replace this with a real API call to a model like Gemini or GPT-4.
-        """
-        print("Using simulated LLM Judge...")
-        # Simulate API call and JSON parsing
-        simulated_scores = {
-            "hallucination_score": (random.uniform(0.6, 0.95), "LLM Judge found the facts to be reasonably grounded in the prompt."),
-            "assumption_control": (random.uniform(0.5, 0.9), "LLM Judge noted minor, acceptable assumptions were made.")
+        
+        print("Using HF model LLM Judge...")
+        # Building Query for LLM Judge
+        query = (
+            f"Prompt: {prompt}\n"
+            f"Response: {response}\n\n"
+            "Evaluate the response and provide:\n"
+            "hallucination_score: float between 0 and 1 (1 = fully grounded, 0 = full hallucination)\n"
+            "assumption_control: float between 0 and 1 (1 = no assumptions, 0 = many unsupported assumptions)\n"
+            "Then explain briefly in one or two sentences."
+        )
+
+        # Calling our LLM
+        try:
+            result = self.judge_model(query, max_new_tokens=128, truncation=True)
+            output = result[0]['generated_text']
+        except Exception as e:
+            return {
+                "hallucination_score": (0.0, f"HF model failed: {e}"),
+                "assumption_control": (0.0, f"HF model failed: {e}")
+            }
+
+        # Parsing scores from model output
+        halluc_score = 0.0
+        assumption_score = 0.0
+        explanation = output
+
+        match_h = re.search(r"hallucination[_ ]?score[: ]?([01](?:\.\d+)?)", output, re.I)
+        match_a = re.search(r"assumption[_ ]?control[: ]?([01](?:\.\d+)?)", output, re.I)
+
+        if match_h:
+            halluc_score = float(match_h.group(1))
+        if match_a:
+            assumption_score = float(match_a.group(1))
+
+        return {
+            "hallucination_score": (halluc_score, explanation),
+            "assumption_control": (assumption_score, explanation)
         }
-        # In a real scenario, you'd handle API errors and fall back to heuristics.
-        return simulated_scores
 
-    # --- Main Evaluation Logic ---
+    # 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]
 
         scores, reasons = {}, {}
 
-        # Use LLM Judge for hallucination and assumption control
+        # 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']
 
-        # Heuristic evaluations for other metrics
+        # 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.")
-        
+
+        # Calculating Overall Score
         scores['overall_score'] = self._calculate_overall_score(scores)
-        reasons['overall_score'] = f"Weighted average based on component scores."
-        
+        reasons['overall_score'] = f" Weighted Average Score based on 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})
-        
+
+        # Updating scores(Eval ID, timestamp and task_type) and reasons(all scores) in result
         result = {"scores": scores, "reasons": reasons}
+
+        #Storing results with corresponding Eval ID in cache
         self.cache[eval_id] = result
+        
         return result
-
+    # 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."""
-        results = []
         
+        results = []
+
+        # 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'),
+                expected_answer=item.get('expected_answer',''),
                 task_type=item.get('task_type', 'general')
             )
-            # Combine with original metadata
+            # 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'),
@@ -100,7 +149,7 @@ class AetherScoreEvaluator:
 
         return results
 
-    # --- Scoring Sub-components ---
+    # Instruction Following Evaluation (Prompt, Response)
     def _evaluate_instruction_following(self, prompt: str, response: str) -> Tuple[float, str]:
         score, checks, passed = 1.0, 0, 0
         
@@ -116,24 +165,35 @@ class AetherScoreEvaluator:
         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
-    
+
+    # Evaluating Coherence (response)
     def _evaluate_coherence(self, response: str) -> Tuple[float, str]:
+
+        # 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."
-        
+
+        # 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)]
-        avg_sim = np.mean(sims) if sims else 0
-        score = max(0.0, min(1.0, avg_sim))
+        
+        score = np.mean(sims)
+        
         reason = f"Average sentence-to-sentence similarity score is {score:.2f} across {len(sentences)} sentences."
         return score, reason
 
+        
+    # 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}."
@@ -142,17 +202,18 @@ class AetherScoreEvaluator:
         elif sim < 0.5:
             reason += " (Low match)"
         return sim, reason
-
+        
+    # Overall Score
     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
+        return total / weight_sum #if weight_sum > 0 else 0.5
 
+    # Explanation Generator, work in progress
     def generate_explanation(self, scores: Dict) -> str:
-        """Generate a high-level summary explanation."""
         explanation = []
         overall = scores.get('overall_score', 0)
         explanation.append(f"Overall Score: {overall:.2f}/1.00 - Reflects a weighted average of all dimensions.")
@@ -165,10 +226,11 @@ class AetherScoreEvaluator:
             explanation.append("⚠️ Low Accuracy: The response significantly differs from the provided expected answer.")
         
         if not explanation[1:]:
-             explanation.append("✅ Solid Performance: The agent performed well across the primary evaluation dimensions.")
+             explanation.append("✅ Great Performance: The agent performed well across the primary evaluation dimensions.")
 
         return "\n".join(explanation)
 
+    # Agent Scores 
     def get_agent_scores_from_results(self, results: List[Dict]) -> Dict[str, List[float]]:
         agent_scores = defaultdict(list)
         for result in results:
@@ -177,10 +239,10 @@ class AetherScoreEvaluator:
             agent_scores[agent_name].append(overall_score)
         return agent_scores
 
-    # --- Helper Methods ---
+    # Helper Functions
     def _generate_eval_id(self, prompt: str, response: str) -> str:
         return hashlib.md5(f"{prompt}{response}".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])