Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-rag/src/lexigram/ai/rag/context_compression/semantic_dedup.py: 98%

1"""Semantic deduplication compression strategies.""" 

2 

3from __future__ import annotations 

4 

5from datetime import UTC, datetime 

6 

7from lexigram.ai.rag.context_compression.base import AbstractCompressor 

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

9 CompressionResult, 

10 CompressionStrategy, 

11) 

12 

13 

14class SemanticDeduplicationCompressor(AbstractCompressor): 

15 """Remove semantically redundant information. 

16 

17 Identifies and removes redundant sentences that convey 

18 the same information as other sentences. 

19 

20 Example: 

21 >>> compressor = SemanticDeduplicationCompressor( 

22 ... similarity_threshold=0.8 

23 ... ) 

24 >>> result = await compressor.compress(context_with_repetition) 

25 """ 

26 

27 def __init__( 

28 self, 

29 similarity_threshold: float = 0.8, 

30 preserve_first: bool = True, 

31 ): 

32 """Initialize semantic deduplication compressor. 

33 

34 Args: 

35 similarity_threshold: Threshold for considering sentences similar (0.0 to 1.0). 

36 preserve_first: Keep first occurrence of similar sentences. 

37 """ 

38 self.similarity_threshold = similarity_threshold 

39 self.preserve_first = preserve_first 

40 

41 async def compress( 

42 self, 

43 context: str | list[str], 

44 query: str | None = None, 

45 **kwargs, 

46 ) -> CompressionResult: 

47 """Compress by removing redundant sentences.""" 

48 original_text = self._normalize_context(context) 

49 original_tokens = self._estimate_tokens(original_text) 

50 

51 # Split into sentences 

52 sentences = self._split_sentences(original_text) 

53 

54 # Deduplicate 

55 unique_sentences = self._deduplicate_sentences(sentences) 

56 

57 # Join unique sentences 

58 compressed_text = " ".join(unique_sentences) 

59 

60 compressed_tokens = self._estimate_tokens(compressed_text) 

61 compression_ratio = ( 

62 compressed_tokens / original_tokens if original_tokens > 0 else 1.0 

63 ) 

64 

65 return CompressionResult( 

66 original_text=original_text, 

67 compressed_text=compressed_text, 

68 original_tokens=original_tokens, 

69 compressed_tokens=compressed_tokens, 

70 compression_ratio=compression_ratio, 

71 strategy=CompressionStrategy.SEMANTIC_DEDUP, 

72 metadata={ 

73 "original_sentences": len(sentences), 

74 "unique_sentences": len(unique_sentences), 

75 "removed_duplicates": len(sentences) - len(unique_sentences), 

76 "similarity_threshold": self.similarity_threshold, 

77 "timestamp": datetime.now(UTC).isoformat(), 

78 }, 

79 ) 

80 

81 def _split_sentences(self, text: str) -> list[str]: 

82 """Split text into sentences.""" 

83 import re 

84 

85 sentences = re.split(r"[.!?]+\s+", text) 

86 return list(map(str.strip, filter(str.strip, sentences))) 

87 

88 def _deduplicate_sentences(self, sentences: list[str]) -> list[str]: 

89 """Remove duplicate/similar sentences.""" 

90 unique = [] 

91 seen_fingerprints: set[str] = set() 

92 

93 for sentence in sentences: 

94 # Create simple fingerprint (for production, use embeddings) 

95 fingerprint = self._create_fingerprint(sentence) 

96 

97 # Check if similar to any seen fingerprint 

98 is_duplicate = False 

99 for seen_fp in seen_fingerprints: 

100 if self._are_similar(fingerprint, seen_fp): 

101 is_duplicate = True 

102 break 

103 

104 if not is_duplicate: 

105 unique.append(sentence) 

106 seen_fingerprints.add(fingerprint) 

107 

108 return unique 

109 

110 def _create_fingerprint(self, sentence: str) -> str: 

111 """Create sentence fingerprint. 

112 

113 Simple approach: normalized word set. 

114 For production, use embeddings. 

115 """ 

116 # Lowercase and split 

117 words = sentence.lower().split() 

118 

119 # Remove stopwords 

120 stopwords = { 

121 "the", 

122 "a", 

123 "an", 

124 "is", 

125 "are", 

126 "was", 

127 "were", 

128 "in", 

129 "on", 

130 "at", 

131 "to", 

132 "for", 

133 "of", 

134 "and", 

135 "or", 

136 "but", 

137 } 

138 words = list(filter(lambda w: w not in stopwords, words)) 

139 

140 # Sort for comparison 

141 return " ".join(sorted(words)) 

142 

143 def _are_similar(self, fp1: str, fp2: str) -> bool: 

144 """Check if two fingerprints are similar.""" 

145 words1 = set(fp1.split()) 

146 words2 = set(fp2.split()) 

147 

148 if not words1 or not words2: 

149 return False 

150 

151 # Jaccard similarity 

152 intersection = len(words1 & words2) 

153 union = len(words1 | words2) 

154 

155 similarity = intersection / union if union > 0 else 0.0 

156 

157 return similarity >= self.similarity_threshold