From 91ba5219d0e5048cc5edc8cbe479594b47c3383e Mon Sep 17 00:00:00 2001 From: ZohaibHassan16 Date: Sat, 21 Feb 2026 18:30:28 +0500 Subject: [PATCH] feat(dedup): implement semantic relationship and triplet dedup v2 (#336) --- .../quality_assurance/test_deduplication.py | 57 +++++ semantica/deduplication/duplicate_detector.py | 203 ++++++++++++------ semantica/deduplication/merge_strategy.py | 35 ++- semantica/deduplication/methods.py | 35 ++- 4 files changed, 258 insertions(+), 72 deletions(-) diff --git a/benchmarks/quality_assurance/test_deduplication.py b/benchmarks/quality_assurance/test_deduplication.py index 19d0c15e..dfad2668 100644 --- a/benchmarks/quality_assurance/test_deduplication.py +++ b/benchmarks/quality_assurance/test_deduplication.py @@ -110,6 +110,35 @@ def generate_entity_cluster(base_name: str, size: int) -> List[Dict[str, Any]]: return entities +def generate_relationship_dataset(size: int) -> List[Dict[str, Any]]: + """ + Generates a dataset of graph relationships/triplets. + Includes exact matches, synonym predicates, and dirty literal strings. + """ + relationships = [] + predicates = ["works_for", "employed_by", "is_employee_of", "has_employer"] + + for i in range(size): + # Base relationship + rel = { + "subject": f"Person_{i % 50}", + "predicate": random.choice(predicates), + "object": f"Company_{i % 10}" + } + relationships.append(rel) + + # Inject semantic duplicates (dirty literals / synonym predicates) + if random.random() < 0.4: + dirty_rel = { + "subject": f"Person_{i % 50}", + "predicate": random.choice(predicates), + "object": f" Company_{i % 10} Inc. " + } + relationships.append(dirty_rel) + + return relationships + + def generate_dataset( num_clusters: int, items_per_cluster: int, worst_case_blocking: bool = False ): @@ -278,4 +307,32 @@ def test_merge_entity_benchmark(benchmark): lambda: merger.merge_entity_group(group, strategy="keep_most_complete"), iterations=10, rounds=10, + ) + + +@pytest.mark.parametrize("mode", ["legacy", "semantic_v2"]) +def test_relationship_dedup_speed(benchmark, mode): + """ + Measures the speed of relationship/triplet deduplication. + Compares the O(N^2) legacy fallback vs the fast canonical hash path. + """ + # Yields ~280 relationships (approx 39,000 comparisons in O(N^2)) + relationships = generate_relationship_dataset(200) + + detector = DuplicateDetector() + options = { + "threshold": 0.85, + "relationship_dedup_mode": mode, + "predicate_synonym_map": { + "works_for": "employed_by", + "is_employee_of": "employed_by", + "has_employer": "employed_by" + }, + "literal_normalization_enabled": True + } + + benchmark.pedantic( + lambda: detector.detect_relationship_duplicates(relationships, **options), + iterations=5, + rounds=10, ) \ No newline at end of file diff --git a/semantica/deduplication/duplicate_detector.py b/semantica/deduplication/duplicate_detector.py index b3d0ff5c..fc89dddd 100644 --- a/semantica/deduplication/duplicate_detector.py +++ b/semantica/deduplication/duplicate_detector.py @@ -32,7 +32,7 @@ Example Usage: >>> detector = DuplicateDetector(similarity_threshold=0.8, confidence_threshold=0.7) >>> duplicates = detector.detect_duplicates(entities) >>> groups = detector.detect_duplicate_groups(entities) - >>> + >>> >>> # Incremental detection >>> new_candidates = detector.incremental_detect(new_entities, existing_entities) @@ -222,37 +222,37 @@ class DuplicateDetector: update_interval = 1 # Update every item for small datasets else: update_interval = max(1, min(10, total_similarities // 100)) - + # Initial progress update - ALWAYS show this remaining = total_similarities self.progress_tracker.update_progress( tracking_id, processed=0, total=total_similarities, - message=f"Creating duplicate candidates... 0/{total_similarities} (remaining: {remaining})" + message=f"Creating duplicate candidates... 0/{total_similarities} (remaining: {remaining})", ) - + for i, (entity1, entity2, score) in enumerate(similarities): candidate = self._create_duplicate_candidate(entity1, entity2, score) # Filter by confidence threshold if candidate.confidence >= self.confidence_threshold: candidates.append(candidate) - + remaining = total_similarities - (i + 1) # Update progress: always update for small datasets, or at intervals for large ones should_update = ( - (i + 1) % update_interval == 0 or - (i + 1) == total_similarities or - i == 0 or - total_similarities <= 10 # Always update for small datasets + (i + 1) % update_interval == 0 + or (i + 1) == total_similarities + or i == 0 + or total_similarities <= 10 # Always update for small datasets ) if should_update: self.progress_tracker.update_progress( tracking_id, processed=i + 1, total=total_similarities, - message=f"Creating duplicate candidates... {i + 1}/{total_similarities} (remaining: {remaining})" + message=f"Creating duplicate candidates... {i + 1}/{total_similarities} (remaining: {remaining})", ) # Sort by confidence (highest first) @@ -326,12 +326,12 @@ class DuplicateDetector: self.logger.info( f"Detecting duplicate groups from {len(entities)} entities" ) - + # Initial progress update self.progress_tracker.update_tracking( - tracking_id, + tracking_id, status="running", - message=f"Starting duplicate detection for {len(entities)} entities..." + message=f"Starting duplicate detection for {len(entities)} entities...", ) # Detect duplicate candidates @@ -358,34 +358,34 @@ class DuplicateDetector: update_interval = 1 # Update every item for small datasets else: update_interval = max(1, min(5, total_groups // 100)) - + # Initial progress update - ALWAYS show this remaining = total_groups self.progress_tracker.update_progress( tracking_id, processed=0, total=total_groups, - message=f"Calculating group metrics... 0/{total_groups} (remaining: {remaining})" + message=f"Calculating group metrics... 0/{total_groups} (remaining: {remaining})", ) - + for i, group in enumerate(groups): group.confidence = self._calculate_group_confidence(group) group.representative = self._select_representative(group) - + remaining = total_groups - (i + 1) # Update progress: always update for small datasets, or at intervals for large ones should_update = ( - (i + 1) % update_interval == 0 or - (i + 1) == total_groups or - i == 0 or - total_groups <= 10 # Always update for small datasets + (i + 1) % update_interval == 0 + or (i + 1) == total_groups + or i == 0 + or total_groups <= 10 # Always update for small datasets ) if should_update: self.progress_tracker.update_progress( tracking_id, processed=i + 1, total=total_groups, - message=f"Calculating group metrics... {i + 1}/{total_groups} (remaining: {remaining})" + message=f"Calculating group metrics... {i + 1}/{total_groups} (remaining: {remaining})", ) self.logger.info( @@ -410,7 +410,7 @@ class DuplicateDetector: self, relationships: List[Dict[str, Any]], **options ) -> List[Tuple[Dict[str, Any], Dict[str, Any]]]: """ - Detect duplicate relationships. + Detect duplicate relationships using opt-in semantic canonicalization. Args: relationships: List of relationships @@ -430,31 +430,59 @@ class DuplicateDetector: try: duplicates = [] threshold = options.get("threshold", 0.9) + mode = options.get("relationship_dedup_mode", "legacy") + + canon_sigs = [] + + if mode == "semantic_v2": + synonym_map = options.get("predicate_synonym_map", {}) + norm_enabled = options.get("literal_normalization_enabled", False) + + for rel in relationships: + subj = str(rel.get("subject", "")) + pred = str(rel.get("predicate", "")).lower() + obj = str(rel.get("object", "")) + + canon_pred = synonym_map.get(pred, pred) + if norm_enabled: + obj = " ".join(obj.lower().split()) + + sig = hash((subj, canon_pred, obj)) + canon_sigs.append(sig) + total_rels = len(relationships) total_pairs = total_rels * (total_rels - 1) // 2 processed = 0 - - # Update interval - if total_pairs <= 10: - update_interval = 1 - else: - update_interval = max(1, min(100, total_pairs // 100)) + update_interval = ( + 1 if total_pairs <= 10 else max(1, min(100, total_pairs // 100)) + ) for i in range(len(relationships)): for j in range(i + 1, len(relationships)): rel1 = relationships[i] rel2 = relationships[j] - if self._relationships_are_duplicates(rel1, rel2, threshold): + is_duplicate = False + + if mode == "semantic_v2" and canon_sigs[i] == canon_sigs[j]: + is_duplicate = True + + else: + is_duplicate = self._relationships_are_duplicates( + rel1, rel2, threshold, mode, options + ) + + if is_duplicate: duplicates.append((rel1, rel2)) - + processed += 1 + if processed % update_interval == 0 or processed == total_pairs: self.progress_tracker.update_progress( tracking_id, processed=processed, total=total_pairs, - message=f"Checking relationships... {processed}/{total_pairs}" + message=f"Checking relationships... {processed}/{total_pairs}", ) self.progress_tracker.stop_tracking( @@ -462,6 +490,7 @@ class DuplicateDetector: status="completed", message=f"Detected {len(duplicates)} duplicate relationships", ) + return duplicates except Exception as e: @@ -526,14 +555,14 @@ class DuplicateDetector: update_interval = 1 # Update every item for small datasets else: update_interval = max(1, min(10, total_comparisons // 100)) - + # Initial progress update - ALWAYS show this remaining = total_comparisons self.progress_tracker.update_progress( tracking_id, processed=0, total=total_comparisons, - message=f"Starting incremental detection... 0/{total_comparisons} (remaining: {remaining})" + message=f"Starting incremental detection... 0/{total_comparisons} (remaining: {remaining})", ) # Compare each new entity with all existing entities @@ -553,22 +582,22 @@ class DuplicateDetector: # Filter by confidence threshold if candidate.confidence >= self.confidence_threshold: candidates.append(candidate) - + processed += 1 remaining = total_comparisons - processed # Update progress: always update for small datasets, or at intervals for large ones should_update = ( - processed % update_interval == 0 or - processed == total_comparisons or - processed == 1 or - total_comparisons <= 10 # Always update for small datasets + processed % update_interval == 0 + or processed == total_comparisons + or processed == 1 + or total_comparisons <= 10 # Always update for small datasets ) if should_update: self.progress_tracker.update_progress( tracking_id, processed=processed, total=total_comparisons, - message=f"Comparing entities... {processed}/{total_comparisons} (remaining: {remaining})" + message=f"Comparing entities... {processed}/{total_comparisons} (remaining: {remaining})", ) # Sort by confidence (highest first) @@ -686,8 +715,12 @@ class DuplicateDetector: groups = [] for candidate in candidates: - entity1_id = self._get_entity_value(candidate.entity1, "id") or id(candidate.entity1) - entity2_id = self._get_entity_value(candidate.entity2, "id") or id(candidate.entity2) + entity1_id = self._get_entity_value(candidate.entity1, "id") or id( + candidate.entity1 + ) + entity2_id = self._get_entity_value(candidate.entity2, "id") or id( + candidate.entity2 + ) group1 = entity_to_group.get(entity1_id) group2 = entity_to_group.get(entity2_id) @@ -707,17 +740,17 @@ class DuplicateDetector: # Add entity2 to group1 if candidate.entity2 not in group1.entities: group1.entities.append(candidate.entity2) - group1.similarity_scores[ - (entity1_id, entity2_id) - ] = candidate.similarity_score + group1.similarity_scores[(entity1_id, entity2_id)] = ( + candidate.similarity_score + ) entity_to_group[entity2_id] = group1 elif group1 is None and group2 is not None: # Add entity1 to group2 if candidate.entity1 not in group2.entities: group2.entities.append(candidate.entity1) - group2.similarity_scores[ - (entity1_id, entity2_id) - ] = candidate.similarity_score + group2.similarity_scores[(entity1_id, entity2_id)] = ( + candidate.similarity_score + ) entity_to_group[entity1_id] = group2 elif group1 != group2: # Merge groups @@ -725,9 +758,9 @@ class DuplicateDetector: [e for e in group2.entities if e not in group1.entities] ) group1.similarity_scores.update(group2.similarity_scores) - group1.similarity_scores[ - (entity1_id, entity2_id) - ] = candidate.similarity_score + group1.similarity_scores[(entity1_id, entity2_id)] = ( + candidate.similarity_score + ) # Update references for entity in group2.entities: @@ -767,11 +800,16 @@ class DuplicateDetector: return best_entity def _relationships_are_duplicates( - self, rel1: Dict[str, Any], rel2: Dict[str, Any], threshold: float + self, + rel1: Dict[str, Any], + rel2: Dict[str, Any], + threshold: float, + mode: str = "legacy", + options: Dict = None, ) -> bool: """Check if two relationships are duplicates.""" - # Exact match - # Handle both dicts and relationship objects if they exist + options = options or {} + def get_rel_val(rel, key): if hasattr(rel, "__dict__"): return getattr(rel, key, None) @@ -779,18 +817,53 @@ class DuplicateDetector: return rel.get(key) return None - if ( - get_rel_val(rel1, "subject") == get_rel_val(rel2, "subject") - and get_rel_val(rel1, "predicate") == get_rel_val(rel2, "predicate") - and get_rel_val(rel1, "object") == get_rel_val(rel2, "object") - ): - return True - - # Fuzzy match for predicate + subj1 = get_rel_val(rel1, "subject") + subj2 = get_rel_val(rel2, "subject") pred1 = str(get_rel_val(rel1, "predicate") or "") pred2 = str(get_rel_val(rel2, "predicate") or "") - similarity = self.similarity_calculator.calculate_string_similarity( - pred1, pred2 + obj1 = str(get_rel_val(rel1, "object") or "") + obj2 = str(get_rel_val(rel2, "object") or "") + + if mode == "legacy": + if subj1 == subj2 and pred1 == pred2 and obj1 == obj2: + return True + similarity = self.similarity_calculator.calculate_string_similarity( + pred1, pred2 + ) + return similarity >= threshold + + if subj1 != subj2: + return False + + synonyms = options.get("predicate_synonym_map", {}) + c_pred1 = synonyms.get(pred1.lower(), pred1.lower()) + c_pred2 = synonyms.get(pred2.lower(), pred2.lower()) + + pred_sim = ( + 1.0 + if c_pred1 == c_pred2 + else self.similarity_calculator.calculate_string_similarity( + c_pred1, c_pred2 + ) ) - return similarity >= threshold + if options.get("literal_normalization_enabled", False): + obj1 = " ".join(obj1.lower().split()) + obj2 = " ".join(obj2.lower().split()) + + obj_sim = ( + 1.0 + if obj1 == obj2 + else self.similarity_calculator.calculate_string_similarity(obj1, obj2) + ) + + # Weighted composition: Predicate is 60% of the match, Object literal is 40% + semantic_score = (pred_sim * 0.6) + (obj_sim * 0.4) + + # Metadata explainability + if semantic_score >= threshold: + if isinstance(rel1, dict) and isinstance(rel2, dict): + rel1.setdefault("metadata", {})["semantic_match_score"] = semantic_score + rel2.setdefault("metadata", {})["semantic_match_score"] = semantic_score + + return semantic_score >= threshold diff --git a/semantica/deduplication/merge_strategy.py b/semantica/deduplication/merge_strategy.py index cc43fc7c..b5c2b29e 100644 --- a/semantica/deduplication/merge_strategy.py +++ b/semantica/deduplication/merge_strategy.py @@ -493,23 +493,46 @@ class MergeStrategyManager: def _merge_relationships( self, entities: List[Dict[str, Any]], base_entity: Dict[str, Any] ) -> List[Dict[str, Any]]: - """Merge relationships from all entities.""" + """Merge relationships from all entities with canonicalization support.""" + all_relationships = [] seen_relationships = set() - + + dedup_mode = self.config.get("relationship_dedup_mode", "legacy") + synonym_map = self.config.get("predicate_synonym_map", {}) + norm_enabled = self.config.get("literal_normalization_enabled", False) + + def _normalize_literal(val: Any) -> Any: + if not norm_enabled or not isinstance(val, str): + return val + + return " ".join(val.lower().split()) + for entity in entities: relationships = entity.get("relationships", []) - + for rel in relationships: - # Create unique key for relationship - rel_key = (rel.get("subject"), rel.get("predicate"), rel.get("object")) - + subj = rel.get("subject") + pred = str(rel.get("predicate", "")) + obj = rel.get("object") + + if dedup_mode == "semantic_v2": + canon_pred = synonym_map.get(pred.lower(), pred).lower() + canon_obj = _normalize_literal(obj) + rel_key = (subj, canon_pred, canon_obj) + + else: + rel_key = (subj, pred, obj) + if rel_key not in seen_relationships: all_relationships.append(rel) seen_relationships.add(rel_key) + return all_relationships + + def _merge_metadata( self, entities: List[Dict[str, Any]], base_entity: Dict[str, Any] ) -> Dict[str, Any]: diff --git a/semantica/deduplication/methods.py b/semantica/deduplication/methods.py index bb5e767c..edba7fba 100644 --- a/semantica/deduplication/methods.py +++ b/semantica/deduplication/methods.py @@ -94,7 +94,7 @@ Author: Semantica Contributors License: MIT """ -from typing import Any, Callable, Dict, List, Optional, Union +from typing import Any, Callable, Dict, List, Optional, Union, Tuple from ..utils.exceptions import ProcessingError from ..utils.logging import get_logger @@ -261,6 +261,38 @@ def detect_duplicates( return detector.detect_duplicates(entities, **kwargs) +def dedup_triplets( + relationships: List[Dict[str, Any]], + mode: str = "semantic_v2", + threshold: float = 0.85, + **kwargs, +) -> List[Tuple[Dict[str, Any], Dict[str, Any]]]: + """ + Detect duplicate relationships/triplets (convenience function). + + Args: + relationships: List of relationship dictionaries to check. + mode: Dedup mode ("legacy" or "semantic_v2") + threshold: Minimum similarity threshold for fuzzy matching. + **kwargs: Additional options for Semantic mode: + - predicate_synonym_map: Dict mapping synonyms to canonical predicates. + - literal_normalization_enabled: Boolean to enable literal normalization. + + Returns: + List of duplicate relationship piars (rel1, rel2). + """ + + custom_method = method_registry.get("detection", "triplets") + + if custom_method: + return custom_method(relationships, mode=mode, threshold=threshold, **kwargs) + + detector = DuplicateDetector(**kwargs) + options = {"threshold": threshold, "relationship_dedup_mode": mode, **kwargs} + + return detector.detect_relationship_duplicates(relationships, **options) + + def merge_entities( entities: List[Dict[str, Any]], method: str = "keep_most_complete", @@ -535,3 +567,4 @@ method_registry.register("similarity", "multi_factor", _multi_factor_similarity) method_registry.register("detection", "pairwise", _pairwise_detection) method_registry.register("merging", "keep_most_complete", _keep_most_complete_merging) method_registry.register("clustering", "graph_based", _graph_based_clustering) +method_registry.register("detection", "triplets", dedup_triplets)