Spaces:
Running
Running
File size: 2,053 Bytes
e70c280 | 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 | """
File: /text_chunker.py
Created Date: Sunday October 19th 2025
Author: Christian Nonis <alch.infoemail@gmail.com>
-----
Last Modified: Sunday October 19th 2025 8:52:43 am
Modified By: the developer formerly known as Christian Nonis at <alch.infoemail@gmail.com>
-----
"""
import gc
import threading
from typing import List
import chunker_cpp
# Prevent multiple threads from accessing the chunker at the same time.
_chunker_lock = threading.Lock()
def chunk_text(
text: str,
max_chunk_size: int = 2000,
min_chunk_size: int = 500,
min_coherence_threshold: float = 0.3,
) -> List[str]:
"""
Chunk text into smaller chunks.
Args:
text: The text to chunk.
max_chunk_size: The maximum size of each chunk.
min_chunk_size: The minimum size of each chunk.
min_coherence_threshold: The minimum coherence threshold for the chunks.
Returns:
A list of chunks.
Raises:
Exception: If the chunking fails.
"""
try:
if not text or len(text.strip()) == 0:
return [text] if text else []
with _chunker_lock:
try:
result = chunker_cpp.chunk_text_semantically(
text, max_chunk_size, min_chunk_size, min_coherence_threshold
)
# Force garbage collection after C++ call
gc.collect()
if not result or not isinstance(result, list):
return [text]
return result
except Exception as cpp_error:
print(f"C++ chunker error: {cpp_error}")
raise
except Exception as e: # pylint: disable=broad-exception-caught
print(f"Chunker failed: {e}, falling back to simple chunking")
chunk_size = min(max_chunk_size, 2000)
chunks = []
for i in range(0, len(text), chunk_size):
chunk = text[i : i + chunk_size]
if chunk.strip():
chunks.append(chunk)
return chunks if chunks else [text]
|