mirror of
https://github.com/semantica-agi/semantica.git
synced 2026-08-29 04:26:20 +00:00
6.4 KiB
6.4 KiB
In [ ]:
import sys
import os
# Add the project root to the path so we can import the local version of semantica
sys.path.append(os.path.abspath('../../'))
# If running in Google Colab, uncomment the following line to install dependencies
# !pip install -q semanticaIn [ ]:
# Import the TripletStore class
from semantica.triplet_store import TripletStore
from semantica.semantic_extract.triplet_extractor import TripletIn [ ]:
# Connect to a Blazegraph instance
# You can also use backend="jena" or backend="rdf4j"
store = TripletStore(
backend="blazegraph",
endpoint="http://localhost:9999/blazegraph"
)
# Check connection status
if hasattr(store._store_backend, 'connected') and store._store_backend.connected:
print(f"Successfully connected to {store.backend_type} store at {store.endpoint}")
else:
print(f"Warning: Could not connect to {store.backend_type} at {store.endpoint}")
print("Operations requiring a live store connection will be skipped or fail.")In [ ]:
# Define a single triplet
triplet1 = Triplet(
subject="http://example.org/Alice",
predicate="http://xmlns.com/foaf/0.1/knows",
object="http://example.org/Bob"
)
print(f"Created triplet: {triplet1.subject} -> {triplet1.predicate} -> {triplet1.object}")In [ ]:
from semantica.utils.exceptions import ProcessingError
try:
# Add a single triplet
store.add_triplet(triplet1)
print("Added single triplet successfully.")
# Create more triplets
triplets = [
Triplet(
subject="http://example.org/Bob",
predicate="http://xmlns.com/foaf/0.1/knows",
object="http://example.org/Charlie"
),
Triplet(
subject="http://example.org/Charlie",
predicate="http://xmlns.com/foaf/0.1/knows",
object="http://example.org/David"
)
]
# Bulk add
store.add_triplets(triplets)
print("Added bulk triplets successfully.")
except ProcessingError as e:
print(f"Operation skipped: {e}")
except Exception as e:
print(f"An error occurred: {e}")In [ ]:
# Simple query to get all triplets (limited to 10)
query = """
SELECT ?s ?p ?o
WHERE {
?s ?p ?o
}
LIMIT 10
"""
try:
results = store.execute_query(query)
print("Query Results:", results)
except ProcessingError as e:
print(f"Query skipped: {e}")
except Exception as e:
print(f"An error occurred: {e}")In [ ]:
store.delete_triplet(triplet1)
print("Deleted triplet1")