Coverage for src/local_deep_research/embeddings/splitters/text_splitter_registry.py: 99%
60 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +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
11from typing import Optional, List, Any, TYPE_CHECKING
13from loguru import logger
15from ...constants import (
16 DEFAULT_LOCAL_SEARCH_CHUNK_OVERLAP,
17 DEFAULT_LOCAL_SEARCH_CHUNK_SIZE,
18 DEFAULT_LOCAL_SEARCH_SPLITTER_TYPE,
19 DEFAULT_LOCAL_SEARCH_TEXT_SEPARATORS,
20)
22# ``langchain_text_splitters`` is imported lazily inside ``get_text_splitter``
23# (see below), NOT at module load. Importing *any* of its submodules runs the
24# package ``__init__``, which eagerly pulls sentence-transformers, torch,
25# spacy, nltk and konlpy — ~4-8 s and hundreds of MB of RSS. Because this
26# module sits on the app-startup import chain (scheduler / blueprints /
27# search engines all import ``LibraryRAGService`` → ``embeddings.splitters``),
28# importing it eagerly added ~19 s to server boot on CI and tripped the
29# UI-test startup watchdog. Deferring the import to call time keeps boot
30# cheap and also means a broken optional splitter dependency (spacy / konlpy)
31# can no longer crash startup — only the indexing path that actually needs
32# it (the original intent of issue #4490, which the concrete-submodule
33# imports did not actually achieve since they still run the package init).
35# ``langchain_text_splitters`` must always be imported parent-first before
36# importing any of its submodules. In CPython, concurrent submodule-first
37# imports can create import-lock order inversion and raise ``_DeadlockError``.
38#
39# Keep these imports function-local and lazy so the package stays off the
40# application startup path. The AST guard in the test suite enforces that
41# every source import of ``langchain_text_splitters.<submodule>`` has a
42# parent-package import earlier in the same executable scope.
44if TYPE_CHECKING:
45 from langchain_core.embeddings import Embeddings
47# Valid splitter type options
48VALID_SPLITTER_TYPES = [
49 "recursive",
50 "token",
51 "sentence",
52 "semantic",
53]
56def get_text_splitter(
57 splitter_type: str = DEFAULT_LOCAL_SEARCH_SPLITTER_TYPE,
58 chunk_size: int = DEFAULT_LOCAL_SEARCH_CHUNK_SIZE,
59 chunk_overlap: int = DEFAULT_LOCAL_SEARCH_CHUNK_OVERLAP,
60 text_separators: Optional[List[str]] = None,
61 embeddings: Optional[Embeddings] = None,
62 **kwargs,
63) -> Any:
64 """
65 Get text splitter based on type.
67 Args:
68 splitter_type: Type of splitter ('recursive', 'token', 'sentence', 'semantic')
69 chunk_size: Maximum size of chunks
70 chunk_overlap: Overlap between chunks
71 text_separators: Custom separators for recursive splitting and semantic
72 safety bounds
73 embeddings: Embeddings instance (required for 'semantic' type)
74 **kwargs: Additional splitter-specific parameters
76 Returns:
77 A text splitter instance
79 Raises:
80 ValueError: If splitter_type is invalid or required parameters are missing
81 ImportError: If required dependencies are not installed
82 """
83 # Normalize splitter type
84 splitter_type = splitter_type.strip().lower()
86 # Validate splitter type
87 if splitter_type not in VALID_SPLITTER_TYPES:
88 logger.error(f"Invalid splitter type: {splitter_type}")
89 raise ValueError(
90 f"Invalid splitter type: {splitter_type}. "
91 f"Must be one of: {VALID_SPLITTER_TYPES}"
92 )
94 logger.info(
95 f"Creating text splitter: type={splitter_type}, "
96 f"chunk_size={chunk_size}, chunk_overlap={chunk_overlap}"
97 )
99 # Create the appropriate splitter. The ``langchain_text_splitters``
100 # imports are intentionally function-local — see the module docstring:
101 # importing the package eagerly pulls torch/sentence-transformers and
102 # must stay off the app-startup import path.
103 #
104 # Enforce the parent-first import invariant before any submodule-level
105 # imports occur. Non-semantic branches warm it here; the semantic branch
106 # warms it below after dependency validation and before constructing its
107 # BoundedSemanticChunker.
108 if splitter_type != "semantic":
109 import langchain_text_splitters # noqa: F401
111 if splitter_type == "token":
112 from langchain_text_splitters.base import TokenTextSplitter
114 return TokenTextSplitter(
115 chunk_size=chunk_size,
116 chunk_overlap=chunk_overlap,
117 )
119 if splitter_type == "sentence":
120 from langchain_text_splitters.sentence_transformers import (
121 SentenceTransformersTokenTextSplitter,
122 )
124 try:
125 return SentenceTransformersTokenTextSplitter(
126 chunk_overlap=chunk_overlap,
127 tokens_per_chunk=chunk_size,
128 )
129 except ValueError as exc:
130 # SentenceTransformersTokenTextSplitter is backed by a fixed
131 # sentence-transformers tokenizer. It raises at construction,
132 # before any chunk is emitted, when tokens_per_chunk (our
133 # chunk_size) exceeds that tokenizer's max_seq_length. The raw
134 # error names an internal model the user never configured, which
135 # reads as an embedding failure. Translate only that specific cap
136 # error into a settings-side message that points at the chunk_size
137 # the user actually set; re-raise anything else unchanged.
138 #
139 # The cap is model-specific (LangChain reads it from
140 # model.max_seq_length; 384 is the default all-mpnet-base-v2
141 # value but not universal), so read it back from LangChain's own
142 # message rather than hardcoding a number that could silently lie
143 # if the tokenizer ever changes. See issue #4800.
144 message = str(exc)
145 if "maximum token limit" not in message:
146 raise
147 cap_match = re.search(r"is:\s*(\d+)", message)
148 if cap_match:
149 cap = cap_match.group(1)
150 limit_phrase = f"capped at {cap} tokens per chunk"
151 fix_phrase = f"Lower the chunk size to {cap} or less"
152 else:
153 limit_phrase = (
154 "capped at the tokenizer's maximum sequence length"
155 )
156 fix_phrase = "Lower the chunk size to fit that limit"
157 raise ValueError(
158 f"chunk_size={chunk_size} is too large for the 'sentence' "
159 f"chunking method: it uses a fixed sentence-transformers "
160 f"tokenizer {limit_phrase}. {fix_phrase}, or choose a "
161 f"different chunking method ('recursive', 'token', or "
162 f"'semantic') for larger chunks."
163 ) from exc
165 if splitter_type == "semantic":
166 # Semantic chunking requires embeddings
167 if embeddings is None:
168 raise ValueError(
169 "Semantic splitter requires 'embeddings' parameter. "
170 "Please provide an embeddings instance."
171 )
173 try:
174 from langchain_experimental.text_splitter import SemanticChunker
175 except ImportError as e:
176 logger.exception("Failed to import SemanticChunker")
177 raise ImportError(
178 "Semantic chunking requires langchain-experimental. "
179 "Install it with: pip install langchain-experimental"
180 ) from e
182 # Enforce the parent-first import invariant before constructing
183 # BoundedSemanticChunker.__init__ imports
184 # langchain_text_splitters.character, so warm the parent package first.
185 import langchain_text_splitters # noqa: F401
187 from .bounded_semantic_chunker import BoundedSemanticChunker
189 # Get breakpoint threshold from kwargs or use default
190 breakpoint_threshold_type = kwargs.get(
191 "breakpoint_threshold_type", "percentile"
192 )
193 breakpoint_threshold_amount = kwargs.get(
194 "breakpoint_threshold_amount", None
195 )
197 # Create semantic chunker
198 chunker_kwargs = {"embeddings": embeddings}
200 if breakpoint_threshold_type: 200 ↛ 205line 200 didn't jump to line 205 because the condition on line 200 was always true
201 chunker_kwargs["breakpoint_threshold_type"] = (
202 breakpoint_threshold_type
203 )
205 if breakpoint_threshold_amount is not None:
206 chunker_kwargs["breakpoint_threshold_amount"] = (
207 breakpoint_threshold_amount
208 )
210 logger.info(
211 f"Creating SemanticChunker with threshold_type={breakpoint_threshold_type}, "
212 f"threshold_amount={breakpoint_threshold_amount}"
213 )
215 semantic_chunker = SemanticChunker(**chunker_kwargs)
216 return BoundedSemanticChunker(
217 semantic_chunker=semantic_chunker,
218 chunk_size=chunk_size,
219 separators=text_separators,
220 )
222 # "recursive" or default
223 from langchain_text_splitters.character import (
224 RecursiveCharacterTextSplitter,
225 )
227 # Use custom separators if provided, otherwise use defaults
228 if text_separators is None:
229 text_separators = list(DEFAULT_LOCAL_SEARCH_TEXT_SEPARATORS)
231 return RecursiveCharacterTextSplitter(
232 chunk_size=chunk_size,
233 chunk_overlap=chunk_overlap,
234 length_function=len,
235 separators=text_separators,
236 )
239def is_semantic_chunker_available() -> bool:
240 """Check if semantic chunking is available."""
241 import importlib.util
243 return importlib.util.find_spec("langchain_experimental") is not None