Spaces:
Sleeping
Sleeping
File size: 1,858 Bytes
c96b98a | 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 | from typing import List
from phi.document.base import Document
from phi.document.chunking.strategy import ChunkingStrategy
class RecursiveChunking(ChunkingStrategy):
"""Chunking strategy that recursively splits text into chunks by finding natural break points"""
def __init__(self, chunk_size: int = 5000, overlap: int = 0):
# overlap must be lesser than chunk size
if overlap >= chunk_size:
raise ValueError(f"Invalid parameters: overlap ({overlap}) must be less than chunk size ({chunk_size}).")
self.chunk_size = chunk_size
self.overlap = overlap
def chunk(self, document: Document) -> List[Document]:
"""Recursively chunk text by finding natural break points"""
if len(document.content) <= self.chunk_size:
return [document]
chunks: List[Document] = []
start = 0
chunk_meta_data = document.meta_data
chunk_number = 1
content = self.clean_text(document.content)
while start < len(content):
end = min(start + self.chunk_size, len(content))
if end < len(content):
for sep in ["\n", "."]:
last_sep = content[start:end].rfind(sep)
if last_sep != -1:
end = start + last_sep + 1
break
chunk = content[start:end]
meta_data = chunk_meta_data.copy()
meta_data["chunk"] = chunk_number
chunk_id = None
if document.id:
chunk_id = f"{document.id}_{chunk_number}"
chunk_number += 1
meta_data["chunk_size"] = len(chunk)
chunks.append(Document(id=chunk_id, name=document.name, meta_data=meta_data, content=chunk))
start = max(start, end - self.overlap)
return chunks
|