Coverage for src/local_deep_research/embeddings/splitters/text_splitter_registry.py: 99%
59 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-20 01:24 +0000
1"""
2Central registry for text splitters.
4This module provides a factory function to create different types of text splitters
5based on configuration, similar to how embeddings_config.py works for embeddings.
6"""
8from __future__ import annotations
10import re
11import threading
12from typing import Optional, List, Any, TYPE_CHECKING
14from loguru import logger
16# ``langchain_text_splitters`` is imported lazily inside ``get_text_splitter``
17# (see below), NOT at module load. Importing *any* of its submodules runs the
18# package ``__init__``, which eagerly pulls sentence-transformers, torch,
19# spacy, nltk and konlpy — ~4-8 s and hundreds of MB of RSS. Because this
20# module sits on the app-startup import chain (scheduler / blueprints /
21# search engines all import ``LibraryRAGService`` → ``embeddings.splitters``),
22# importing it eagerly added ~19 s to server boot on CI and tripped the
23# UI-test startup watchdog. Deferring the import to call time keeps boot
24# cheap and also means a broken optional splitter dependency (spacy / konlpy)
25# can no longer crash startup — only the indexing path that actually needs
26# it (the original intent of issue #4490, which the concrete-submodule
27# imports did not actually achieve since they still run the package init).
29# Serializes the cold ``langchain_text_splitters`` import. Its package
30# ``__init__`` eagerly imports ~14 splitter submodules with enough internal
31# cross-referencing that importing different submodules from multiple threads
32# at once observes a partially-initialized package and raises
33# ``ImportError: cannot import name ... from partially initialized module``.
34# The module-level import this replaced warmed ``sys.modules`` single-threaded
35# at boot; deferring it to call time reintroduced the race for the RAG
36# auto-index ``ThreadPoolExecutor`` and the per-user document scheduler, which
37# call ``get_text_splitter`` from several threads. Warm the package once under
38# this lock before the function-local submodule imports; once warm the lock is
39# uncontended (a ``sys.modules`` hit).
40_LANGCHAIN_TEXT_SPLITTERS_IMPORT_LOCK = threading.Lock()
42if TYPE_CHECKING:
43 from langchain_core.embeddings import Embeddings
45# Valid splitter type options
46VALID_SPLITTER_TYPES = [
47 "recursive",
48 "token",
49 "sentence",
50 "semantic",
51]
54def get_text_splitter(
55 splitter_type: str = "recursive",
56 chunk_size: int = 1000,
57 chunk_overlap: int = 200,
58 text_separators: Optional[List[str]] = None,
59 embeddings: Optional[Embeddings] = None,
60 **kwargs,
61) -> Any:
62 """
63 Get text splitter based on type.
65 Args:
66 splitter_type: Type of splitter ('recursive', 'token', 'sentence', 'semantic')
67 chunk_size: Maximum size of chunks
68 chunk_overlap: Overlap between chunks
69 text_separators: Custom separators (only used for 'recursive' type)
70 embeddings: Embeddings instance (required for 'semantic' type)
71 **kwargs: Additional splitter-specific parameters
73 Returns:
74 A text splitter instance
76 Raises:
77 ValueError: If splitter_type is invalid or required parameters are missing
78 ImportError: If required dependencies are not installed
79 """
80 # Normalize splitter type
81 splitter_type = splitter_type.strip().lower()
83 # Validate splitter type
84 if splitter_type not in VALID_SPLITTER_TYPES:
85 logger.error(f"Invalid splitter type: {splitter_type}")
86 raise ValueError(
87 f"Invalid splitter type: {splitter_type}. "
88 f"Must be one of: {VALID_SPLITTER_TYPES}"
89 )
91 logger.info(
92 f"Creating text splitter: type={splitter_type}, "
93 f"chunk_size={chunk_size}, chunk_overlap={chunk_overlap}"
94 )
96 # Create the appropriate splitter. The ``langchain_text_splitters``
97 # imports are intentionally function-local — see the module docstring:
98 # importing the package eagerly pulls torch/sentence-transformers and
99 # must stay off the app-startup import path.
100 #
101 # Every branch except "semantic" (which uses langchain_experimental)
102 # imports from ``langchain_text_splitters``. Warm the whole package once
103 # under the lock first so concurrent first-callers can't observe a
104 # half-initialized package (see the lock's definition); after this the
105 # per-branch ``from ... import`` are plain ``sys.modules`` lookups.
106 if splitter_type != "semantic":
107 with _LANGCHAIN_TEXT_SPLITTERS_IMPORT_LOCK:
108 import langchain_text_splitters # noqa: F401
110 if splitter_type == "token":
111 from langchain_text_splitters.base import TokenTextSplitter
113 return TokenTextSplitter(
114 chunk_size=chunk_size,
115 chunk_overlap=chunk_overlap,
116 )
118 if splitter_type == "sentence":
119 from langchain_text_splitters.sentence_transformers import (
120 SentenceTransformersTokenTextSplitter,
121 )
123 try:
124 return SentenceTransformersTokenTextSplitter(
125 chunk_overlap=chunk_overlap,
126 tokens_per_chunk=chunk_size,
127 )
128 except ValueError as exc:
129 # SentenceTransformersTokenTextSplitter is backed by a fixed
130 # sentence-transformers tokenizer. It raises at construction,
131 # before any chunk is emitted, when tokens_per_chunk (our
132 # chunk_size) exceeds that tokenizer's max_seq_length. The raw
133 # error names an internal model the user never configured, which
134 # reads as an embedding failure. Translate only that specific cap
135 # error into a settings-side message that points at the chunk_size
136 # the user actually set; re-raise anything else unchanged.
137 #
138 # The cap is model-specific (LangChain reads it from
139 # model.max_seq_length; 384 is the default all-mpnet-base-v2
140 # value but not universal), so read it back from LangChain's own
141 # message rather than hardcoding a number that could silently lie
142 # if the tokenizer ever changes. See issue #4800.
143 message = str(exc)
144 if "maximum token limit" not in message:
145 raise
146 cap_match = re.search(r"is:\s*(\d+)", message)
147 if cap_match:
148 cap = cap_match.group(1)
149 limit_phrase = f"capped at {cap} tokens per chunk"
150 fix_phrase = f"Lower the chunk size to {cap} or less"
151 else:
152 limit_phrase = (
153 "capped at the tokenizer's maximum sequence length"
154 )
155 fix_phrase = "Lower the chunk size to fit that limit"
156 raise ValueError(
157 f"chunk_size={chunk_size} is too large for the 'sentence' "
158 f"chunking method: it uses a fixed sentence-transformers "
159 f"tokenizer {limit_phrase}. {fix_phrase}, or choose a "
160 f"different chunking method ('recursive', 'token', or "
161 f"'semantic') for larger chunks."
162 ) from exc
164 if splitter_type == "semantic":
165 # Semantic chunking requires embeddings
166 if embeddings is None:
167 raise ValueError(
168 "Semantic splitter requires 'embeddings' parameter. "
169 "Please provide an embeddings instance."
170 )
172 try:
173 # Try to import experimental semantic chunker
174 from langchain_experimental.text_splitter import SemanticChunker
176 # Get breakpoint threshold from kwargs or use default
177 breakpoint_threshold_type = kwargs.get(
178 "breakpoint_threshold_type", "percentile"
179 )
180 breakpoint_threshold_amount = kwargs.get(
181 "breakpoint_threshold_amount", None
182 )
184 # Create semantic chunker
185 chunker_kwargs = {"embeddings": embeddings}
187 if breakpoint_threshold_type: 187 ↛ 192line 187 didn't jump to line 192 because the condition on line 187 was always true
188 chunker_kwargs["breakpoint_threshold_type"] = (
189 breakpoint_threshold_type
190 )
192 if breakpoint_threshold_amount is not None:
193 chunker_kwargs["breakpoint_threshold_amount"] = (
194 breakpoint_threshold_amount
195 )
197 logger.info(
198 f"Creating SemanticChunker with threshold_type={breakpoint_threshold_type}, "
199 f"threshold_amount={breakpoint_threshold_amount}"
200 )
202 return SemanticChunker(**chunker_kwargs)
204 except ImportError as e:
205 logger.exception("Failed to import SemanticChunker")
206 raise ImportError(
207 "Semantic chunking requires langchain-experimental. "
208 "Install it with: pip install langchain-experimental"
209 ) from e
211 else: # "recursive" or default
212 from langchain_text_splitters.character import (
213 RecursiveCharacterTextSplitter,
214 )
216 # Use custom separators if provided, otherwise use defaults
217 if text_separators is None:
218 text_separators = ["\n\n", "\n", ". ", " ", ""]
220 return RecursiveCharacterTextSplitter(
221 chunk_size=chunk_size,
222 chunk_overlap=chunk_overlap,
223 length_function=len,
224 separators=text_separators,
225 )
228def is_semantic_chunker_available() -> bool:
229 """Check if semantic chunking is available."""
230 import importlib.util
232 return importlib.util.find_spec("langchain_experimental") is not None