Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-rag/src/lexigram/ai/rag/chunking/strategies/recursive.py: 18%

1"""Recursive chunking strategy.""" 

2 

3from __future__ import annotations 

4 

5import re 

6from typing import Any 

7 

8from lexigram.ai.rag.chunking.base import AbstractChunker 

9from lexigram.ai.rag.chunking.types import Chunk 

10 

11 

12class RecursiveChunker(AbstractChunker): 

13 """Recursive chunking using multiple separators. 

14 

15 Tries to split on separators in order of preference (e.g., paragraphs, then 

16 sentences, then words) to maintain semantic coherence. 

17 

18 Example: 

19 >>> chunker = RecursiveChunker( 

20 ... chunk_size=1000, 

21 ... separators=[r'\\n\\n', r'\\n', r'\\. ', r' '] 

22 ... ) 

23 >>> chunks = chunker.chunk("Document text...") 

24 """ 

25 

26 def __init__( 

27 self, 

28 chunk_size: int = 1000, 

29 overlap: int = 200, 

30 separators: list[str] | None = None, 

31 is_regex: bool = True, 

32 ): 

33 """Initialize recursive chunker. 

34 

35 Args: 

36 chunk_size: Target chunk size in characters 

37 overlap: Overlap between chunks 

38 separators: List of separators to try (in order of preference) 

39 is_regex: Whether separators are regex patterns 

40 """ 

41 self.chunk_size = chunk_size 

42 self.overlap = overlap 

43 self.is_regex = is_regex 

44 

45 # Default separators: paragraph, line, sentence, word 

46 self.separators = separators or [ 

47 r"\n\n", # Paragraph 

48 r"\n", # Line 

49 r"\.\s", # Sentence (period + space) 

50 r" ", # Word 

51 ] 

52 

53 if overlap >= chunk_size: 

54 msg = "overlap must be less than chunk_size" 

55 raise ValueError(msg) 

56 

57 def chunk(self, text: str, metadata: dict[str, Any] | None = None) -> list[Chunk]: 

58 """Split text recursively. 

59 

60 Args: 

61 text: Text to chunk 

62 metadata: Optional metadata 

63 

64 Returns: 

65 List of chunks 

66 """ 

67 if not text: 

68 return [] 

69 

70 chunks: list[Chunk] = [] 

71 chunk_index = 0 

72 step = self.chunk_size - self.overlap 

73 

74 if step <= 0: 

75 step = 1 

76 

77 position = 0 

78 while position < len(text): 

79 end = min(position + self.chunk_size, len(text)) 

80 

81 # Try to find a good separator position 

82 if end < len(text): 

83 best_split = end 

84 for sep in self.separators: 

85 if self.is_regex: 

86 # Find last match of separator before end 

87 matches = list(re.finditer(sep, text[position:end])) 

88 if matches: 

89 best_split = position + matches[-1].end() 

90 break 

91 else: 

92 pos = text.rfind(sep, position, end) 

93 if pos > position: 

94 best_split = pos + len(sep) 

95 break 

96 end = best_split 

97 

98 chunk_text = text[position:end] 

99 if chunk_text.strip(): 

100 chunks.append( 

101 Chunk( 

102 text=chunk_text, 

103 start_index=position, 

104 end_index=end, 

105 chunk_index=chunk_index, 

106 metadata=metadata, 

107 ), 

108 ) 

109 chunk_index += 1 

110 

111 position += step 

112 

113 return chunks