import re import numpy as np from typing import Dict, List, Optional, Tuple import json from collections import defaultdict import spacy from transformers import pipeline, AutoTokenizer, AutoModelForSequenceClassification import torch from sentence_transformers import SentenceTransformer from sklearn.metrics.pairwise import cosine_similarity import hashlib from datetime import datetime class AetherScoreEvaluator: """ Core evaluation engine for AetherScore framework Implements multi-dimensional scoring for AI agent responses """ def __init__(self): """Initialize evaluation models and components""" # Load models for evaluation self.init_models() # Initialize scoring weights self.weights = { 'instruction_following': 0.25, 'hallucination_score': 0.20, 'assumption_control': 0.20, 'coherence': 0.20, 'accuracy': 0.15 } # Evaluation cache for performance self.cache = {} def init_models(self): """Initialize NLP models for evaluation""" try: # Load spaCy for linguistic analysis self.nlp = spacy.load("en_core_web_sm") except: import subprocess subprocess.run(["python", "-m", "spacy", "download", "en_core_web_sm"]) self.nlp = spacy.load("en_core_web_sm") # Sentence transformer for semantic similarity self.sentence_model = SentenceTransformer('all-MiniLM-L6-v2') # Sentiment/coherence analyzer self.coherence_pipeline = pipeline( "text-classification", model="distilbert-base-uncased-finetuned-sst-2-english", device=-1 # CPU ) # Fact checking model (simulated for demo) self.fact_checker = FactChecker() def evaluate_single( self, prompt: str, response: str, expected_answer: Optional[str] = None, task_type: str = "general", metadata: Optional[Dict] = None ) -> Dict: """ Evaluate a single agent response Args: prompt: The input prompt given to the agent response: The agent's response expected_answer: Optional ground truth answer task_type: Type of task (general, QA, summarization, etc.) metadata: Additional metadata for evaluation Returns: Dictionary containing scores for each dimension """ # Generate evaluation ID eval_id = self._generate_eval_id(prompt, response) # Check cache if eval_id in self.cache: return self.cache[eval_id] scores = {} # 1. Instruction Following Score scores['instruction_following'] = self._evaluate_instruction_following( prompt, response, task_type ) # 2. Hallucination Detection scores['hallucination_score'] = self._detect_hallucination( prompt, response ) # 3. Assumption Control scores['assumption_control'] = self._evaluate_assumption_control( prompt, response ) # 4. Coherence Score scores['coherence'] = self._evaluate_coherence(response) # 5. Accuracy Score (if ground truth available) if expected_answer: scores['accuracy'] = self._evaluate_accuracy( response, expected_answer, task_type ) else: scores['accuracy'] = 0.5 # Neutral score when no ground truth # Calculate overall score scores['overall_score'] = self._calculate_overall_score(scores) # Add metadata scores['eval_id'] = eval_id scores['timestamp'] = datetime.now().isoformat() scores['task_type'] = task_type # Cache result self.cache[eval_id] = scores return scores def evaluate_batch( self, data: List[Dict], mode: str = "comprehensive" ) -> List[Dict]: """ Evaluate multiple agent responses in batch Args: data: List of evaluation items mode: Evaluation mode (comprehensive, fast, detailed) Returns: List of evaluation results """ results = [] for item in data: # Extract fields task_id = item.get('task_id', f"task_{len(results)}") agent_name = item.get('agent_name', 'Unknown') prompt = item.get('prompt', '') response = item.get('response', item.get('model_answer', '')) expected = item.get('expected_answer', None) task_type = item.get('task_type', 'general') # Evaluate scores = self.evaluate_single( prompt=prompt, response=response, expected_answer=expected, task_type=task_type ) # Add to results result = { 'task_id': task_id, 'agent_name': agent_name, 'scores': scores, 'prompt': prompt[:100] + '...' if len(prompt) > 100 else prompt, 'response': response[:100] + '...' if len(response) > 100 else response } results.append(result) return results def _evaluate_instruction_following( self, prompt: str, response: str, task_type: str ) -> float: """Evaluate how well the response follows instructions""" score = 0.0 checks_passed = 0 total_checks = 0 # Extract instructions from prompt instructions = self._extract_instructions(prompt) # Check each instruction for instruction in instructions: total_checks += 1 if self._check_instruction_compliance(instruction, response, task_type): checks_passed += 1 # Check format requirements if "list" in prompt.lower() or "enumerate" in prompt.lower(): total_checks += 1 if self._has_list_format(response): checks_passed += 1 if "step" in prompt.lower() or "steps" in prompt.lower(): total_checks += 1 if self._has_step_format(response): checks_passed += 1 # Check length requirements length_req = self._extract_length_requirement(prompt) if length_req: total_checks += 1 if self._check_length_compliance(response, length_req): checks_passed += 1 # Calculate final score if total_checks > 0: score = checks_passed / total_checks else: # Use semantic similarity as fallback score = self._semantic_similarity(prompt, response) return min(1.0, max(0.0, score)) def _detect_hallucination(self, prompt: str, response: str) -> float: """ Detect potential hallucinations in the response Returns a score where 1.0 means no hallucination detected """ hallucination_indicators = 0 total_checks = 5 # Check 1: Unsupported statistics/numbers numbers_in_response = re.findall(r'\b\d+\.?\d*%?\b', response) numbers_in_prompt = re.findall(r'\b\d+\.?\d*%?\b', prompt) unsupported_numbers = set(numbers_in_response) - set(numbers_in_prompt) if len(unsupported_numbers) > 2: # Allow some flexibility hallucination_indicators += 1 # Check 2: Unverifiable claims claim_patterns = [ r"studies show", r"research indicates", r"according to \w+", r"experts say", r"it is proven", r"scientifically" ] unverified_claims = 0 for pattern in claim_patterns: if re.search(pattern, response.lower()) and not re.search(pattern, prompt.lower()): unverified_claims += 1 if unverified_claims > 1: hallucination_indicators += 1 # Check 3: Fact checking (simulated) fact_check_score = self.fact_checker.check(response, prompt) if fact_check_score < 0.5: hallucination_indicators += 1 # Check 4: Temporal inconsistencies dates_response = re.findall(r'\b(19|20)\d{2}\b', response) dates_prompt = re.findall(r'\b(19|20)\d{2}\b', prompt) if dates_response and not dates_prompt: hallucination_indicators += 1 # Check 5: Entity consistency entities_response = self._extract_entities(response) entities_prompt = self._extract_entities(prompt) new_entities = entities_response - entities_prompt if len(new_entities) > len(entities_prompt) * 0.5: # Too many new entities hallucination_indicators += 1 # Calculate score (higher is better - less hallucination) score = 1.0 - (hallucination_indicators / total_checks) return max(0.0, score) def _evaluate_assumption_control(self, prompt: str, response: str) -> float: """Evaluate if the response makes unwarranted assumptions""" assumption_penalty = 0.0 # Check for assumption indicators assumption_phrases = [ "i assume", "assuming that", "presumably", "probably", "i guess", "i suppose", "it seems like", "based on my understanding" ] # Count assumption indicators for phrase in assumption_phrases: count = response.lower().count(phrase) assumption_penalty += count * 0.1 # Check for speculation speculation_words = ["might", "could", "maybe", "perhaps", "possibly"] speculation_count = sum(response.lower().count(word) for word in speculation_words) # Penalize excessive speculation if speculation_count > 3: assumption_penalty += 0.2 # Check if response goes beyond prompt scope response_topics = self._extract_topics(response) prompt_topics = self._extract_topics(prompt) out_of_scope = len(response_topics - prompt_topics) if out_of_scope > 2: assumption_penalty += 0.3 # Calculate final score score = 1.0 - min(1.0, assumption_penalty) return score def _evaluate_coherence(self, response: str) -> float: """Evaluate the coherence and logical flow of the response""" if not response: return 0.0 coherence_score = 0.0 # Split into sentences doc = self.nlp(response) sentences = [sent.text for sent in doc.sents] if len(sentences) < 2: # Single sentence - check basic coherence return 0.7 if len(response.split()) > 5 else 0.5 # Check sentence-to-sentence coherence coherence_scores = [] for i in range(len(sentences) - 1): emb1 = self.sentence_model.encode([sentences[i]]) emb2 = self.sentence_model.encode([sentences[i + 1]]) similarity = cosine_similarity(emb1, emb2)[0][0] coherence_scores.append(similarity) # Average coherence avg_coherence = np.mean(coherence_scores) if coherence_scores else 0.5 # Check for logical connectors connectors = ['therefore', 'however', 'moreover', 'furthermore', 'additionally', 'consequently', 'thus', 'hence', 'meanwhile', 'nevertheless'] connector_count = sum(1 for conn in connectors if conn in response.lower()) connector_bonus = min(0.2, connector_count * 0.05) # Check for structural coherence has_introduction = any(phrase in response.lower()[:100] for phrase in ['first', 'to begin', 'initially']) has_conclusion = any(phrase in response.lower()[-100:] for phrase in ['finally', 'in conclusion', 'to sum up', 'overall']) structure_bonus = 0.1 if has_introduction else 0 structure_bonus += 0.1 if has_conclusion else 0 # Final coherence score coherence_score = min(1.0, avg_coherence + connector_bonus + structure_bonus) return coherence_score def _evaluate_accuracy( self, response: str, expected_answer: str, task_type: str ) -> float: """Evaluate accuracy against ground truth""" if not expected_answer: return 0.5 # Normalize both strings response_norm = self._normalize_text(response) expected_norm = self._normalize_text(expected_answer) # Exact match check if response_norm == expected_norm: return 1.0 # Task-specific evaluation if task_type == "QA": return self._evaluate_qa_accuracy(response_norm, expected_norm) elif task_type == "summarization": return self._evaluate_summary_accuracy(response, expected_answer) elif task_type == "code": return self._evaluate_code_accuracy(response, expected_answer) else: # General semantic similarity return self._semantic_similarity(response, expected_answer) def _evaluate_qa_accuracy(self, response: str, expected: str) -> float: """Evaluate QA-specific accuracy""" # Check if answer is contained if expected in response or response in expected: return 0.9 # Check for numerical answers numbers_response = re.findall(r'\b\d+\.?\d*\b', response) numbers_expected = re.findall(r'\b\d+\.?\d*\b', expected) if numbers_expected and numbers_response: if set(numbers_expected) == set(numbers_response): return 0.95 # Semantic similarity return self._semantic_similarity(response, expected) def _evaluate_summary_accuracy(self, response: str, expected: str) -> float: """Evaluate summarization accuracy""" # Extract key points from both response_points = self._extract_key_points(response) expected_points = self._extract_key_points(expected) if not expected_points: return self._semantic_similarity(response, expected) # Calculate coverage covered = sum(1 for point in expected_points if any(self._semantic_similarity(point, rp) > 0.7 for rp in response_points)) coverage_score = covered / len(expected_points) # Check for brevity brevity_score = min(1.0, len(expected) / max(len(response), 1)) return 0.7 * coverage_score + 0.3 * brevity_score def _evaluate_code_accuracy(self, response: str, expected: str) -> float: """Evaluate code accuracy""" # Remove comments and whitespace response_clean = re.sub(r'#.*?\n', '', response) response_clean = re.sub(r'\s+', ' ', response_clean) expected_clean = re.sub(r'#.*?\n', '', expected) expected_clean = re.sub(r'\s+', ' ', expected_clean) # Check for key code elements if response_clean == expected_clean: return 1.0 # Check for similar structure return self._semantic_similarity(response_clean, expected_clean) def _calculate_overall_score(self, scores: Dict) -> float: """Calculate weighted overall score""" total = 0.0 weight_sum = 0.0 for metric, weight in self.weights.items(): if metric in scores: total += scores[metric] * weight weight_sum += weight if weight_sum > 0: return total / weight_sum return 0.5 def generate_explanation( self, scores: Dict, prompt: str, response: str ) -> str: """Generate human-readable explanation of scores""" explanation = [] overall = scores.get('overall_score', 0) explanation.append(f"Overall Score: {overall:.2f}/1.00") # Instruction following if_score = scores.get('instruction_following', 0) if if_score < 0.5: explanation.append(f"⚠️ Low instruction following ({if_score:.2f}): The response may not fully address the prompt requirements.") elif if_score > 0.8: explanation.append(f"✅ Good instruction following ({if_score:.2f}): The response addresses the prompt well.") # Hallucination hall_score = scores.get('hallucination_score', 0) if hall_score < 0.5: explanation.append(f"⚠️ Potential hallucinations detected ({hall_score:.2f}): The response may contain unsupported claims.") elif hall_score > 0.8: explanation.append(f"✅ Low hallucination risk ({hall_score:.2f}): The response appears factually grounded.") # Assumptions assume_score = scores.get('assumption_control', 0) if assume_score < 0.5: explanation.append(f"⚠️ Excessive assumptions ({assume_score:.2f}): The response makes unwarranted assumptions.") elif assume_score > 0.8: explanation.append(f"✅ Good assumption control ({assume_score:.2f}): The response stays within scope.") # Coherence coh_score = scores.get('coherence', 0) if coh_score < 0.5: explanation.append(f"⚠️ Low coherence ({coh_score:.2f}): The response lacks logical flow.") elif coh_score > 0.8: explanation.append(f"✅ High coherence ({coh_score:.2f}): The response is well-structured and logical.") # Accuracy acc_score = scores.get('accuracy', 0) if acc_score != 0.5: # Not default if acc_score < 0.5: explanation.append(f"⚠️ Low accuracy ({acc_score:.2f}): The response diverges from expected answer.") elif acc_score > 0.8: explanation.append(f"✅ High accuracy ({acc_score:.2f}): The response matches expected answer well.") return "\n".join(explanation) # Helper methods def _generate_eval_id(self, prompt: str, response: str) -> str: """Generate unique evaluation ID""" content = f"{prompt}{response}" return hashlib.md5(content.encode()).hexdigest()[:12] def _extract_instructions(self, prompt: str) -> List[str]: """Extract specific instructions from prompt""" instructions = [] # Look for imperative sentences doc = self.nlp(prompt) for sent in doc.sents: if any(token.dep_ == "ROOT" and token.pos_ == "VERB" for token in sent): instructions.append(sent.text) return instructions def _check_instruction_compliance( self, instruction: str, response: str, task_type: str ) -> bool: """Check if specific instruction is followed""" instruction_lower = instruction.lower() response_lower = response.lower() # Check for specific keywords if "explain" in instruction_lower: return len(response) > 50 if "list" in instruction_lower: return bool(re.search(r'(\d+\.|\*|-)\s', response)) if "summarize" in instruction_lower: return len(response) < len(instruction) * 3 if "compare" in instruction_lower: return any(word in response_lower for word in ['whereas', 'while', 'but', 'however']) # Default check using semantic similarity return self._semantic_similarity(instruction, response) > 0.5 def _has_list_format(self, text: str) -> bool: """Check if text has list formatting""" patterns = [ r'^\d+\.', # Numbered list r'^[*•-]', # Bullet points r'^\([a-z]\)', # Letter lists ] lines = text.split('\n') list_lines = sum(1 for line in lines if any(re.match(pattern, line.strip()) for pattern in patterns)) return list_lines >= 2 def _has_step_format(self, text: str) -> bool: """Check if text has step-by-step formatting""" step_indicators = ['step', 'first', 'second', 'then', 'next', 'finally'] return sum(1 for indicator in step_indicators if indicator in text.lower()) >= 2 def _extract_length_requirement(self, prompt: str) -> Optional[int]: """Extract length requirements from prompt""" patterns = [ r'(\d+)\s*words', r'(\d+)\s*sentences', r'(\d+)\s*paragraphs' ] for pattern in patterns: match = re.search(pattern, prompt.lower()) if match: return int(match.group(1)) return None def _check_length_compliance(self, response: str, requirement: int) -> bool: """Check if response meets length requirement""" word_count = len(response.split()) # Allow 20% deviation return requirement * 0.8 <= word_count <= requirement * 1.2 def _semantic_similarity(self, text1: str, text2: str) -> float: """Calculate semantic similarity between texts""" emb1 = self.sentence_model.encode([text1]) emb2 = self.sentence_model.encode([text2]) similarity = cosine_similarity(emb1, emb2)[0][0] return float(similarity) def _extract_entities(self, text: str) -> set: """Extract named entities from text""" doc = self.nlp(text) return {ent.text.lower() for ent in doc.ents} def _extract_topics(self, text: str) -> set: """Extract main topics from text""" doc = self.nlp(text) topics = set() # Extract noun phrases for chunk in doc.noun_chunks: if len(chunk.text) > 3: topics.add(chunk.text.lower()) # Extract important nouns for token in doc: if token.pos_ in ['NOUN', 'PROPN'] and len(token.text) > 3: topics.add(token.text.lower()) return topics def _extract_key_points(self, text: str) -> List[str]: """Extract key points from text""" doc = self.nlp(text) key_points = [] for sent in doc.sents: # Look for sentences with important verbs if any(token.pos_ == 'VERB' and token.dep_ == 'ROOT' for token in sent): key_points.append(sent.text) return key_points[:5] # Limit to top 5 points def _normalize_text(self, text: str) -> str: """Normalize text for comparison""" # Lower case text = text.lower() # Remove extra whitespace text = re.sub(r'\s+', ' ', text) # Remove punctuation text = re.sub(r'[^\w\s]', '', text) return text.strip() class FactChecker: """Simulated fact checker for demonstration""" def check(self, response: str, prompt: str) -> float: """ Check facts in response against prompt context Returns score between 0 and 1 """ # This is a simplified implementation # In production, this would use external knowledge bases # Check if response contradicts prompt if "not" in response.lower() and "not" not in prompt.lower(): return 0.3 # Check for consistency response_nums = set(re.findall(r'\b\d+\b', response)) prompt_nums = set(re.findall(r'\b\d+\b', prompt)) if response_nums and not response_nums.intersection(prompt_nums): return 0.4 # Default moderate score return 0.7