From 53db5bbdc0237c3266824488a509071e699cddcb Mon Sep 17 00:00:00 2001 From: KaifAhmad1 Date: Sun, 28 Dec 2025 20:32:56 +0530 Subject: [PATCH] Add progress tracking with ETA to all long-running operations - Fixed ConflictDetector to use update_progress() with counts/ETA for type, temporal, and logical conflict detection - Fixed NERExtractor batch operations to show progress with ETA - Fixed RelationExtractor batch operations to show progress with ETA - All modules now display clear progress bars with percentage, counts, and estimated time remaining --- .../02_Threat_Intelligence_Hybrid_RAG.ipynb | 28 +++- semantica/conflicts/conflict_detector.py | 145 ++++++++++++++---- semantica/semantic_extract/ner_extractor.py | 55 +++++-- .../semantic_extract/relation_extractor.py | 69 ++++++--- 4 files changed, 235 insertions(+), 62 deletions(-) diff --git a/cookbook/use_cases/cybersecurity/02_Threat_Intelligence_Hybrid_RAG.ipynb b/cookbook/use_cases/cybersecurity/02_Threat_Intelligence_Hybrid_RAG.ipynb index e1d7a628..00294040 100644 --- a/cookbook/use_cases/cybersecurity/02_Threat_Intelligence_Hybrid_RAG.ipynb +++ b/cookbook/use_cases/cybersecurity/02_Threat_Intelligence_Hybrid_RAG.ipynb @@ -63,7 +63,7 @@ }, { "cell_type": "code", - "execution_count": 1, + "execution_count": 2, "metadata": {}, "outputs": [ { @@ -102,7 +102,7 @@ }, { "cell_type": "code", - "execution_count": 2, + "execution_count": 3, "metadata": {}, "outputs": [], "source": [ @@ -134,7 +134,25 @@ "name": "stdout", "output_type": "stream", "text": [ - "Ingesting from 6 feed sources...\n", + "Ingesting from 6 feed sources...\n" + ] + }, + { + "data": { + "text/html": [ + "

🧠 Semantica - 📊 Current Progress

StatusActionModuleSubmoduleProgressETARateTime
❌Semantica is parsing🔍 parseDocumentParser---0.00s
❌Semantica is parsing🔍 parseDocumentParser---0.00s
❌Semantica is parsing🔍 parseDocumentParser---0.00s
❌Semantica is parsing🔍 parseDocumentParser---0.01s
❌Semantica is parsing🔍 parseDocumentParser---0.00s
✅Semantica is normalizing🔧 normalizeTextNormalizer100.0%--0.00s
✅Semantica is extracting🎯 semantic_extractNERExtractor100.0%--0.48s
✅Semantica is extracting🎯 semantic_extractRelationExtractor100.0%--0.44s
🔄Semantica is deduplicating🔄 deduplicationDuplicateDetector---519.18s
🔄Semantica is deduplicating🔄 deduplicationSimilarityCalculator---0.02s
" + ], + "text/plain": [ + "" + ] + }, + "metadata": {}, + "output_type": "display_data" + }, + { + "name": "stdout", + "output_type": "stream", + "text": [ " [1/6] US-CERT Alerts: 10 documents\n", " [2/6] SANS ISC: 10 documents\n", " [3/6] Krebs on Security: 10 documents\n", @@ -319,7 +337,7 @@ }, { "cell_type": "code", - "execution_count": null, + "execution_count": 7, "metadata": {}, "outputs": [ { @@ -380,7 +398,7 @@ }, { "cell_type": "code", - "execution_count": 9, + "execution_count": 8, "metadata": {}, "outputs": [ { diff --git a/semantica/conflicts/conflict_detector.py b/semantica/conflicts/conflict_detector.py index b12fa68b..b9839233 100644 --- a/semantica/conflicts/conflict_detector.py +++ b/semantica/conflicts/conflict_detector.py @@ -614,7 +614,7 @@ class ConflictDetector: file=None, module="conflicts", submodule="ConflictDetector", - message="Detecting type conflicts", + message=f"Detecting type conflicts in {len(entities)} entities", ) try: @@ -622,8 +622,10 @@ class ConflictDetector: # Group entities by ID entity_groups: Dict[str, List[Dict[str, Any]]] = {} + total_entities = len(entities) + update_interval = max(1, total_entities // 20) # Update every 5% - for entity in entities: + for i, entity in enumerate(entities): entity_id = entity.get("id") or entity.get("entity_id") if not entity_id: continue @@ -631,9 +633,21 @@ class ConflictDetector: if entity_id not in entity_groups: entity_groups[entity_id] = [] entity_groups[entity_id].append(entity) + + # Update progress periodically + if (i + 1) % update_interval == 0 or (i + 1) == total_entities: + self.progress_tracker.update_progress( + tracking_id, + processed=i + 1, + total=total_entities, + message=f"Grouping entities... {i + 1}/{total_entities}" + ) # Check each entity group for type conflicts - for entity_id, entity_list in entity_groups.items(): + total_groups = len(entity_groups) + group_update_interval = max(1, total_groups // 20) # Update every 5% + + for j, (entity_id, entity_list) in enumerate(entity_groups.items()): if len(entity_list) < 2: continue # Need at least 2 sources to have conflict @@ -688,6 +702,15 @@ class ConflictDetector: f"Type conflict detected: {entity_id} conflicting types: " f"{unique_types}" ) + + # Update progress periodically for group checking + if (j + 1) % group_update_interval == 0 or (j + 1) == total_groups: + self.progress_tracker.update_progress( + tracking_id, + processed=j + 1, + total=total_groups, + message=f"Checking entity groups for type conflicts... {j + 1}/{total_groups}" + ) self.progress_tracker.stop_tracking( tracking_id, @@ -718,7 +741,7 @@ class ConflictDetector: file=None, module="conflicts", submodule="ConflictDetector", - message="Detecting temporal conflicts", + message=f"Detecting temporal conflicts in {len(entities)} entities", ) try: @@ -738,8 +761,10 @@ class ConflictDetector: # Group entities by ID entity_groups: Dict[str, List[Dict[str, Any]]] = {} + total_entities = len(entities) + update_interval = max(1, total_entities // 20) # Update every 5% - for entity in entities: + for i, entity in enumerate(entities): entity_id = entity.get("id") or entity.get("entity_id") if not entity_id: continue @@ -747,9 +772,21 @@ class ConflictDetector: if entity_id not in entity_groups: entity_groups[entity_id] = [] entity_groups[entity_id].append(entity) + + # Update progress periodically + if (i + 1) % update_interval == 0 or (i + 1) == total_entities: + self.progress_tracker.update_progress( + tracking_id, + processed=i + 1, + total=total_entities, + message=f"Grouping entities... {i + 1}/{total_entities}" + ) # Check each entity group for temporal conflicts - for entity_id, entity_list in entity_groups.items(): + total_groups = len(entity_groups) + group_update_interval = max(1, total_groups // 20) # Update every 5% + + for j, (entity_id, entity_list) in enumerate(entity_groups.items()): if len(entity_list) < 2: continue @@ -834,6 +871,15 @@ class ConflictDetector: f"Temporal conflict detected: {entity_id}.{prop_name} " f"has conflicting values: {unique_values}" ) + + # Update progress periodically for group checking + if (j + 1) % group_update_interval == 0 or (j + 1) == total_groups: + self.progress_tracker.update_progress( + tracking_id, + processed=j + 1, + total=total_groups, + message=f"Checking entity groups for temporal conflicts... {j + 1}/{total_groups}" + ) self.progress_tracker.stop_tracking( tracking_id, @@ -864,7 +910,7 @@ class ConflictDetector: file=None, module="conflicts", submodule="ConflictDetector", - message="Detecting logical conflicts", + message=f"Detecting logical conflicts in {len(entities)} entities", ) try: @@ -880,8 +926,10 @@ class ConflictDetector: # Group entities by ID entity_groups: Dict[str, List[Dict[str, Any]]] = {} + total_entities = len(entities) + update_interval = max(1, total_entities // 20) # Update every 5% - for entity in entities: + for i, entity in enumerate(entities): entity_id = entity.get("id") or entity.get("entity_id") if not entity_id: continue @@ -889,9 +937,21 @@ class ConflictDetector: if entity_id not in entity_groups: entity_groups[entity_id] = [] entity_groups[entity_id].append(entity) + + # Update progress periodically + if (i + 1) % update_interval == 0 or (i + 1) == total_entities: + self.progress_tracker.update_progress( + tracking_id, + processed=i + 1, + total=total_entities, + message=f"Grouping entities... {i + 1}/{total_entities}" + ) # Check each entity group for logical conflicts - for entity_id, entity_list in entity_groups.items(): + total_groups = len(entity_groups) + group_update_interval = max(1, total_groups // 20) # Update every 5% + + for j, (entity_id, entity_list) in enumerate(entity_groups.items()): if len(entity_list) < 2: continue @@ -953,6 +1013,15 @@ class ConflictDetector: f"{type1_str} and {type2_str}" ) break + + # Update progress periodically for group checking + if (j + 1) % group_update_interval == 0 or (j + 1) == total_groups: + self.progress_tracker.update_progress( + tracking_id, + processed=j + 1, + total=total_groups, + message=f"Checking entity groups for logical conflicts... {j + 1}/{total_groups}" + ) self.progress_tracker.stop_tracking( tracking_id, @@ -997,7 +1066,7 @@ class ConflictDetector: file=None, module="conflicts", submodule="ConflictDetector", - message="Detecting all conflicts", + message=f"Detecting all conflicts in {len(entities)} entities", ) try: @@ -1012,31 +1081,55 @@ class ConflictDetector: if (e.get("type") or e.get("entity_type")) == entity_type ] + # Track overall progress across multiple detection methods + detection_steps = [] + # Detect value conflicts (for common properties) if self.conflict_fields: for entity_type_key, fields in self.conflict_fields.items(): if not entity_type or entity_type_key == entity_type: for field_name in fields: - conflicts = self.detect_value_conflicts( - filtered_entities, field_name, entity_type - ) - all_conflicts.extend(conflicts) + detection_steps.append(("value", field_name)) else: - # Detect entity-wide conflicts - conflicts = self.detect_entity_conflicts(filtered_entities, entity_type) - all_conflicts.extend(conflicts) + detection_steps.append(("entity_wide", None)) - # Detect type conflicts - type_conflicts = self.detect_type_conflicts(filtered_entities) - all_conflicts.extend(type_conflicts) + # Add other detection steps + detection_steps.extend([ + ("type", None), + ("temporal", None), + ("logical", None) + ]) - # Detect temporal conflicts - temporal_conflicts = self.detect_temporal_conflicts(filtered_entities) - all_conflicts.extend(temporal_conflicts) + total_steps = len(detection_steps) + update_interval = max(1, total_steps // 10) # Update every 10% - # Detect logical conflicts - logical_conflicts = self.detect_logical_conflicts(filtered_entities) - all_conflicts.extend(logical_conflicts) + for step_idx, (step_type, step_param) in enumerate(detection_steps): + if step_type == "value": + conflicts = self.detect_value_conflicts( + filtered_entities, step_param, entity_type + ) + all_conflicts.extend(conflicts) + elif step_type == "entity_wide": + conflicts = self.detect_entity_conflicts(filtered_entities, entity_type) + all_conflicts.extend(conflicts) + elif step_type == "type": + conflicts = self.detect_type_conflicts(filtered_entities) + all_conflicts.extend(conflicts) + elif step_type == "temporal": + conflicts = self.detect_temporal_conflicts(filtered_entities) + all_conflicts.extend(conflicts) + elif step_type == "logical": + conflicts = self.detect_logical_conflicts(filtered_entities) + all_conflicts.extend(conflicts) + + # Update progress periodically + if (step_idx + 1) % update_interval == 0 or (step_idx + 1) == total_steps: + self.progress_tracker.update_progress( + tracking_id, + processed=step_idx + 1, + total=total_steps, + message=f"Detecting conflicts... {step_idx + 1}/{total_steps} steps completed" + ) self.progress_tracker.stop_tracking( tracking_id, diff --git a/semantica/semantic_extract/ner_extractor.py b/semantica/semantic_extract/ner_extractor.py index 3e02a993..bf5bdca3 100644 --- a/semantica/semantic_extract/ner_extractor.py +++ b/semantica/semantic_extract/ner_extractor.py @@ -155,20 +155,53 @@ class NERExtractor: Union[List[Entity], List[List[Entity]]]: Extracted entities """ if isinstance(text, list): - # Handle batch extraction - results = [] - for item in text: - if isinstance(item, dict) and "content" in item: - results.append(self.extract_entities(item["content"], **kwargs)) - elif isinstance(item, str): - results.append(self.extract_entities(item, **kwargs)) - else: - # Try converting to string + # Handle batch extraction with progress tracking + tracking_id = self.progress_tracker.start_tracking( + module="semantic_extract", + submodule="NERExtractor", + message=f"Batch extracting entities from {len(text)} documents", + ) + + try: + results = [] + total_items = len(text) + update_interval = max(1, total_items // 20) # Update every 5% + + for idx, item in enumerate(text, 1): try: - results.append(self.extract_entities(str(item), **kwargs)) + if isinstance(item, dict) and "content" in item: + results.append(self.extract_entities(item["content"], **kwargs)) + elif isinstance(item, str): + results.append(self.extract_entities(item, **kwargs)) + else: + # Try converting to string + try: + results.append(self.extract_entities(str(item), **kwargs)) + except Exception: + results.append([]) except Exception: results.append([]) - return results + + # Update progress periodically + if idx % update_interval == 0 or idx == total_items: + self.progress_tracker.update_progress( + tracking_id, + processed=idx, + total=total_items, + message=f"Processing documents... {idx}/{total_items}" + ) + + self.progress_tracker.stop_tracking( + tracking_id, + status="completed", + message=f"Extracted entities from {len(results)} documents", + ) + return results + except Exception as e: + self.progress_tracker.stop_tracking( + tracking_id, status="failed", message=str(e) + ) + raise else: return self.extract_entities(text, **kwargs) diff --git a/semantica/semantic_extract/relation_extractor.py b/semantica/semantic_extract/relation_extractor.py index 59bc32f9..51487d34 100644 --- a/semantica/semantic_extract/relation_extractor.py +++ b/semantica/semantic_extract/relation_extractor.py @@ -180,28 +180,57 @@ class RelationExtractor: Union[List[Relation], List[List[Relation]]]: Extracted relations """ if isinstance(text, list) and isinstance(entities, list): - # Handle batch extraction - results = [] - # Ensure lists are same length - min_len = min(len(text), len(entities)) - for i in range(min_len): - doc_item = text[i] - ent_item = entities[i] + # Handle batch extraction with progress tracking + tracking_id = self.progress_tracker.start_tracking( + module="semantic_extract", + submodule="RelationExtractor", + message=f"Batch extracting relations from {len(text)} documents", + ) + + try: + results = [] + # Ensure lists are same length + min_len = min(len(text), len(entities)) + update_interval = max(1, min_len // 20) # Update every 5% - doc_text = "" - if isinstance(doc_item, dict) and "content" in doc_item: - doc_text = doc_item["content"] - elif isinstance(doc_item, str): - doc_text = doc_item - else: - doc_text = str(doc_item) + for i in range(min_len): + doc_item = text[i] + ent_item = entities[i] + + doc_text = "" + if isinstance(doc_item, dict) and "content" in doc_item: + doc_text = doc_item["content"] + elif isinstance(doc_item, str): + doc_text = doc_item + else: + doc_text = str(doc_item) + + # Ensure ent_item is a list of entities + if not isinstance(ent_item, list): + ent_item = [] # Should not happen if entities is List[List[Entity]] + + results.append(self.extract_relations(doc_text, ent_item, **kwargs)) + + # Update progress periodically + if (i + 1) % update_interval == 0 or (i + 1) == min_len: + self.progress_tracker.update_progress( + tracking_id, + processed=i + 1, + total=min_len, + message=f"Processing documents... {i + 1}/{min_len}" + ) - # Ensure ent_item is a list of entities - if not isinstance(ent_item, list): - ent_item = [] # Should not happen if entities is List[List[Entity]] - - results.append(self.extract_relations(doc_text, ent_item, **kwargs)) - return results + self.progress_tracker.stop_tracking( + tracking_id, + status="completed", + message=f"Extracted relations from {len(results)} documents", + ) + return results + except Exception as e: + self.progress_tracker.stop_tracking( + tracking_id, status="failed", message=str(e) + ) + raise elif isinstance(text, str) and isinstance(entities, list): # Single text, single list of entities (standard case) return self.extract_relations(text, entities, **kwargs)