Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-rag/src/lexigram/ai/rag/synthesis/quality/relevance.py: 29%

1"""Relevance filter for response quality. 

2 

3This module implements relevance filtering to verify that synthesized 

4responses answer the original query. 

5""" 

6 

7from __future__ import annotations 

8 

9import re 

10from typing import TYPE_CHECKING 

11 

12if TYPE_CHECKING: 

13 from lexigram.contracts.ai import EmbeddingClientProtocol 

14 

15from lexigram.logging import ( 

16 get_logger, 

17) 

18 

19logger = get_logger(__name__) 

20 

21 

22class RelevanceFilter: 

23 """Filter responses by relevance to query. 

24 

25 This component checks if the response actually answers the query. 

26 

27 Attributes: 

28 threshold: Relevance threshold (0-1) 

29 embedding_client: Optional embedding client for semantic similarity 

30 """ 

31 

32 def __init__( 

33 self, 

34 threshold: float = 0.6, 

35 embedding_client: EmbeddingClientProtocol | None = None, 

36 ): 

37 """Initialize the relevance filter. 

38 

39 Args: 

40 threshold: Relevance threshold 

41 embedding_client: Optional embedding client for semantic check 

42 """ 

43 self.threshold = threshold 

44 self.embedding_client = embedding_client 

45 

46 def _extract_keywords(self, text: str) -> set[str]: 

47 """Extract keywords from text. 

48 

49 Args: 

50 text: Input text 

51 

52 Returns: 

53 Set of keywords 

54 """ 

55 words = re.findall(r"\b\w+\b", text.lower()) 

56 

57 stop_words = { 

58 "the", 

59 "a", 

60 "an", 

61 "and", 

62 "or", 

63 "but", 

64 "in", 

65 "on", 

66 "at", 

67 "to", 

68 "for", 

69 "of", 

70 "with", 

71 "by", 

72 "from", 

73 "as", 

74 "is", 

75 "was", 

76 "are", 

77 "been", 

78 "be", 

79 "have", 

80 "has", 

81 "had", 

82 "do", 

83 "does", 

84 "did", 

85 } 

86 

87 return {w for w in words if w not in stop_words and len(w) > 2} 

88 

89 async def check_relevance( 

90 self, 

91 query: str, 

92 response: str, 

93 ) -> float: 

94 """Check response relevance to query. 

95 

96 Args: 

97 query: The original query 

98 response: The synthesized response 

99 

100 Returns: 

101 Relevance score (0-1) 

102 """ 

103 if not query or not response: 

104 return 0.0 

105 

106 # If embedding client available, use semantic similarity 

107 if self.embedding_client: 

108 try: 

109 query_emb = await self.embedding_client.embed([query]) 

110 response_emb = await self.embedding_client.embed([response]) 

111 

112 # Cosine similarity 

113 import numpy as np 

114 

115 similarity = float( 

116 np.dot(query_emb, response_emb) 

117 / (np.linalg.norm(query_emb) * np.linalg.norm(response_emb)), 

118 ) 

119 return max(0.0, min(1.0, similarity)) 

120 except (ValueError, TypeError, RuntimeError) as e: 

121 logger.debug("Embedding similarity failed: %s", e) 

122 # Fall back to keyword matching 

123 

124 # Keyword-based relevance 

125 query_keywords = self._extract_keywords(query) 

126 response_keywords = self._extract_keywords(response) 

127 

128 if not query_keywords: 

129 return 1.0 

130 

131 overlap = len(query_keywords & response_keywords) 

132 return overlap / len(query_keywords) 

133 

134 async def is_relevant( 

135 self, 

136 query: str, 

137 response: str, 

138 ) -> bool: 

139 """Check if response meets relevance threshold. 

140 

141 Args: 

142 query: The original query 

143 response: The synthesized response 

144 

145 Returns: 

146 True if response is relevant 

147 """ 

148 score = await self.check_relevance(query, response) 

149 return score >= self.threshold