diff --git a/.claude/settings.local.json b/.claude/settings.local.json index 94a6ddd..0c846bb 100644 --- a/.claude/settings.local.json +++ b/.claude/settings.local.json @@ -31,7 +31,8 @@ "Bash(docker-compose -f docker-compose.neo4j.yml up -d)", "Bash(python test_phase4_integration.py)", "Bash(git add *)", - "Bash(git commit *)" + "Bash(git commit *)", + "Bash(python test_phase5_entity_resolver.py)" ] } } diff --git a/ontology_platform/ont_platform/core/graph/__init__.py b/ontology_platform/ont_platform/core/graph/__init__.py index e69de29..4c0d43d 100644 --- a/ontology_platform/ont_platform/core/graph/__init__.py +++ b/ontology_platform/ont_platform/core/graph/__init__.py @@ -0,0 +1,13 @@ +"""Graph module: Neo4j adapter, RDF conversion, entity resolution, pattern matching.""" + +from .neo4j_adapter import Neo4jAdapter, Neo4jConfig +from .rdf_converter import RDFToPropertyGraphConverter +from .entity_resolver import EntityResolver, EntityCluster + +__all__ = [ + "Neo4jAdapter", + "Neo4jConfig", + "RDFToPropertyGraphConverter", + "EntityResolver", + "EntityCluster", +] diff --git a/ontology_platform/ont_platform/core/graph/entity_resolver.py b/ontology_platform/ont_platform/core/graph/entity_resolver.py new file mode 100644 index 0000000..9f33228 --- /dev/null +++ b/ontology_platform/ont_platform/core/graph/entity_resolver.py @@ -0,0 +1,323 @@ +"""Entity Resolver: Semantic duplicate detection and merging (Phase 5). + +Detects duplicate entities using: +- Vector similarity (all-MiniLM-L6-v2) +- Jaro-Winkler text similarity +- Label normalization + +Merges duplicates and consolidates evidence. +""" + +import logging +import re +import string +from typing import Optional, List, Dict, Any +from dataclasses import dataclass +from datetime import datetime +from difflib import SequenceMatcher + +try: + from textdistance import JaroWinkler + HAS_TEXTDISTANCE = True +except ImportError: + HAS_TEXTDISTANCE = False + +import numpy as np +from sentence_transformers import SentenceTransformer + +logger = logging.getLogger(__name__) + + +@dataclass +class EntityCluster: + """Result of entity clustering.""" + + cluster_id: str + canonical_id: int + duplicates: List[int] + confidence: float + reason: str # "vector_similarity", "text_similarity", or "combined" + metadata: Dict[str, Any] + + +class EntityResolver: + """Detects and resolves duplicate entities using vector and text similarity.""" + + def __init__( + self, + vector_threshold: float = 0.85, + text_threshold: float = 0.88, + model_name: str = "all-MiniLM-L6-v2", + ): + """ + Initialize entity resolver. + + Args: + vector_threshold: Similarity threshold for vector matching (0-1) + text_threshold: Similarity threshold for text matching (0-1) + model_name: SentenceTransformer model name + """ + self.vector_threshold = vector_threshold + self.text_threshold = text_threshold + self.model_name = model_name + self.embedder = None + self.jaro_winkler = JaroWinkler() if HAS_TEXTDISTANCE else None + + async def initialize_embedder(self) -> bool: + """Load embedding model.""" + try: + self.embedder = SentenceTransformer(self.model_name) + logger.info(f"Loaded embedder: {self.model_name}") + return True + except Exception as e: + logger.error(f"Failed to load embedder: {e}") + return False + + async def detect_duplicates( + self, + entities: List[Dict[str, Any]], + batch_size: int = 1000, + ) -> List[EntityCluster]: + """ + Detect duplicate entities using 2-stage matching. + + Args: + entities: List of entity dicts with id, label, type + batch_size: Batch size for processing + + Returns: + List of EntityCluster objects + """ + if not self.embedder: + logger.warning("Embedder not initialized, skipping duplicate detection") + return [] + + clusters = [] + processed = set() + + # Normalize all labels + normalized = {} + for entity in entities: + entity_id = entity.get("id") + label = entity.get("label", "") + normalized[entity_id] = self._normalize_label(label) + + # Extract embeddings for all entities + labels = [e.get("label", "") for e in entities] + embeddings = self._embed_batch(labels) + embedding_map = {e["id"]: emb for e, emb in zip(entities, embeddings)} + + # Stage 1: Vector similarity matching + similarity_pairs = self._compute_vector_similarities( + entities, + embeddings, + self.vector_threshold, + ) + + # Stage 2: Text similarity refinement + for entity_id1, entity_id2, vec_similarity in similarity_pairs: + if entity_id1 in processed or entity_id2 in processed: + continue + + # Text similarity check + label1 = normalized[entity_id1] + label2 = normalized[entity_id2] + text_similarity = self._compute_text_similarity(label1, label2) + + # Combine scores (weighted average) + combined_similarity = 0.6 * vec_similarity + 0.4 * text_similarity + + if combined_similarity >= self.text_threshold: + # Determine canonical entity (by ID or confidence) + canonical_id = min(entity_id1, entity_id2) + duplicate_id = max(entity_id1, entity_id2) + + cluster = EntityCluster( + cluster_id=f"C_{canonical_id}_{duplicate_id}", + canonical_id=canonical_id, + duplicates=[duplicate_id], + confidence=min(combined_similarity, 0.99), # Cap at 0.99 + reason="combined" if text_similarity >= 0.5 else "vector_similarity", + metadata={ + "vector_similarity": vec_similarity, + "text_similarity": text_similarity, + "combined_similarity": combined_similarity, + }, + ) + clusters.append(cluster) + processed.add(entity_id1) + processed.add(entity_id2) + + logger.info(f"Detected {len(clusters)} duplicate clusters from {len(entities)} entities") + return clusters + + async def resolve_cluster( + self, + cluster: EntityCluster, + entities_map: Dict[int, Dict[str, Any]], + merge_strategy: str = "highest_confidence", + ) -> Dict[str, Any]: + """ + Merge duplicate entities in a cluster. + + Args: + cluster: EntityCluster to resolve + entities_map: Map of entity_id → entity dict + merge_strategy: "highest_confidence", "most_evidenced", or "earliest" + + Returns: + Merged entity dict + """ + canonical_id = cluster.canonical_id + duplicate_ids = cluster.duplicates + + if canonical_id not in entities_map: + logger.warning(f"Canonical entity {canonical_id} not found") + return {} + + canonical = entities_map[canonical_id].copy() + + # Collect all aliases from all entities in cluster + all_aliases = set(canonical.get("aliases", [])) + all_aliases.add(canonical.get("label", "")) # Add canonical label as alias + + evidence_list = list(canonical.get("evidence", [])) + + for dup_id in duplicate_ids: + if dup_id not in entities_map: + logger.warning(f"Duplicate entity {dup_id} not found") + continue + + duplicate = entities_map[dup_id] + all_aliases.add(duplicate.get("label", "")) # Add duplicate label + all_aliases.update(duplicate.get("aliases", [])) # Add duplicate aliases + evidence_list.extend(duplicate.get("evidence", [])) + + # Update canonical entity + canonical["aliases"] = sorted(list(all_aliases)) + canonical["evidence"] = evidence_list + canonical["merged_from"] = duplicate_ids + canonical["merged_at"] = datetime.utcnow().isoformat() + canonical["merge_confidence"] = cluster.confidence + + return canonical + + def _normalize_label(self, label: str) -> str: + """ + Normalize label for comparison. + + - Lowercase + - Replace hyphens/underscores with spaces + - Remove punctuation + - Collapse whitespace + - Remove common prefixes/suffixes + """ + # Lowercase + text = label.lower() + + # Replace hyphens and underscores with spaces + text = text.replace("-", " ").replace("_", " ") + + # Remove common articles and prepositions + for word in ["the ", "a ", "an "]: + if text.startswith(word): + text = text[len(word) :] + + # Remove special characters (keep alphanumeric and spaces) + text = re.sub(r"[^a-z0-9\s]", "", text) + + # Collapse whitespace + text = re.sub(r"\s+", " ", text).strip() + + return text + + def _jaro_winkler_similarity(self, s1: str, s2: str) -> float: + """ + Calculate Jaro-Winkler similarity (0-1, 1=identical). + + Falls back to SequenceMatcher if textdistance not available. + """ + if self.jaro_winkler: + return self.jaro_winkler.similarity(s1, s2) + + # Fallback: use SequenceMatcher + return SequenceMatcher(None, s1, s2).ratio() + + def _compute_text_similarity(self, label1: str, label2: str) -> float: + """Compute text similarity between normalized labels.""" + # Exact match + if label1 == label2: + return 1.0 + + # Jaro-Winkler similarity + jw_sim = self._jaro_winkler_similarity(label1, label2) + + # Token overlap (for multi-word labels) + tokens1 = set(label1.split()) + tokens2 = set(label2.split()) + if tokens1 and tokens2: + overlap = len(tokens1 & tokens2) / len(tokens1 | tokens2) + else: + overlap = 0 + + # Weighted combination + return 0.7 * jw_sim + 0.3 * overlap + + def _embed_batch(self, texts: List[str]) -> List[List[float]]: + """Compute embeddings for batch of texts.""" + if not self.embedder: + raise RuntimeError("Embedder not initialized") + + return self.embedder.encode(texts, convert_to_tensor=False).tolist() + + def _compute_vector_similarities( + self, + entities: List[Dict[str, Any]], + embeddings: List[List[float]], + threshold: float, + ) -> List[tuple]: + """ + Compute vector similarities between entities. + + Returns: + List of (entity_id1, entity_id2, similarity) where similarity >= threshold + """ + similarities = [] + embeddings_array = np.array(embeddings) + + # Compute cosine similarity matrix + norms = np.linalg.norm(embeddings_array, axis=1, keepdims=True) + normalized = embeddings_array / (norms + 1e-8) + similarity_matrix = np.dot(normalized, normalized.T) + + # Extract pairs above threshold + for i in range(len(entities)): + for j in range(i + 1, len(entities)): + sim = float(similarity_matrix[i][j]) + if sim >= threshold: + entity_id1 = entities[i]["id"] + entity_id2 = entities[j]["id"] + similarities.append((entity_id1, entity_id2, sim)) + + logger.info(f"Found {len(similarities)} vector similarities >= {threshold}") + return similarities + + def get_resolution_report(self, clusters: List[EntityCluster]) -> Dict[str, Any]: + """Generate report on duplicate resolution.""" + total_duplicates = sum(len(c.duplicates) for c in clusters) + avg_confidence = ( + np.mean([c.confidence for c in clusters]) if clusters else 0 + ) + + reasons = {} + for cluster in clusters: + reason = cluster.reason + reasons[reason] = reasons.get(reason, 0) + 1 + + return { + "total_clusters": len(clusters), + "total_duplicates": total_duplicates, + "avg_confidence": float(avg_confidence), + "by_reason": reasons, + "timestamp": datetime.utcnow().isoformat(), + } diff --git a/ontology_platform/ont_platform/core/graph/neo4j_adapter.py b/ontology_platform/ont_platform/core/graph/neo4j_adapter.py index 2d7c903..378ce93 100644 --- a/ontology_platform/ont_platform/core/graph/neo4j_adapter.py +++ b/ontology_platform/ont_platform/core/graph/neo4j_adapter.py @@ -1,18 +1,19 @@ -"""Neo4j adapter for Phase 4: Graph projection and search. +"""Neo4j adapter for Phase 4-5: Graph projection, search, and GraphRAG. Provides: - Connection management - Basic RDF → Property Graph conversion - Vector embedding and indexing - Search APIs (vector search) +- Phase 5: Batch operations, transactions, indexes, pattern matching Design: Lightweight, extensible for Phase 5+ enhancements. """ import logging -from typing import Optional, List, Dict, Any +from typing import Optional, List, Dict, Any, Tuple import asyncio -from neo4j import AsyncGraphDatabase +from neo4j import AsyncGraphDatabase, Transaction from sentence_transformers import SentenceTransformer logger = logging.getLogger(__name__) @@ -355,6 +356,219 @@ class Neo4jAdapter: "entity_nodes": entities_count, } + async def batch_create_entity_nodes( + self, + entities: List[Dict[str, Any]], + entity_type: str = "Entity", + batch_size: int = 1000, + ) -> Dict[str, int]: + """ + Batch create entity nodes (Phase 5 optimization). + + Args: + entities: List of entity dicts + entity_type: Node label + batch_size: Size of each batch + + Returns: + {"created": int, "failed": int, "total": int} + """ + if not self._driver: + raise RuntimeError("Not connected to Neo4j") + + created = 0 + failed = 0 + total = len(entities) + + # Process in batches + for i in range(0, len(entities), batch_size): + batch = entities[i : i + batch_size] + try: + # Compute embeddings for batch + labels = [e.get("label", "") for e in batch] + embeddings = self._get_embeddings(labels) + + async with self._driver.session(database=self.config.database) as session: + query = f""" + UNWIND $batch AS entity + MERGE (e:{entity_type} {{id: entity.id}}) + SET e.label = entity.label, + e.type = entity.entity_type, + e.confidence = entity.confidence, + e.embedding = entity.embedding + RETURN count(e) AS count + """ + params = { + "batch": [ + { + "id": entity.get("id"), + "label": entity.get("label"), + "entity_type": entity.get("type", "concept"), + "confidence": entity.get("confidence", 0.5), + "embedding": embeddings[idx], + } + for idx, entity in enumerate(batch) + ] + } + result = await session.run(query, params) + record = await result.single() + created += record["count"] + except Exception as e: + logger.warning(f"Batch creation failed for {len(batch)} entities: {e}") + failed += len(batch) + + logger.info(f"Batch created {created}/{total} entity nodes ({failed} failed)") + return {"created": created, "failed": failed, "total": total} + + async def batch_create_relation_edges( + self, + relations: List[Dict[str, Any]], + batch_size: int = 1000, + ) -> Dict[str, int]: + """ + Batch create relation edges (Phase 5 optimization). + + Args: + relations: List of relation dicts + batch_size: Size of each batch + + Returns: + {"created": int, "failed": int, "total": int} + """ + if not self._driver: + raise RuntimeError("Not connected to Neo4j") + + created = 0 + failed = 0 + total = len(relations) + + for i in range(0, len(relations), batch_size): + batch = relations[i : i + batch_size] + try: + async with self._driver.session(database=self.config.database) as session: + query = """ + UNWIND $batch AS rel + MATCH (source {id: rel.source_id}) + MATCH (target {id: rel.target_id}) + MERGE (source)-[r:RELATES {predicate: rel.predicate}]->(target) + SET r.confidence = rel.confidence + RETURN count(r) AS count + """ + params = { + "batch": [ + { + "source_id": r.get("source_id"), + "target_id": r.get("target_id"), + "predicate": r.get("predicate"), + "confidence": r.get("confidence", 0.5), + } + for r in batch + ] + } + result = await session.run(query, params) + record = await result.single() + created += record["count"] + except Exception as e: + logger.warning(f"Batch relation creation failed: {e}") + failed += len(batch) + + logger.info(f"Batch created {created}/{total} relation edges ({failed} failed)") + return {"created": created, "failed": failed, "total": total} + + async def create_indexes(self) -> Dict[str, bool]: + """ + Create performance indexes (Phase 5). + + Returns: + {"entity_id": bool, "embedding": bool, "label": bool} + """ + if not self._driver: + raise RuntimeError("Not connected to Neo4j") + + indexes = {} + async with self._driver.session(database=self.config.database) as session: + # Entity ID index + try: + await session.run("CREATE INDEX entity_id_idx IF NOT EXISTS FOR (e:Entity) ON (e.id)") + indexes["entity_id"] = True + logger.info("Created entity ID index") + except Exception as e: + logger.warning(f"Failed to create entity ID index: {e}") + indexes["entity_id"] = False + + # Label index (for text search) + try: + await session.run("CREATE INDEX entity_label_idx IF NOT EXISTS FOR (e:Entity) ON (e.label)") + indexes["label"] = True + logger.info("Created entity label index") + except Exception as e: + logger.warning(f"Failed to create label index: {e}") + indexes["label"] = False + + # Confidence index + try: + await session.run("CREATE INDEX entity_confidence_idx IF NOT EXISTS FOR (e:Entity) ON (e.confidence)") + indexes["confidence"] = True + logger.info("Created confidence index") + except Exception as e: + logger.warning(f"Failed to create confidence index: {e}") + indexes["confidence"] = False + + return indexes + + async def execute_cypher( + self, + cypher: str, + params: Optional[Dict[str, Any]] = None, + ) -> List[Dict[str, Any]]: + """ + Execute arbitrary Cypher query (read-only recommended). + + Args: + cypher: Cypher query string + params: Query parameters + + Returns: + List of result records as dicts + """ + if not self._driver: + raise RuntimeError("Not connected to Neo4j") + + results = [] + async with self._driver.session(database=self.config.database) as session: + result = await session.run(cypher, params or {}) + async for record in result: + results.append(dict(record)) + + return results + + async def delete_project_nodes(self, project_id: Optional[int] = None) -> int: + """ + Delete all nodes for a project (Phase 5 cleanup). + + Args: + project_id: Project ID (if None, delete all) + + Returns: + Number of nodes deleted + """ + if not self._driver: + raise RuntimeError("Not connected to Neo4j") + + async with self._driver.session(database=self.config.database) as session: + if project_id: + query = "MATCH (n {project_id: $project_id}) DETACH DELETE n RETURN count(n) AS count" + result = await session.run(query, project_id=project_id) + else: + query = "MATCH (n) DETACH DELETE n RETURN count(n) AS count" + result = await session.run(query) + + record = await result.single() + deleted = record["count"] if record else 0 + + logger.info(f"Deleted {deleted} nodes") + return deleted + async def close(self): """Close Neo4j connection.""" if self._driver: diff --git a/ontology_platform/ont_platform/core/graph/rdf_converter.py b/ontology_platform/ont_platform/core/graph/rdf_converter.py new file mode 100644 index 0000000..3dc1e77 --- /dev/null +++ b/ontology_platform/ont_platform/core/graph/rdf_converter.py @@ -0,0 +1,308 @@ +"""RDF ↔ Property Graph converter (Phase 5). + +Converts between: +- RDF triples (Subject-Predicate-Object) +- Neo4j Property Graph (Nodes with properties + Relationships) + +Maintains ontology metadata and enables bidirectional conversion. +""" + +import logging +from typing import Optional, List, Dict, Any, Tuple +from datetime import datetime + +logger = logging.getLogger(__name__) + + +class RDFToPropertyGraphConverter: + """Converts RDF triples to Neo4j Property Graph and vice versa.""" + + def __init__( + self, + namespace_base: str = "http://example.org/", + project_id: Optional[int] = None, + ): + """ + Initialize RDF converter. + + Args: + namespace_base: Base namespace for URIs + project_id: Optional project ID for isolation + """ + self.namespace_base = namespace_base + self.project_id = project_id + self.standard_prefixes = { + "rdf": "http://www.w3.org/1999/02/22-rdf-syntax-ns#", + "rdfs": "http://www.w3.org/2000/01/rdf-schema#", + "owl": "http://www.w3.org/2002/07/owl#", + "xsd": "http://www.w3.org/2001/XMLSchema#", + "foaf": "http://xmlns.com/foaf/0.1/", + "skos": "http://www.w3.org/2004/02/skos/core#", + } + + async def convert_triples_to_graph( + self, + triples: List[Tuple[str, str, str]], + namespace_map: Optional[Dict[str, str]] = None, + confidence: float = 0.9, + ) -> Dict[str, Any]: + """ + Convert RDF triples to Neo4j graph structure. + + Args: + triples: List of (subject, predicate, object) tuples + namespace_map: Optional custom namespace mappings + confidence: Default confidence for all triples + + Returns: + { + "nodes": List of node dicts (id, label, type, properties), + "edges": List of edge dicts (source, target, predicate, confidence), + "warnings": List of warnings + } + """ + nodes = {} + edges = [] + warnings = [] + + # Merge namespaces + namespaces = {**self.standard_prefixes} + if namespace_map: + namespaces.update(namespace_map) + + # Track unique URIs + uri_to_node = {} + + # Process triples + for subject, predicate, obj in triples: + try: + # Normalize URIs + subject_uri = self._normalize_uri(subject, namespaces) + predicate_uri = self._normalize_uri(predicate, namespaces) + obj_uri = self._normalize_uri(obj, namespaces) + + # Create subject node + if subject_uri not in uri_to_node: + subject_node_id = self._uri_to_node_id(subject_uri) + nodes[subject_node_id] = { + "id": subject_node_id, + "uri": subject_uri, + "label": self._extract_label(subject_uri), + "type": "Entity", + "confidence": confidence, + "project_id": self.project_id, + "created_at": datetime.utcnow().isoformat(), + } + uri_to_node[subject_uri] = subject_node_id + else: + subject_node_id = uri_to_node[subject_uri] + + # Create object node (if it's a URI, not a literal) + obj_node_id = None + if self._is_uri(obj_uri): + if obj_uri not in uri_to_node: + obj_node_id = self._uri_to_node_id(obj_uri) + nodes[obj_node_id] = { + "id": obj_node_id, + "uri": obj_uri, + "label": self._extract_label(obj_uri), + "type": "Entity", + "confidence": confidence, + "project_id": self.project_id, + "created_at": datetime.utcnow().isoformat(), + } + uri_to_node[obj_uri] = obj_node_id + else: + obj_node_id = uri_to_node[obj_uri] + + # Create edge for URI object + edges.append({ + "source_id": subject_node_id, + "target_id": obj_node_id, + "predicate": predicate_uri, + "confidence": confidence, + "created_at": datetime.utcnow().isoformat(), + }) + else: + # Store literal value as property on subject + if "properties" not in nodes[subject_node_id]: + nodes[subject_node_id]["properties"] = {} + predicate_key = self._extract_label(predicate_uri) + nodes[subject_node_id]["properties"][predicate_key] = obj + + except Exception as e: + warnings.append(f"Failed to convert triple ({subject}, {predicate}, {obj}): {e}") + + logger.info(f"Converted {len(triples)} triples → {len(nodes)} nodes, {len(edges)} edges") + + return { + "nodes": list(nodes.values()), + "edges": edges, + "warnings": warnings, + } + + async def to_rdf_triples( + self, + nodes: List[Dict[str, Any]], + edges: List[Dict[str, Any]], + ) -> List[Tuple[str, str, str]]: + """ + Convert Neo4j graph back to RDF triples. + + Args: + nodes: List of node dicts + edges: List of edge dicts + + Returns: + List of (subject, predicate, object) tuples + """ + triples = [] + + # Build node ID to URI map + id_to_uri = {n["id"]: n.get("uri", f"{self.namespace_base}{n['id']}") for n in nodes} + + # Convert edges to triples + for edge in edges: + subject_uri = id_to_uri.get(edge["source_id"]) + object_uri = id_to_uri.get(edge["target_id"]) + + if subject_uri and object_uri: + predicate_uri = edge.get("predicate", f"{self.namespace_base}related") + triples.append((subject_uri, predicate_uri, object_uri)) + + # Convert node properties to triples + for node in nodes: + subject_uri = node.get("uri", f"{self.namespace_base}{node['id']}") + + # Add label as rdfs:label + if "label" in node: + triples.append(( + subject_uri, + "http://www.w3.org/2000/01/rdf-schema#label", + node["label"], + )) + + # Add other properties + if "properties" in node: + for key, value in node["properties"].items(): + predicate_uri = f"{self.namespace_base}{key}" + triples.append((subject_uri, predicate_uri, str(value))) + + logger.info(f"Converted graph → {len(triples)} RDF triples") + return triples + + def _normalize_uri(self, uri: str, namespace_map: Dict[str, str]) -> str: + """ + Expand prefixed URIs to full URIs. + + Examples: + - "foaf:name" → "http://xmlns.com/foaf/0.1/name" + - "http://example.org/name" → "http://example.org/name" (unchanged) + """ + if "://" in uri: + return uri # Already a full URI + + if ":" in uri: + prefix, local = uri.split(":", 1) + if prefix in namespace_map: + return f"{namespace_map[prefix]}{local}" + + # Return as-is if no prefix found + return uri + + def _is_uri(self, value: str) -> bool: + """Check if value is a URI (not a literal).""" + return "://" in value or ":" in value + + def _extract_label(self, uri: str) -> str: + """ + Extract readable label from URI. + + Examples: + - "http://example.org/John_Doe" → "John Doe" + - "http://example.org/hasName" → "hasName" + """ + if "#" in uri: + return uri.split("#")[-1].replace("_", " ") + elif "/" in uri: + return uri.split("/")[-1].replace("_", " ") + else: + return uri + + def _uri_to_node_id(self, uri: str) -> str: + """ + Convert URI to node ID (hash-based). + + Format: "E_" + """ + import hashlib + hash_val = hashlib.md5(uri.encode()).hexdigest()[:8] + return f"E_{hash_val}" + + def validate_rdf_consistency( + self, + triples: List[Tuple[str, str, str]], + ) -> Dict[str, Any]: + """ + Validate RDF triples for common issues. + + Returns: + { + "is_valid": bool, + "errors": List[str], + "warnings": List[str], + "statistics": { + "triple_count": int, + "unique_subjects": int, + "unique_predicates": int, + "unique_objects": int, + } + } + """ + errors = [] + warnings = [] + subjects = set() + predicates = set() + objects = set() + + for subject, predicate, obj in triples: + subjects.add(subject) + predicates.add(predicate) + objects.add(obj) + + # Check for empty values + if not subject or not predicate or not obj: + errors.append(f"Empty value in triple: ({subject}, {predicate}, {obj})") + + # Check for malformed URIs + if "://" in subject and not self._is_valid_uri(subject): + warnings.append(f"Potentially malformed subject URI: {subject}") + + if "://" in predicate and not self._is_valid_uri(predicate): + warnings.append(f"Potentially malformed predicate URI: {predicate}") + + is_valid = len(errors) == 0 + + return { + "is_valid": is_valid, + "errors": errors, + "warnings": warnings, + "statistics": { + "triple_count": len(triples), + "unique_subjects": len(subjects), + "unique_predicates": len(predicates), + "unique_objects": len(objects), + }, + } + + def _is_valid_uri(self, uri: str) -> bool: + """Basic URI validation.""" + try: + if not uri.startswith(("http://", "https://", "urn:", "file://")): + return False + # Check for common invalid patterns + if " " in uri or "\n" in uri or "\t" in uri: + return False + return True + except Exception: + return False diff --git a/test_phase5_entity_resolver.py b/test_phase5_entity_resolver.py new file mode 100644 index 0000000..45d32d3 --- /dev/null +++ b/test_phase5_entity_resolver.py @@ -0,0 +1,264 @@ +#!/usr/bin/env python3 +"""Phase 5 Entity Resolver tests.""" + +import asyncio +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).parent / "ontology_platform")) + +from ont_platform.core.graph.entity_resolver import EntityResolver, EntityCluster + + +async def test_normalize_label(): + """Test label normalization.""" + print("\n[TEST 1] Label Normalization") + + resolver = EntityResolver() + + test_cases = [ + ("iPhone Pro Max", "iphone pro max"), + ("The Apple Inc.", "apple inc"), + ("Test-Entity", "test entity"), + ("UPPERCASE LABEL", "uppercase label"), + ("Label with spaces", "label with spaces"), + ] + + for input_label, expected in test_cases: + result = resolver._normalize_label(input_label) + status = "[OK]" if result == expected else "[FAIL]" + print(f" {status} '{input_label}' -> '{result}' (expected: '{expected}')") + assert result == expected, f"Expected '{expected}', got '{result}'" + + print(" [PASS]") + + +async def test_jaro_winkler_similarity(): + """Test Jaro-Winkler text similarity.""" + print("\n[TEST 2] Jaro-Winkler Similarity") + + resolver = EntityResolver() + + test_cases = [ + ("iphone", "iphone", 1.0), # Exact match + ("iphone", "iPhone", None), # Will be normalized before comparison + ("apple", "aplicant", None), # Similar but not identical + ("test", "best", None), # Partial match + ] + + for s1, s2, expected_range in test_cases: + sim = resolver._jaro_winkler_similarity(s1, s2) + print(f" Similarity('{s1}', '{s2}') = {sim:.3f}") + + if expected_range == 1.0: + assert sim == 1.0, f"Expected 1.0, got {sim}" + elif expected_range == 0.0: + assert sim == 0.0, f"Expected 0.0, got {sim}" + + print(" [PASS]") + + +async def test_text_similarity(): + """Test combined text similarity.""" + print("\n[TEST 3] Text Similarity (Jaro-Winkler + Token Overlap)") + + resolver = EntityResolver() + + test_cases = [ + ("machine learning", "machine learning", 1.0), + ("machine learning", "learning machine", 0.6), # Same tokens, different order + ("apple", "apple inc", 0.5), # Partial match + ("test", "best", 0.4), # Phonetically similar + ] + + for label1, label2, min_expected in test_cases: + sim = resolver._compute_text_similarity(label1, label2) + status = "[OK]" if sim >= min_expected else "[FAIL]" + print(f" {status} TextSim('{label1}', '{label2}') = {sim:.3f} (>= {min_expected})") + assert sim >= min_expected, f"Expected >= {min_expected}, got {sim}" + + print(" [PASS]") + + +async def test_embedder_initialization(): + """Test embedding model initialization.""" + print("\n[TEST 4] Embedder Initialization") + + resolver = EntityResolver(model_name="all-MiniLM-L6-v2") + + success = await resolver.initialize_embedder() + assert success, "Failed to initialize embedder" + + assert resolver.embedder is not None, "Embedder not loaded" + print(" [OK] Embedder loaded successfully") + + # Test embedding computation + texts = ["machine learning", "artificial intelligence"] + embeddings = resolver._embed_batch(texts) + + assert len(embeddings) == 2, f"Expected 2 embeddings, got {len(embeddings)}" + assert len(embeddings[0]) == 384, f"Expected 384-dim vectors, got {len(embeddings[0])}-dim" + + print(" [OK] Generated 384-dim embeddings for 2 texts") + print(" [PASS]") + + +async def test_detect_duplicates(): + """Test duplicate detection with vector similarity.""" + print("\n[TEST 5] Duplicate Detection (Vector + Text)") + + resolver = EntityResolver( + vector_threshold=0.85, + text_threshold=0.88, + ) + + success = await resolver.initialize_embedder() + assert success, "Failed to initialize embedder" + + # Create test entities with intentional duplicates + entities = [ + {"id": 1, "label": "Apple Inc.", "type": "Company"}, + {"id": 2, "label": "Apple Inc", "type": "Company"}, # Duplicate (slightly different) + {"id": 3, "label": "Microsoft", "type": "Company"}, + {"id": 4, "label": "Microsoft Corp", "type": "Company"}, # Duplicate + {"id": 5, "label": "Google", "type": "Company"}, + ] + + clusters = await resolver.detect_duplicates(entities) + + print(f" Detected {len(clusters)} duplicate clusters") + for cluster in clusters: + print( + f" Cluster: {cluster.canonical_id} ← {cluster.duplicates} " + f"(confidence: {cluster.confidence:.3f}, reason: {cluster.reason})" + ) + + # We expect to find some duplicates + assert len(clusters) > 0, "Should detect at least 1 duplicate cluster" + + print(" [PASS]") + + +async def test_resolve_cluster(): + """Test entity merging.""" + print("\n[TEST 6] Cluster Resolution (Entity Merging)") + + resolver = EntityResolver() + + # Create test entities + entities_map = { + 1: { + "id": 1, + "label": "Apple Inc.", + "type": "Company", + "aliases": ["Apple"], + "evidence": [{"text": "Founded in 1976"}], + }, + 2: { + "id": 2, + "label": "Apple", + "type": "Company", + "aliases": ["AAPL"], + "evidence": [{"text": "Technology company"}], + }, + } + + cluster = EntityCluster( + cluster_id="C_1_2", + canonical_id=1, + duplicates=[2], + confidence=0.92, + reason="combined", + metadata={}, + ) + + merged = await resolver.resolve_cluster(cluster, entities_map) + + assert merged["id"] == 1, "Canonical ID should be preserved" + assert 2 in merged["merged_from"], "Should record merged_from" + assert len(merged["aliases"]) >= 3, f"Should consolidate aliases (got {len(merged['aliases'])})" + assert len(merged["evidence"]) >= 2, "Should consolidate evidence" + + print(f" [OK] Merged entity with {len(merged['aliases'])} aliases, {len(merged['evidence'])} evidence") + print(f" [OK] Aliases: {merged['aliases']}") + print(" [PASS]") + + +async def test_resolution_report(): + """Test resolution report generation.""" + print("\n[TEST 7] Resolution Report") + + resolver = EntityResolver() + + clusters = [ + EntityCluster( + cluster_id="C_1", + canonical_id=1, + duplicates=[2, 3], + confidence=0.90, + reason="combined", + metadata={}, + ), + EntityCluster( + cluster_id="C_2", + canonical_id=4, + duplicates=[5], + confidence=0.85, + reason="vector_similarity", + metadata={}, + ), + ] + + report = resolver.get_resolution_report(clusters) + + assert report["total_clusters"] == 2, "Should have 2 clusters" + assert report["total_duplicates"] == 3, "Should have 3 total duplicates (2+1)" + assert "combined" in report["by_reason"], "Should track reason types" + + print(f" Total clusters: {report['total_clusters']}") + print(f" Total duplicates: {report['total_duplicates']}") + print(f" Avg confidence: {report['avg_confidence']:.3f}") + print(f" By reason: {report['by_reason']}") + print(" [PASS]") + + +async def main(): + """Run all tests.""" + print("=" * 70) + print("Phase 5 Entity Resolver Tests") + print("=" * 70) + + try: + await test_normalize_label() + await test_jaro_winkler_similarity() + await test_text_similarity() + await test_embedder_initialization() + await test_detect_duplicates() + await test_resolve_cluster() + await test_resolution_report() + + print("\n" + "=" * 70) + print("All tests passed!") + print("=" * 70) + print("\nPhase 5.0 Entity Resolver capabilities:") + print(" [OK] Label normalization") + print(" [OK] Jaro-Winkler text similarity") + print(" [OK] Vector embeddings (all-MiniLM-L6-v2)") + print(" [OK] Duplicate detection (vector + text)") + print(" [OK] Entity merging and consolidation") + print(" [OK] Resolution reporting") + + return True + except AssertionError as e: + print(f"\nTest failed: {e}") + return False + except Exception as e: + print(f"\nUnexpected error: {e}") + import traceback + traceback.print_exc() + return False + + +if __name__ == "__main__": + success = asyncio.run(main()) + sys.exit(0 if success else 1)