Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-rag/src/lexigram/ai/rag/evaluation/evaluator.py: 95%

1"""Main RAG evaluation orchestrator.""" 

2 

3from __future__ import annotations 

4 

5from typing import Any 

6 

7from lexigram.ai.rag.evaluation.base import EvaluatorBase 

8from lexigram.ai.rag.evaluation.types import ( 

9 EvaluationResult, 

10 MetricType, 

11 RAGEvaluationReport, 

12) 

13 

14 

15class RAGEvaluator: 

16 """Comprehensive RAG evaluation orchestrator. 

17 

18 Coordinates multiple evaluators to produce a complete evaluation report. 

19 """ 

20 

21 def __init__( 

22 self, 

23 evaluators: list[EvaluatorBase] | None = None, 

24 weights: dict[MetricType, float] | None = None, 

25 ): 

26 """Initialize RAG evaluator. 

27 

28 Args: 

29 evaluators: List of evaluators to use. 

30 weights: Optional weights for computing overall score. 

31 """ 

32 self.evaluators = evaluators or [] 

33 self.weights = weights or {} 

34 

35 def add_evaluator(self, evaluator: EvaluatorBase) -> Any: 

36 """Add an evaluator. 

37 

38 Args: 

39 evaluator: Evaluator to add. 

40 """ 

41 self.evaluators.append(evaluator) 

42 

43 async def evaluate( 

44 self, 

45 query: str, 

46 retrieved_docs: list[Any], 

47 generated_answer: str, 

48 reference_answer: str | None = None, 

49 **kwargs, 

50 ) -> RAGEvaluationReport: 

51 """Run all evaluators and produce a report. 

52 

53 Args: 

54 query: The query. 

55 retrieved_docs: Retrieved documents. 

56 generated_answer: Generated answer. 

57 reference_answer: Optional ground truth. 

58 **kwargs: Additional parameters for evaluators. 

59 

60 Returns: 

61 Complete evaluation report. 

62 """ 

63 results = [] 

64 

65 # Run all evaluators 

66 for evaluator in self.evaluators: 

67 result = await evaluator.evaluate( 

68 query=query, 

69 retrieved_docs=retrieved_docs, 

70 generated_answer=generated_answer, 

71 reference_answer=reference_answer, 

72 **kwargs, 

73 ) 

74 results.append(result) 

75 

76 # Calculate overall score 

77 overall_score = self._calculate_overall_score(results) 

78 

79 return RAGEvaluationReport( 

80 query=query, 

81 retrieved_docs=retrieved_docs, 

82 generated_answer=generated_answer, 

83 reference_answer=reference_answer, 

84 results=results, 

85 overall_score=overall_score, 

86 metadata=kwargs.get("metadata", {}), 

87 ) 

88 

89 def _calculate_overall_score(self, results: list[EvaluationResult]) -> float: 

90 """Calculate weighted overall score. 

91 

92 Args: 

93 results: Individual evaluation results. 

94 

95 Returns: 

96 Overall score (0.0 to 1.0). 

97 """ 

98 if not results: 

99 return 0.0 

100 

101 # If no weights specified, use equal weights 

102 if not self.weights: 

103 # Invert hallucination rate (lower is better) 

104 total = 0.0 

105 count = 0 

106 for result in results: 

107 if result.metric_type == MetricType.HALLUCINATION_RATE: 

108 total += 1.0 - result.score # Invert 

109 else: 

110 total += result.score 

111 count += 1 

112 return total / count if count > 0 else 0.0 

113 

114 # Use weighted average 

115 weighted_sum = 0.0 

116 total_weight = 0.0 

117 

118 for result in results: 

119 weight = self.weights.get(result.metric_type, 0.0) 

120 if weight > 0: 

121 score = result.score 

122 # Invert hallucination rate 

123 if result.metric_type == MetricType.HALLUCINATION_RATE: 

124 score = 1.0 - score 

125 weighted_sum += score * weight 

126 total_weight += weight 

127 

128 return weighted_sum / total_weight if total_weight > 0 else 0.0