semantica/cookbook/use_cases/blockchain/01_DeFi_Protocol_Intelligence.ipynb

DeFi Protocol Intelligence - Risk Assessment & Ontology Reasoning

Overview

This notebook demonstrates DeFi protocol intelligence using Semantica with focus on risk assessment, ontology-based reasoning, and relationship analysis. The pipeline ingests DeFi data from multiple sources, extracts protocol entities, builds knowledge graphs, and assesses risks using graph reasoning.

Key Features

• Risk Assessment Focus: Emphasizes KG construction and reasoning for risk evaluation
• Ontology-Based Reasoning: Uses domain ontologies for DeFi protocol analysis
• Relationship Analysis: Analyzes protocol relationships and dependencies
• Comprehensive Data Sources: Multiple RSS feeds, APIs, and databases
• Modular Architecture: Direct use of Semantica modules without core orchestrator

Learning Objectives

• Ingest DeFi data from multiple sources (RSS feeds, APIs, databases)
• Extract DeFi entities (Protocols, Tokens, Pools, Transactions, Risks)
• Build and analyze DeFi knowledge graphs
• Generate and utilize DeFi ontologies
• Perform risk assessment using graph reasoning
• Store and query DeFi data using vector stores and graph stores

Pipeline Flow

graph TD
    A[Data Ingestion] --> B[Document Parsing]
    B --> C[Text Processing]
    C --> D[Entity Extraction]
    D --> E[Relationship Extraction]
    E --> F[Deduplication]
    F --> G[Conflict Detection]
    G --> H[Knowledge Graph]
    H --> I[Embeddings]
    I --> J[Vector Store]
    H --> K[Ontology Generation]
    K --> L[Reasoning & Risk]
    J --> M[GraphRAG Queries]
    L --> M
    H --> N[Graph Store]
    K --> O[Triplet Store]
    M --> P[Visualization]
    N --> P
    O --> P
    P --> Q[Export]

Installation

In [ ]:

%pip install -qU semantica networkx matplotlib plotly pandas faiss-cpu beautifulsoup4 groq sentence-transformers scikit-learn

Configuration & Setup

In [ ]:

import os

os.environ["GROQ_API_KEY"] = os.getenv("GROQ_API_KEY", "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")

# Configuration constants
EMBEDDING_DIMENSION = 384
EMBEDDING_MODEL = "all-MiniLM-L6-v2"
CHUNK_SIZE = 1000
CHUNK_OVERLAP = 200

Ingesting DeFi Data from Multiple Sources

In [ ]:

from semantica.ingest import FeedIngestor, FileIngestor, WebIngestor
import os
from contextlib import redirect_stderr
from io import StringIO

os.makedirs("data", exist_ok=True)

feed_sources = [
    # Crypto News RSS Feeds
    ("CoinDesk", "https://www.coindesk.com/arc/outboundfeeds/rss/"),
    ("CoinTelegraph", "https://cointelegraph.com/rss"),
    ("Decrypt", "https://decrypt.co/feed"),
    ("The Block", "https://www.theblock.co/rss.xml"),
    ("CryptoSlate", "https://cryptoslate.com/feed/"),
    ("CryptoNews", "https://cryptonews.com/news/feed/"),
]

feed_ingestor = FeedIngestor()
all_documents = []

print(f"Ingesting from {len(feed_sources)} feed sources...")
for i, (feed_name, feed_url) in enumerate(feed_sources, 1):
    try:
        with redirect_stderr(StringIO()):
            feed_data = feed_ingestor.ingest_feed(feed_url, validate=False)
        
        feed_count = 0
        for item in feed_data.items:
            if not item.content:
                item.content = item.description or item.title or ""
            if item.content:
                if not hasattr(item, 'metadata'):
                    item.metadata = {}
                item.metadata['source'] = feed_name
                all_documents.append(item)
                feed_count += 1
        
        if feed_count > 0:
            print(f"  [{i}/{len(feed_sources)}] {feed_name}: {feed_count} documents")
    except Exception:
        continue

if not all_documents:
    defi_data = """
    Uniswap is a decentralized exchange protocol with high liquidity pools. It uses automated market makers (AMMs) for token swaps.
    Aave is a lending protocol that offers variable and stable interest rates. Users can deposit assets to earn yield.
    Compound is a money market protocol for lending and borrowing cryptocurrencies. It uses algorithmic interest rates.
    MakerDAO uses collateralized debt positions (CDPs) for stablecoin generation. DAI is the stablecoin created.
    Curve Finance is a decentralized exchange optimized for stablecoin trading with low slippage.
    Yearn Finance aggregates yield farming strategies across multiple DeFi protocols.
    SushiSwap is a decentralized exchange and automated market maker with yield farming features.
    Balancer is a protocol for programmable liquidity and automated portfolio management.
    """
    with open("data/defi_protocols.txt", "w") as f:
        f.write(defi_data)
    file_ingestor = FileIngestor()
    all_documents = file_ingestor.ingest("data/defi_protocols.txt")

documents = all_documents
print(f"Ingested {len(documents)} documents")

In [ ]:

from semantica.parse import DocumentParser

parser = DocumentParser()

print(f"Parsing {len(documents)} documents...")
parsed_documents = []
for i, doc in enumerate(documents, 1):
    try:
        parsed = parser.parse(
            doc.content if hasattr(doc, 'content') else str(doc),
            content_type="text"
        )
        parsed_documents.append(parsed)
    except Exception:
        parsed_documents.append(doc)
    if i % 50 == 0 or i == len(documents):
        print(f"  Parsed {i}/{len(documents)} documents...")

documents = parsed_documents

Normalizing and Chunking DeFi Documents

In [ ]:

from semantica.normalize import TextNormalizer
from semantica.split import TextSplitter

normalizer = TextNormalizer()
splitter = TextSplitter(
    method="entity_aware",
    ner_method="spacy",
    chunk_size=CHUNK_SIZE,
    chunk_overlap=CHUNK_OVERLAP
)

print(f"Normalizing {len(documents)} documents...")
normalized_documents = []
for i, doc in enumerate(documents, 1):
    normalized_text = normalizer.normalize(
        doc.content if hasattr(doc, 'content') else str(doc),
        clean_html=True,
        normalize_entities=True,
        remove_extra_whitespace=True,
        lowercase=False
    )
    normalized_documents.append(normalized_text)
    if i % 50 == 0 or i == len(documents):
        print(f"  Normalized {i}/{len(documents)} documents...")

print(f"Chunking {len(normalized_documents)} documents...")
chunked_documents = []
for i, doc_text in enumerate(normalized_documents, 1):
    try:
        with redirect_stderr(StringIO()):
            chunks = splitter.split(doc_text)
        chunked_documents.extend(chunks)
    except Exception:
        simple_splitter = TextSplitter(method="recursive", chunk_size=CHUNK_SIZE, chunk_overlap=CHUNK_OVERLAP)
        chunks = simple_splitter.split(doc_text)
        chunked_documents.extend(chunks)
    if i % 50 == 0 or i == len(normalized_documents):
        print(f"  Chunked {i}/{len(normalized_documents)} documents ({len(chunked_documents)} chunks so far)")

print(f"Created {len(chunked_documents)} chunks from {len(normalized_documents)} documents")

Extracting DeFi Entities

In [ ]:

from semantica.semantic_extract import NERExtractor

# Initialize NERExtractor with ML method only (spaCy)
# Note: ML method extracts standard NER labels (PERSON, ORG, GPE, etc.)
entity_extractor = NERExtractor(
    method=["ml"],
    min_confidence=0.5
)

# Extract all entities (ML method doesn't support custom entity types)
# We'll filter/classify them after extraction
print(f"Extracting entities from {len(chunked_documents)} chunks using ML (spaCy)...")
batch_results = entity_extractor.extract(chunked_documents)

# Flatten results (extract() returns List[List[Entity]] for batch input)
all_entities = [entity for entity_list in batch_results for entity in entity_list]

# Use Semantica's classify_entities to group by standard labels
classified = entity_extractor.classify_entities(all_entities)

# Filter entities for DeFi domain - look for protocol/token names in ORG entities
# and common DeFi-related terms
protocol_keywords = ["uniswap", "aave", "compound", "makerdao", "curve", "yearn", 
                     "sushiswap", "balancer", "protocol", "defi", "dapp"]
token_keywords = ["token", "coin", "crypto", "btc", "eth", "dai", "usdc", "usdt"]
risk_keywords = ["risk", "vulnerability", "exploit", "hack", "attack", "breach"]

protocols = [
    e for e in all_entities 
    if e.label == "ORG" or any(kw in e.text.lower() for kw in protocol_keywords)
]
tokens = [
    e for e in all_entities 
    if any(kw in e.text.lower() for kw in token_keywords) or e.label == "MONEY"
]
risks = [
    e for e in all_entities 
    if any(kw in e.text.lower() for kw in risk_keywords)
]

print(f"\n✅ Extraction complete!")
print(f"   Total entities: {len(all_entities)}")
print(f"   Standard labels: {list(classified.keys())}")
print(f"   Protocols (filtered): {len(protocols)}")
print(f"   Tokens (filtered): {len(tokens)}")
print(f"   Risks (filtered): {len(risks)}")

Extracting DeFi Relationships

In [ ]:

from semantica.semantic_extract import RelationExtractor

# Use ML-based dependency parsing to avoid rate limits
relation_extractor = RelationExtractor(
    method="dependency",  # ML/NLP method - no API calls needed
    verbose=True
)

all_relationships = []
error_count = 0
print(f"Extracting relationships from {len(chunked_documents)} chunks...")

for i, chunk in enumerate(chunked_documents, 1):
    chunk_text = chunk.text if hasattr(chunk, 'text') else str(chunk)
    try:
        relationships = relation_extractor.extract_relations(
            chunk_text,
            entities=all_entities,
            relation_types=["uses", "governs", "provides", "has_risk", "interacts_with", "depends_on"],
            verbose=True
        )
        all_relationships.extend(relationships)
    except Exception as e:
        error_count += 1
        if error_count <= 3:
            print(f"  Warning: Error on chunk {i}: {str(e)[:100]}")
    
    if i % 20 == 0 or i == len(chunked_documents):
        print(f"  Processed {i}/{len(chunked_documents)} chunks ({len(all_relationships)} relationships found)")

if error_count > 0:
    print(f"  Note: {error_count} chunks had errors during relation extraction")

print(f"Extracted {len(all_relationships)} relationships")

Resolving Duplicate Entities

Detecting and Resolving Conflicts

• Entity & Relationship Conflict Detection: Detects conflicts in both entity properties (protocol names, addresses) and relationships (protocol interactions, dependencies) from multiple data sources to ensure data consistency across the DeFi knowledge graph.

• Credibility-Weighted Resolution: Uses credibility-weighted strategy that considers source reliability and extraction confidence scores, prioritizing high-confidence sources for critical DeFi protocol information while aggregating evidence from multiple sources.

In [ ]:

from semantica.conflicts import ConflictDetector, ConflictResolver

conflict_detector = ConflictDetector()
conflict_resolver = ConflictResolver()

# Convert entities to dictionaries for conflict detection
entity_dicts = [
    {
        "id": e.id if hasattr(e, 'id') else e.text,
        "text": e.text,
        "label": e.label,
        "type": e.label,
        "confidence": e.confidence if hasattr(e, 'confidence') else 1.0,
        "metadata": e.metadata if hasattr(e, 'metadata') else {}
    }
    for e in all_entities
]

# Convert relationships to dictionaries for conflict detection
relationship_dicts = [
    {
        "id": f"{r.subject.text}_{r.predicate}_{r.object.text}",
        "source_id": r.subject.text,
        "target_id": r.object.text,
        "type": r.predicate,
        "subject": r.subject.text,
        "object": r.object.text,
        "predicate": r.predicate,
        "confidence": r.confidence if hasattr(r, 'confidence') else 1.0,
        "metadata": r.metadata if hasattr(r, 'metadata') else {}
    }
    for r in all_relationships
]

# Detect conflicts in both entities and relationships
all_conflicts = []

# 1. Detect entity conflicts (duplicate protocols, conflicting properties)
print(f"Detecting entity conflicts in {len(entity_dicts)} entities...")
entity_conflicts = conflict_detector.detect_entity_conflicts(entity_dicts)
all_conflicts.extend(entity_conflicts)
print(f"Detected {len(entity_conflicts)} entity conflicts")

# 2. Detect relationship conflicts (conflicting protocol interactions)
print(f"Detecting relationship conflicts in {len(relationship_dicts)} relationships...")
relationship_conflicts = conflict_detector.detect_relationship_conflicts(relationship_dicts)
all_conflicts.extend(relationship_conflicts)
print(f"Detected {len(relationship_conflicts)} relationship conflicts")

# Resolve all conflicts using credibility-weighted strategy
resolved_entities = entity_dicts.copy()
resolved_relationships = relationship_dicts.copy()

if all_conflicts:
    print(f"Resolving {len(all_conflicts)} conflicts using credibility-weighted strategy...")
    resolved = conflict_resolver.resolve_conflicts(
        all_conflicts,
        strategy="credibility_weighted"  # Weight by source credibility and confidence
    )
    
    # Apply resolved values back to entities and relationships
    for result in resolved:
        if result.resolved and result.resolved_value is not None:
            if result.metadata.get("entity_id"):
                # Entity conflict - update entity
                entity_id = result.metadata.get("entity_id")
                property_name = result.metadata.get("property_name")
                for entity in resolved_entities:
                    if entity.get("id") == entity_id and property_name:
                        entity[property_name] = result.resolved_value
            elif result.metadata.get("relationship_id"):
                # Relationship conflict - update relationship
                rel_id = result.metadata.get("relationship_id")
                property_name = result.metadata.get("property_name")
                for rel in resolved_relationships:
                    if rel.get("id") == rel_id and property_name:
                        rel[property_name] = result.resolved_value
    
    print(f"Resolved {len([r for r in resolved if r.resolved])} conflicts")
    print(f"Applied resolutions to {len(resolved_entities)} entities and {len(resolved_relationships)} relationships")
else:
    print("No conflicts detected")

Building DeFi Knowledge Graph

In [ ]:

from semantica.kg import GraphBuilder

# Conflicts already resolved - disable conflict detection in GraphBuilder
graph_builder = GraphBuilder(
    entity_resolution_strategy="fuzzy",
    resolve_conflicts=False  # Conflicts already resolved in previous cell
)

kg_sources = [{
    "entities": [
        {"id": e.get("id", e.get("text")), "text": e.get("text"), "type": e.get("type", e.get("label"))}
        for e in resolved_entities
    ],
    "relationships": [
        {
            "source": r.get("source_id", r.get("subject")),
            "target": r.get("target_id", r.get("object")),
            "type": r.get("type", r.get("predicate"))
        }
        for r in resolved_relationships
    ]
}]

kg = graph_builder.build(kg_sources)

entities_count = len(kg.get('entities', []))
relationships_count = len(kg.get('relationships', []))
print(f"Graph: {entities_count} entities, {relationships_count} relationships")

Generating Embeddings for Protocols and Tokens

In [ ]:

from semantica.embeddings import EmbeddingGenerator

embedding_gen = EmbeddingGenerator(
    provider="sentence_transformers",
    model=EMBEDDING_MODEL
)

print(f"Generating embeddings for {len(protocols)} protocols and {len(tokens)} tokens...")
protocol_texts = [p.text for p in protocols]
protocol_embeddings = embedding_gen.generate_embeddings(protocol_texts)

token_texts = [t.text for t in tokens]
token_embeddings = embedding_gen.generate_embeddings(token_texts)

print(f"Generated {len(protocol_embeddings)} protocol embeddings and {len(token_embeddings)} token embeddings")

Populating Vector Store

In [ ]:

from semantica.vector_store import VectorStore

vector_store = VectorStore(backend="faiss", dimension=EMBEDDING_DIMENSION)

print(f"Storing {len(protocol_embeddings)} protocol vectors and {len(token_embeddings)} token vectors...")
protocol_ids = vector_store.store_vectors(
    vectors=protocol_embeddings,
    metadata=[{"type": "protocol", "name": p.text, "label": p.label} for p in protocols]
)

token_ids = vector_store.store_vectors(
    vectors=token_embeddings,
    metadata=[{"type": "token", "name": t.text, "label": t.label} for t in tokens]
)

print(f"Stored {len(protocol_ids)} protocol vectors and {len(token_ids)} token vectors")

Generating DeFi Ontology

In [ ]:

from semantica.ontology import OntologyGenerator

ontology_gen = OntologyGenerator(base_uri="https://defi.example.org/ontology/")
ontology = ontology_gen.generate_from_graph(kg)

print(f"Generated DeFi ontology with {len(ontology.get('classes', []))} classes")

Reasoning and Risk Assessment

In [ ]:

from semantica.reasoning import Reasoner
from semantica.kg import GraphAnalyzer

reasoner = Reasoner()
reasoner.add_rule("IF Protocol has_risk Risk AND Risk severity high THEN Protocol risk_level critical")
reasoner.add_rule("IF Protocol depends_on Protocol AND Protocol has_risk Risk THEN Protocol inherits Risk")

inferred_facts = reasoner.infer_facts(kg)

# Find paths from Protocols to Risks using GraphAnalyzer
graph_analyzer = GraphAnalyzer(kg)
protocols = [e.get("id") or e.get("text") for e in kg.get("entities", []) if e.get("type") == "Protocol"]
risks = [e.get("id") or e.get("text") for e in kg.get("entities", []) if e.get("type") == "Risk"]

risk_paths = []
for protocol in protocols[:10]:
    for risk in risks[:5]:
        path = graph_analyzer.connectivity_analyzer.calculate_shortest_paths(kg, source=protocol, target=risk)
        if path.get("exists") and path.get("distance", -1) <= 2:
            risk_paths.append(path)

print(f"Inferred {len(inferred_facts)} facts")
print(f"Found {len(risk_paths)} risk paths")

Storing Knowledge Graph (Optional)

In [ ]:

from semantica.graph_store import GraphStore

# Optional: Store to persistent graph database
# graph_store = GraphStore(backend="neo4j", uri="bolt://localhost:7687", user="neo4j", password="password")
# graph_store.store_graph(kg)

print("Graph store configured (commented out for demo)")

Storing Ontology as RDF Triplets (Optional)

In [ ]:

from semantica.triplet_store import TripletStore

# Store knowledge graph and ontology as RDF triplets
# Note: Requires Blazegraph running on localhost:9999
try:
    triplet_store = TripletStore(backend="blazegraph", endpoint="http://localhost:9999/blazegraph")
    result = triplet_store.store(knowledge_graph=kg, ontology=ontology)
    
    if result.get('success'):
        print(f"✓ Stored {result.get('processed', 0)}/{result.get('total', 0)} triplets successfully")
    else:
        print(f"⚠ Stored {result.get('processed', 0)}/{result.get('total', 0)} triplets ({result.get('failed', 0)} failed)")
except Exception as e:
    print(f"⚠ Could not connect to Blazegraph: {str(e)[:100]}")
    print("   To use triplet store, start Blazegraph on localhost:9999")
    print("   Skipping triplet storage for this demo")

GraphRAG: Hybrid Vector + Graph Queries

In [ ]:

from semantica.context import AgentContext

context = AgentContext(vector_store=vector_store, knowledge_graph=kg)

query = "What protocols have high risk?"
results = context.retrieve(
    query,
    max_results=10,
    use_graph=True,
    expand_graph=True,
    include_entities=True,
    include_relationships=True
)

print(f"GraphRAG query: '{query}'")
print(f"\nRetrieved {len(results)} results:\n")
for i, result in enumerate(results[:5], 1):
    print(f"{i}. Score: {result.get('score', 0):.3f}")
    print(f"   Content: {result.get('content', '')[:200]}...")
    if result.get('related_entities'):
        print(f"   Related entities: {len(result['related_entities'])}")
    print()

Visualizing the DeFi Knowledge Graph

In [ ]:

from semantica.visualization import KGVisualizer
import plotly.graph_objects as go

# Print graph statistics for context
num_entities = len(kg.get("entities", []))
num_relationships = len(kg.get("relationships", []))
print(f"📊 Knowledge Graph Statistics:")
print(f"   Entities: {num_entities}")
print(f"   Relationships: {num_relationships}")
print(f"   Density: {num_relationships / max(num_entities * (num_entities - 1) / 2, 1):.4f}\n")

# Create visualizer with highly optimized settings for clarity and interactivity
visualizer = KGVisualizer(
    layout="force",
    node_size=20,  # Larger nodes for better visibility
    edge_width=1.5,
    color_scheme="vibrant",  # Colorful and distinct
    k=3.0,  # More spacing between nodes
    iterations=150,  # More iterations for stable layout
    temperature=0.7,
    cooling_factor=0.99
)

# Generate interactive Plotly figure with maximum interactivity
fig = visualizer.visualize_network(
    kg,
    output="interactive",
    algorithm="kamada_kawai",  # Best algorithm for complex graphs
    node_color_by="type",  # Color by entity type
    node_size_by=None,
    hover_data=["type", "id", "label"],  # Rich hover information
    scale=2.5,  # Large scale for clear spacing
    seed=42,  # Reproducible
    show_detailed_edges=True  # Show edge labels
)

# Enhance the visualization with better interactivity and explanations
if fig:
    # Update layout with comprehensive interactive features
    fig.update_layout(
        title={
            "text": "🔗 DeFi Protocol Knowledge Graph - Interactive Visualization",
            "x": 0.5,
            "xanchor": "center",
            "font": {"size": 20, "color": "#2c3e50"}
        },
        showlegend=True,
        hovermode="closest",  # Show closest node on hover
        margin=dict(b=40, l=40, r=40, t=80),
        xaxis=dict(
            showgrid=False,
            zeroline=False,
            showticklabels=False,
            title=""
        ),
        yaxis=dict(
            showgrid=False,
            zeroline=False,
            showticklabels=False,
            title=""
        ),
        plot_bgcolor="rgba(250, 250, 250, 1)",  # Light gray background
        paper_bgcolor="white",
        font=dict(family="Arial, sans-serif", size=12),
        # Add annotations for explanation
        annotations=[
            dict(
                text="💡 <b>How to interact:</b><br>"
                     "• Hover over nodes to see details<br>"
                     "• Click and drag to pan<br>"
                     "• Use mouse wheel to zoom<br>"
                     "• Double-click to reset view<br>"
                     "• Colors represent entity types",
                xref="paper",
                yref="paper",
                x=0.02,
                y=0.98,
                xanchor="left",
                yanchor="top",
                bgcolor="rgba(255, 255, 255, 0.9)",
                bordercolor="rgba(0, 0, 0, 0.2)",
                borderwidth=1,
                font=dict(size=10, color="#34495e"),
                showarrow=False
            )
        ],
        # Enhanced hover template
        hoverlabel=dict(
            bgcolor="rgba(255, 255, 255, 0.95)",
            bordercolor="#3498db",
            font_size=12,
            font_family="Arial"
        ),
        # Make it more responsive
        autosize=True,
        height=800,
        width=None
    )
    
    # Update traces for better interactivity
    for trace in fig.data:
        if hasattr(trace, 'marker'):
            # Enhance node visibility
            trace.marker.line.width = 2
            trace.marker.line.color = "white"
            trace.marker.opacity = 0.9
        if hasattr(trace, 'text'):
            # Make labels more readable
            trace.textfont.size = 11
            trace.textfont.color = "#2c3e50"
            trace.textposition = "middle center"
    
    # Add modebar with useful tools
    fig.update_layout(
        modebar_add=[
            "zoom2d",
            "pan2d",
            "select2d",
            "lasso2d",
            "zoomIn2d",
            "zoomOut2d",
            "autoScale2d",
            "resetScale2d"
        ]
    )
    
    # Display the enhanced interactive graph
    fig.show(config={
        "displayModeBar": True,
        "displaylogo": False,
        "modeBarButtonsToAdd": ["drawline", "drawopenpath", "drawclosedpath", "drawcircle", "drawrect", "eraseshape"],
        "toImageButtonOptions": {
            "format": "png",
            "filename": "defi_kg",
            "height": 800,
            "width": 1200,
            "scale": 2
        }
    })
    
    print("\n✅ Interactive visualization displayed!")
    print("   Use the toolbar above to zoom, pan, and interact with the graph")
else:
    print("⚠️ Could not generate visualization")

Exporting Results

In [ ]:

from semantica.export import GraphExporter, RDFExporter

# Export knowledge graph to graph formats
graph_exporter = GraphExporter()
graph_exporter.export(kg, output_path="defi_protocol_kg.json", format="json")
graph_exporter.export(kg, output_path="defi_protocol_kg.graphml", format="graphml")

# Export ontology to RDF/TTL format using RDFExporter
rdf_exporter = RDFExporter()
rdf_string = rdf_exporter.export_to_rdf(ontology, format="turtle")
with open("defi_ontology.ttl", "w", encoding="utf-8") as f:
    f.write(rdf_string)

print("✅ Exported knowledge graph to JSON and GraphML formats")
print("✅ Exported ontology to RDF/TTL format")