""" Quality assessment component for evaluating system performance. """ from typing import Dict, Any, List, Tuple from dataclasses import dataclass import time from src.utils.logging_config import logger @dataclass class QualityMetrics: """Quality metrics for system evaluation.""" accuracy: float completeness: float relevance: float efficiency: float robustness: float overall_score: float class QualityAssessor: """Assesses the quality of system outputs and performance.""" def __init__(self, config: Dict[str, Any] = None): """Initialize the quality assessor.""" self.config = config or {} self.quality_history = [] def assess_query_processing(self, query: str, results: Dict[str, Any]) -> QualityMetrics: """Assess the quality of query processing results.""" # Accuracy: How well did we understand the query? accuracy = self._assess_understanding_accuracy(query, results) # Completeness: Did we find and execute relevant APIs? completeness = self._assess_result_completeness(results) # Relevance: How relevant were the matched APIs? relevance = self._assess_api_relevance(results) # Efficiency: How efficiently did we process the query? efficiency = self._assess_processing_efficiency(results) # Robustness: How well did we handle errors? robustness = self._assess_error_handling(results) # Overall score (weighted average) overall_score = ( accuracy * 0.25 + completeness * 0.25 + relevance * 0.20 + efficiency * 0.15 + robustness * 0.15 ) metrics = QualityMetrics( accuracy=accuracy, completeness=completeness, relevance=relevance, efficiency=efficiency, robustness=robustness, overall_score=overall_score ) # Store in history self.quality_history.append({ 'timestamp': time.time(), 'query': query, 'metrics': metrics, 'results': results }) logger.info(f"Quality assessment: {overall_score:.3f} (A:{accuracy:.2f} C:{completeness:.2f} R:{relevance:.2f} E:{efficiency:.2f} R:{robustness:.2f})") return metrics def _assess_understanding_accuracy(self, query: str, results: Dict[str, Any]) -> float: """Assess how accurately we understood the query.""" query_info = results.get('query', {}) # Base score from confidence confidence = query_info.get('confidence', 0.5) # Bonus for extracting keywords keywords = query_info.get('keywords', []) keyword_bonus = min(0.2, len(keywords) * 0.02) # Bonus for correct intent classification intent = query_info.get('intent', '') intent_bonus = 0.1 if intent and intent != 'unknown' else 0 return min(1.0, confidence + keyword_bonus + intent_bonus) def _assess_result_completeness(self, results: Dict[str, Any]) -> float: """Assess completeness of results.""" api_results = results.get('results', {}) executed_count = api_results.get('executed_count', 0) if executed_count == 0: return 0.0 # Base score from execution count base_score = min(1.0, executed_count / 5.0) # Normalize to 5 API calls # Bonus for successful executions successful_count = api_results.get('successful_calls', 0) success_rate = successful_count / executed_count if executed_count > 0 else 0 return base_score * success_rate def _assess_api_relevance(self, results: Dict[str, Any]) -> float: """Assess relevance of matched APIs.""" api_matches = results.get('api_matches', {}) matches = api_matches.get('matches', []) if not matches: return 0.0 # Average confidence of matches confidences = [m.get('confidence', 0) for m in matches] avg_confidence = sum(confidences) / len(confidences) # Bonus for multiple high-confidence matches high_confidence_count = sum(1 for c in confidences if c > 0.8) diversity_bonus = min(0.2, high_confidence_count * 0.05) return min(1.0, avg_confidence + diversity_bonus) def _assess_processing_efficiency(self, results: Dict[str, Any]) -> float: """Assess processing efficiency.""" # Check if we have timing information metadata = results.get('metadata', {}) processing_time = metadata.get('processing_time', 1.0) # Efficiency based on processing time (lower is better) # Target: under 100ms for excellent, under 500ms for good if processing_time < 0.1: return 1.0 elif processing_time < 0.5: return 0.8 elif processing_time < 1.0: return 0.6 elif processing_time < 2.0: return 0.4 else: return 0.2 def _assess_error_handling(self, results: Dict[str, Any]) -> float: """Assess error handling robustness.""" api_results = results.get('results', {}) executed_count = api_results.get('executed_count', 0) successful_count = api_results.get('successful_calls', 0) if executed_count == 0: return 0.5 # Neutral score if no APIs executed # Success rate as base robustness measure success_rate = successful_count / executed_count # Check for graceful error handling results_data = api_results.get('data', []) error_handling_score = 0.0 for result in results_data: if not result.get('success', True): # Check if error was handled gracefully if result.get('error') and len(result.get('error', '')) > 0: error_handling_score += 0.1 # Points for proper error reporting return min(1.0, success_rate + error_handling_score) def get_quality_trends(self, window_size: int = 10) -> Dict[str, Any]: """Get quality trends over recent assessments.""" if len(self.quality_history) < 2: return {'trend': 'insufficient_data'} recent_history = self.quality_history[-window_size:] # Calculate trends scores = [h['metrics'].overall_score for h in recent_history] if len(scores) >= 2: trend = scores[-1] - scores[0] avg_score = sum(scores) / len(scores) return { 'trend': 'improving' if trend > 0.05 else 'declining' if trend < -0.05 else 'stable', 'trend_value': trend, 'average_score': avg_score, 'latest_score': scores[-1], 'assessment_count': len(recent_history) } return {'trend': 'insufficient_data'} def get_improvement_recommendations(self) -> List[str]: """Get recommendations for system improvement.""" if not self.quality_history: return ["Insufficient data for recommendations"] recommendations = [] recent_metrics = [h['metrics'] for h in self.quality_history[-5:]] # Analyze weak areas avg_accuracy = sum(m.accuracy for m in recent_metrics) / len(recent_metrics) avg_completeness = sum(m.completeness for m in recent_metrics) / len(recent_metrics) avg_relevance = sum(m.relevance for m in recent_metrics) / len(recent_metrics) avg_efficiency = sum(m.efficiency for m in recent_metrics) / len(recent_metrics) avg_robustness = sum(m.robustness for m in recent_metrics) / len(recent_metrics) if avg_accuracy < 0.7: recommendations.append("Improve natural language understanding - consider better models or training") if avg_completeness < 0.7: recommendations.append("Expand API operation database and improve matching algorithms") if avg_relevance < 0.7: recommendations.append("Enhance API relevance scoring and semantic matching") if avg_efficiency < 0.7: recommendations.append("Optimize processing pipeline for better performance") if avg_robustness < 0.7: recommendations.append("Strengthen error handling and retry mechanisms") if not recommendations: recommendations.append("System performing well - consider advanced optimizations") return recommendations