Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-rag/src/lexigram/ai/rag/pipeline/stages/quality.py: 21%

1"""Quality assurance stage for response validation.""" 

2 

3from __future__ import annotations 

4 

5from lexigram.ai.rag.config import QualityAssuranceConfig 

6from lexigram.ai.rag.pipeline.types import PipelineContext 

7from lexigram.ai.rag.synthesis import ( 

8 ConfidenceScorer, 

9 FaithfulnessChecker, 

10 HallucinationChecker, 

11 QualityMetrics, 

12 RelevanceFilter, 

13) 

14from lexigram.logging import ( 

15 get_logger, 

16) 

17 

18logger = get_logger(__name__) 

19 

20 

21class QualityAssuranceStage: 

22 """Pipeline stage for response quality assurance. 

23 

24 This stage validates the synthesized response against quality 

25 thresholds including faithfulness, relevance, and confidence. 

26 """ 

27 

28 def __init__(self, config: QualityAssuranceConfig): 

29 """Initialize the quality assurance stage. 

30 

31 Args: 

32 config: Quality assurance configuration 

33 """ 

34 self.config = config 

35 self.faithfulness_checker = FaithfulnessChecker() 

36 self.relevance_filter = RelevanceFilter() 

37 self.hallucination_detector = HallucinationChecker( 

38 strict_mode=config.hallucination_strict_mode, 

39 ) 

40 self.confidence_scorer = ConfidenceScorer() 

41 

42 @property 

43 def name(self) -> str: 

44 """Get stage name.""" 

45 return "quality_assurance" 

46 

47 async def process(self, context: PipelineContext) -> PipelineContext: 

48 """Process the quality assurance stage. 

49 

50 Args: 

51 context: Pipeline context with synthesis result 

52 

53 Returns: 

54 Updated context with quality metrics 

55 """ 

56 if not self.config.enabled: 

57 logger.info("Quality assurance stage disabled, skipping") 

58 return context 

59 

60 # Check if we have a synthesis result 

61 if context.synthesis_result is None: 

62 logger.warning("No synthesis result available for quality check") 

63 context.add_warning("No synthesis result for quality assurance") 

64 return context 

65 

66 result = context.synthesis_result 

67 query = result.query 

68 response = result.response 

69 chunks = result.context_chunks 

70 

71 try: 

72 logger.info( 

73 "Starting quality assurance", 

74 extra={ 

75 "request_id": context.request_id, 

76 "response_length": len(response), 

77 }, 

78 ) 

79 

80 # Calculate quality metrics 

81 metrics = await self.confidence_scorer.calculate_quality_metrics( 

82 query, 

83 response, 

84 chunks, 

85 ) 

86 

87 # Store metrics in context 

88 context.quality_metrics = metrics 

89 

90 # Also update synthesis result 

91 result.quality_metrics = metrics 

92 

93 logger.info( 

94 "Quality metrics calculated", 

95 extra={ 

96 "request_id": context.request_id, 

97 "faithfulness": metrics.faithfulness, 

98 "relevance": metrics.relevance, 

99 "coherence": metrics.coherence, 

100 "confidence": metrics.confidence, 

101 "has_hallucinations": metrics.has_hallucinations, 

102 }, 

103 ) 

104 

105 # Check quality thresholds 

106 quality_issues = self._check_quality_thresholds(metrics) 

107 

108 if quality_issues: 

109 logger.warning( 

110 "Quality issues detected", 

111 extra={ 

112 "request_id": context.request_id, 

113 "issues": quality_issues, 

114 }, 

115 ) 

116 

117 for issue in quality_issues: 

118 context.add_warning(f"Quality issue: {issue}") 

119 

120 # Handle low quality response 

121 if self.config.reject_low_quality: 

122 msg = ( 

123 f"Response quality below threshold: {', '.join(quality_issues)}" 

124 ) 

125 raise ValueError( 

126 msg, 

127 ) 

128 

129 except Exception as e: 

130 logger.exception( 

131 "Quality assurance failed", 

132 extra={ 

133 "request_id": context.request_id, 

134 "error": str(e), 

135 }, 

136 ) 

137 raise 

138 

139 return context 

140 

141 def _check_quality_thresholds(self, metrics: QualityMetrics) -> list[str]: 

142 """Check if quality metrics meet configured thresholds. 

143 

144 Args: 

145 metrics: Quality metrics to check 

146 

147 Returns: 

148 List of quality issues (empty if all thresholds met) 

149 """ 

150 issues = [] 

151 

152 if metrics.faithfulness < self.config.min_faithfulness: 

153 issues.append( 

154 f"Low faithfulness: {metrics.faithfulness:.2f} " 

155 f"< {self.config.min_faithfulness:.2f}", 

156 ) 

157 

158 if metrics.relevance < self.config.min_relevance: 

159 issues.append( 

160 f"Low relevance: {metrics.relevance:.2f} " 

161 f"< {self.config.min_relevance:.2f}", 

162 ) 

163 

164 if metrics.confidence < self.config.min_confidence: 

165 issues.append( 

166 f"Low confidence: {metrics.confidence:.2f} " 

167 f"< {self.config.min_confidence:.2f}", 

168 ) 

169 

170 if self.config.hallucination_detection_enabled and metrics.has_hallucinations: 

171 issues.append( 

172 f"Hallucinations detected: {metrics.hallucination_count} claims", 

173 ) 

174 

175 return issues