Coverage for src/qdrant_loader/core/text_processing/semantic_analyzer.py: 94%
230 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-20 10:15 +0000
1"""Semantic analysis module for text processing."""
3from __future__ import annotations
5import hashlib
6import logging
7import threading
8from dataclasses import dataclass
9from typing import TYPE_CHECKING, Any
11from gensim import corpora
12from gensim.models import LdaModel
13from gensim.parsing.preprocessing import preprocess_string
14from qdrant_loader.core.text_processing import spacy_model_cache
16if TYPE_CHECKING:
17 from spacy.tokens import Doc
19logger = logging.getLogger(__name__)
22def is_meaningful_text(text: str) -> bool:
23 """Check if text contains meaningful content (letters or digits).
25 Returns False for text that only contains:
26 - Punctuation marks: ., #, @, |, -, _, etc.
27 - Whitespace characters
28 - Special symbols without semantic meaning (---, ..., |||, etc.)
30 """
31 # Check if text contains at least one alphanumeric character
32 return any(c.isalnum() for c in text)
35@dataclass
36class SemanticAnalysisResult:
37 """Container for semantic analysis results."""
39 entities: list[dict[str, Any]]
40 pos_tags: list[dict[str, Any]]
41 dependencies: list[dict[str, Any]]
42 topics: list[dict[str, Any]]
43 key_phrases: list[str]
44 document_similarity: dict[str, float]
47class SemanticAnalyzer:
48 """Advanced semantic analysis for text processing."""
50 def __init__(
51 self,
52 spacy_model: str = "en_core_web_md",
53 num_topics: int = 5,
54 passes: int = 10,
55 min_topic_freq: int = 2,
56 ):
57 """Initialize the semantic analyzer.
59 Args:
60 spacy_model: Name of the spaCy model to use
61 num_topics: Number of topics for LDA
62 passes: Number of passes for LDA training
63 min_topic_freq: Minimum frequency for topic terms
64 """
65 self.logger = logging.getLogger(__name__)
67 # Initialize spaCy. Cached and shared across instances -- a
68 # SemanticAnalyzer is constructed fresh per document (via
69 # ChunkProcessor), and spacy.load() is too expensive to repeat for
70 # every one of them.
71 def _load_nlp():
72 import spacy
73 from spacy.cli.download import download as spacy_download
75 try:
76 nlp = spacy.load(spacy_model)
77 except OSError:
78 self.logger.info(f"Downloading spaCy model {spacy_model}...")
79 spacy_download(spacy_model)
80 nlp = spacy.load(spacy_model)
81 return nlp
83 self.nlp = spacy_model_cache.get_or_load(
84 ("semantic_analyzer", spacy_model), _load_nlp
85 )
87 # Initialize LDA parameters
88 self.num_topics = num_topics
89 self.passes = passes
90 self.min_topic_freq = min_topic_freq
92 # Initialize LDA model
93 self.lda_model = None
94 self.dictionary = None
96 # Front-loaded (per-document) topic model: trained once over all of a
97 # document's chunks via fit_topic_model(), then inferred per chunk. Until it
98 # is fitted, _extract_topics falls back to the legacy single-chunk path.
99 self._topic_model_fitted = False
100 # A corpus smaller than this trains a degenerate model, so we skip fitting
101 # and let those chunks use the per-chunk fallback instead.
102 self._min_topic_corpus_docs = 3
103 # How many of a chunk's most-probable topics to surface, and how many terms
104 # per topic — matched to the legacy producer's output shape.
105 self._topics_per_chunk = 3
106 self._topic_top_n = 10
108 # Cache for processed documents
109 self._doc_cache: dict = {}
110 self._doc_cache_lock = threading.Lock()
112 def _build_cache_key(
113 self, text: str, doc_id: str | None, include_enhanced: bool
114 ) -> tuple[str, bool, str] | None:
115 """Build a cache key that includes a content fingerprint.
117 Including a fingerprint prevents stale cache hits when the same doc_id
118 is reused with different content.
119 """
120 if not doc_id:
121 return None
123 text_fingerprint = hashlib.sha256(text.encode("utf-8")).hexdigest()
124 return (doc_id, include_enhanced, text_fingerprint)
126 def analyze_text(
127 self,
128 text: str,
129 doc_id: str | None = None,
130 include_enhanced: bool = False,
131 ) -> SemanticAnalysisResult:
132 """Perform comprehensive semantic analysis on text.
134 Args:
135 text: Text to analyze
136 doc_id: Optional document ID for caching
137 include_enhanced: Whether to compute enhanced NLP fields
138 (pos_tags, dependencies, document_similarity)
140 Returns:
141 SemanticAnalysisResult containing all analysis results
142 """
143 # Check cache
144 cache_key = self._build_cache_key(text, doc_id, include_enhanced)
146 # Protected read
147 with self._doc_cache_lock:
148 cached = self._doc_cache.get(cache_key) if cache_key else None
150 if cached is not None:
151 if include_enhanced:
152 # Compute similarity OUTSIDE the lock (can be slow)
153 doc_similarity = self._calculate_document_similarity(
154 text, doc_id=doc_id
155 )
156 refreshed = SemanticAnalysisResult(
157 entities=cached.entities,
158 pos_tags=cached.pos_tags,
159 dependencies=cached.dependencies,
160 topics=cached.topics,
161 key_phrases=cached.key_phrases,
162 document_similarity=doc_similarity,
163 )
164 # Protected write-back
165 with self._doc_cache_lock:
166 self._doc_cache[cache_key] = refreshed
167 return refreshed
168 return cached
170 # Process with spaCy
171 doc = self.nlp(text)
173 # Extract entities with linking
174 entities = self._extract_entities(doc)
176 if include_enhanced:
177 # Get part-of-speech tags
178 pos_tags = self._get_pos_tags(doc)
180 # Get dependency parse
181 dependencies = self._get_dependencies(doc)
182 else:
183 pos_tags = []
184 dependencies = []
186 # Extract topics
187 topics = self._extract_topics(text)
189 # Extract key phrases
190 key_phrases = self._extract_key_phrases(doc)
192 # Calculate document similarity
193 doc_similarity = (
194 self._calculate_document_similarity(text, doc_id=doc_id)
195 if include_enhanced
196 else {}
197 )
199 # Create result
200 result = SemanticAnalysisResult(
201 entities=entities,
202 pos_tags=pos_tags,
203 dependencies=dependencies,
204 topics=topics,
205 key_phrases=key_phrases,
206 document_similarity=doc_similarity,
207 )
209 # Protected write
210 if cache_key:
211 with self._doc_cache_lock:
212 self._doc_cache[cache_key] = result
214 return result
216 def _extract_entities(self, doc: Doc) -> list[dict[str, Any]]:
217 """Extract named entities with linking, filtering garbage entities.
219 Filters out entities that:
220 - Only contain punctuation/symbols (., #, |, etc.)
221 - Don't have any alphanumeric characters
222 - Are just whitespace
224 Args:
225 doc: spaCy document
227 Returns:
228 List of entity dictionaries with linking information
229 """
230 entities = []
231 for ent in doc.ents:
232 # Filter entities that only contain punctuation/symbols
233 if not is_meaningful_text(ent.text):
234 continue
236 # Get entity context
237 start_sent = ent.sent.start
238 end_sent = ent.sent.end
239 context = doc[start_sent:end_sent].text
241 # Get entity description
242 description = self.nlp.vocab.strings[ent.label_]
244 # Get related entities (also filter meaningless ones)
245 related = []
246 for token in ent.sent:
247 if token.ent_type_ and token.text != ent.text:
248 # Only add related entities with meaningful text
249 if is_meaningful_text(token.text):
250 related.append(
251 {
252 "text": token.text,
253 "type": token.ent_type_,
254 "relation": token.dep_,
255 }
256 )
258 entities.append(
259 {
260 "text": ent.text,
261 "label": ent.label_,
262 "start": ent.start_char,
263 "end": ent.end_char,
264 "description": description,
265 "context": context,
266 "related_entities": related,
267 }
268 )
270 return entities
272 def _get_pos_tags(self, doc: Doc) -> list[dict[str, Any]]:
273 """Get part-of-speech tags with detailed information, filtering noise tokens.
275 Filters out multiple types of noise:
276 - Whitespace tokens (is_space=True)
277 - Punctuation tokens (is_punct=True)
278 - Symbol-only tokens without alphanumeric content (e.g., ---, ..., |||)
280 This is especially important for Excel tables and structured data.
282 Args:
283 doc: spaCy document
285 Returns:
286 List of POS tag dictionaries (excluding spaces, punctuation, and symbols)
287 """
288 pos_tags = []
289 for token in doc:
290 # Skip whitespace and punctuation - they pollute metadata
291 if token.is_space or token.is_punct:
292 continue
294 # Also skip tokens with no meaningful content (e.g., ---, ...)
295 # This catches edge cases where spaCy doesn't mark as punct
296 if not is_meaningful_text(token.text):
297 continue
299 pos_tags.append(
300 {
301 "text": token.text,
302 "pos": token.pos_,
303 "tag": token.tag_,
304 "lemma": token.lemma_,
305 "is_stop": token.is_stop,
306 }
307 )
308 return pos_tags
310 def _get_dependencies(self, doc: Doc) -> list[dict[str, Any]]:
311 """Get dependency parse information with filtering.
313 Filters out:
314 - Whitespace tokens (is_space=True)
315 - Punctuation tokens (is_punct=True)
316 - Symbol-only tokens without alphanumeric content
317 - Children that are punctuation or meaningless symbols
319 Args:
320 doc: spaCy document
322 Returns:
323 List of dependency dictionaries (excluding noise tokens)
324 """
325 dependencies = []
326 for token in doc:
327 # Skip whitespace and punctuation tokens
328 if token.is_space or token.is_punct:
329 continue
331 # Skip tokens with no meaningful content (e.g., ---, ...)
332 if not is_meaningful_text(token.text):
333 continue
335 # Filter children to only include meaningful tokens
336 meaningful_children = [
337 child.text
338 for child in token.children
339 if not child.is_space
340 and not child.is_punct
341 and is_meaningful_text(child.text)
342 ]
344 dependencies.append(
345 {
346 "text": token.text,
347 "dep": token.dep_,
348 "head": token.head.text,
349 "head_pos": token.head.pos_,
350 "children": meaningful_children,
351 }
352 )
353 return dependencies
355 def fit_topic_model(self, texts: list[str]) -> None:
356 """Train one document-level LDA over all of a document's chunk texts.
358 Front-loading the model once lets :meth:`_extract_topics` *infer* each
359 chunk's topics against a shared corpus model, instead of training a
360 degenerate single-document LDA per chunk. When the corpus is too small to be
361 meaningful the model is left unfitted, so those chunks fall back to the
362 legacy per-chunk path.
364 Args:
365 texts: The contents of every chunk produced for one document.
366 """
367 try:
368 processed = [preprocess_string(text) for text in texts]
369 processed = [tokens for tokens in processed if len(tokens) >= 5]
371 if len(processed) < self._min_topic_corpus_docs:
372 self.logger.debug(
373 "Corpus too small to fit a document-level topic model "
374 f"({len(processed)} usable chunks); using per-chunk fallback"
375 )
376 self.dictionary = None
377 self.lda_model = None
378 self._topic_model_fitted = False
379 return
381 dictionary = corpora.Dictionary(processed)
382 corpus = [dictionary.doc2bow(tokens) for tokens in processed]
383 self.dictionary = dictionary
384 self.lda_model = LdaModel(
385 corpus,
386 num_topics=min(self.num_topics, len(processed)),
387 passes=self.passes,
388 id2word=dictionary,
389 random_state=42, # For reproducibility
390 alpha=0.1, # Fixed positive value for document-topic density
391 eta=0.01, # Fixed positive value for topic-word density
392 )
393 self._topic_model_fitted = True
394 except Exception as e:
395 self.logger.warning(
396 f"Topic model fit failed; using per-chunk fallback: {e}",
397 exc_info=True,
398 )
399 self.dictionary = None
400 self.lda_model = None
401 self._topic_model_fitted = False
403 def _extract_topics(self, text: str) -> list[dict[str, Any]]:
404 """Extract topics for one chunk.
406 Uses the front-loaded document-level model when one has been fitted (see
407 :meth:`fit_topic_model`), inferring the chunk's dominant topics against it
408 without retraining. Otherwise falls back to the legacy per-chunk model.
410 Args:
411 text: Text to analyze
413 Returns:
414 List of topic dictionaries:
415 ``{"id", "terms": [{"term", "weight"}], "coherence"}``.
416 """
417 try:
418 # Preprocess text
419 processed_text = preprocess_string(text)
421 # Skip topic extraction for very short texts
422 if len(processed_text) < 5:
423 self.logger.debug("Text too short for topic extraction")
424 return [
425 {
426 "id": 0,
427 "terms": [{"term": "general", "weight": 1.0}],
428 "coherence": 0.5,
429 }
430 ]
432 # A document-level model was front-loaded: infer this chunk's topics
433 # against it, without retraining or mutating the shared model.
434 if (
435 self._topic_model_fitted
436 and self.lda_model is not None
437 and self.dictionary is not None
438 ):
439 return self._infer_chunk_topics(processed_text)
441 # Unfitted (the markdown path, or a corpus too small to fit): legacy
442 # per-chunk model. Degenerate by construction, but retained so behavior
443 # for unfitted callers is unchanged.
444 temp_dictionary = corpora.Dictionary([processed_text])
445 corpus = [temp_dictionary.doc2bow(processed_text)]
447 # Create a fresh LDA model for this specific text
448 current_lda_model = LdaModel(
449 corpus,
450 num_topics=min(
451 self.num_topics, len(processed_text) // 2
452 ), # Ensure reasonable topic count
453 passes=self.passes,
454 id2word=temp_dictionary,
455 random_state=42, # For reproducibility
456 alpha=0.1, # Fixed positive value for document-topic density
457 eta=0.01, # Fixed positive value for topic-word density
458 )
460 # Get topics
461 topics = []
462 for topic_id, topic in current_lda_model.print_topics():
463 # Parse topic terms
464 terms = []
465 for term in topic.split("+"):
466 try:
467 weight, word = term.strip().split("*")
468 terms.append({"term": word.strip('"'), "weight": float(weight)})
469 except ValueError:
470 # Skip malformed terms
471 continue
473 topics.append(
474 {
475 "id": topic_id,
476 "terms": terms,
477 "coherence": self._calculate_topic_coherence(terms),
478 }
479 )
481 return (
482 topics
483 if topics
484 else [
485 {
486 "id": 0,
487 "terms": [{"term": "general", "weight": 1.0}],
488 "coherence": 0.5,
489 }
490 ]
491 )
493 except Exception as e:
494 self.logger.warning(f"Topic extraction failed: {e}", exc_info=True)
495 # Return fallback topic
496 return [
497 {
498 "id": 0,
499 "terms": [{"term": "general", "weight": 1.0}],
500 "coherence": 0.5,
501 }
502 ]
504 def _infer_chunk_topics(self, processed_text: list[str]) -> list[dict[str, Any]]:
505 """Infer a chunk's dominant topics against the front-loaded model.
507 Args:
508 processed_text: The chunk's preprocessed tokens.
510 Returns:
511 The chunk's most-probable topics, in the legacy producer's shape. An
512 empty bag-of-words (all tokens out of the trained vocabulary) still
513 yields the model's topics ranked by the prior distribution.
514 """
515 bow = self.dictionary.doc2bow(processed_text)
516 distribution = self.lda_model.get_document_topics(bow, minimum_probability=0.0)
517 ranked = sorted(distribution, key=lambda pair: pair[1], reverse=True)
519 topics: list[dict[str, Any]] = []
520 for topic_id, _probability in ranked[: self._topics_per_chunk]:
521 terms = [
522 {"term": word, "weight": float(weight)}
523 for word, weight in self.lda_model.show_topic(
524 topic_id, topn=self._topic_top_n
525 )
526 ]
527 topics.append(
528 {
529 "id": int(topic_id),
530 "terms": terms,
531 "coherence": self._calculate_topic_coherence(terms),
532 }
533 )
535 return (
536 topics
537 if topics
538 else [
539 {
540 "id": 0,
541 "terms": [{"term": "general", "weight": 1.0}],
542 "coherence": 0.5,
543 }
544 ]
545 )
547 def _extract_key_phrases(self, doc: Doc) -> list[str]:
548 """Extract key phrases from text.
550 Args:
551 doc: spaCy document
553 Returns:
554 List of key phrases
555 """
556 key_phrases = []
558 # Extract noun phrases
559 for chunk in doc.noun_chunks:
560 if len(chunk.text.split()) >= 2: # Only multi-word phrases
561 key_phrases.append(chunk.text)
563 # Extract named entities
564 for ent in doc.ents:
565 if ent.label_ in ["ORG", "PRODUCT", "WORK_OF_ART", "LAW"]:
566 key_phrases.append(ent.text)
568 return list(set(key_phrases)) # Remove duplicates
570 def _calculate_document_similarity(
571 self, text: str, doc_id: str | None = None
572 ) -> dict[str, float]:
573 """Calculate similarity with other processed documents.
575 Args:
576 text: Text to compare
577 doc_id: Optional current document ID to exclude from results
579 Returns:
580 Dictionary of document similarities
581 """
582 similarities = {}
583 skipped_ids = {doc_id} if doc_id else set()
585 doc = self.nlp(text)
587 # Check if the model has word vectors
588 has_vectors = self.nlp.vocab.vectors_length > 0
590 with self._doc_cache_lock:
591 cached_items = list(self._doc_cache.items())
593 for cache_key, cached_result in cached_items:
594 cached_doc_id = cache_key[0] if isinstance(cache_key, tuple) else cache_key
595 if cached_doc_id is None or cached_doc_id in skipped_ids:
596 continue
598 # Check if cached_result has entities and the first entity has context
599 if not cached_result.entities or not cached_result.entities[0].get(
600 "context"
601 ):
602 continue
604 cached_doc = self.nlp(cached_result.entities[0]["context"])
606 if has_vectors:
607 # Use spaCy's built-in similarity which uses word vectors
608 similarity = doc.similarity(cached_doc)
609 else:
610 # Use alternative similarity calculation for models without word vectors
611 # This avoids the spaCy warning about missing word vectors
612 similarity = self._calculate_alternative_similarity(doc, cached_doc)
614 similarities[cached_doc_id] = float(similarity)
615 skipped_ids.add(cached_doc_id)
617 return similarities
619 def _calculate_alternative_similarity(self, doc1: Doc, doc2: Doc) -> float:
620 """Calculate similarity for models without word vectors.
622 Uses token overlap and shared entities as similarity metrics.
624 Args:
625 doc1: First document
626 doc2: Second document
628 Returns:
629 Similarity score between 0 and 1
630 """
631 # Extract lemmatized tokens (excluding stop words and punctuation)
632 tokens1 = {
633 token.lemma_.lower()
634 for token in doc1
635 if not token.is_stop and not token.is_punct and token.is_alpha
636 }
637 tokens2 = {
638 token.lemma_.lower()
639 for token in doc2
640 if not token.is_stop and not token.is_punct and token.is_alpha
641 }
643 # Calculate token overlap (Jaccard similarity)
644 if not tokens1 and not tokens2:
645 return 1.0 # Both empty
646 if not tokens1 or not tokens2:
647 return 0.0 # One empty
649 intersection = len(tokens1.intersection(tokens2))
650 union = len(tokens1.union(tokens2))
651 token_similarity = intersection / union if union > 0 else 0.0
653 # Extract named entities
654 entities1 = {ent.text.lower() for ent in doc1.ents}
655 entities2 = {ent.text.lower() for ent in doc2.ents}
657 # Calculate entity overlap
658 entity_similarity = 0.0
659 if entities1 or entities2:
660 entity_intersection = len(entities1.intersection(entities2))
661 entity_union = len(entities1.union(entities2))
662 entity_similarity = (
663 entity_intersection / entity_union if entity_union > 0 else 0.0
664 )
666 # Combine token and entity similarities (weighted average)
667 # Token similarity gets more weight as it's more comprehensive
668 combined_similarity = 0.7 * token_similarity + 0.3 * entity_similarity
670 return combined_similarity
672 def _calculate_topic_coherence(self, terms: list[dict[str, Any]]) -> float:
673 """Calculate topic coherence score.
675 Args:
676 terms: List of topic terms with weights
678 Returns:
679 Coherence score between 0 and 1
680 """
681 # Simple coherence based on term weights
682 weights = [term["weight"] for term in terms]
683 return sum(weights) / len(weights) if weights else 0.0
685 def clear_cache(self):
686 """Clear instance-owned caches and model state."""
687 with self._doc_cache_lock:
688 self._doc_cache.clear()
690 # Reset topic model state for this analyzer instance.
691 self._topic_model_fitted = False
692 self.lda_model = None
693 self.dictionary = None
695 logger.debug("Semantic analyzer instance caches cleared")
697 def shutdown(self):
698 """Shutdown the semantic analyzer and release all resources.
700 This method should be called when the analyzer is no longer needed
701 to ensure proper cleanup of all resources.
702 """
703 self.clear_cache()
705 if hasattr(self, "nlp"):
706 del self.nlp
708 logger.debug("Semantic analyzer shutdown completed")