Spaces:
Running
Running
File size: 2,744 Bytes
09801ca | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 | # Document processor module
"""
Text processing with support for both fixed-size and semantic chunking.
Default: Semantic chunking for better RAG accuracy.
"""
import re
from typing import List
class Processor:
"""
Document processor with configurable chunking strategy.
Strategies:
- "semantic": Sentence-boundary aware chunking (default, better for RAG)
- "fixed": Fixed-size character chunking (legacy)
"""
def __init__(self, size: int = 700, overlap: int = 150, strategy: str = "semantic"):
"""
Args:
size: Max chunk size in characters
overlap: Overlap size (characters for fixed, sentences for semantic)
strategy: "semantic" or "fixed"
"""
self.size = size
self.overlap = overlap
self.strategy = strategy
# Initialize semantic chunker if needed
self._semantic_chunker = None
if strategy == "semantic":
try:
from ingestion.semantic_chunker import SemanticChunker
self._semantic_chunker = SemanticChunker(
max_chunk_size=size,
min_chunk_size=100,
overlap_sentences=max(1, overlap // 100),
preserve_tables=True
)
except ImportError:
print("⚠️ Semantic chunker not available, using fixed strategy")
self.strategy = "fixed"
@staticmethod
def clean(text: str) -> str:
"""Clean text while preserving paragraph structure"""
# Normalize whitespace but keep paragraph breaks
text = re.sub(r'[ \t]+', ' ', text)
text = re.sub(r'\n{3,}', '\n\n', text)
return text.strip()
def chunk(self, text: str) -> List[str]:
"""
Split text into chunks using configured strategy
Args:
text: Document text to chunk
Returns:
List of text chunks
"""
if not text or not text.strip():
return []
if self.strategy == "semantic" and self._semantic_chunker:
return self._semantic_chunk(text)
else:
return self._fixed_chunk(text)
def _semantic_chunk(self, text: str) -> List[str]:
"""Semantic chunking using sentence boundaries"""
return self._semantic_chunker.chunk_texts(text)
def _fixed_chunk(self, text: str) -> List[str]:
"""Legacy fixed-size chunking"""
chunks = []
start = 0
while start < len(text):
end = start + self.size
chunks.append(text[start:end])
start += self.size - self.overlap
return chunks
|