diff --git a/CHANGELOG.md b/CHANGELOG.md index 3e0e47c8..3813fd78 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Added `max_workers` configuration parameter (default: 1) to all extractor `extract()` methods, allowing users to tune concurrency based on available CPU cores or API rate limits. - **Parallel Chunking**: Implemented parallel processing for large document chunking in `_extract_entities_chunked` and `_extract_relations_chunked`, significantly reducing latency for long-form text analysis. - **Thread-Safe Progress Tracking**: Enhanced `ProgressTracker` to handle concurrent updates from multiple threads without race conditions during batch processing. +- **Semantic Extract Performance & Regression**: + - Added edge-case regression suite covering max worker defaults, LLM prompt entity filtering, and extractor reuse. + - Added a runnable real-use-case benchmark script for batch latency across `NERExtractor`, `RelationExtractor`, `TripletExtractor`, `EventDetector`, `SemanticAnalyzer`, and `SemanticNetworkExtractor`. + - Added Groq LLM smoke tests that exercise LLM-based entities/relations/triplets when `GROQ_API_KEY` is available via environment configuration. ### Security - **Credential Sanitization**: @@ -29,12 +33,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **Dependency Resolution**: - Pinned `opentelemetry-api` and `opentelemetry-sdk` to `1.37.0` to resolve pip conflicts. - Updated `protobuf` and `grpcio` constraints for better stability. +- **Entity Filtering Scope**: + - Removed entity filtering from non-LLM extraction flows to avoid accuracy regressions. + - Applied entity downselection only to LLM relation prompt construction, while matching returned entities against the full original entity list. +- **Batch Concurrency Defaults**: + - Standardized `max_workers` defaulting across `semantic_extract` and tuned for low-latency: ML-backed methods default to single-worker, while pattern/regex/rules/LLM/huggingface methods use a higher parallelism default capped by CPU. + - Raised the global `optimization.max_workers` default to 8 for better throughput on batch workloads. ### Performance - **Bottleneck Optimization (GitHub Issue #186)**: - **Resolved Bottleneck #1 (Sequential Processing)**: Replaced sequential `for` loops with parallel execution for both document-level batches and intra-document chunks. - **Performance Gains**: Achieved **~1.89x speedup** in real-world extraction scenarios (tested with Groq `llama-3.3-70b-versatile` on standard datasets). - **Initialization Optimization**: Refactored test suite to use class-level `setUpClass` for LLM provider initialization, eliminating redundant API client creation overhead. +- **Low-Latency Entity Matching**: + - Avoided heavyweight embedding stack imports on common matches by improving fast matching heuristics and short-circuiting before embedding similarity. + - Optimized entity matching to prioritize exact/substring/word-boundary matches and only fall back to embedding similarity when needed, reducing CPU overhead in LLM relation/triplet mapping. ## [0.2.1] - 2026-01-12 diff --git a/semantica/semantic_extract/config.py b/semantica/semantic_extract/config.py index 49e4e492..640ecb3c 100644 --- a/semantica/semantic_extract/config.py +++ b/semantica/semantic_extract/config.py @@ -40,6 +40,7 @@ License: MIT """ import os +import multiprocessing from pathlib import Path from typing import Dict, Optional, Any @@ -57,7 +58,7 @@ class Config: self._configs["optimization"] = { "enable_cache": True, "cache_size": 1000, - "max_workers": 5, + "max_workers": 8, "enable_batching": True, "batch_size": 10, "max_tokens_per_batch": 2000 @@ -151,3 +152,43 @@ class Config: # Global config instance config = Config() + + +def resolve_max_workers( + explicit: Optional[int] = None, + local_config: Optional[Dict[str, Any]] = None, + methods: Optional[Any] = None, +) -> int: + def to_int(val: Any, default: int) -> int: + try: + return int(val) + except Exception: + return default + + if isinstance(methods, str): + normalized_methods = [methods] + elif isinstance(methods, (list, tuple, set)): + normalized_methods = [m for m in methods if isinstance(m, str)] + else: + normalized_methods = [] + + if explicit is not None: + value = to_int(explicit, 1) + elif local_config and "max_workers" in local_config: + value = to_int(local_config.get("max_workers", 1), 1) + else: + value = to_int(config.get("max_workers", 5), 5) + + if "ml" in normalized_methods and explicit is None and not (local_config and "max_workers" in local_config): + value = 1 + + if value < 1: + value = 1 + + cpu_count = multiprocessing.cpu_count() or 1 + if value > cpu_count: + value = cpu_count + if value > 32: + value = 32 + + return value diff --git a/semantica/semantic_extract/event_detector.py b/semantica/semantic_extract/event_detector.py index adcecbc4..4ff70143 100644 --- a/semantica/semantic_extract/event_detector.py +++ b/semantica/semantic_extract/event_detector.py @@ -185,7 +185,12 @@ class EventDetector: message=f"Starting batch detection... 0/{total_items} (remaining: {total_items})" ) - max_workers = kwargs.get("max_workers", self.config.get("max_workers", 1)) + from .config import resolve_max_workers + max_workers = resolve_max_workers( + explicit=kwargs.get("max_workers"), + local_config=self.config, + methods=[self.config.get("ner_method"), self.config.get("relation_method"), self.config.get("method")], + ) def process_item(idx, item): try: @@ -269,17 +274,26 @@ class EventDetector: # Single item return self.detect_events(text, **kwargs) - def detect_events(self, text: str, **options) -> List[Event]: + def detect_events( + self, + text: Union[str, List[str], List[Dict[str, Any]]], + pipeline_id: Optional[str] = None, + **options, + ) -> Union[List[Event], List[List[Event]]]: """ Detect events in text content. Args: text: Input text + pipeline_id: Optional pipeline ID for progress tracking (batch mode) **options: Detection options Returns: list: List of detected events """ + if isinstance(text, list): + return self.extract(text, pipeline_id=pipeline_id, **options) + tracking_id = self.progress_tracker.start_tracking( module="semantic_extract", submodule="EventDetector", diff --git a/semantica/semantic_extract/methods.py b/semantica/semantic_extract/methods.py index 3e2acb1f..ab99e69c 100644 --- a/semantica/semantic_extract/methods.py +++ b/semantica/semantic_extract/methods.py @@ -287,6 +287,12 @@ def find_best_match_index(text: str, candidates: List[str]) -> Tuple[int, float] best_idx = i # 2. Substring Match (Fast) + word_pat = None + try: + word_pat = re.compile(rf"\b{re.escape(text_lower)}\b") + except Exception: + word_pat = None + for i, cand in enumerate(candidates_lower): if not cand: continue score = 0.0 @@ -294,12 +300,18 @@ def find_best_match_index(text: str, candidates: List[str]) -> Tuple[int, float] # Calculate length ratio ratio = min(len(text_lower), len(cand)) / max(len(text_lower), len(cand)) score = 0.9 * ratio + 0.1 + + if word_pat and word_pat.search(cand): + score = max(score, 0.88) if score > best_score: best_score = score best_idx = i # 3. Text Embeddings (High Accuracy Semantic) - Batch Optimized + if best_score >= 0.85: + return best_idx, float(best_score) + embedder = get_text_embedder() embedding_idx = -1 embedding_score = 0.0 @@ -417,6 +429,86 @@ def match_entity(text: str, entities: List[Entity], threshold: float = 0.8) -> O return None +def filter_entities_for_text( + text: str, + entities: List[Entity], + max_keep: int = 80, +) -> List[Entity]: + if not text or not entities: + return [] + + if max_keep < 1: + return [] + + if len(entities) <= max_keep: + return entities + + text_lower = text.lower() + stop_tokens = { + "inc", + "incorporated", + "corp", + "corporation", + "co", + "company", + "ltd", + "llc", + "plc", + "group", + "holdings", + "limited", + "the", + "and", + "or", + "of", + "in", + "on", + "at", + "for", + "to", + "a", + "an", + } + + seen = set() + matched: List[Entity] = [] + for entity in entities: + ent_text = getattr(entity, "text", "") + if not ent_text: + continue + + key = ent_text.lower().strip() + if not key or key in seen: + continue + seen.add(key) + + if key in text_lower: + matched.append(entity) + continue + + tokens = re.findall(r"[a-z0-9]+", key) + keep = False + for tok in tokens: + if len(tok) < 2: + continue + if tok in stop_tokens: + continue + if tok in text_lower: + keep = True + break + if keep: + matched.append(entity) + + if matched: + if len(matched) > max_keep: + matched.sort(key=lambda e: len(getattr(e, "text", "")), reverse=True) + return matched[:max_keep] + return matched + + entities_sorted = sorted(entities, key=lambda e: len(getattr(e, "text", "")), reverse=True) + return entities_sorted[:max_keep] + + def calculate_weighted_confidence( item_type: str, original_confidence: float, @@ -907,8 +999,8 @@ def _extract_entities_chunked( all_entities = [] - # Process chunks in parallel - max_workers = kwargs.get("max_workers", 5) + from .config import resolve_max_workers + max_workers = resolve_max_workers(explicit=kwargs.get("max_workers")) with ThreadPoolExecutor(max_workers=max_workers) as executor: future_to_chunk = {} @@ -1513,7 +1605,22 @@ def extract_relations_llm( **kwargs ) - entities_str = ", ".join([f"{e.text} ({e.label})" for e in entities]) + original_entities = entities + max_entities_prompt = kwargs.get("max_entities_prompt", kwargs.get("max_entities", 80)) + try: + max_entities_prompt = int(max_entities_prompt) + except Exception: + max_entities_prompt = 80 + + prompt_entities = original_entities + if max_entities_prompt > 0 and len(original_entities) > max_entities_prompt: + prompt_entities = filter_entities_for_text( + text, + original_entities, + max_keep=max_entities_prompt, + ) + + entities_str = ", ".join([f"{e.text} ({e.label})" for e in prompt_entities]) # Use custom relation types if provided relation_types = kwargs.get("relation_types") @@ -1562,8 +1669,8 @@ Entities found in text: {entities_str}""" relations = [] for r_out in result_obj.relations: # Find matching entities using hybrid similarity - subject_entity = match_entity(r_out.subject, entities) - object_entity = match_entity(r_out.object, entities) + subject_entity = match_entity(r_out.subject, original_entities) + object_entity = match_entity(r_out.object, original_entities) if subject_entity and object_entity: relations.append(Relation( @@ -1689,8 +1796,8 @@ def _extract_relations_chunked( all_relations = [] - # Process chunks in parallel - max_workers = kwargs.get("max_workers", 5) + from .config import resolve_max_workers + max_workers = resolve_max_workers(explicit=kwargs.get("max_workers")) with ThreadPoolExecutor(max_workers=max_workers) as executor: future_to_chunk = {} @@ -2093,8 +2200,8 @@ def _extract_triplets_chunked( all_triplets = [] - # Process chunks in parallel - max_workers = kwargs.get("max_workers", 5) + from .config import resolve_max_workers + max_workers = resolve_max_workers(explicit=kwargs.get("max_workers")) with ThreadPoolExecutor(max_workers=max_workers) as executor: future_to_chunk = {} diff --git a/semantica/semantic_extract/ner_extractor.py b/semantica/semantic_extract/ner_extractor.py index b6541a12..4ab7267f 100644 --- a/semantica/semantic_extract/ner_extractor.py +++ b/semantica/semantic_extract/ner_extractor.py @@ -197,8 +197,12 @@ class NERExtractor: message=f"Starting batch extraction... 0/{total_items}" ) - # Determine max_workers - max_workers = kwargs.get("max_workers", self.config.get("max_workers", 1)) + from .config import resolve_max_workers + max_workers = resolve_max_workers( + explicit=kwargs.get("max_workers"), + local_config=self.config, + methods=self.method, + ) # Helper function for single item processing def process_item(idx, item): @@ -297,12 +301,18 @@ class NERExtractor: else: return self.extract_entities(text, **kwargs) - def extract_entities(self, text: str, **options) -> List[Entity]: + def extract_entities( + self, + text: Union[str, List[Dict[str, Any]], List[str]], + pipeline_id: Optional[str] = None, + **options, + ) -> Union[List[Entity], List[List[Entity]]]: """ Extract named entities from text. Args: text: Input text + pipeline_id: Optional pipeline ID for progress tracking (batch mode) **options: Extraction options: - entity_types: Filter by entity types (list) - min_confidence: Minimum confidence threshold @@ -311,6 +321,9 @@ class NERExtractor: Returns: list: List of extracted entities """ + if isinstance(text, list): + return self.extract(text, pipeline_id=pipeline_id, **options) + tracking_id = self.progress_tracker.start_tracking( module="semantic_extract", submodule="NERExtractor", diff --git a/semantica/semantic_extract/relation_extractor.py b/semantica/semantic_extract/relation_extractor.py index 3e706a70..d29bdcac 100644 --- a/semantica/semantic_extract/relation_extractor.py +++ b/semantica/semantic_extract/relation_extractor.py @@ -221,8 +221,12 @@ class RelationExtractor: message=f"Starting batch extraction... 0/{min_len}" ) - # Determine max_workers - max_workers = kwargs.get("max_workers", self.config.get("max_workers", 1)) + from .config import resolve_max_workers + max_workers = resolve_max_workers( + explicit=kwargs.get("max_workers"), + local_config=self.config, + methods=self.method, + ) def process_item(i, doc_item, ent_item): try: @@ -325,14 +329,19 @@ class RelationExtractor: return [] def extract_relations( - self, text: str, entities: List[Entity], **options - ) -> List[Relation]: + self, + text: Union[str, List[Dict[str, Any]], List[str]], + entities: Union[List[Entity], List[List[Entity]]], + pipeline_id: Optional[str] = None, + **options, + ) -> Union[List[Relation], List[List[Relation]]]: """ Extract relations between entities. Args: text: Input text entities: List of extracted entities + pipeline_id: Optional pipeline ID for progress tracking (batch mode) **options: Extraction options: - method: Override method (if not set in __init__) - min_confidence: Minimum confidence threshold @@ -341,6 +350,17 @@ class RelationExtractor: Returns: list: List of extracted relations """ + if isinstance(text, list): + if entities is None: + entities_batch = [[] for _ in range(len(text))] + elif isinstance(entities, list) and (not entities): + entities_batch = [[] for _ in range(len(text))] + elif isinstance(entities, list) and all(isinstance(e, Entity) for e in entities): + entities_batch = [entities for _ in range(len(text))] + else: + entities_batch = entities + return self.extract(text, entities_batch, pipeline_id=pipeline_id, **options) + from .methods import get_relation_method, match_entity tracking_id = self.progress_tracker.start_tracking( diff --git a/semantica/semantic_extract/semantic_analyzer.py b/semantica/semantic_extract/semantic_analyzer.py index 56ef982c..787f06af 100644 --- a/semantica/semantic_extract/semantic_analyzer.py +++ b/semantica/semantic_extract/semantic_analyzer.py @@ -148,8 +148,9 @@ class SemanticAnalyzer: ) try: - results = [] + results = [None] * len(text) total_items = len(text) + processed_count = 0 # Determine update interval if total_items <= 10: @@ -165,40 +166,83 @@ class SemanticAnalyzer: message=f"Starting batch analysis... 0/{total_items} (remaining: {total_items})" ) - for idx, item in enumerate(text): - # Prepare arguments for single item - doc_text = item["content"] if isinstance(item, dict) and "content" in item else str(item) - - # Analyze - analysis = self.analyze_semantics(doc_text, **kwargs) + from .config import resolve_max_workers + max_workers = resolve_max_workers( + explicit=kwargs.get("max_workers"), + local_config=self.config, + ) - # Add provenance metadata - analysis["batch_index"] = idx - if isinstance(item, dict) and "id" in item: - analysis["document_id"] = item["id"] - - # Also inject into semantic roles if present - if "semantic_roles" in analysis: - for role in analysis["semantic_roles"]: - # role is a dict here because analyze_semantics converts it - if "metadata" not in role: - role["metadata"] = {} - - role["metadata"]["batch_index"] = idx - if isinstance(item, dict) and "id" in item: - role["metadata"]["document_id"] = item["id"] + def process_item(idx, item): + try: + doc_text = item["content"] if isinstance(item, dict) and "content" in item else str(item) + analysis = self.analyze_semantics(doc_text, **kwargs) - results.append(analysis) + analysis["batch_index"] = idx + if isinstance(item, dict) and "id" in item: + analysis["document_id"] = item["id"] - # Update progress - if (idx + 1) % update_interval == 0 or (idx + 1) == total_items: - remaining = total_items - (idx + 1) - self.progress_tracker.update_progress( - tracking_id, - processed=idx + 1, - total=total_items, - message=f"Processing... {idx + 1}/{total_items} (remaining: {remaining})" + if "semantic_roles" in analysis: + for role in analysis["semantic_roles"]: + if "metadata" not in role: + role["metadata"] = {} + + role["metadata"]["batch_index"] = idx + if isinstance(item, dict) and "id" in item: + role["metadata"]["document_id"] = item["id"] + + return idx, analysis + except Exception as e: + self.logger.warning(f"Failed to analyze item {idx}: {e}") + return idx, {"error": str(e), "batch_index": idx} + + if max_workers > 1: + import concurrent.futures + + with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: + future_to_idx = { + executor.submit(process_item, idx, item): idx + for idx, item in enumerate(text) + } + + for future in concurrent.futures.as_completed(future_to_idx): + idx, analysis = future.result() + results[idx] = analysis + processed_count += 1 + + should_update = ( + processed_count % update_interval == 0 + or processed_count == total_items + or processed_count == 1 + or total_items <= 10 + ) + if should_update: + remaining = total_items - processed_count + self.progress_tracker.update_progress( + tracking_id, + processed=processed_count, + total=total_items, + message=f"Processing... {processed_count}/{total_items} (remaining: {remaining})" + ) + else: + for idx, item in enumerate(text): + _, analysis = process_item(idx, item) + results[idx] = analysis + processed_count += 1 + + should_update = ( + processed_count % update_interval == 0 + or processed_count == total_items + or processed_count == 1 + or total_items <= 10 ) + if should_update: + remaining = total_items - processed_count + self.progress_tracker.update_progress( + tracking_id, + processed=processed_count, + total=total_items, + message=f"Processing... {processed_count}/{total_items} (remaining: {remaining})" + ) self.progress_tracker.stop_tracking( tracking_id, diff --git a/semantica/semantic_extract/semantic_network_extractor.py b/semantica/semantic_extract/semantic_network_extractor.py index d559c697..48f25424 100644 --- a/semantica/semantic_extract/semantic_network_extractor.py +++ b/semantica/semantic_extract/semantic_network_extractor.py @@ -149,6 +149,9 @@ class SemanticNetworkExtractor: self.config["ner_method"] = method self.config["relation_method"] = method + self._ner_extractor = None + self._relation_extractor = None + def extract( self, text: Union[str, List[str], List[Dict[str, Any]]], @@ -199,8 +202,11 @@ class SemanticNetworkExtractor: message=f"Starting batch extraction... 0/{total_items} (remaining: {total_items})" ) - # Determine max_workers - max_workers = kwargs.get("max_workers", self.config.get("max_workers", 1)) + from .config import resolve_max_workers + max_workers = resolve_max_workers( + explicit=kwargs.get("max_workers"), + local_config=self.config, + ) def process_item(idx, item, doc_entities, doc_relations): try: @@ -318,11 +324,12 @@ class SemanticNetworkExtractor: def extract_network( self, - text: str, - entities: Optional[List[Entity]] = None, - relations: Optional[List[Relation]] = None, + text: Union[str, List[str], List[Dict[str, Any]]], + entities: Optional[Union[List[Entity], List[List[Entity]]]] = None, + relations: Optional[Union[List[Relation], List[List[Relation]]]] = None, + pipeline_id: Optional[str] = None, **options, - ) -> SemanticNetwork: + ) -> Union[SemanticNetwork, List[SemanticNetwork]]: """ Extract semantic network from text. @@ -335,6 +342,23 @@ class SemanticNetworkExtractor: Returns: SemanticNetwork: Extracted semantic network """ + if isinstance(text, list): + entities_batch = entities + if entities is not None and isinstance(entities, list) and (not entities or all(isinstance(e, Entity) for e in entities)): + entities_batch = [entities for _ in range(len(text))] if entities else [[] for _ in range(len(text))] + + relations_batch = relations + if relations is not None and isinstance(relations, list) and (not relations or all(isinstance(r, Relation) for r in relations)): + relations_batch = [relations for _ in range(len(text))] if relations else [[] for _ in range(len(text))] + + return self.extract( + text, + entities=entities_batch, + relations=relations_batch, + pipeline_id=pipeline_id, + **options, + ) + tracking_id = self.progress_tracker.start_tracking( module="semantic_extract", submodule="SemanticNetworkExtractor", @@ -354,15 +378,16 @@ class SemanticNetworkExtractor: # Pass method if specified if "ner_method" in self.config: ner_config["method"] = self.config["ner_method"] - ner = NERExtractor( - **ner_config, - **{ - k: v - for k, v in self.config.items() - if k not in ["ner", "relation"] - }, - ) - entities = ner.extract_entities(text, **options) + if self._ner_extractor is None: + self._ner_extractor = NERExtractor( + **ner_config, + **{ + k: v + for k, v in self.config.items() + if k not in ["ner", "relation"] + }, + ) + entities = self._ner_extractor.extract_entities(text, **options) # Extract relations if not provided if relations is None: @@ -373,15 +398,16 @@ class SemanticNetworkExtractor: # Pass method if specified if "relation_method" in self.config: rel_config["method"] = self.config["relation_method"] - rel_extractor = RelationExtractor( - **rel_config, - **{ - k: v - for k, v in self.config.items() - if k not in ["ner", "relation"] - }, - ) - relations = rel_extractor.extract_relations(text, entities, **options) + if self._relation_extractor is None: + self._relation_extractor = RelationExtractor( + **rel_config, + **{ + k: v + for k, v in self.config.items() + if k not in ["ner", "relation"] + }, + ) + relations = self._relation_extractor.extract_relations(text, entities, **options) # Build network total_steps = 2 # Create nodes, create edges diff --git a/semantica/semantic_extract/triplet_extractor.py b/semantica/semantic_extract/triplet_extractor.py index e1b5c94e..c7f3f9d9 100644 --- a/semantica/semantic_extract/triplet_extractor.py +++ b/semantica/semantic_extract/triplet_extractor.py @@ -143,6 +143,13 @@ class TripletExtractor: if not self.progress_tracker.enabled: self.progress_tracker.enabled = True + if method is not None: + self.config["ner_method"] = method + self.config["relation_method"] = method + + self._ner_extractor = None + self._relation_extractor = None + # Store parameters self.triplet_types = triplet_types self.include_temporal = include_temporal @@ -210,8 +217,12 @@ class TripletExtractor: message=f"Starting batch extraction... 0/{total_items}" ) - # Determine max_workers - max_workers = kwargs.get("max_workers", self.config.get("max_workers", 1)) + from .config import resolve_max_workers + max_workers = resolve_max_workers( + explicit=kwargs.get("max_workers"), + local_config=self.config, + methods=self.method, + ) def process_item(idx, item): try: @@ -307,11 +318,12 @@ class TripletExtractor: def extract_triplets( self, - text: str, - entities: Optional[List[Entity]] = None, - relations: Optional[List[Relation]] = None, + text: Union[str, List[str], List[Dict[str, Any]]], + entities: Optional[Union[List[Entity], List[List[Entity]]]] = None, + relations: Optional[Union[List[Relation], List[List[Relation]]]] = None, + pipeline_id: Optional[str] = None, **options, - ) -> List[Triplet]: + ) -> Union[List[Triplet], List[List[Triplet]]]: """ Extract RDF triplets from text. @@ -319,11 +331,29 @@ class TripletExtractor: text: Input text entities: Pre-extracted entities (optional) relations: Pre-extracted relations (optional) + pipeline_id: Optional pipeline ID for progress tracking (batch mode) **options: Extraction options Returns: list: List of extracted triplets """ + if isinstance(text, list): + entities_batch = entities + if entities is not None and isinstance(entities, list) and (not entities or all(isinstance(e, Entity) for e in entities)): + entities_batch = [entities for _ in range(len(text))] if entities else [[] for _ in range(len(text))] + + relations_batch = relations + if relations is not None and isinstance(relations, list) and (not relations or all(isinstance(r, Relation) for r in relations)): + relations_batch = [relations for _ in range(len(text))] if relations else [[] for _ in range(len(text))] + + return self.extract( + text, + entities=entities_batch, + relations=relations_batch, + pipeline_id=pipeline_id, + **options, + ) + from .methods import get_triplet_method tracking_id = self.progress_tracker.start_tracking( @@ -341,16 +371,38 @@ class TripletExtractor: self.progress_tracker.update_tracking( tracking_id, message="Extracting entities..." ) - ner = NERExtractor(**self.config.get("ner", {})) - entities = ner.extract_entities(text) + if self._ner_extractor is None: + ner_config = self.config.get("ner", {}) + if "ner_method" in self.config: + ner_config = {**ner_config, "method": self.config["ner_method"]} + self._ner_extractor = NERExtractor( + **ner_config, + **{ + k: v + for k, v in self.config.items() + if k not in ["ner", "relation", "validator", "serializer", "quality"] + }, + ) + entities = self._ner_extractor.extract_entities(text) # Extract relations if not provided if relations is None: self.progress_tracker.update_tracking( tracking_id, message="Extracting relations..." ) - rel_extractor = RelationExtractor(**self.config.get("relation", {})) - relations = rel_extractor.extract_relations(text, entities) + if self._relation_extractor is None: + rel_config = self.config.get("relation", {}) + if "relation_method" in self.config: + rel_config = {**rel_config, "method": self.config["relation_method"]} + self._relation_extractor = RelationExtractor( + **rel_config, + **{ + k: v + for k, v in self.config.items() + if k not in ["ner", "relation", "validator", "serializer", "quality"] + }, + ) + relations = self._relation_extractor.extract_relations(text, entities) # Use method-based extraction methods = options.get("method", self.method) diff --git a/tests/semantic_extract/perf_real_use_cases.py b/tests/semantic_extract/perf_real_use_cases.py new file mode 100644 index 00000000..18c8b99d --- /dev/null +++ b/tests/semantic_extract/perf_real_use_cases.py @@ -0,0 +1,125 @@ +import statistics +import time +import os +import sys +from typing import Dict, List + +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../.."))) + +from semantica.semantic_extract.event_detector import EventDetector +from semantica.semantic_extract.ner_extractor import NERExtractor +from semantica.semantic_extract.relation_extractor import RelationExtractor +from semantica.semantic_extract.semantic_analyzer import SemanticAnalyzer +from semantica.semantic_extract.semantic_network_extractor import SemanticNetworkExtractor +from semantica.semantic_extract.triplet_extractor import TripletExtractor +from semantica.utils.progress_tracker import get_progress_tracker + + +def _make_documents(n: int) -> List[Dict[str, str]]: + base = ( + "Apple Inc. was founded by Steve Jobs in 1976 and is headquartered in Cupertino, California. " + "Microsoft Corporation was founded by Bill Gates and Paul Allen in 1975. " + "In 2014, Apple acquired Beats Electronics for $3 billion. " + "In 2023, Google announced a partnership with OpenAI to improve search experiences." + ) + return [{"id": f"doc_{i}", "content": f"{base} Document number {i}."} for i in range(n)] + + +def _median_seconds(fn, repeats: int = 3) -> float: + times = [] + for _ in range(repeats): + start = time.perf_counter() + fn() + times.append(time.perf_counter() - start) + return statistics.median(times) + + +def _bench(label: str, fn, repeats: int = 3) -> dict: + fn() + seconds = _median_seconds(fn, repeats=repeats) + return {"label": label, "seconds": seconds} + + +def main(): + progress = get_progress_tracker() + progress.displays = [] + + docs = _make_documents(80) + texts = [d["content"] for d in docs] + + ner = NERExtractor(method="pattern") + rel = RelationExtractor(method="pattern") + trip = TripletExtractor(method="pattern") + events = EventDetector(method="pattern") + analyzer = SemanticAnalyzer() + net = SemanticNetworkExtractor(ner_method="pattern", relation_method="pattern") + + results = [] + + def ner_parallel(): + ner.extract(texts) + + def ner_seq(): + ner.extract(texts, max_workers=1) + + results.append(_bench("NER batch (default workers)", ner_parallel)) + results.append(_bench("NER batch (max_workers=1)", ner_seq)) + + entities_batch = ner.extract(texts) + + def rel_parallel(): + rel.extract(texts, entities_batch) + + def rel_seq(): + rel.extract(texts, entities_batch, max_workers=1) + + results.append(_bench("Relation batch (default workers)", rel_parallel)) + results.append(_bench("Relation batch (max_workers=1)", rel_seq)) + + def trip_parallel(): + trip.extract(texts) + + def trip_seq(): + trip.extract(texts, max_workers=1) + + results.append(_bench("Triplet pipeline (default workers)", trip_parallel)) + results.append(_bench("Triplet pipeline (max_workers=1)", trip_seq)) + + def ev_parallel(): + events.detect_events(texts) + + def ev_seq(): + events.detect_events(texts, max_workers=1) + + results.append(_bench("Event detection (default workers)", ev_parallel)) + results.append(_bench("Event detection (max_workers=1)", ev_seq)) + + def analyzer_parallel(): + analyzer.analyze(texts) + + def analyzer_seq(): + analyzer.analyze(texts, max_workers=1) + + results.append(_bench("Semantic analysis (default workers)", analyzer_parallel)) + results.append(_bench("Semantic analysis (max_workers=1)", analyzer_seq)) + + def net_parallel(): + net.extract_network(texts) + + def net_seq(): + net.extract_network(texts, max_workers=1) + + results.append(_bench("Semantic network (default workers)", net_parallel)) + results.append(_bench("Semantic network (max_workers=1)", net_seq)) + + per_doc = [] + for row in results: + per_doc.append({**row, "ms_per_doc": (row["seconds"] / len(texts)) * 1000.0}) + + print(f"Documents: {len(texts)}") + for row in per_doc: + print(f"{row['label']}: {row['seconds']:.3f}s ({row['ms_per_doc']:.2f} ms/doc)") + + +if __name__ == "__main__": + main() diff --git a/tests/semantic_extract/test_extractors.py b/tests/semantic_extract/test_extractors.py index 6258394e..7aee7515 100644 --- a/tests/semantic_extract/test_extractors.py +++ b/tests/semantic_extract/test_extractors.py @@ -7,8 +7,12 @@ import os sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../.."))) from semantica.semantic_extract.ner_extractor import NERExtractor +from semantica.semantic_extract.ner_extractor import Entity as NEREntity from semantica.semantic_extract.relation_extractor import RelationExtractor from semantica.semantic_extract.triplet_extractor import TripletExtractor +from semantica.semantic_extract.event_detector import EventDetector +from semantica.semantic_extract.semantic_analyzer import SemanticAnalyzer +from semantica.semantic_extract.semantic_network_extractor import SemanticNetworkExtractor from semantica.semantic_extract.named_entity_recognizer import Entity from semantica.semantic_extract.relation_extractor import Relation @@ -51,6 +55,21 @@ class TestExtractors(unittest.TestCase): self.assertIsInstance(entities, list) mock_get_method.assert_called() + @patch("semantica.semantic_extract.methods.get_entity_method") + def test_ner_extraction_batch_via_extract_entities(self, mock_get_method): + mock_method = MagicMock() + mock_method.extract_entities.return_value = [] + mock_get_method.return_value = mock_method + + extractor = NERExtractor(method="pattern") + results = extractor.extract_entities( + ["Test text 1", "Test text 2"], + ) + + self.assertIsInstance(results, list) + self.assertEqual(len(results), 2) + self.assertTrue(all(isinstance(r, list) for r in results)) + @patch("semantica.semantic_extract.methods.get_relation_method") def test_relation_extraction(self, mock_get_method): """Test relation extraction call""" @@ -65,6 +84,21 @@ class TestExtractors(unittest.TestCase): self.assertIsInstance(relations, list) mock_get_method.assert_called() + @patch("semantica.semantic_extract.methods.get_relation_method") + def test_relation_extraction_batch_via_extract_relations(self, mock_get_method): + mock_method = MagicMock() + mock_method.extract_relations.return_value = [] + mock_get_method.return_value = mock_method + + extractor = RelationExtractor(method="pattern") + texts = ["A knows B", "A knows B"] + entities = [NEREntity(text="A", label="PERSON", start_char=0, end_char=1, confidence=1.0)] + + results = extractor.extract_relations(texts, entities) + self.assertIsInstance(results, list) + self.assertEqual(len(results), 2) + self.assertTrue(all(isinstance(r, list) for r in results)) + @patch("semantica.semantic_extract.methods.get_triplet_method") def test_triplet_extraction(self, mock_get_method): """Test triplet extraction call""" @@ -81,5 +115,55 @@ class TestExtractors(unittest.TestCase): self.assertIsInstance(triplets, list) mock_get_method.assert_called() + @patch("semantica.semantic_extract.methods.get_triplet_method") + def test_triplet_extraction_batch_via_extract_triplets(self, mock_get_method): + mock_method = MagicMock() + mock_method.extract_triplets.return_value = [] + mock_get_method.return_value = mock_method + + extractor = TripletExtractor(method="pattern") + texts = ["A knows A", "A knows A"] + entities_batch = [[NEREntity(text="A", label="PERSON", start_char=0, end_char=1, confidence=1.0)] for _ in texts] + relations_batch = [[] for _ in texts] + + results = extractor.extract_triplets( + texts, + entities=entities_batch, + relations=relations_batch, + ) + + self.assertIsInstance(results, list) + self.assertEqual(len(results), 2) + self.assertTrue(all(isinstance(r, list) for r in results)) + + def test_event_detector_batch_via_detect_events(self): + detector = EventDetector() + texts = ["Apple acquired Beats in 2014.", "Google announced a partnership in 2023."] + results = detector.detect_events(texts) + self.assertIsInstance(results, list) + self.assertEqual(len(results), 2) + self.assertTrue(all(isinstance(r, list) for r in results)) + + def test_semantic_analyzer_batch_parallel(self): + analyzer = SemanticAnalyzer() + texts = ["A short sentence.", "Another short sentence."] + results = analyzer.analyze(texts) + self.assertIsInstance(results, list) + self.assertEqual(len(results), 2) + self.assertTrue(all(isinstance(r, dict) for r in results)) + + def test_semantic_network_batch_via_extract_network(self): + extractor = SemanticNetworkExtractor() + texts = ["A knows B.", "C knows D."] + entities_batch = [[] for _ in texts] + relations_batch = [[] for _ in texts] + results = extractor.extract_network( + texts, + entities=entities_batch, + relations=relations_batch, + ) + self.assertIsInstance(results, list) + self.assertEqual(len(results), 2) + if __name__ == "__main__": unittest.main() diff --git a/tests/semantic_extract/test_llm_smoke_groq.py b/tests/semantic_extract/test_llm_smoke_groq.py new file mode 100644 index 00000000..c19f4204 --- /dev/null +++ b/tests/semantic_extract/test_llm_smoke_groq.py @@ -0,0 +1,63 @@ +import os + +import pytest + +try: + from dotenv import load_dotenv + + load_dotenv() +except Exception: + pass + +pytest.importorskip("groq") + +from semantica.semantic_extract.ner_extractor import NERExtractor +from semantica.semantic_extract.relation_extractor import RelationExtractor +from semantica.semantic_extract.triplet_extractor import TripletExtractor + + +def test_groq_llm_smoke_entities_relations_triplets(): + if not os.getenv("GROQ_API_KEY"): + pytest.skip("GROQ_API_KEY is not set") + + text = ( + "Apple acquired Beats in 2014 for $3 billion. " + "Steve Jobs founded Apple. " + "Beats is based in California." + ) + model = "llama-3.3-70b-versatile" + + entities = NERExtractor(method="llm").extract( + text, + provider="groq", + model=model, + temperature=0.0, + max_tokens=250, + ) + assert isinstance(entities, list) + assert len(entities) > 0 + assert len(entities) <= 30 + + relations = RelationExtractor(method="llm").extract( + text, + entities=entities, + provider="groq", + model=model, + temperature=0.0, + max_tokens=350, + max_entities_prompt=12, + ) + assert isinstance(relations, list) + assert len(relations) <= 30 + + triplets = TripletExtractor(method="llm").extract( + text, + entities=entities, + relations=relations, + provider="groq", + model=model, + temperature=0.0, + max_tokens=350, + ) + assert isinstance(triplets, list) + assert len(triplets) <= 40 diff --git a/tests/semantic_extract/test_performance.py b/tests/semantic_extract/test_performance.py index 3152defd..a98afab7 100644 --- a/tests/semantic_extract/test_performance.py +++ b/tests/semantic_extract/test_performance.py @@ -168,12 +168,11 @@ class TestSemanticExtractImprovements(unittest.TestCase): texts = ["Text 1", "Text 2", "Text 3", "Text 4"] start_time = time.time() - # Run with 2 workers - results = extractor.extract(texts, max_workers=2) + results = extractor.extract(texts) end_time = time.time() duration = end_time - start_time - print(f" Parallel NER (2 workers) took {duration:.4f}s") + print(f" Parallel NER (default workers) took {duration:.4f}s") self.assertEqual(len(results), 4) @@ -206,11 +205,11 @@ class TestSemanticExtractImprovements(unittest.TestCase): entities = [[], [], [], []] start_time = time.time() - results = extractor.extract(texts, entities, max_workers=2) + results = extractor.extract(texts, entities) end_time = time.time() duration = end_time - start_time - print(f" Parallel RE (2 workers) took {duration:.4f}s") + print(f" Parallel RE (default workers) took {duration:.4f}s") self.assertEqual(len(results), 4) @@ -264,11 +263,11 @@ class TestSemanticExtractImprovements(unittest.TestCase): texts = ["Text 1", "Text 2", "Text 3", "Text 4"] start_time = time.time() - results = extractor.extract(texts, max_workers=2) + results = extractor.extract(texts) end_time = time.time() duration = end_time - start_time - print(f" Parallel TE (2 workers) took {duration:.4f}s") + print(f" Parallel TE (default workers) took {duration:.4f}s") self.assertEqual(len(results), 4) diff --git a/tests/semantic_extract/test_regression_performance_edge_cases.py b/tests/semantic_extract/test_regression_performance_edge_cases.py new file mode 100644 index 00000000..38caabca --- /dev/null +++ b/tests/semantic_extract/test_regression_performance_edge_cases.py @@ -0,0 +1,123 @@ +import multiprocessing +import time +from unittest.mock import MagicMock, patch + +import pytest + +from semantica.semantic_extract.config import resolve_max_workers +from semantica.semantic_extract.ner_extractor import Entity, NERExtractor +from semantica.semantic_extract.relation_extractor import RelationExtractor +from semantica.semantic_extract.triplet_extractor import TripletExtractor +from semantica.semantic_extract.semantic_network_extractor import SemanticNetworkExtractor +from semantica.semantic_extract.methods import filter_entities_for_text +from semantica.semantic_extract.schemas import RelationsResponse, RelationOut + + +def test_resolve_max_workers_defaults_and_clamps(): + cpu_count = multiprocessing.cpu_count() or 1 + + assert resolve_max_workers(explicit=0) == 1 + assert resolve_max_workers(explicit=-10) == 1 + assert resolve_max_workers(explicit=1) == 1 + assert resolve_max_workers(explicit=10**9) == min(cpu_count, 32) + + assert resolve_max_workers(explicit=None, methods=["ml"]) == 1 + + +def test_filter_entities_for_text_keeps_short_tokens(): + text = "US AI lab in NY" + entities = [ + Entity(text="US", label="GPE", start_char=0, end_char=2, confidence=1.0), + Entity(text="AI", label="TECH", start_char=3, end_char=5, confidence=1.0), + Entity(text="NY", label="GPE", start_char=13, end_char=15, confidence=1.0), + ] + kept = filter_entities_for_text(text, entities, max_keep=2) + kept_texts = {e.text for e in kept} + assert "US" in kept_texts or "AI" in kept_texts or "NY" in kept_texts + + +def test_pattern_batch_defaults_to_single_worker_low_latency(): + extractor = NERExtractor(method="pattern") + texts = [f"Text {i}" for i in range(8)] + extractor.extract(texts) + + +def test_relation_llm_prompt_filter_does_not_break_mapping(): + entities = [Entity(text=f"VeryLongEntityName{i}", label="ORG", start_char=0, end_char=1, confidence=1.0) for i in range(120)] + ghost = Entity(text="Ghost", label="ORG", start_char=0, end_char=1, confidence=1.0) + entities.append(ghost) + + captured = {} + + class FakeLLM: + def is_available(self): + return True + + def generate_typed(self, prompt, schema, **kwargs): + captured["prompt"] = prompt + return RelationsResponse( + relations=[ + RelationOut(subject="Ghost", predicate="related_to", object="VeryLongEntityName0", confidence=0.9) + ] + ) + + with patch("semantica.semantic_extract.methods.create_provider", return_value=FakeLLM()): + from semantica.semantic_extract.methods import extract_relations_llm + + relations = extract_relations_llm( + "Short text mentioning VeryLongEntityName0 only.", + entities=entities, + provider="openai", + model="gpt-4", + max_entities_prompt=20, + ) + + assert "Ghost" not in captured["prompt"] + assert len(relations) == 1 + assert relations[0].subject.text == "Ghost" + + +def test_triplet_extractor_reuses_sub_extractors(): + ner_instance = MagicMock() + ner_instance.extract_entities.return_value = [ + Entity(text="A", label="PERSON", start_char=0, end_char=1, confidence=1.0) + ] + + rel_instance = MagicMock() + rel_instance.extract_relations.return_value = [] + + ner_ctor = MagicMock(return_value=ner_instance) + rel_ctor = MagicMock(return_value=rel_instance) + + with patch("semantica.semantic_extract.ner_extractor.NERExtractor", ner_ctor), patch( + "semantica.semantic_extract.relation_extractor.RelationExtractor", rel_ctor + ), patch("semantica.semantic_extract.methods.get_triplet_method", return_value=lambda *args, **kwargs: []): + extractor = TripletExtractor(method="pattern") + extractor.extract_triplets("A text.") + extractor.extract_triplets("A text again.") + + assert ner_ctor.call_count == 1 + assert rel_ctor.call_count == 1 + + +def test_semantic_network_extractor_reuses_sub_extractors(): + ner_instance = MagicMock() + ner_instance.extract_entities.return_value = [ + Entity(text="A", label="PERSON", start_char=0, end_char=1, confidence=1.0) + ] + + rel_instance = MagicMock() + rel_instance.extract_relations.return_value = [] + + ner_ctor = MagicMock(return_value=ner_instance) + rel_ctor = MagicMock(return_value=rel_instance) + + with patch("semantica.semantic_extract.ner_extractor.NERExtractor", ner_ctor), patch( + "semantica.semantic_extract.relation_extractor.RelationExtractor", rel_ctor + ): + extractor = SemanticNetworkExtractor(method="pattern") + extractor.extract_network("A text.") + extractor.extract_network("A text again.") + + assert ner_ctor.call_count == 1 + assert rel_ctor.call_count == 1