Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-rag/src/lexigram/ai/rag/synthesis/context/deduplicator.py: 18%

1"""Context deduplicator for synthesis. 

2 

3This module implements context deduplication to remove redundant chunks 

4before synthesis. 

5""" 

6 

7from __future__ import annotations 

8 

9from typing import TYPE_CHECKING 

10 

11if TYPE_CHECKING: 

12 from lexigram.contracts.ai import EmbeddingClientProtocol 

13 

14from lexigram.ai.rag.synthesis.types import ContextChunk 

15 

16 

17class ContextDeduplicator: 

18 """Remove redundant context chunks. 

19 

20 This component identifies and removes duplicate or highly similar chunks 

21 to reduce redundancy and token usage. 

22 

23 Attributes: 

24 similarity_threshold: Threshold for considering chunks similar (0-1) 

25 use_embeddings: Whether to use embeddings for similarity 

26 embedding_client: Optional embedding client 

27 """ 

28 

29 def __init__( 

30 self, 

31 similarity_threshold: float = 0.9, 

32 use_embeddings: bool = False, 

33 embedding_client: EmbeddingClientProtocol | None = None, 

34 ): 

35 """Initialize the context deduplicator. 

36 

37 Args: 

38 similarity_threshold: Similarity threshold 

39 use_embeddings: Whether to use embeddings 

40 embedding_client: Optional embedding client 

41 """ 

42 self.similarity_threshold = similarity_threshold 

43 self.use_embeddings = use_embeddings 

44 self.embedding_client = embedding_client 

45 

46 def _calculate_text_similarity(self, text1: str, text2: str) -> float: 

47 """Calculate text similarity using simple overlap. 

48 

49 Args: 

50 text1: First text 

51 text2: Second text 

52 

53 Returns: 

54 Similarity score (0-1) 

55 """ 

56 # Simple word-based Jaccard similarity 

57 import re 

58 

59 words1 = set(re.findall(r"\b\w+\b", text1.lower())) 

60 words2 = set(re.findall(r"\b\w+\b", text2.lower())) 

61 

62 if not words1 or not words2: 

63 return 0.0 

64 

65 intersection = len(words1 & words2) 

66 union = len(words1 | words2) 

67 

68 return intersection / union if union > 0 else 0.0 

69 

70 async def deduplicate_chunks( 

71 self, 

72 chunks: list[ContextChunk], 

73 ) -> list[ContextChunk]: 

74 """Remove duplicate chunks. 

75 

76 Args: 

77 chunks: Chunks to deduplicate 

78 

79 Returns: 

80 Deduplicated list of chunks 

81 """ 

82 if not chunks: 

83 return [] 

84 

85 unique_chunks: list[ContextChunk] = [] 

86 seen_texts: set[str] = set() 

87 

88 for chunk in chunks: 

89 # Check exact duplicates 

90 if chunk.text in seen_texts: 

91 continue 

92 

93 # Check similarity to existing chunks 

94 is_duplicate = False 

95 

96 for unique_chunk in unique_chunks: 

97 similarity = self._calculate_text_similarity( 

98 chunk.text, 

99 unique_chunk.text, 

100 ) 

101 

102 if similarity >= self.similarity_threshold: 

103 # Keep the one with higher score 

104 chunk_score = chunk.score if chunk.score is not None else 0.0 

105 unique_score = ( 

106 unique_chunk.score if unique_chunk.score is not None else 0.0 

107 ) 

108 if chunk_score > unique_score: 

109 unique_chunks.remove(unique_chunk) 

110 seen_texts.discard(unique_chunk.text) 

111 else: 

112 is_duplicate = True 

113 break 

114 

115 if not is_duplicate: 

116 unique_chunks.append(chunk) 

117 seen_texts.add(chunk.text) 

118 

119 return unique_chunks