diff --git a/cookbook/advanced/02_Advanced_Graph_Analytics.ipynb b/cookbook/advanced/02_Advanced_Graph_Analytics.ipynb index 2ee1d0de..b28d4696 100644 --- a/cookbook/advanced/02_Advanced_Graph_Analytics.ipynb +++ b/cookbook/advanced/02_Advanced_Graph_Analytics.ipynb @@ -4,39 +4,20 @@ "cell_type": "markdown", "metadata": {}, "source": [ - "[![Open In Colab](https://colab.research.google.com/assets/colab-badge.svg)](https://colab.research.google.com/github/Hawksight-AI/semantica/blob/main/cookbook/advanced/02_Advanced_Graph_Analytics.ipynb)\n", + "# Ultimate Graph Analytics Masterclass\n", "\n", - "# Advanced Graph Analytics\n", + "Welcome to the **comprehensive walkthrough** of Semantica's Graph Analytics capabilities. This notebook goes beyond simple graph construction to demonstrate a full-lifecycle production pipeline.\n", "\n", - "## Overview\n", + "We will simulate a messy, real-world scenario involving a **Startup Ecosystem** (Investors, Startups, Founders) and guide you through every step of the process:\n", "\n", - "This notebook demonstrates advanced graph analytics using GraphAnalyzer, CentralityCalculator, CommunityDetector, ConnectivityAnalyzer, GraphValidator, Deduplicator, and **GraphStore** for persistent storage.\n", + "1. **Validation**: Catching bad data before it enters the graph.\n", + "2. **Cleaning**: Deduplicating entities and resolving conflicts.\n", + "3. **Structural Analysis**: Understanding the shape and health of your network.\n", + "4. **Deep Analytics**: Centrality, Communities, and Path Finding.\n", + "5. **Temporal Analytics**: Time-traveling through your graph data.\n", + "6. **Provenance**: Tracking where your data came from.\n", "\n", - "\n", - "**Documentation**: [API Reference](https://semantica.readthedocs.io/reference/kg/)\n", - "\n", - "### Learning Objectives\n", - "\n", - "- Use GraphAnalyzer for comprehensive graph analysis\n", - "- Use CentralityCalculator for advanced centrality measures\n", - "- Use CommunityDetector for community detection\n", - "- Use ConnectivityAnalyzer for connectivity analysis\n", - "- Use GraphValidator and Deduplicator for graph quality\n", - "- **Use GraphStore to persist graphs to Neo4j or FalkorDB**\n", - "\n", - "## Installation\n", - "\n", - "Install Semantica from PyPI:\n", - "\n", - "```bash\n", - "pip install semantica\n", - "# Or with all optional dependencies:\n", - "pip install semantica[all]\n", - "```\n", - "\n", - "---\n", - "\n", - "## Workflow: Graph Analysis → Centrality → Communities → Connectivity → Validation → Deduplication → **Persist to Graph Store**\n" + "Let's dive in!" ] }, { @@ -45,9 +26,39 @@ "metadata": {}, "outputs": [], "source": [ - "%pip install -U \"semantica[all]\"\n", - "import semantica\n", - "print(semantica.__version__)\n" + "import logging\n", + "import json\n", + "from datetime import datetime\n", + "\n", + "# Set up logging to see what's happening under the hood\n", + "logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')\n", + "\n", + "# Import all the powerful tools from Semantica\n", + "from semantica.kg import (\n", + " GraphBuilder,\n", + " GraphAnalyzer,\n", + " GraphValidator,\n", + " ConnectivityAnalyzer,\n", + " CentralityCalculator,\n", + " CommunityDetector,\n", + " TemporalGraphQuery,\n", + " ProvenanceTracker\n", + ")\n", + "from semantica.deduplication import DuplicateDetector\n", + "from semantica.conflicts import ConflictDetector, ConflictResolver" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## 1. The Scenario: A Messy Startup Ecosystem\n", + "\n", + "We have data from multiple sources (scrapers, news, user submissions). It's messy:\n", + "- **Duplicates**: \"TechFlow AI\" and \"TechFlow Inc.\"\n", + "- **Conflicts**: Different revenue numbers for the same company.\n", + "- **Errors**: Relationships pointing to non-existent nodes (dangling edges).\n", + "- **History**: Investment rounds happening at different times." ] }, { @@ -56,221 +67,320 @@ "metadata": {}, "outputs": [], "source": [ - "from semantica.kg import GraphBuilder, GraphAnalyzer, CentralityCalculator, CommunityDetector, ConnectivityAnalyzer\n", - "from semantica.deduplication import DuplicateDetector, EntityMerger, MergeStrategy\n", - "\n", - "builder = GraphBuilder()\n", - "analyzer = GraphAnalyzer()\n", - "\n", - "entities = [\n", - " {\"id\": \"e1\", \"type\": \"Organization\", \"name\": \"Apple Inc.\", \"properties\": {}},\n", - " {\"id\": \"e2\", \"type\": \"Person\", \"name\": \"Tim Cook\", \"properties\": {}},\n", - " {\"id\": \"e3\", \"type\": \"Location\", \"name\": \"Cupertino\", \"properties\": {}}\n", + "# Our \"Raw\" Messy Data\n", + "raw_entities = [\n", + " {\"id\": \"startup_1\", \"type\": \"Startup\", \"name\": \"TechFlow AI\", \"revenue\": 1000000, \"founded\": \"2021-01-01\"},\n", + " {\"id\": \"startup_2\", \"type\": \"Startup\", \"name\": \"GreenEnergy Co\", \"revenue\": 500000, \"founded\": \"2020-05-15\"},\n", + " {\"id\": \"startup_1_dup\", \"type\": \"Startup\", \"name\": \"TechFlow Inc.\", \"revenue\": 1200000, \"founded\": \"2021-01-01\"}, # Duplicate!\n", + " {\"id\": \"investor_1\", \"type\": \"Investor\", \"name\": \"Venture Capital X\"},\n", + " {\"id\": \"founder_1\", \"type\": \"Person\", \"name\": \"Alice Chen\"},\n", + " {\"id\": \"founder_2\", \"type\": \"Person\", \"name\": \"Bob Smith\"}\n", "]\n", "\n", - "relationships = [\n", - " {\"source\": \"e2\", \"target\": \"e1\", \"type\": \"CEO_of\", \"properties\": {}},\n", - " {\"source\": \"e1\", \"target\": \"e3\", \"type\": \"located_in\", \"properties\": {}}\n", - "]\n", - "\n", - "kg = builder.build(entities, relationships)\n", - "\n", - "metrics = analyzer.compute_metrics(kg)\n", - "\n", - "print(f\"Graph metrics:\")\n", - "print(f\" Entities: {metrics.get('entity_count', 0)}\")\n", - "print(f\" Relationships: {metrics.get('relationship_count', 0)}\")\n", - "print(f\" Density: {metrics.get('density', 0):.3f}\")\n" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "## Step 2: Advanced Centrality Measures\n", - "\n", - "Calculate multiple centrality measures.\n" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "centrality_calculator = CentralityCalculator()\n", - "\n", - "degree_centrality_result = centrality_calculator.calculate_degree_centrality(kg)\n", - "degree_centrality = degree_centrality_result.get('centrality', {})\n", - "betweenness_centrality_result = centrality_calculator.calculate_betweenness_centrality(kg)\n", - "betweenness_centrality = betweenness_centrality_result.get('centrality', {})\n", - "\n", - "print(f\"Degree centrality: {len(degree_centrality)} entities\")\n", - "print(f\"Betweenness centrality: {len(betweenness_centrality)} entities\")\n" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "## Step 3: Community Detection\n", - "\n", - "Detect communities in the graph.\n" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "community_detector = CommunityDetector()\n", - "\n", - "communities = community_detector.detect_communities(kg)\n", - "\n", - "print(f\"Detected {len(communities)} communities\")\n", - "for i, community in enumerate(communities[:3], 1):\n", - " print(f\" Community {i}: {len(community)} entities\")\n" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "## Step 4: Connectivity Analysis\n", - "\n", - "Analyze graph connectivity.\n" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "connectivity_analyzer = ConnectivityAnalyzer()\n", - "\n", - "connectivity = connectivity_analyzer.analyze_connectivity(kg)\n", - "\n", - "print(f\"Connectivity analysis:\")\n", - "print(f\" Is connected: {connectivity.get('is_connected', False)}\")\n", - "print(f\" Components: {len(connectivity.get('components', []))}\")\n" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "## Step 5: Graph Validation and Deduplication\n", - "\n", - "Validate and deduplicate the graph.\n" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "graph_validator = GraphValidator()\n", - "\n", - "validation_result = graph_validator.validate(kg)\n", - "\n", - "print(f\"Graph validation: {validation_result.get('valid', False)}\")\n", - "print(f\"Issues found: {len(validation_result.get('issues', []))}\")\n", - "\n", - "# For deduplication, use semantica.deduplication module:\n", - "# from semantica.deduplication import DuplicateDetector, EntityMerger, MergeStrategy\n", - "# detector = DuplicateDetector(similarity_threshold=0.8)\n", - "# duplicate_groups = detector.detect_duplicate_groups(kg.get('entities', []))\n", - "# merger = EntityMerger()\n", - "# merge_operations = merger.merge_duplicates(kg.get('entities', []), strategy=MergeStrategy.KEEP_MOST_COMPLETE)\n" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "## Step 6: Persist to Graph Store\n", - "\n", - "Store the analyzed graph in a persistent graph database using GraphStore.\n" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "from semantica.graph_store import GraphStore\n", - "\n", - "# Option 1: Neo4j (requires Neo4j server running)\n", - "graph_store = GraphStore(backend=\"neo4j\", uri=\"bolt://localhost:7687\", user=\"neo4j\", password=\"password\")\n", - "graph_store.connect()\n", - "\n", - "# Store entities as nodes and track node ID mapping\n", - "node_id_map = {}\n", - "for entity in entities:\n", - " node = graph_store.create_node(\n", - " labels=[entity[\"type\"]],\n", - " properties={\"name\": entity[\"name\"], \"original_id\": entity[\"id\"]}\n", - " )\n", - " node_id_map[entity[\"id\"]] = node.get(\"id\")\n", - " print(f\"Stored node: {entity['name']} (ID: {node.get('id')})\")\n", - "\n", - "# Store relationships using mapped node IDs\n", - "for rel in relationships:\n", - " source_id = node_id_map.get(rel[\"source\"])\n", - " target_id = node_id_map.get(rel[\"target\"])\n", + "raw_relationships = [\n", + " # Valid Relationships\n", + " {\"source\": \"founder_1\", \"target\": \"startup_1\", \"type\": \"FOUNDED\", \"valid_from\": \"2021-01-01\"},\n", + " {\"source\": \"investor_1\", \"target\": \"startup_1\", \"type\": \"INVESTED_IN\", \"amount\": 5000000, \"valid_from\": \"2023-06-01\"},\n", " \n", - " if source_id is not None and target_id is not None:\n", - " relationship = graph_store.create_relationship(\n", - " start_node_id=source_id,\n", - " end_node_id=target_id,\n", - " rel_type=rel[\"type\"],\n", - " properties=rel.get(\"properties\", {})\n", - " )\n", - " print(f\"Stored relationship: {rel['source']} -{rel['type']}-> {rel['target']}\")\n", - " else:\n", - " print(f\"Warning: Could not find node IDs for relationship {rel['source']} -> {rel['target']}\")\n", + " # Dangling Edge (Error!)\n", + " {\"source\": \"founder_2\", \"target\": \"startup_999\", \"type\": \"FOUNDED\", \"valid_from\": \"2020-05-15\"}, \n", + " \n", + " # Temporal Data (History)\n", + " {\"source\": \"founder_1\", \"target\": \"startup_2\", \"type\": \"ADVISED\", \"valid_from\": \"2020-01-01\", \"valid_until\": \"2021-01-01\"}\n", + "]\n", "\n", - "# Query using Cypher\n", - "results = graph_store.execute_query(\"MATCH (n) RETURN n.name, labels(n) LIMIT 10\")\n", - "print(f\"\\nQuery results: {len(results.get('records', []))} nodes\")\n", - "\n", - "# Get statistics\n", - "stats = graph_store.get_stats()\n", - "print(f\"\\nGraph store statistics:\")\n", - "print(f\" Node count: {stats.get('node_count', 'N/A')}\")\n", - "print(f\" Relationship count: {stats.get('relationship_count', 'N/A')}\")\n", - "print(f\" Label counts: {stats.get('label_counts', {})}\")\n", - "\n", - "graph_store.close()\n" + "print(f\"Loaded {len(raw_entities)} raw entities and {len(raw_relationships)} raw relationships.\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ - "## Summary\n", + "## 2. Phase 1: Validation (The Gatekeeper)\n", "\n", - "You've learned advanced graph analytics:\n", + "Before we do anything, we must validate the graph. Bad data in = Bad insights out.\n", + "We use `GraphValidator` to check for:\n", + "- **Structural Integrity**: Are all relationship targets present?\n", + "- **Schema Compliance**: Do entities have required fields?\n", + "- **Consistency**: Are IDs unique?" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# Initialize Validator\n", + "validator = GraphValidator()\n", "\n", - "- **GraphAnalyzer**: Comprehensive graph analysis and metrics\n", - "- **CentralityCalculator**: Multiple centrality measures\n", - "- **CommunityDetector**: Community detection\n", - "- **ConnectivityAnalyzer**: Connectivity analysis\n", - "- **GraphValidator**: Graph validation\n", - "- **Deduplicator**: Graph deduplication\n", - "- **GraphStore**: Persist graphs to Neo4j or FalkorDB\n" + "# Create a temporary graph object for validation\n", + "temp_graph = {\"entities\": raw_entities, \"relationships\": raw_relationships}\n", + "\n", + "# Run Validation\n", + "print(\"Running Validation Check...\")\n", + "validation_result = validator.validate(temp_graph)\n", + "\n", + "if not validation_result.is_valid:\n", + " print(\"Validation Failed! Issues found:\")\n", + " for issue in validation_result.issues:\n", + " print(f\" - [{issue.severity.name}] {issue.message} (Code: {issue.code})\")\n", + " \n", + " # AUTOMATIC FIX: If it's a dangling edge, remove it\n", + " if issue.code == \"DANGLING_EDGE\":\n", + " print(\" Auto-Fixing: Removing invalid relationship...\")\n", + " raw_relationships = [r for r in raw_relationships \n", + " if r['target'] != issue.details.get('target_id')]\n", + "else:\n", + " print(\"Graph is valid!\")\n", + "\n", + "# Re-validate to confirm fix\n", + "print(\"\\nRe-validating after fixes...\")\n", + "temp_graph = {\"entities\": raw_entities, \"relationships\": raw_relationships}\n", + "if validator.validate(temp_graph).is_valid:\n", + " print(\"Graph is now clean and valid!\")" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## 3. Phase 2: Deduplication & Conflict Resolution\n", + "\n", + "We have \"TechFlow AI\" and \"TechFlow Inc.\". These are likely the same company.\n", + "We also have conflicting revenue data." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# 1. Detect Duplicates\n", + "print(\"Scanning for duplicates...\")\n", + "deduper = DuplicateDetector(similarity_threshold=0.7) # 70% similarity threshold\n", + "duplicates = deduper.detect_duplicates(raw_entities)\n", + "\n", + "for candidate in duplicates:\n", + " print(f\"Found potential duplicate pair (Score: {candidate.similarity_score:.2f}):\")\n", + " print(f\" - {candidate.entity1['name']} (ID: {candidate.entity1['id']})\")\n", + " print(f\" - {candidate.entity2['name']} (ID: {candidate.entity2['id']})\")\n", + " \n", + " # MERGE STRATEGY: Keep entity1, merge data from entity2\n", + " print(\" Merging entities...\")\n", + " # (In a real app, you'd use EntityMerger, but here's the logic:)\n", + " # We keep startup_1 and discard startup_1_dup, but we note the conflict\n", + " \n", + "# 2. Detect Conflicts\n", + "print(\"\\nChecking for data conflicts...\")\n", + "conflict_detector = ConflictDetector()\n", + "\n", + "# Simulating a conflict check between the two versions of TechFlow\n", + "# To check conflicts, we treat them as the same entity (same ID)\n", + "entity_a = raw_entities[0].copy()\n", + "entity_b = raw_entities[2].copy()\n", + "entity_b['id'] = entity_a['id'] # Force same ID for conflict detection\n", + "\n", + "conflicts = conflict_detector.detect_conflicts([entity_a, entity_b])\n", + "\n", + "for conflict in conflicts:\n", + " print(f\" Conflict detected in field '{conflict.property_name}':\")\n", + " print(f\" Values: {conflict.conflicting_values}\")\n", + " \n", + " # RESOLUTION: Trust the higher number (optimistic!)\n", + " if conflict.property_name == \"revenue\":\n", + " # values are strings or ints, need to handle types\n", + " vals = [float(v) for v in conflict.conflicting_values if v is not None]\n", + " resolved_val = max(vals)\n", + " print(f\" Resolved to: {resolved_val}\")\n", + " raw_entities[0]['revenue'] = resolved_val\n", + "\n", + "# Final Cleanup: Remove the duplicate entity from our list\n", + "clean_entities = [e for e in raw_entities if e['id'] != 'startup_1_dup']\n", + "clean_relationships = raw_relationships # (We'd normally re-link relationships too)\n", + "\n", + "print(f\"\\nCleaned Data: {len(clean_entities)} entities remaining.\")" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## 4. Phase 3: Building the Knowledge Graph\n", + "\n", + "Now that our data is clean, we build the official graph object." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# Manual Graph Construction (since we already cleaned it)\n", + "kg = {\n", + " \"entities\": clean_entities,\n", + " \"relationships\": clean_relationships,\n", + " \"metadata\": {\n", + " \"created_at\": datetime.now().isoformat(),\n", + " \"source\": \"Manual Advanced Pipeline\"\n", + " }\n", + "}\n", + "print(\"Knowledge Graph Assembled Successfully!\")" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## 5. Phase 4: Advanced Analytics\n", + "\n", + "This is where the magic happens. We'll use multiple analyzers to extract insights." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# Initialize the Master Analyzer\n", + "analyzer = GraphAnalyzer(enable_temporal=True)\n", + "\n", + "# 1. Structural Analysis (Connectivity)\n", + "print(\"\\n--- Connectivity Analysis ---\")\n", + "connectivity = analyzer.analyze_connectivity(kg)\n", + "print(f\" • Graph Connected? {'Yes' if connectivity['is_connected'] else 'No'}\")\n", + "print(f\" • Connected Components: {connectivity['num_components']}\")\n", + "\n", + "# 2. Centrality (Who is important?)\n", + "print(\"\\n--- Centrality Analysis ---\")\n", + "centrality_result = analyzer.calculate_centrality(kg, centrality_type=\"degree\")\n", + "degree_data = centrality_result[\"centrality_measures\"][\"degree\"]\n", + "\n", + "# Get pre-calculated rankings\n", + "top_nodes = degree_data[\"rankings\"][:3]\n", + "\n", + "print(\" • Top Influencers (Degree Centrality):\")\n", + "for item in top_nodes:\n", + " print(f\" - {item['node']}: {item['score']:.2f}\")\n", + "\n", + "# 3. Community Detection (Clustering)\n", + "print(\"\\n--- Community Detection ---\")\n", + "community_result = analyzer.detect_communities(kg, algorithm=\"louvain\")\n", + "communities = community_result[\"communities\"]\n", + "\n", + "print(f\" • Detected {len(communities)} communities.\")\n", + "for i, comm in enumerate(communities):\n", + " # comm is a set of node IDs\n", + " members = list(comm)\n", + " print(f\" Community {i+1}: {', '.join(members)}\")" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## 6. Phase 5: Temporal Analytics (Time Travel)\n", + "\n", + "Static graphs are boring. Real worlds change. Let's analyze the **evolution** of our ecosystem." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "temporal_engine = TemporalGraphQuery(temporal_granularity=\"year\")\n", + "\n", + "# 1. Time Travel Query: What did the world look like in 2020?\n", + "print(\"\\n--- Time Travel: 2020 ---\")\n", + "snapshot_2020 = temporal_engine.query_at_time(kg, query=\"*\", at_time=\"2020-06-01\")\n", + "print(f\" Active Relationships in 2020: {len(snapshot_2020['relationships'])}\")\n", + "for rel in snapshot_2020['relationships']:\n", + " print(f\" - {rel['source']} --[{rel['type']}]--> {rel['target']}\")\n", + "\n", + "# 2. Time Travel Query: What about 2023?\n", + "print(\"\\n--- Time Travel: 2023 ---\")\n", + "snapshot_2023 = temporal_engine.query_at_time(kg, query=\"*\", at_time=\"2023-07-01\")\n", + "print(f\" Active Relationships in 2023: {len(snapshot_2023['relationships'])}\")\n", + "for rel in snapshot_2023['relationships']:\n", + " print(f\" - {rel['source']} --[{rel['type']}]--> {rel['target']}\")\n", + " \n", + "# Notice how 'ADVISED' might disappear if it ended, and 'INVESTED_IN' appears!" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## 7. Phase 6: Provenance (Data Lineage)\n", + "\n", + "Finally, in a production system, you need to know **where** a fact came from. This is crucial for trust." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "tracker = ProvenanceTracker()\n", + "\n", + "# Let's pretend we're tracking the source of our data\n", + "tracker.track_entity(\"startup_1\", source=\"Crunchbase_API_v2\", metadata={\"confidence\": 0.95})\n", + "tracker.track_entity(\"startup_1\", source=\"Manual_Entry_User_Bob\", metadata={\"confidence\": 1.0})\n", + "\n", + "print(\"\\n--- Provenance Report: TechFlow AI ---\")\n", + "lineage = tracker.get_lineage(\"startup_1\")\n", + "print(f\" Entity: startup_1\")\n", + "print(f\" First Seen: {lineage['first_seen']}\")\n", + "print(f\" Sources:\")\n", + "for src in lineage['sources']:\n", + " print(f\" - {src['source']} (at {src['timestamp']})\")" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Conclusion\n", + "\n", + "You have just walked through a complete, advanced Knowledge Graph pipeline:\n", + "\n", + "1. **Validated** messy input data.\n", + "2. **Cleaned** duplicates and conflicts.\n", + "3. **Analyzed** structure and community dynamics.\n", + "4. **Queried** across time dimensions.\n", + "5. **Tracked** data lineage.\n", + "\n", + "This represents the state-of-the-art in modern KG Engineering using Semantica." ] } ], "metadata": { + "kernelspec": { + "display_name": "Python 3", + "language": "python", + "name": "python3" + }, "language_info": { - "name": "python" - } + "codemirror_mode": { + "name": "ipython", + "version": 3 + }, + "file_extension": ".py", + "mimetype": "text/x-python", + "name": "python", + "nbconvert_exporter": "python", + "pygments_lexer": "ipython3", + "version": "3.8.5" + }, + "nbformat": 4, + "nbformat_minor": 4 }, "nbformat": 4, - "nbformat_minor": 2 + "nbformat_minor": 5 } diff --git a/semantica/kg/__init__.py b/semantica/kg/__init__.py index b6e494ef..e398ee00 100644 --- a/semantica/kg/__init__.py +++ b/semantica/kg/__init__.py @@ -112,6 +112,7 @@ from .connectivity_analyzer import ConnectivityAnalyzer from .entity_resolver import EntityResolver from .graph_analyzer import GraphAnalyzer from .graph_builder import GraphBuilder +from .graph_validator import GraphValidator from .provenance_tracker import ProvenanceTracker from .registry import MethodRegistry, method_registry from .seed_manager import SeedManager @@ -126,6 +127,7 @@ __all__ = [ "GraphBuilder", "EntityResolver", "GraphAnalyzer", + "GraphValidator", "TemporalGraphQuery", "TemporalPatternDetector", "TemporalVersionManager", diff --git a/semantica/kg/graph_validator.py b/semantica/kg/graph_validator.py new file mode 100644 index 00000000..58e632b6 --- /dev/null +++ b/semantica/kg/graph_validator.py @@ -0,0 +1,280 @@ +""" +Graph Validator Module + +This module provides comprehensive validation capabilities for knowledge graphs in the +Semantica framework. It ensures graph integrity, schema compliance, and structural consistency. + +Key Features: + - Schema validation (required fields, data types) + - Structural integrity (dangling edges, self-loops) + - Type checking (entity and relationship types) + - Cycle detection + - Orphan node detection + - detailed validation reporting + +Main Classes: + - GraphValidator: Main validation engine + +Example Usage: + >>> from semantica.kg import GraphValidator + >>> validator = GraphValidator() + >>> result = validator.validate(kg) + >>> if not result.is_valid: + ... print(result.issues) + +Author: Semantica Contributors +License: MIT +""" + +from typing import Any, Dict, List, Optional, Set, Union +from dataclasses import dataclass, field +from enum import Enum +import networkx as nx + +from ..utils.logging import get_logger + +class ValidationSeverity(Enum): + """Severity levels for validation issues.""" + INFO = "info" + WARNING = "warning" + ERROR = "error" + CRITICAL = "critical" + +@dataclass +class ValidationIssue: + """Represents a single validation issue.""" + code: str + message: str + severity: ValidationSeverity + element_id: Optional[str] = None + element_type: Optional[str] = None + details: Dict[str, Any] = field(default_factory=dict) + + def to_dict(self) -> Dict[str, Any]: + """Convert issue to dictionary.""" + return { + "code": self.code, + "message": self.message, + "severity": self.severity.value, + "element_id": self.element_id, + "element_type": self.element_type, + "details": self.details + } + +@dataclass +class ValidationResult: + """Container for validation results.""" + is_valid: bool + issues: List[ValidationIssue] + stats: Dict[str, int] = field(default_factory=dict) + + def get_issues_by_severity(self, severity: ValidationSeverity) -> List[ValidationIssue]: + """Filter issues by severity.""" + return [i for i in self.issues if i.severity == severity] + + def to_dict(self) -> Dict[str, Any]: + """Convert result to dictionary.""" + return { + "is_valid": self.is_valid, + "issues": [i.to_dict() for i in self.issues], + "stats": self.stats + } + +class GraphValidator: + """ + Comprehensive Knowledge Graph Validator. + + Validates graph structure, schema compliance, and data integrity. + """ + + def __init__(self, schema: Optional[Dict[str, Any]] = None, strict: bool = False): + """ + Initialize the validator. + + Args: + schema: Optional schema definition to validate against. + Should contain 'entity_types' and 'relationship_types'. + strict: If True, treats warnings as errors. + """ + self.logger = get_logger("graph_validator") + self.schema = schema or {} + self.strict = strict + + # Default required fields + self.required_entity_fields = {"id", "type", "name"} + self.required_rel_fields = {"source", "target", "type"} + + def validate(self, graph: Dict[str, Any]) -> ValidationResult: + """ + Run all validation checks on the graph. + + Args: + graph: The knowledge graph dictionary (must contain 'entities' and 'relationships'). + + Returns: + ValidationResult object containing success status and list of issues. + """ + issues: List[ValidationIssue] = [] + + # 1. Basic Structure Check + if not isinstance(graph, dict): + issues.append(ValidationIssue( + code="INVALID_FORMAT", + message="Graph must be a dictionary.", + severity=ValidationSeverity.CRITICAL + )) + return ValidationResult(False, issues) + + entities = graph.get("entities", []) + relationships = graph.get("relationships", []) + + if not isinstance(entities, list): + issues.append(ValidationIssue( + code="INVALID_ENTITIES", + message="'entities' must be a list.", + severity=ValidationSeverity.CRITICAL + )) + return ValidationResult(False, issues) + + if not isinstance(relationships, list): + issues.append(ValidationIssue( + code="INVALID_RELATIONSHIPS", + message="'relationships' must be a list.", + severity=ValidationSeverity.CRITICAL + )) + return ValidationResult(False, issues) + + # 2. Entity Validation + entity_ids = set() + for entity in entities: + # Check required fields + missing = self.required_entity_fields - set(entity.keys()) + if missing: + issues.append(ValidationIssue( + code="MISSING_FIELD", + message=f"Entity missing required fields: {missing}", + severity=ValidationSeverity.ERROR, + element_id=entity.get("id", "unknown"), + element_type="entity" + )) + + # Check ID uniqueness + eid = entity.get("id") + if eid: + if eid in entity_ids: + issues.append(ValidationIssue( + code="DUPLICATE_ID", + message=f"Duplicate entity ID found: {eid}", + severity=ValidationSeverity.CRITICAL, + element_id=eid, + element_type="entity" + )) + entity_ids.add(eid) + + # Schema Check (if schema provided) + if self.schema and "entity_types" in self.schema: + etype = entity.get("type") + if etype and etype not in self.schema["entity_types"]: + issues.append(ValidationIssue( + code="INVALID_TYPE", + message=f"Unknown entity type: {etype}", + severity=ValidationSeverity.WARNING, + element_id=eid, + element_type="entity" + )) + + # 3. Relationship Validation + for i, rel in enumerate(relationships): + # Check required fields + missing = self.required_rel_fields - set(rel.keys()) + if missing: + issues.append(ValidationIssue( + code="MISSING_FIELD", + message=f"Relationship missing required fields: {missing}", + severity=ValidationSeverity.ERROR, + element_type="relationship", + details={"index": i} + )) + continue + + src = rel.get("source") + tgt = rel.get("target") + + # Check Dangling Edges + if src not in entity_ids: + issues.append(ValidationIssue( + code="DANGLING_EDGE", + message=f"Source entity ID not found: {src}", + severity=ValidationSeverity.ERROR, + element_id=f"{src}->{tgt}", + element_type="relationship" + )) + if tgt not in entity_ids: + issues.append(ValidationIssue( + code="DANGLING_EDGE", + message=f"Target entity ID not found: {tgt}", + severity=ValidationSeverity.ERROR, + element_id=f"{src}->{tgt}", + element_type="relationship" + )) + + # Check Self-Loops (Warning) + if src == tgt: + issues.append(ValidationIssue( + code="SELF_LOOP", + message=f"Self-loop detected on entity: {src}", + severity=ValidationSeverity.INFO, + element_id=src, + element_type="relationship" + )) + + # 4. Structural Analysis (Cycles, connectivity) + # Only run if graph is small enough or requested? + # For now, let's do a quick cycle check using NetworkX + try: + nx_graph = nx.DiGraph() + nx_graph.add_nodes_from(entity_ids) + nx_graph.add_edges_from([(r["source"], r["target"]) for r in relationships if r.get("source") in entity_ids and r.get("target") in entity_ids]) + + # Check for cycles + try: + cycles = list(nx.simple_cycles(nx_graph)) + if cycles: + issues.append(ValidationIssue( + code="CYCLE_DETECTED", + message=f"Graph contains {len(cycles)} cycles.", + severity=ValidationSeverity.INFO, # Cycles aren't always bad + details={"count": len(cycles)} + )) + except Exception: + pass # Skip if too complex + + # Check for Orphans (Isolated nodes) + isolates = list(nx.isolates(nx_graph)) + if isolates: + issues.append(ValidationIssue( + code="ORPHAN_NODES", + message=f"Found {len(isolates)} orphan nodes (no relationships).", + severity=ValidationSeverity.WARNING, + details={"count": len(isolates), "ids": isolates[:10]} # Limit output + )) + + except Exception as e: + self.logger.warning(f"Structural validation failed: {e}") + + # Determine Validity + error_count = len([i for i in issues if i.severity in [ValidationSeverity.ERROR, ValidationSeverity.CRITICAL]]) + if self.strict: + warning_count = len([i for i in issues if i.severity == ValidationSeverity.WARNING]) + is_valid = (error_count + warning_count) == 0 + else: + is_valid = error_count == 0 + + stats = { + "total_entities": len(entities), + "total_relationships": len(relationships), + "issues_found": len(issues), + "errors": error_count + } + + return ValidationResult(is_valid, issues, stats) diff --git a/test_notebook_logic.py b/test_notebook_logic.py new file mode 100644 index 00000000..08845035 --- /dev/null +++ b/test_notebook_logic.py @@ -0,0 +1,197 @@ + +import logging +import json +from datetime import datetime + +# Set up logging to see what's happening under the hood +# Using WARNING to keep the output clean for the notebook demonstration +logging.basicConfig(level=logging.WARNING, format='%(asctime)s - %(levelname)s - %(message)s') + +# Import all the powerful tools from Semantica +from semantica.kg import ( + GraphBuilder, + GraphAnalyzer, + GraphValidator, + ConnectivityAnalyzer, + CentralityCalculator, + CommunityDetector, + TemporalGraphQuery, + ProvenanceTracker +) +from semantica.deduplication import DuplicateDetector +from semantica.conflicts import ConflictDetector, ConflictResolver + +# Our "Raw" Messy Data +raw_entities = [ + {"id": "startup_1", "type": "Startup", "name": "TechFlow AI", "revenue": 1000000, "founded": "2021-01-01"}, + {"id": "startup_2", "type": "Startup", "name": "GreenEnergy Co", "revenue": 500000, "founded": "2020-05-15"}, + {"id": "startup_1_dup", "type": "Startup", "name": "TechFlow Inc.", "revenue": 1200000, "founded": "2021-01-01"}, # Duplicate! + {"id": "investor_1", "type": "Investor", "name": "Venture Capital X"}, + {"id": "founder_1", "type": "Person", "name": "Alice Chen"}, + {"id": "founder_2", "type": "Person", "name": "Bob Smith"} +] + +raw_relationships = [ + # Valid Relationships + {"source": "founder_1", "target": "startup_1", "type": "FOUNDED", "valid_from": "2021-01-01"}, + {"source": "investor_1", "target": "startup_1", "type": "INVESTED_IN", "amount": 5000000, "valid_from": "2023-06-01"}, + + # Dangling Edge (Error!) + {"source": "founder_2", "target": "startup_999", "type": "FOUNDED", "valid_from": "2020-05-15"}, + + # Temporal Data (History) + {"source": "founder_1", "target": "startup_2", "type": "ADVISED", "valid_from": "2020-01-01", "valid_until": "2021-01-01"} +] + +print(f"Loaded {len(raw_entities)} raw entities and {len(raw_relationships)} raw relationships.") + +# Initialize Validator +validator = GraphValidator() + +# Create a temporary graph object for validation +temp_graph = {"entities": raw_entities, "relationships": raw_relationships} + +# Run Validation +print("Running Validation Check...") +validation_result = validator.validate(temp_graph) + +if not validation_result.is_valid: + print("Validation Failed! Issues found:") + for issue in validation_result.issues: + print(f" - [{issue.severity.name}] {issue.message} (Code: {issue.code})") + + # AUTOMATIC FIX: If it's a dangling edge, remove it + if issue.code == "DANGLING_EDGE": + print(" Auto-Fixing: Removing invalid relationship...") + raw_relationships = [r for r in raw_relationships + if r['target'] != issue.details.get('target_id')] +else: + print("Graph is valid!") + +# Re-validate to confirm fix +print("\nRe-validating after fixes...") +temp_graph = {"entities": raw_entities, "relationships": raw_relationships} +if validator.validate(temp_graph).is_valid: + print("Graph is now clean and valid!") + +# 1. Detect Duplicates +print("Scanning for duplicates...") +deduper = DuplicateDetector(similarity_threshold=0.7) # 70% similarity threshold +duplicates = deduper.detect_duplicates(raw_entities) + +for candidate in duplicates: + print(f"Found potential duplicate pair (Score: {candidate.similarity_score:.2f}):") + print(f" - {candidate.entity1['name']} (ID: {candidate.entity1['id']})") + print(f" - {candidate.entity2['name']} (ID: {candidate.entity2['id']})") + + # MERGE STRATEGY: Keep entity1, merge data from entity2 + print(" Merging entities...") + # (In a real app, you'd use EntityMerger, but here's the logic:) + # We keep startup_1 and discard startup_1_dup, but we note the conflict + +# 2. Detect Conflicts +print("\nChecking for data conflicts...") +conflict_detector = ConflictDetector() + +# Simulating a conflict check between the two versions of TechFlow +# To check conflicts, we treat them as the same entity (same ID) +entity_a = raw_entities[0].copy() +entity_b = raw_entities[2].copy() +entity_b['id'] = entity_a['id'] # Force same ID for conflict detection + +conflicts = conflict_detector.detect_conflicts([entity_a, entity_b]) + +for conflict in conflicts: + print(f" Conflict detected in field '{conflict.property_name}':") + print(f" Values: {conflict.conflicting_values}") + + # RESOLUTION: Trust the higher number (optimistic!) + if conflict.property_name == "revenue": + # values are strings or ints, need to handle types + vals = [float(v) for v in conflict.conflicting_values if v is not None] + resolved_val = max(vals) + print(f" Resolved to: {resolved_val}") + raw_entities[0]['revenue'] = resolved_val + +# Final Cleanup: Remove the duplicate entity from our list +clean_entities = [e for e in raw_entities if e['id'] != 'startup_1_dup'] +clean_relationships = raw_relationships # (We'd normally re-link relationships too) + +print(f"\nCleaned Data: {len(clean_entities)} entities remaining.") + + +# Manual Graph Construction (since we already cleaned it) +kg = { + "entities": clean_entities, + "relationships": clean_relationships, + "metadata": { + "created_at": datetime.now().isoformat(), + "source": "Manual Advanced Pipeline" + } +} +print("Knowledge Graph Assembled Successfully!") + +# Initialize the Master Analyzer +analyzer = GraphAnalyzer(enable_temporal=True) + +# 1. Structural Analysis (Connectivity) +print("\n--- Connectivity Analysis ---") +connectivity = analyzer.analyze_connectivity(kg) +print(f" • Graph Connected? {'Yes' if connectivity['is_connected'] else 'No'}") +print(f" • Connected Components: {connectivity['num_components']}") + +# 2. Centrality (Who is important?) +print("\n--- Centrality Analysis ---") +centrality_result = analyzer.calculate_centrality(kg, centrality_type="degree") +degree_data = centrality_result["centrality_measures"]["degree"] + +# Get pre-calculated rankings +top_nodes = degree_data["rankings"][:3] + +print(" • Top Influencers (Degree Centrality):") +for item in top_nodes: + print(f" - {item['node']}: {item['score']:.2f}") + +# 3. Community Detection (Clustering) +print("\n--- Community Detection ---") +communities = analyzer.detect_communities(kg, algorithm="louvain") +community_result = communities +communities = community_result["communities"] + +print(f" • Detected {len(communities)} communities.") +for i, comm in enumerate(communities): + # comm is a set of node IDs + members = list(comm) + print(f" Community {i+1}: {', '.join(members)}") + +temporal_engine = TemporalGraphQuery(temporal_granularity="year") + +# 1. Time Travel Query: What did the world look like in 2020? +print("\n--- Time Travel: 2020 ---") +snapshot_2020 = temporal_engine.query_at_time(kg, query="*", at_time="2020-06-01") +print(f" Active Relationships in 2020: {len(snapshot_2020['relationships'])}") +for rel in snapshot_2020['relationships']: + print(f" - {rel['source']} --[{rel['type']}]--> {rel['target']}") + +# 2. Time Travel Query: What about 2023? +print("\n--- Time Travel: 2023 ---") +snapshot_2023 = temporal_engine.query_at_time(kg, query="*", at_time="2023-07-01") +print(f" Active Relationships in 2023: {len(snapshot_2023['relationships'])}") +for rel in snapshot_2023['relationships']: + print(f" - {rel['source']} --[{rel['type']}]--> {rel['target']}") + +# Notice how 'ADVISED' might disappear if it ended, and 'INVESTED_IN' appears! + +tracker = ProvenanceTracker() + +# Let's pretend we're tracking the source of our data +tracker.track_entity("startup_1", source="Crunchbase_API_v2", metadata={"confidence": 0.95}) +tracker.track_entity("startup_1", source="Manual_Entry_User_Bob", metadata={"confidence": 1.0}) + +print("\n--- Provenance Report: TechFlow AI ---") +lineage = tracker.get_lineage("startup_1") +print(f" Entity: startup_1") +print(f" First Seen: {lineage['first_seen']}") +print(f" Sources:") +for src in lineage['sources']: + print(f" - {src['source']} (at {src['timestamp']})")