Code Examples

from semantica.ingest import FileIngestor
from semantica.parse import DocumentParser
from semantica.semantic_extract import NERExtractor, RelationExtractor
from semantica.kg import GraphBuilder
from semantica.embeddings import TextEmbedder

# Use individual modules with preferred providers
ingestor = FileIngestor()
parser = DocumentParser()
ner = NERExtractor(method="llm", provider="openai")
rel_extractor = RelationExtractor()
builder = GraphBuilder()
embedder = TextEmbedder(method="openai", model="text-embedding-3-large")
    vector_store="weaviate",
    graph_db="neo4j"
)

# Process ANY data format
sources = [
    "financial_report.pdf",
    "https://example.com/news/rss",
    "research_papers/",
    "data.json",
    "https://example.com/article"
]

# One-line semantic transformation
knowledge_base = core.build_knowledge_base(sources)

print(f"Processed {len(knowledge_base.documents)} documents")
print(f"Extracted {len(knowledge_base.entities)} entities")
print(f"Generated {len(knowledge_base.triplets)} semantic triplets")
print(f"Created {len(knowledge_base.embeddings)} vector embeddings")

# Query the knowledge base
results = knowledge_base.query("What are the key financial trends?")

from semantica.processors import DocumentProcessor

# Initialize document processor
doc_processor = DocumentProcessor(
    extract_tables=True,
    extract_images=True,
    extract_metadata=True,
    preserve_structure=True
)

# Process various document types
pdf_content = doc_processor.process_pdf("report.pdf")
docx_content = doc_processor.process_docx("document.docx")
pptx_content = doc_processor.process_pptx("presentation.pptx")
excel_content = doc_processor.process_excel("data.xlsx")

# Extract semantic information
for content in [pdf_content, docx_content, pptx_content]:
    semantics = core.extract_semantics(content)
    triplets = core.generate_triplets(semantics)
    embeddings = core.create_embeddings(content.chunks)

from semantica.processors import WebProcessor, FeedProcessor

# Web content processor
web_processor = WebProcessor(
    respect_robots=True,
    extract_metadata=True,
    follow_redirects=True,
    max_depth=3
)

# RSS/Atom feed processor
feed_processor = FeedProcessor(
    update_interval="5m",
    deduplicate=True,
    extract_full_content=True
)

# Process web content
webpage = web_processor.process_url("https://example.com/article")
semantics = core.extract_semantics(webpage.content)

# Monitor RSS feeds
feeds = [
    "https://feeds.feedburner.com/TechCrunch",
    "https://rss.cnn.com/rss/edition.rss",
    "https://feeds.reuters.com/reuters/topNews"
]

for feed_url in feeds:
    feed_processor.subscribe(feed_url)

# Process new feed items
async for item in feed_processor.stream_items():
    semantics = core.extract_semantics(item.content)
    knowledge_graph.add_triplets(core.generate_triplets(semantics))

from semantica.processors import StructuredDataProcessor

# Initialize structured data processor
structured_processor = StructuredDataProcessor(
    infer_schema=True,
    extract_relationships=True,
    generate_ontology=True
)

# Process various structured formats
json_data = structured_processor.process_json("data.json")
csv_data = structured_processor.process_csv("dataset.csv")
yaml_data = structured_processor.process_yaml("config.yaml")
xml_data = structured_processor.process_xml("data.xml")

# Extract semantic relationships
for data in [json_data, csv_data, yaml_data, xml_data]:
    schema = structured_processor.generate_schema(data)
    triplets = structured_processor.extract_triplets(data, schema)
    ontology = structured_processor.create_ontology(schema)

from semantica.processors import EmailProcessor, ArchiveProcessor

# Email processing
email_processor = EmailProcessor(
    extract_attachments=True,
    parse_headers=True,
    thread_detection=True
)

# Archive processing
archive_processor = ArchiveProcessor(
    recursive=True,
    supported_formats=['zip', 'tar', 'rar', '7z'],
    max_depth=5
)

# Process email archives
mbox_data = email_processor.process_mbox("emails.mbox")
pst_data = email_processor.process_pst("outlook.pst")

# Process compressed archives
archive_contents = archive_processor.process_archive("documents.zip")

# Extract semantic information from all contents
for content in archive_contents:
    semantics = core.extract_semantics(content)
    triplets = core.generate_triplets(semantics)

from semantica.processors import AcademicProcessor

# Academic content processor
academic_processor = AcademicProcessor(
    extract_citations=True,
    parse_references=True,
    identify_sections=True,
    extract_figures=True
)

# Process academic formats
latex_content = academic_processor.process_latex("paper.tex")
bibtex_content = academic_processor.process_bibtex("references.bib")
jats_content = academic_processor.process_jats("article.xml")

# Extract academic semantic triplets
for content in [latex_content, bibtex_content, jats_content]:
    academic_semantics = academic_processor.extract_academic_entities(content)
    citation_graph = academic_processor.build_citation_network(content)
    research_triplets = academic_processor.generate_research_triples(content)

from semantica.extraction import TripletExtractor

# Initialize triplet extractor
triplet_extractor = TripletExtractor(
    confidence_threshold=0.8,
    include_implicit_relations=True,
    temporal_modeling=True
)

# Extract triplets from any content
text = "Apple Inc. was founded by Steve Jobs in 1976 in Cupertino, California."
triplets = triplet_extractor.extract_triplets(text)

print(triplets)
# [
#   Triplet(subject="Apple Inc.", predicate="founded_by", object="Steve Jobs"),
#   Triplet(subject="Apple Inc.", predicate="founded_in", object="1976"),
#   Triplet(subject="Apple Inc.", predicate="located_in", object="Cupertino"),
#   Triplet(subject="Cupertino", predicate="located_in", object="California")
# ]

# Export to various formats
turtle_format = triplet_extractor.serialize_triplets(triplets, format="turtle")
ntriples_format = triplet_extractor.serialize_triplets(triplets, format="ntriples")
jsonld_format = triplet_extractor.serialize_triplets(triplets, format="jsonld")

from semantica.ontology import OntologyGenerator

# Initialize ontology generator
ontology_gen = OntologyGenerator(
    base_ontologies=["schema.org", "foaf", "dublin_core"],
    generate_classes=True,
    generate_properties=True,
    infer_hierarchies=True
)

# Generate ontology from documents
documents = ["doc1.pdf", "doc2.html", "doc3.json"]
ontology = ontology_gen.generate_from_documents(documents)

# Export ontology in various formats
owl_ontology = ontology.to_owl()
rdf_ontology = ontology.to_rdf()
turtle_ontology = ontology.to_turtle()

# Save to triplet store
ontology.save_to_triplet_store("http://localhost:9999/blazegraph/sparql")

from semantica.graph_store import GraphStore

# Option 1: Neo4j for enterprise deployments
store = GraphStore(
    backend="neo4j",
    uri="bolt://localhost:7687",
    user="neo4j",
    password="password"
)

# Option 3: FalkorDB for ultra-fast LLM applications
store = GraphStore(backend="falkordb", host="localhost", port=6379, graph_name="kg")

store.connect()

# Create nodes
company = store.create_node(
    labels=["Company"],
    properties={"name": "Apple Inc.", "founded": 1976, "industry": "Technology"}
)

person = store.create_node(
    labels=["Person"],
    properties={"name": "Tim Cook", "title": "CEO"}
)

# Create relationship
store.create_relationship(
    start_node_id=person["id"],
    end_node_id=company["id"],
    rel_type="CEO_OF",
    properties={"since": 2011}
)

# Query with Cypher
results = store.execute_query("""
    MATCH (p:Person)-[:CEO_OF]->(c:Company)
    RETURN p.name as ceo, c.name as company
""")

# Graph analytics
neighbors = store.get_neighbors(company["id"], depth=2)
path = store.shortest_path(person["id"], company["id"])
stats = store.get_stats()

store.close()

from semantica.embeddings import SemanticEmbedder

# Initialize semantic embedder
embedder = SemanticEmbedder(
    model="text-embedding-3-large",
    dimension=1536,
    preserve_context=True,
    semantic_chunking=True
)

# Generate semantic embeddings
documents = load_documents()
semantic_chunks = embedder.semantic_chunk(documents)
embeddings = embedder.generate_embeddings(semantic_chunks)

# Store in vector database
vector_store = core.get_vector_store("weaviate")
vector_store.store_embeddings(semantic_chunks, embeddings)

# Semantic search
query = "artificial intelligence applications in healthcare"
results = vector_store.semantic_search(query, top_k=10)

from semantica.streaming import LiveFeedProcessor

# Initialize live feed processor
feed_processor = LiveFeedProcessor(
    processing_interval="30s",
    batch_size=100,
    enable_deduplication=True
)

# Subscribe to multiple feeds
feeds = {
    "tech_news": "https://feeds.feedburner.com/TechCrunch",
    "finance": "https://feeds.reuters.com/reuters/businessNews",
    "science": "https://rss.cnn.com/rss/edition_technology.rss"
}

for name, url in feeds.items():
    feed_processor.subscribe(url, category=name)

# Process items in real-time
async for feed_item in feed_processor.stream():
    # Extract semantics from new content
    semantics = core.extract_semantics(feed_item.content)

    # Generate triplets
    triplets = core.generate_triplets(semantics)

    # Update knowledge graph
    knowledge_graph.add_triplets(triplets)

    # Create embeddings for search
    embeddings = core.create_embeddings([feed_item.content])
    vector_store.add_embeddings(embeddings)

    print(f"Processed: {feed_item.title} from {feed_item.source}")

from semantica.streaming import StreamProcessor

# Kafka integration
kafka_processor = StreamProcessor(
    platform="kafka",
    bootstrap_servers=["localhost:9092"],
    topics=["documents", "web_content", "feeds"]
)

# RabbitMQ integration
rabbitmq_processor = StreamProcessor(
    platform="rabbitmq",
    host="localhost",
    port=5672,
    queues=["semantic_processing"]
)

# Process streaming data
async for message in kafka_processor.consume():
    content = message.value

    # Determine content type and process accordingly
    if message.headers.get("content_type") == "application/pdf":
        processed = doc_processor.process_pdf_bytes(content)
    elif message.headers.get("content_type") == "text/html":
        processed = web_processor.process_html(content)
    else:
        processed = content

    # Extract semantics and build knowledge
    semantics = core.extract_semantics(processed)
    triplets = core.generate_triplets(semantics)
    knowledge_graph.add_triplets(triplets)

from semantica.domains.cyber import CyberIntelProcessor

# Initialize cybersecurity processor
cyber_processor = CyberIntelProcessor(
    threat_feeds=[
        "https://feeds.feedburner.com/CyberSecurityNewsDaily",
        "https://www.us-cert.gov/ncas/current-activity.xml"
    ],
    formats=["pdf", "html", "xml", "json"],
    extract_iocs=True,
    map_to_mitre=True
)

# Process various cybersecurity sources
sources = [
    "threat_report.pdf",
    "https://security-blog.com/rss",
    "vulnerability_data.json",
    "incident_reports/"
]

cyber_knowledge = cyber_processor.build_threat_intelligence(sources)

# Generate STIX bundles
stix_bundle = cyber_knowledge.to_stix()
print(f"Generated STIX bundle with {len(stix_bundle.objects)} objects")

# Export to threat intelligence platforms
cyber_knowledge.export_to_misp()
cyber_knowledge.export_to_opencti()

from semantica.domains.biomedical import BiomedicalProcessor

# Initialize biomedical processor
bio_processor = BiomedicalProcessor(
    pubmed_integration=True,
    extract_drug_interactions=True,
    map_to_mesh=True,
    clinical_trial_detection=True
)

# Process biomedical literature
sources = [
    "research_papers/",
    "https://pubmed.ncbi.nlm.nih.gov/rss/",
    "clinical_reports.pdf",
    "drug_databases.json"
]

biomedical_knowledge = bio_processor.build_medical_knowledge_base(sources)

# Generate medical ontology
medical_ontology = biomedical_knowledge.generate_ontology()

# Export to medical databases
biomedical_knowledge.export_to_umls()
biomedical_knowledge.export_to_bioportal()

from semantica.domains.finance import FinancialProcessor

# Initialize financial processor
finance_processor = FinancialProcessor(
    sec_filings=True,
    news_sentiment=True,
    market_data_integration=True,
    regulatory_compliance=True
)

# Process financial data sources
sources = [
    "earnings_reports/",
    "https://feeds.finance.yahoo.com/rss/",
    "sec_filings.xml",
    "market_data.csv",
    "financial_news/"
]

financial_knowledge = finance_processor.build_financial_knowledge_graph(sources)

# Generate financial semantic triplets
triplets = financial_knowledge.extract_financial_triplets()

# Export to financial analysis platforms
financial_knowledge.export_to_bloomberg_api()
financial_knowledge.export_to_refinitiv()

from semantica.deployment import ScaleManager

# Kubernetes deployment configuration
k8s_config = {
    "replicas": 5,
    "resources": {
        "cpu": "2000m",
        "memory": "8Gi",
        "gpu": "1"
    },
    "auto_scaling": {
        "min_replicas": 2,
        "max_replicas": 20,
        "cpu_threshold": 70
    }
}

# Deploy to Kubernetes
scale_manager = ScaleManager()
deployment = scale_manager.deploy_kubernetes(config=k8s_config)

# Monitor performance
metrics = deployment.get_metrics()
print(f"Processing rate: {metrics.documents_per_second} docs/sec")
print(f"Memory usage: {metrics.memory_usage_percent}%")

from semantica.pipeline import PipelineBuilder

# Build custom processing pipeline
pipeline = PipelineBuilder() \
    .add_input_sources(["pdf", "html", "rss", "json"]) \
    .add_preprocessing([
        "text_cleaning",
        "language_detection", 
        "content_extraction"
    ]) \
    .add_semantic_processing([
        "entity_extraction",
        "relation_extraction",
        "triplet_generation",
        "ontology_mapping"
    ]) \
    .add_enrichment([
        "context_expansion",
        "cross_reference_resolution",
        "metadata_enhancement"
    ]) \
    .add_output_formats([
        "knowledge_graph",
        "vector_embeddings",
        "rdf_triplets",
        "json_ld"
    ]) \
    .build()

# Process data through custom pipeline
results = pipeline.process(input_sources)

from semantica.monitoring import AnalyticsDashboard

# Initialize analytics dashboard
dashboard = AnalyticsDashboard(
    port=8080,
    enable_real_time=True,
    metrics=[
        "processing_rate",
        "extraction_accuracy",
        "memory_usage",
        "knowledge_graph_growth"
    ]
)

# Start monitoring
dashboard.start()

# Custom metrics
dashboard.add_custom_metric("semantic_quality_score", 
                          lambda: core.get_semantic_quality_score())

# Alert configuration
dashboard.add_alert(
    condition="processing_rate < 100",
    action="scale_up_workers",
    notification="slack://alerts-channel"
)

from semantica.quality import QualityAssurance

# Initialize quality assurance
qa = QualityAssurance(
    validation_rules=[
        "entity_consistency",
        "triplet_validity",
        "schema_compliance",
        "ontology_alignment"
    ],
    confidence_thresholds={
        "entity_extraction": 0.8,
        "relation_extraction": 0.7,
        "triplet_generation": 0.9
    }
)

# Validate processing results
validation_report = qa.validate(processing_results)
print(f"Overall quality score: {validation_report.quality_score:.2%}")
print(f"Issues found: {len(validation_report.issues)}")

# Continuous quality monitoring
qa.enable_continuous_monitoring()

from semantica.schema import SchemaManager, BusinessEntity
from pydantic import BaseModel
from typing import List, Optional

# Define your business schema upfront
class Employee(BusinessEntity):
    name: str
    employee_id: str
    department: str
    role: str
    manager: Optional[str] = None
    email: str
    hire_date: str

class Department(BusinessEntity):
    name: str
    budget: float
    head: str
    location: str

class Product(BusinessEntity):
    name: str
    sku: str
    department: str
    owner: str
    price: float
    launch_date: str

# Initialize schema manager with your business entities
schema_manager = SchemaManager()
schema_manager.register_entities([Employee, Department, Product])

# Process documents with schema enforcement
core = Semantica(schema_manager=schema_manager)
results = core.process_with_schema("hr_documents/", strict_mode=True)

# Only entities matching your schema are extracted and validated
print(f"Extracted {len(results.employees)} employees")
print(f"Extracted {len(results.departments)} departments") 
print(f"Schema violations: {len(results.violations)}")

from semantica.knowledge import SeedManager

# Initialize with known business entities
seed_manager = SeedManager()

# Load seed data from various sources
seed_manager.load_from_csv("employees.csv", entity_type="Employee")
seed_manager.load_from_json("departments.json", entity_type="Department")
seed_manager.load_from_database("products", connection_string="postgresql://...")

# Seed the knowledge graph
knowledge_graph = core.create_knowledge_graph(seed_data=seed_manager.get_seeds())

# Process new documents - will match against seeded entities
new_documents = ["meeting_notes.pdf", "project_reports/", "emails.mbox"]
results = core.process_documents(new_documents, 
                                knowledge_graph=knowledge_graph,
                                enable_entity_linking=True)

# Results show both seeded and newly discovered entities
print(f"Seeded entities: {len(knowledge_graph.seeded_entities)}")
print(f"Newly discovered: {len(results.new_entities)}")
print(f"Linked to existing: {len(results.linked_entities)}")

from semantica.deduplication import EntityDeduplicator

# Configure deduplication rules for each entity type
dedup_config = {
    "Employee": {
        "match_fields": ["email", "employee_id"],
        "fuzzy_fields": ["name"],
        "similarity_threshold": 0.85,
        "merge_strategy": "most_recent"
    },
    "Product": {
        "match_fields": ["sku"],
        "fuzzy_fields": ["name"],
        "similarity_threshold": 0.90,
        "merge_strategy": "highest_confidence"
    },
    "Department": {
        "match_fields": ["name"],
        "similarity_threshold": 0.95,
        "merge_strategy": "manual_review"
    }
}

# Initialize deduplicator
deduplicator = EntityDeduplicator(config=dedup_config)

# Process documents with automatic deduplication
results = core.process_documents(
    sources=["hr_data/", "finance_reports/", "project_docs/"],
    deduplicator=deduplicator,
    enable_auto_merge=True
)

# Review deduplication results
print(f"Duplicates found: {len(results.duplicates)}")
print(f"Auto-merged: {len(results.auto_merged)}")
print(f"Requires manual review: {len(results.manual_review_needed)}")

# Access detailed merge information
for merge in results.auto_merged:
    print(f"Merged {merge.entity_type}: {merge.canonical_name}")
    print(f"  Sources: {', '.join(merge.source_documents)}")
    print(f"  Confidence: {merge.confidence:.2%}")

from semantica.conflicts import ConflictDetector

# Configure conflict detection rules
conflict_detector = ConflictDetector(
    track_provenance=True,
    conflict_fields={
        "Employee": ["salary", "department", "role", "manager"],
        "Product": ["price", "owner", "department"],
        "Department": ["budget", "head", "location"]
    },
    confidence_threshold=0.7
)

# Process with conflict detection enabled
results = core.process_documents(
    sources=["q1_report.pdf", "hr_database.csv", "manager_updates.docx"],
    conflict_detector=conflict_detector
)

# Review detected conflicts
for conflict in results.conflicts:
    print(f"\n🚨 CONFLICT DETECTED: {conflict.entity_name}")
    print(f"Field: {conflict.field}")
    print(f"Conflicting values:")

    for claim in conflict.claims:
        print(f"  • '{claim.value}' from {claim.source_document}")
        print(f"    Page: {claim.page_number}, Confidence: {claim.confidence:.2%}")
        print(f"    Context: {claim.context}")

    print(f"Recommended action: {conflict.recommended_action}")

# Export conflicts for manual resolution
conflict_report = results.export_conflicts_report()
conflict_report.save_to_excel("conflicts_review.xlsx")

# Resolve conflicts programmatically or through UI
resolution_rules = {
    "Employee.salary": "use_most_recent",
    "Product.price": "use_highest_confidence", 
    "Department.budget": "require_manual_review"
}

resolved_conflicts = conflict_detector.resolve_conflicts(
    results.conflicts, 
    rules=resolution_rules
)

from semantica.validation import BusinessRuleEngine

# Define business rules
rules = BusinessRuleEngine()

# Add validation rules
rules.add_rule(
    name="employee_department_exists",
    condition="Employee.department must exist in Department entities",
    severity="error"
)

rules.add_rule(
    name="salary_range_check", 
    condition="Employee.salary must be between $30,000 and $500,000",
    severity="warning"
)

rules.add_rule(
    name="product_owner_validation",
    condition="Product.owner must be an existing Employee",
    severity="error"
)

rules.add_rule(
    name="department_budget_consistency",
    condition="Department.budget should align with sum of employee salaries",
    severity="info"
)

# Process with business rule validation
results = core.process_documents(
    sources=["company_data/"],
    validation_engine=rules,
    fail_on_errors=False
)

# Review validation results
validation_report = results.validation_report

print(f"Total violations: {len(validation_report.violations)}")
print(f"Errors: {validation_report.errors}")
print(f"Warnings: {validation_report.warnings}")
print(f"Info: {validation_report.info}")

# Get detailed violation information
for violation in validation_report.violations:
    print(f"\n❌ {violation.rule_name}")
    print(f"Entity: {violation.entity_name} ({violation.entity_type})")
    print(f"Issue: {violation.description}")
    print(f"Source: {violation.source_document}")
    print(f"Suggested fix: {violation.suggested_resolution}")

from semantica.ui import ConflictResolutionDashboard

# Start interactive dashboard
dashboard = ConflictResolutionDashboard(
    knowledge_graph=knowledge_graph,
    conflicts=results.conflicts,
    port=8080
)

# Dashboard features:
# - Side-by-side source comparison
# - Confidence score visualization
# - One-click conflict resolution
# - Bulk resolution with rules
# - Export resolved data

dashboard.start()
print("Dashboard available at http://localhost:8080")

# Programmatic resolution after dashboard review
resolved_data = dashboard.get_resolved_conflicts()
knowledge_graph.apply_resolutions(resolved_data)