Coverage for src/qdrant_loader/core/chunking/chunking_service.py: 100%
69 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-20 10:15 +0000
1"""Service for chunking documents."""
3import logging
4from pathlib import Path
6from qdrant_loader.config import GlobalConfig, Settings
7from qdrant_loader.core.chunking.strategy import (
8 BaseChunkingStrategy,
9 CodeChunkingStrategy,
10 DefaultChunkingStrategy,
11 DoclingChunkingStrategy,
12 HTMLChunkingStrategy,
13 JSONChunkingStrategy,
14 MarkdownChunkingStrategy,
15)
16from qdrant_loader.core.document import Document
17from qdrant_loader.core.monitoring.ingestion_metrics import IngestionMonitor
18from qdrant_loader.utils.logging import LoggingConfig
21class ChunkingService:
22 """Service for chunking documents into smaller pieces."""
24 def __new__(cls, config: GlobalConfig, settings: Settings):
25 """Create a new instance of ChunkingService.
27 Args:
28 config: Global configuration
29 settings: Application settings
30 """
31 instance = super().__new__(cls)
32 instance.__init__(config, settings)
33 return instance
35 def __init__(self, config: GlobalConfig, settings: Settings):
36 """Initialize the chunking service.
38 Args:
39 config: Global configuration
40 settings: Application settings
41 """
42 self.config = config
43 self.settings = settings
44 self.validate_config()
45 self.logger = LoggingConfig.get_logger(__name__)
47 # Initialize metrics directory
48 metrics_dir = Path.cwd() / "metrics"
49 metrics_dir.mkdir(parents=True, exist_ok=True)
50 self.monitor = IngestionMonitor(str(metrics_dir.absolute()))
52 # Initialize strategies
53 self.strategies: dict[str, type[BaseChunkingStrategy]] = {
54 "md": MarkdownChunkingStrategy,
55 "html": HTMLChunkingStrategy,
56 # JSON files
57 "json": JSONChunkingStrategy,
58 # Programming languages
59 "py": CodeChunkingStrategy,
60 "java": CodeChunkingStrategy,
61 "js": CodeChunkingStrategy,
62 "ts": CodeChunkingStrategy,
63 "go": CodeChunkingStrategy,
64 "rs": CodeChunkingStrategy,
65 "cpp": CodeChunkingStrategy,
66 "c": CodeChunkingStrategy,
67 "cs": CodeChunkingStrategy,
68 "php": CodeChunkingStrategy,
69 "rb": CodeChunkingStrategy,
70 "kt": CodeChunkingStrategy,
71 "swift": CodeChunkingStrategy,
72 "scala": CodeChunkingStrategy,
73 # Add more strategies here as needed
74 }
76 # Default strategy for unknown file types
77 self.default_strategy = DefaultChunkingStrategy(settings=self.settings)
79 def validate_config(self) -> None:
80 """Validate the configuration.
82 Raises:
83 ValueError: If chunk size or overlap parameters are invalid.
84 """
85 if self.config.chunking.chunk_size <= 0:
86 raise ValueError("Chunk size must be greater than 0")
87 if self.config.chunking.chunk_overlap < 0:
88 raise ValueError("Chunk overlap must be non-negative")
89 if self.config.chunking.chunk_overlap >= self.config.chunking.chunk_size:
90 raise ValueError("Chunk overlap must be less than chunk size")
92 def _get_strategy(self, document: Document) -> BaseChunkingStrategy:
93 """Get the appropriate chunking strategy for a document.
95 Args:
96 document: The document to chunk
98 Returns:
99 The appropriate chunking strategy for the document type
100 """
101 # Check if this is a converted file
102 conversion_method = document.metadata.get("conversion_method")
103 if conversion_method == "docling":
104 # Files converted with docling carry a structured DoclingDocument; chunk it
105 # natively (structure-aware) rather than re-parsing exported markdown.
106 self.logger.info(
107 "Using docling strategy for converted file",
108 original_file_type=document.metadata.get("original_file_type"),
109 conversion_method=conversion_method,
110 document_id=document.id,
111 document_title=document.title,
112 )
113 return DoclingChunkingStrategy(self.settings)
114 if conversion_method == "markitdown":
115 # Files converted with MarkItDown are now in markdown format
116 self.logger.info(
117 "Using markdown strategy for converted file",
118 original_file_type=document.metadata.get("original_file_type"),
119 conversion_method=conversion_method,
120 document_id=document.id,
121 document_title=document.title,
122 )
123 return MarkdownChunkingStrategy(self.settings)
124 elif conversion_method == "markitdown_fallback":
125 # Fallback documents are also in markdown format
126 self.logger.info(
127 "Using markdown strategy for fallback converted file",
128 original_file_type=document.metadata.get("original_file_type"),
129 conversion_method=conversion_method,
130 conversion_failed=document.metadata.get("conversion_failed", False),
131 document_id=document.id,
132 document_title=document.title,
133 )
134 return MarkdownChunkingStrategy(self.settings)
136 # Get file extension from the document content type
137 file_type = document.content_type.lower()
139 self.logger.debug(
140 "Selecting chunking strategy",
141 file_type=file_type,
142 available_strategies=list(self.strategies.keys()),
143 document_id=document.id,
144 document_source=document.source,
145 document_title=document.title,
146 conversion_method=conversion_method,
147 )
149 # Get strategy class for file type
150 strategy_class = self.strategies.get(file_type)
152 if strategy_class:
153 self.logger.debug(
154 "Using specific strategy for this file type",
155 file_type=file_type,
156 strategy=strategy_class.__name__,
157 document_id=document.id,
158 document_title=document.title,
159 )
160 return strategy_class(self.settings)
162 self.logger.debug(
163 "No specific strategy found for this file type, using default text chunking strategy",
164 file_type=file_type,
165 document_id=document.id,
166 document_title=document.title,
167 )
168 return self.default_strategy
170 def chunk_document(self, document: Document) -> list[Document]:
171 """Chunk a document into smaller pieces.
173 Args:
174 document: The document to chunk
176 Returns:
177 List of chunked documents
178 """
179 self.logger.debug(
180 "Starting document chunking",
181 extra={
182 "doc_id": document.id,
183 "source": document.source,
184 "source_type": document.source_type,
185 "content_size": len(document.content),
186 "content_type": document.content_type,
187 },
188 )
190 if not document.content:
191 # Return a single empty chunk if document has no content
192 empty_doc = document.model_copy()
193 empty_doc.metadata.update({"chunk_index": 0, "total_chunks": 1})
194 self.logger.debug(
195 "Empty document, returning single empty chunk",
196 extra={"doc_id": document.id, "chunk_id": empty_doc.id},
197 )
198 return [empty_doc]
200 # Get the appropriate strategy for the document type
201 strategy = self._get_strategy(document)
203 # Optimized: Only log detailed chunking info when debug logging is enabled
204 if logging.getLogger().isEnabledFor(logging.DEBUG):
205 self.logger.debug(
206 "Selected chunking strategy",
207 extra={
208 "doc_id": document.id,
209 "strategy": strategy.__class__.__name__,
210 "content_type": document.content_type,
211 },
212 )
214 try:
215 # Chunk the document using the selected strategy
216 chunked_docs = strategy.chunk_document(document)
218 # Add contextual embedding
219 prefix = document.build_contextual_content()
220 for chunk in chunked_docs:
221 chunk.contextual_content = f"{prefix}" if prefix else ""
223 # Optimized: Only calculate and log detailed metrics when debug logging is enabled
224 if logging.getLogger().isEnabledFor(logging.DEBUG):
225 self.logger.debug(
226 "Document chunking completed",
227 extra={
228 "doc_id": document.id,
229 "chunk_count": len(chunked_docs),
230 "avg_chunk_size": (
231 sum(len(d.content) for d in chunked_docs)
232 / len(chunked_docs)
233 if chunked_docs
234 else 0
235 ),
236 },
237 )
238 return chunked_docs
239 except Exception as e:
240 self.logger.error(
241 f"Error chunking document {document.id}: {str(e)}",
242 extra={
243 "doc_id": document.id,
244 "error": str(e),
245 "error_type": type(e).__name__,
246 "strategy": strategy.__class__.__name__,
247 },
248 )
249 raise