mengzhx/semantica
1
0
Fork 0
mirror of https://github.com/semantica-agi/semantica.git synced 2026-09-15 04:00:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #119 from Hawksight-AI/utils

Browse Source 
Add Progress Tracking with ETA to Long-Running Operations
This commit is contained in:
Mohd Kaif
2025-12-28 20:36:46 +05:30
committed by GitHub
parent 8313cd73a0 53db5bbdc0
commit 9e0aa28eb1
4 changed files with 235 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cookbook/use_cases/cybersecurity/02_Threat_Intelligence_Hybrid_RAG.ipynb
+23 -5
View File
@@ -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": [
"<div style='font-family: monospace;'><h4>🧠 Semantica - 📊 Current Progress</h4><table style='width: 100%; border-collapse: collapse;'><tr><th>Status</th><th>Action</th><th>Module</th><th>Submodule</th><th>Progress</th><th>ETA</th><th>Rate</th><th>Time</th></tr><tr><td>❌</td><td>Semantica is parsing</td><td>🔍 parse</td><td>DocumentParser</td><td>-</td><td>-</td><td>-</td><td>0.00s</td></tr><tr><td>❌</td><td>Semantica is parsing</td><td>🔍 parse</td><td>DocumentParser</td><td>-</td><td>-</td><td>-</td><td>0.00s</td></tr><tr><td>❌</td><td>Semantica is parsing</td><td>🔍 parse</td><td>DocumentParser</td><td>-</td><td>-</td><td>-</td><td>0.00s</td></tr><tr><td>❌</td><td>Semantica is parsing</td><td>🔍 parse</td><td>DocumentParser</td><td>-</td><td>-</td><td>-</td><td>0.01s</td></tr><tr><td>❌</td><td>Semantica is parsing</td><td>🔍 parse</td><td>DocumentParser</td><td>-</td><td>-</td><td>-</td><td>0.00s</td></tr><tr><td>✅</td><td>Semantica is normalizing</td><td>🔧 normalize</td><td>TextNormalizer</td><td>100.0%</td><td>-</td><td>-</td><td>0.00s</td></tr><tr><td>✅</td><td>Semantica is extracting</td><td>🎯 semantic_extract</td><td>NERExtractor</td><td>100.0%</td><td>-</td><td>-</td><td>0.48s</td></tr><tr><td>✅</td><td>Semantica is extracting</td><td>🎯 semantic_extract</td><td>RelationExtractor</td><td>100.0%</td><td>-</td><td>-</td><td>0.44s</td></tr><tr><td>🔄</td><td>Semantica is deduplicating</td><td>🔄 deduplication</td><td>DuplicateDetector</td><td>-</td><td>-</td><td>-</td><td>519.18s</td></tr><tr><td>🔄</td><td>Semantica is deduplicating</td><td>🔄 deduplication</td><td>SimilarityCalculator</td><td>-</td><td>-</td><td>-</td><td>0.02s</td></tr></table></div>"
],
"text/plain": [
"<IPython.core.display.HTML object>"
]
},
"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": [
{
semantica/conflicts/conflict_detector.py
+119 -26
View File
@@ -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,
semantica/semantic_extract/ner_extractor.py
+44 -11
View File
@@ -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)
semantica/semantic_extract/relation_extractor.py
+49 -20
View File
@@ -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)