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

feat: implement ontology alignment API(#324)

Browse Source
This commit is contained in:
ZohaibHassan16
2026-03-07 17:24:05 +05:00
parent e150f43ee4
commit 2be45a01f1
8 changed files with 445 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/reference/ontology.md
+49 -1
View File
@@ -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
docs/reference/triplet_store.md
+39
View File
@@ -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)
```
semantica/ontology/engine.py
+101
View File
@@ -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)
semantica/ontology/namespace_manager.py
+25
View File
@@ -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."""
semantica/ontology/reuse_manager.py
+64
View File
@@ -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,
semantica/triplet_store/query_engine.py
+73
View File
@@ -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 { <http://ex.org/Person> <http://other.org/Human> }
"""
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)."""
tests/ontology/test_ontology_advanced.py
+58
View File
@@ -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()
tests/triplet_store/test_triplet_store.py
+36
View File
@@ -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 { <http://ex.org/1> <http://ex.org/2> }")
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()