From 2be45a01f1f31d2158a97bb454814341b022682f Mon Sep 17 00:00:00 2001 From: ZohaibHassan16 Date: Sat, 7 Mar 2026 17:24:05 +0500 Subject: [PATCH] feat: implement ontology alignment API(#324) --- docs/reference/ontology.md | 50 ++++++++++- docs/reference/triplet_store.md | 39 +++++++++ semantica/ontology/engine.py | 101 ++++++++++++++++++++++ semantica/ontology/namespace_manager.py | 25 ++++++ semantica/ontology/reuse_manager.py | 64 ++++++++++++++ semantica/triplet_store/query_engine.py | 73 ++++++++++++++++ tests/ontology/test_ontology_advanced.py | 58 +++++++++++++ tests/triplet_store/test_triplet_store.py | 36 ++++++++ 8 files changed, 445 insertions(+), 1 deletion(-) diff --git a/docs/reference/ontology.md b/docs/reference/ontology.md index 5c46d816..fb3cc9ab 100644 --- a/docs/reference/ontology.md +++ b/docs/reference/ontology.md @@ -250,6 +250,54 @@ ontology: --- +## Ontology Alignment + +Semantica supports mapping and connecting different ontologies to unify data across systems, standards, and domains. This enables cross-system interoperability, allowing a single semantic layer to span multiple standards (e.g., internal models and industry standards). + +Alignments are represented using standard RDF predicates such as `owl:equivalentClass`, `owl:equivalentProperty`, and `skos:exactMatch`. + +### Creating and Managing Alignments + +You can create and query alignments programmatically using the `OntologyEngine`: + +```python +from semantica.ontology.engine import OntologyEngine + +# Ensure your engine is initialized with a TripletStore instance +engine = OntologyEngine(store=my_triplet_store) + +# Create an alignment between an internal class and a standard schema +engine.create_alignment( + source_uri="http://internal.org/ontology/Employee" + target_uri="[http://schema.org/Person]", + predicate="[http://www.w3.org/2002/07/owl#equivalentClass]" +) + +# Retrieve all bidirectional alignments for a specific entity +alignments = engine.get_alignments("[http://internal.org/ontology/Employee](http://internal.org/ontology/Employee)") +``` +### Automated Alignment Suggestions + +When importing or merging external ontologies, the ReuseManager can automatically suggest alignments based on heuristic matching (such as identical labels with differing URIs). + +```python +from semantica.ontology.reuse_manager import ReuseManager + +manager = ReuseManager() + +# Merge ontologies and auto-compute alignment suggestions +merged_ontology = manager.merge_ontology_data( + target=internal_ontology, + source=industry_ontology, + compute_alignments=True +) + +# Suggestions are stored in merged_ontology["suggested_alignments"] +``` + +For executing SPARQL queries that utilize these alignments to retrieve cross-ontology results, see the [Triplet Store Alignment-Aware Queries](triplet_store.md#alignment-aware-queries) + + ## Integration Examples ### Schema-First Knowledge Graph @@ -312,4 +360,4 @@ Interactive tutorials to learn ontology generation and management: - **[Unstructured to Ontology](https://github.com/Hawksight-AI/semantica/blob/main/cookbook/advanced/12_Unstructured_to_Ontology.ipynb)**: Generate ontologies automatically from unstructured data - **Topics**: Automatic ontology generation, 6-stage pipeline, OWL validation - **Difficulty**: Advanced - - **Use Cases**: Domain modeling, automatic schema generation + - **Use Cases**: Domain modeling, automatic schema generation \ No newline at end of file diff --git a/docs/reference/triplet_store.md b/docs/reference/triplet_store.md index 321af524..f42fb9c4 100644 --- a/docs/reference/triplet_store.md +++ b/docs/reference/triplet_store.md @@ -152,6 +152,8 @@ SPARQL query execution and optimization engine. |--------|-------------|-----------| | `execute(query)` | Execute SPARQL query | Query execution | | `optimize(query)` | Optimize SPARQL query | Query rewriting | +| `expand_entity_uri(uri, store, ...)` | Expand aligned entity URIs | Bidirectional SPARQL lookup | +| `build_values_clause(var, uris)` | Generate VALUES clause | String formatting | --- @@ -204,3 +206,40 @@ LIMIT 10 """ results = store.execute_query(query) ``` +### Alignment-Aware Queries + +In complex enterprise environments with multiple data sources, you may want queries to seamlessly retrieve instances across aligned classes. For example, retrieving all http://schema.org/Person instances when querying for your internal http://internal.org/ontology/Employee class. + +The QueryEngine provides helper methods to expand entity URIs based on stored alignments (e.g., owl:equivalentClass, owl:sameAs, skos:exactMatch) and safely inject them into your queries using SPARQL VALUES clauses. + +Expanding URIs in Queries +You can expand a URI and build an alignment-aware query dynamically: + +```python +from semantica.triplet_store.query_engine import QueryEngine + +engine = QueryEngine() + +# i) Expand the base URI to include all aligned equivalents +expanded_uris = engine.expand_entity_uri( + entity_uri="[http://internal.org/ontology/Employee](http://internal.org/ontology/Employee)", + store_backend=store_backend, + use_alignments=True +) + +# ii) Build a SPARQL VALUES clause +values_clause = engine.build_values_clause("entity_class", expanded_uris) +# Result: VALUES ?entity_class { [http://internal.org/ontology/Employee](http://internal.org/ontology/Employee) [http://schema.org/Person](http://schema.org/Person) } + +# iii) Inject the clause into your query template +query = f""" +SELECT ?instance ?name WHERE {{ + {values_clause} + ?instance a ?entity_class . + ?instance [http://schema.org/name](http://schema.org/name) ?name . +}} +""" + +# Execute the query to retrieve results across all aligned ontologies +results = engine.execute_query(query, store_backend) +``` diff --git a/semantica/ontology/engine.py b/semantica/ontology/engine.py index 6635d03b..6c2bbc1b 100644 --- a/semantica/ontology/engine.py +++ b/semantica/ontology/engine.py @@ -10,6 +10,7 @@ from .owl_generator import OWLGenerator from .ontology_evaluator import OntologyEvaluator from .ontology_validator import OntologyValidator from .llm_generator import LLMOntologyGenerator +from ..semantic_extract.triplet_extractor import Triplet class OntologyEngine: @@ -25,6 +26,7 @@ class OntologyEngine: self.evaluator = OntologyEvaluator(**config) self.validator = OntologyValidator(**config) self.llm = LLMOntologyGenerator(**config) + self.store = config.get("store") def from_data(self, data: Dict[str, Any], **options) -> Dict[str, Any]: tracking_id = self.progress.start_tracking( @@ -56,6 +58,103 @@ class OntologyEngine: **options, ) -> List[Dict[str, Any]]: return self.propgen.infer_properties(entities, relationships, classes, **options) + + def create_alignment(self, source_uri: str, target_uri: str, predicate: str, **options) -> None: + """ + Creates an alignment between two ontology entities and stores it. + """ + if not self.store: + raise ProcessingError("TripletStore instance not configured in OntologyEngine.") + + tracking_id = self.progress.start_tracking( + module="ontology", + submodule="OntologyEngine", + message=f"Creating alignment: {source_uri} -> {target_uri}" + ) + try: + triplet = Triplet(subject=source_uri, predicate=predicate, object=target_uri) + self.store.add_triplet(triplet, **options) + + self.progress.stop_tracking(tracking_id, status="completed", message="Alignment created") + except Exception as e: + self.progress.stop_tracking(tracking_id, status="failed", message=str(e)) + self.logger.error(f"Failed to create alignment: {e}") + raise ProcessingError(f"Alignment creation failed: {e}") + + def get_alignments(self, entity_uri: str, **options) -> List[Dict[str, Any]]: + """ + Retrieves all alignments for a specific entity URI (bidirectional). + """ + if not self.store: + raise ProcessingError("TripletStore instance not configured in OntologyEngine.") + + query = f""" + SELECT ?s ?p ?o WHERE {{ + {{ <{entity_uri}> ?p ?o . BIND(<{entity_uri}> AS ?s) }} + UNION + {{ ?s ?p <{entity_uri}> . BIND(<{entity_uri}> AS ?o) }} + + FILTER ( + STRSTARTS(STR(?p), "http://www.w3.org/2002/07/owl#") || + STRSTARTS(STR(?p), "http://www.w3.org/2004/02/skos/core#") + ) + }} + """ + try: + results = self.store.execute_query(query, **options) + + alignments = [] + if hasattr(results, 'bindings'): + for b in results.bindings: + alignments.append({ + "source": b.get("s", {}).get("value") if isinstance(b.get("s"), dict) else b.get("s"), + "predicate": b.get("p", {}).get("value") if isinstance(b.get("p"), dict) else b.get("p"), + "target": b.get("o", {}).get("value") if isinstance(b.get("o"), dict) else b.get("o") + }) + return alignments + except Exception as e: + self.logger.error(f"Failed to get alignments for {entity_uri}: {e}") + raise ProcessingError(f"Failed to get alignments: {e}") + + def list_alignments(self, ontology_uri: Optional[str] = None, **options) -> List[Dict[str, Any]]: + """ + Lists all alignments, optionally filtered by an ontology URI. + """ + if not self.store: + raise ProcessingError("TripletStore instance not configured in OntologyEngine.") + + filter_clause = "" + if ontology_uri: + filter_clause = f'FILTER(STRSTARTS(STR(?s), "{ontology_uri}") || STRSTARTS(STR(?o), "{ontology_uri}"))' + + query = f""" + SELECT ?s ?p ?o WHERE {{ + ?s ?p ?o . + FILTER ( + STRSTARTS(STR(?p), "http://www.w3.org/2002/07/owl#equivalent") || + STRSTARTS(STR(?p), "http://www.w3.org/2002/07/owl#sameAs") || + STRSTARTS(STR(?p), "http://www.w3.org/2004/02/skos/core#") + ) + {filter_clause} + }} + """ + try: + results = self.store.execute_query(query, **options) + + alignments = [] + if hasattr(results, 'bindings'): + for b in results.bindings: + alignments.append({ + "source": b.get("s", {}).get("value") if isinstance(b.get("s"), dict) else b.get("s"), + "predicate": b.get("p", {}).get("value") if isinstance(b.get("p"), dict) else b.get("p"), + "target": b.get("o", {}).get("value") if isinstance(b.get("o"), dict) else b.get("o") + }) + return alignments + except Exception as e: + self.logger.error(f"Failed to list alignments: {e}") + raise ProcessingError(f"Failed to list alignments: {e}") + + def evaluate(self, ontology: Dict[str, Any], **options): return self.evaluator.evaluate_ontology(ontology, **options) @@ -68,4 +167,6 @@ class OntologyEngine: def export_owl(self, ontology: Dict[str, Any], path: str, format: str = "turtle"): return self.owl.export_owl(ontology, path, format=format) + + diff --git a/semantica/ontology/namespace_manager.py b/semantica/ontology/namespace_manager.py index fdacbfd9..614492a7 100644 --- a/semantica/ontology/namespace_manager.py +++ b/semantica/ontology/namespace_manager.py @@ -206,6 +206,31 @@ class NamespaceManager: Dictionary of prefix -> URI mappings """ return dict(self.namespaces) + + def get_alignment_predicates(self) -> Dict[str, str]: + """ + Get standard alignment predicates for ontology mapping. + + Returns: + Dictionary mapping common alignment types to their full URIs. + """ + owl_ns = self.get_namespace("owl") + skos_ns = self.get_namespace("skos") + + return { + #OWL alignments + "equivalentClass": f"{owl_ns}equivalentClass", + "equivalentProperty": f"{owl_ns}equivalentProperty", + "sameAs": f"{owl_ns}sameAs", + #SKOS alignments + "exactMatch": f"{skos_ns}exactMatch", + "closeMatch": f"{skos_ns}closeMatch", + "broadMatch": f"{skos_ns}broadMatch", + "narrowMatch": f"{skos_ns}narrowMatch", + "relatedMatch": f"{skos_ns}relatedMatch", + + } + def _to_pascal_case(self, name: str) -> str: """Convert name to PascalCase.""" diff --git a/semantica/ontology/reuse_manager.py b/semantica/ontology/reuse_manager.py index 2c9b546e..16ec266d 100644 --- a/semantica/ontology/reuse_manager.py +++ b/semantica/ontology/reuse_manager.py @@ -363,6 +363,60 @@ class ReuseManager: def list_known_ontologies(self) -> List[str]: """List known ontology URIs.""" return list(self.known_ontologies.keys()) + + def suggest_alignments( + self, target: Dict[str, Any], source: Dict[str, Any], **options + ) -> List[Dict[str, str]]: + """ + Suggest alignments between a source and target ontology based on heuristics. + Currently matches identical class/property names with differing URIs. + + Args: + target: Target ontology dictionary + source: Source ontology dictionary + **options: Additional heuristics configurations + + Returns: + List of alignment dictionaries (source_uri, target_uri, predicate, reason) + """ + suggestions = [] + + # Helper to find matches based on identical names + def find_matches(target_items, source_items, entity_type): + predicate = ( + "http://www.w3.org/2002/07/owl#equivalentClass" + if entity_type == "class" + else "http://www.w3.org/2002/07/owl#equivalentProperty" + ) + + for s_item in source_items: + s_uri = s_item.get("uri") + s_name = s_item.get("name", "").strip().lower() + if not s_uri or not s_name: + continue + + for t_item in target_items: + t_uri = t_item.get("uri") + t_name = t_item.get("name", "").strip().lower() + + if not t_uri or not t_name: + continue + + # If names match exactly but URIs are different, suggest an alignment + if s_name == t_name and s_uri != t_uri: + suggestions.append({ + "source_uri": s_uri, + "target_uri": t_uri, + "predicate": predicate, + "reason": f"Exact label match for {entity_type}: '{s_item.get('name')}'" + }) + + + find_matches(target.get("classes", []), source.get("classes", []), "class") + find_matches(target.get("properties", []), source.get("properties", []), "property") + + return suggestions + def merge_ontology_data( self, target: Dict[str, Any], source: Dict[str, Any], **options @@ -437,6 +491,16 @@ class ReuseManager: for imp in source["imports"]: if imp not in target["imports"]: target["imports"].append(imp) + + if options.get("compute_alignments", False): + self.progress_tracker.update_tracking( + tracking_id, message="Computing suggested alignments..." + ) + suggested = self.suggest_alignments(target, source, **options) + if suggested: + if "suggested_alignments" not in target: + target["suggested_alignments"] = [] + target["suggested_alignments"].extend(suggested) self.progress_tracker.stop_tracking( tracking_id, diff --git a/semantica/triplet_store/query_engine.py b/semantica/triplet_store/query_engine.py index b5ac307c..e90a9869 100644 --- a/semantica/triplet_store/query_engine.py +++ b/semantica/triplet_store/query_engine.py @@ -267,6 +267,79 @@ class QueryEngine: execution_steps=execution_steps, metadata={"optimization_enabled": self.enable_optimization}, ) + + def expand_entity_uri(self, entity_uri: str, store_backend: Any, use_alignments: bool = False) -> List[str]: + """ + Expand an entity URI to include all aligned/equivalent entities. + + Args: + entity_uri: The original URI to expand + store_backend: Triplet store backend to query + use_alignments: If False, returns only the original URI + + Returns: + List of URIs including the original and any aligned entities + """ + if not use_alignments: + return [entity_uri] + + tracking_id = self.progress_tracker.start_tracking( + module="triplet_store", + submodule="QueryEngine", + message=f"Expanding alignments for: {entity_uri}" + ) + + # SPARQL query to find bidirectional alignments + query = f""" + SELECT DISTINCT ?aligned WHERE {{ + {{ <{entity_uri}> ?p ?aligned }} + UNION + {{ ?aligned ?p <{entity_uri}> }} + + FILTER ( + STRSTARTS(STR(?p), "http://www.w3.org/2002/07/owl#equivalent") || + STRSTARTS(STR(?p), "http://www.w3.org/2002/07/owl#sameAs") || + STRSTARTS(STR(?p), "http://www.w3.org/2004/02/skos/core#") + ) + }} + """ + + expanded_uris = set([entity_uri]) + try: + if hasattr(store_backend, "execute_sparql"): + result_data = store_backend.execute_sparql(query) + for binding in result_data.get("bindings", []): + val = binding.get("aligned", {}) + uri = val.get("value") if isinstance(val, dict) else val + if uri: + expanded_uris.add(uri) + + self.progress_tracker.stop_tracking( + tracking_id, + status="completed", + message=f"Expanded to {len(expanded_uris)} URIs" + ) + except Exception as e: + self.logger.error(f"Failed to expand alignments for {entity_uri}: {e}") + self.progress_tracker.stop_tracking(tracking_id, status="failed", message=str(e)) + # Fallback to returning just the original URI if the expansion query fails + + return list(expanded_uris) + + def build_values_clause(self, variable_name: str, uris: List[str]) -> str: + """ + Helper to generate a SPARQL VALUES clause for a list of URIs. + Allows higher-level components to build alignment-aware queries. + + Example: + uris = engine.expand_entity_uri("http://ex.org/Person", store, use_alignments=True) + clause = engine.build_values_clause("subject", uris) + # Returns: VALUES ?subject { } + """ + if not uris: + return "" + formatted_uris = " ".join([f"<{uri}>" for uri in uris]) + return f"VALUES ?{variable_name} {{ {formatted_uris} }}" def _validate_query(self, query: str) -> bool: """Validate SPARQL query syntax (basic).""" diff --git a/tests/ontology/test_ontology_advanced.py b/tests/ontology/test_ontology_advanced.py index 8371ed7c..25d257d8 100644 --- a/tests/ontology/test_ontology_advanced.py +++ b/tests/ontology/test_ontology_advanced.py @@ -8,6 +8,8 @@ from semantica.ontology.ontology_evaluator import OntologyEvaluator, EvaluationR from semantica.ontology.competency_questions import CompetencyQuestionsManager, CompetencyQuestion from semantica.change_management import VersionManager, OntologyVersion from semantica.ontology.associative_class import AssociativeClassBuilder, AssociativeClass +from semantica.ontology.reuse_manager import ReuseManager +from semantica.ontology.engine import OntologyEngine class TestOntologyAdvanced(unittest.TestCase): @@ -151,6 +153,62 @@ class TestOntologyAdvanced(unittest.TestCase): self.assertEqual(len(is_valid), 0) except Exception: pass + + def test_reuse_manager_suggest_alignments(self): + manager = ReuseManager() + target = { + "classes": [{"uri": "http://target.org/Person", "name": "Person"}], + "properties": [{"uri": "http://target.org/hasName", "name": "has name"}] + } + source = { + "classes": [{"uri": "http://source.org/Person", "name": "Person"}], + "properties": [{"uri": "http://source.org/hasName", "name": "has name"}] + } + + suggestions = manager.suggest_alignments(target, source) + + self.assertEqual(len(suggestions), 2) + self.assertEqual(suggestions[0]["predicate"], "http://www.w3.org/2002/07/owl#equivalentClass") + self.assertEqual(suggestions[0]["source_uri"], "http://source.org/Person") + self.assertEqual(suggestions[1]["predicate"], "http://www.w3.org/2002/07/owl#equivalentProperty") + + def test_reuse_manager_merge_with_alignments(self): + manager = ReuseManager() + target = {"classes": [{"uri": "http://target.org/Dog", "name": "Dog"}]} + source = {"classes": [{"uri": "http://source.org/Dog", "name": "Dog"}]} + + merged = manager.merge_ontology_data(target, source, compute_alignments=True) + + self.assertIn("suggested_alignments", merged) + self.assertEqual(len(merged["suggested_alignments"]), 1) + self.assertEqual(merged["suggested_alignments"][0]["target_uri"], "http://target.org/Dog") + + def test_engine_create_alignment(self): + mock_store = MagicMock() + engine = OntologyEngine(store=mock_store) + + engine.create_alignment("http://source.org/1", "http://target.org/2", "http://www.w3.org/2002/07/owl#sameAs") + + mock_store.add_triplet.assert_called_once() + args, kwargs = mock_store.add_triplet.call_args + self.assertEqual(args[0].subject, "http://source.org/1") + self.assertEqual(args[0].object, "http://target.org/2") + self.assertEqual(args[0].predicate, "http://www.w3.org/2002/07/owl#sameAs") + + def test_engine_get_alignments(self): + mock_store = MagicMock() + mock_result = MagicMock() + # Mocking the SPARQL binding response format + mock_result.bindings = [ + {"s": {"value": "http://source.org/1"}, "p": {"value": "http://owl#sameAs"}, "o": {"value": "http://target.org/2"}} + ] + mock_store.execute_query.return_value = mock_result + + engine = OntologyEngine(store=mock_store) + alignments = engine.get_alignments("http://source.org/1") + + self.assertEqual(len(alignments), 1) + self.assertEqual(alignments[0]["target"], "http://target.org/2") if __name__ == '__main__': unittest.main() diff --git a/tests/triplet_store/test_triplet_store.py b/tests/triplet_store/test_triplet_store.py index d6e679ee..5a6e0dad 100644 --- a/tests/triplet_store/test_triplet_store.py +++ b/tests/triplet_store/test_triplet_store.py @@ -98,3 +98,39 @@ class TestTripletStore(unittest.TestCase): self.assertTrue(result["success"]) mock_backend_instance.delete_triplet.assert_called_once_with(triplet) + + def test_query_engine_build_values_clause(self): + engine = QueryEngine() + uris = ["http://ex.org/1", "http://ex.org/2"] + + clause = engine.build_values_clause("subject", uris) + self.assertEqual(clause, "VALUES ?subject { }") + + empty_clause = engine.build_values_clause("subject", []) + self.assertEqual(empty_clause, "") + + def test_query_engine_expand_entity_uri_disabled(self): + engine = QueryEngine() + mock_backend = MagicMock() + + result = engine.expand_entity_uri("http://ex.org/1", mock_backend, use_alignments=False) + + # Should return only the original URI and NOT query the store + self.assertEqual(result, ["http://ex.org/1"]) + mock_backend.execute_sparql.assert_not_called() + + def test_query_engine_expand_entity_uri_enabled(self): + engine = QueryEngine() + mock_backend = MagicMock() + + # Mock the backend returning an aligned URI + mock_backend.execute_sparql.return_value = { + "bindings": [{"aligned": {"value": "http://ex.org/aligned_entity"}}] + } + + result = engine.expand_entity_uri("http://ex.org/original", mock_backend, use_alignments=True) + + self.assertIn("http://ex.org/original", result) + self.assertIn("http://ex.org/aligned_entity", result) + self.assertEqual(len(result), 2) + mock_backend.execute_sparql.assert_called_once()