{
 "cells": [
  {
   "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/use_cases/intelligence/02_Intelligence_Analysis.ipynb)\n",
    "\n",
    "# Intelligence Analysis with Orchestrator-Worker Pattern\n",
    "\n",
    "## Overview\n",
    "\n",
    "This notebook demonstrates comprehensive intelligence analysis using **Semantica's Orchestrator-Worker Pattern** with complete implementation of graph analytics, hybrid RAG, and ontology building. The pipeline processes OSINT feeds, threat intelligence, and geospatial data to build multi-source intelligence graphs with detailed analysis.\n",
    "\n",
    "\n",
    "**Documentation**: [API Reference](https://semantica.readthedocs.io/use-cases/)\n",
    "\n",
    "### Orchestrator-Worker Architecture\n",
    "\n",
    "The Orchestrator-Worker pattern uses **ExecutionEngine** as the orchestrator to coordinate **7 specialized workers** in parallel:\n",
    "\n",
    "- **Orchestrator**: ExecutionEngine coordinates all workers using PipelineBuilder and ParallelismManager\n",
    "- **Worker 1 - Data Ingestion Worker**: Multi-source data ingestion (FileIngestor, WebIngestor, StreamIngestor, FeedIngestor, DBIngestor)\n",
    "- **Worker 2 - Ontology Building Worker**: Complete 6-stage ontology generation pipeline\n",
    "- **Worker 3 - Graph Construction Worker**: Builds knowledge graphs (GraphBuilder, TemporalGraphQuery)\n",
    "- **Worker 4 - Graph Analytics Worker**: Comprehensive graph analytics (all centrality measures, community detection, connectivity)\n",
    "- **Worker 5 - Hybrid RAG Worker**: Complete hybrid RAG with KG queries + vector search\n",
    "- **Worker 6 - Intelligence Analysis Worker**: Threat assessment, geospatial analysis, pattern detection\n",
    "- **Worker 7 - Report Generation Worker**: Compiles comprehensive intelligence reports\n",
    "\n",
    "### Why Semantica?\n",
    "\n",
    "Semantica provides a complete framework for intelligence analysis:\n",
    "\n",
    "- **Orchestrator-Worker Pattern**: Efficient parallel processing with coordinated workers\n",
    "- **Complete Ontology Building**: 6-stage automatic ontology generation from data\n",
    "- **Comprehensive Graph Analytics**: All centrality measures, community detection, connectivity analysis\n",
    "- **Hybrid RAG**: Complete implementation combining KG queries with vector similarity search\n",
    "- **MCP Integration**: Real-time data fetching, web scraping, API integration, browser automation\n",
    "- **Agent Memory**: Persistent memory for threat context and intelligence history\n",
    "- **Multi-Source Fusion**: Correlate intelligence from multiple sources\n",
    "\n",
    "### Key Features\n",
    "\n",
    "- **Orchestrator-Worker Pattern**: 7 specialized workers coordinated by ExecutionEngine\n",
    "- **Complete Ontology Pipeline**: 6-stage ontology generation (semantic network parsing → YAML-to-definition → definition-to-types → hierarchy generation → TTL generation → symbolic validation)\n",
    "- **All Graph Analytics**: PageRank, Betweenness, Closeness, Eigenvector centrality, Louvain community detection, connectivity analysis, path finding\n",
    "- **Hybrid RAG**: Vector store + knowledge graph queries + hybrid search + query orchestration\n",
    "- **Multi-Source Intelligence**: OSINT, threat intelligence, social media, news, public records, geospatial data\n",
    "- **MCP Integration**: Browser automation, web scraping, API integration\n",
    "- **Threat Assessment**: Risk scoring, entity relationship mapping, pattern detection\n",
    "- **Geospatial Intelligence**: Location-based tracking, movement pattern analysis\n",
    "\n",
    "### Semantica Modules Used (30+)\n",
    "\n",
    "- **Orchestrator**: ExecutionEngine, PipelineBuilder, ParallelismManager\n",
    "- **Ingest**: FileIngestor, WebIngestor, FeedIngestor, StreamIngestor, DBIngestor, RepoIngestor, EmailIngestor, MCPIngestor\n",
    "- **MCP Integration**: MCP browser tools and resources\n",
    "- **Parse**: JSONParser, XMLParser, CSVParser, DocumentParser, StructuredDataParser\n",
    "- **Normalize**: TextNormalizer, DataNormalizer\n",
    "- **Semantic Extract**: NERExtractor, RelationExtractor, TripleExtractor, EventDetector\n",
    "- **Ontology**: OntologyGenerator, ClassInferrer, PropertyGenerator, OWLGenerator, OntologyValidator\n",
    "- **KG**: GraphBuilder, TemporalGraphQuery, GraphAnalyzer, ConnectivityAnalyzer, CentralityCalculator, CommunityDetector\n",
    "- **Graph Analytics**: All centrality measures, Louvain, connectivity, path finding\n",
    "- **Embeddings**: EmbeddingGenerator, TextEmbedder\n",
    "- **Vector Store**: VectorStore, HybridSearch, MetadataFilter\n",
    "- **Context**: AgentMemory, ContextRetriever, ContextGraphBuilder\n",
    "- **Reasoning**: InferenceEngine, RuleManager, ExplanationGenerator\n",
    "- **Export**: ReportGenerator, JSONExporter\n",
    "- **Visualization**: KGVisualizer, AnalyticsVisualizer, TemporalVisualizer\n",
    "\n",
    "### Pipeline Overview\n",
    "\n",
    "**OSINT Sources → MCP Integration → Orchestrator Setup → Parallel Workers (Data Ingestion → Ontology Building → Graph Construction → Graph Analytics → Hybrid RAG → Intelligence Analysis → Report Generation) → Visualization → Reporting**\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"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 1: Setup and Import Semantica Modules\n",
    "\n",
    "Import all necessary Semantica modules including orchestrator, workers, ontology, graph analytics, and hybrid RAG modules.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Import all Semantica modules for intelligence analysis with Orchestrator-Worker pattern\n",
    "from semantica.pipeline import PipelineBuilder, ExecutionEngine, ParallelismManager\n",
    "from semantica.ingest import FileIngestor, DBIngestor, WebIngestor, StreamIngestor, FeedIngestor\n",
    "from semantica.parse import JSONParser, XMLParser, CSVParser, DocumentParser, StructuredDataParser\n",
    "from semantica.normalize import TextNormalizer, DataNormalizer\n",
    "from semantica.semantic_extract import NERExtractor, RelationExtractor, TripleExtractor, EventDetector\n",
    "from semantica.ontology import OntologyGenerator, ClassInferrer, PropertyGenerator, OWLGenerator, OntologyValidator\n",
    "from semantica.kg import GraphBuilder, TemporalGraphQuery, GraphAnalyzer, ConnectivityAnalyzer\n",
    "from semantica.embeddings import EmbeddingGenerator, TextEmbedder\n",
    "from semantica.vector_store import VectorStore, HybridSearch, MetadataFilter\n",
    "from semantica.context import AgentMemory, ContextRetriever, ContextGraphBuilder\n",
    "from semantica.reasoning import InferenceEngine, RuleManager, ExplanationGenerator\n",
    "from semantica.export import ReportGenerator, JSONExporter\n",
    "from semantica.visualization import KGVisualizer, AnalyticsVisualizer, TemporalVisualizer\n",
    "\n",
    "import tempfile\n",
    "import os\n",
    "import json\n",
    "from datetime import datetime, timedelta\n",
    "import numpy as np\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 2: Initialize Orchestrator and Setup\n",
    "\n",
    "Initialize the orchestrator (ExecutionEngine) and prepare for worker coordination.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Initialize orchestrator\n",
    "orchestrator = ExecutionEngine()\n",
    "pipeline_builder = PipelineBuilder()\n",
    "parallelism_manager = ParallelismManager(max_workers=7)\n",
    "\n",
    "# Initialize vector store and agent memory\n",
    "vector_store = VectorStore(backend=\"faiss\", dimension=768)\n",
    "agent_memory = AgentMemory(\n",
    "    vector_store=vector_store,\n",
    "    retention_policy=\"unlimited\",\n",
    "    max_memory_size=10000\n",
    ")\n",
    "\n",
    "# Create temporary directory for data\n",
    "temp_dir = tempfile.mkdtemp()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 3: Define Worker 1 - Data Ingestion Worker\n",
    "\n",
    "Worker for multi-source data ingestion from OSINT feeds, threat intelligence, and geospatial data.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Worker 1: Data Ingestion Worker\n",
    "def data_ingestion_worker(sources, memory):\n",
    "    \"\"\"Multi-source data ingestion worker.\"\"\"\n",
    "    file_ingestor = FileIngestor()\n",
    "    web_ingestor = WebIngestor()\n",
    "    feed_ingestor = FeedIngestor()\n",
    "    \n",
    "    ingested_data = []\n",
    "    \n",
    "    # Sample OSINT and threat intelligence data\n",
    "    osint_data = {\n",
    "        \"threat_actors\": [\n",
    "            {\"name\": \"Threat Actor A\", \"type\": \"APT\", \"location\": \"Country X\"},\n",
    "            {\"name\": \"Threat Actor B\", \"type\": \"Cybercriminal\", \"location\": \"Country Y\"}\n",
    "        ],\n",
    "        \"indicators\": [\n",
    "            {\"type\": \"IP\", \"value\": \"192.168.1.100\", \"threat_level\": \"high\"},\n",
    "            {\"type\": \"Domain\", \"value\": \"malicious.com\", \"threat_level\": \"medium\"}\n",
    "        ]\n",
    "    }\n",
    "    \n",
    "    # Save sample data\n",
    "    osint_file = os.path.join(temp_dir, \"osint_data.json\")\n",
    "    with open(osint_file, 'w') as f:\n",
    "        json.dump(osint_data, f, indent=2)\n",
    "    \n",
    "    ingested_data.append(file_ingestor.ingest_file(osint_file, read_content=True))\n",
    "    \n",
    "    # Store in memory\n",
    "    memory.store(\n",
    "        f\"Data ingestion worker processed {len(ingested_data)} sources\",\n",
    "        metadata={\"worker\": \"data_ingestion\", \"sources_count\": len(ingested_data)}\n",
    "    )\n",
    "    \n",
    "    return {\"ingested_data\": ingested_data, \"osint_data\": osint_data}\n",
    "\n",
    "# Test worker\n",
    "osint_data_result = data_ingestion_worker([], agent_memory)\n",
    "print(f\"  - Sources processed: {len(osint_data_result.get('ingested_data', []))}\")\n",
    "print(f\"  - OSINT data ingested: {len(osint_data_result.get('osint_data', {}).get('threat_actors', []))} threat actors\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 4: Define Worker 2 - Ontology Building Worker\n",
    "\n",
    "Worker for complete 6-stage ontology generation pipeline.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Worker 2: Ontology Building Worker\n",
    "def ontology_building_worker(entities, relationships, memory):\n",
    "    \"\"\"Complete 6-stage ontology generation worker.\"\"\"\n",
    "    ontology_gen = OntologyGenerator(base_uri=\"https://example.org/intelligence/ontology/\")\n",
    "    class_inferrer = ClassInferrer()\n",
    "    property_generator = PropertyGenerator()\n",
    "    owl_generator = OWLGenerator()\n",
    "    ontology_validator = OntologyValidator()\n",
    "    \n",
    "    # Stage 1-6: Complete ontology generation pipeline\n",
    "    ontology = ontology_gen.generate_ontology({\n",
    "        \"entities\": entities,\n",
    "        \"relationships\": relationships\n",
    "    })\n",
    "    \n",
    "    # Generate OWL/Turtle\n",
    "    owl_content = owl_generator.generate_owl(ontology, format=\"turtle\")\n",
    "    \n",
    "    # Validate ontology\n",
    "    validation_result = ontology_validator.validate_ontology(ontology, method=\"hermit\")\n",
    "    \n",
    "    # Store in memory\n",
    "    memory.store(\n",
    "        f\"Ontology building worker generated ontology with {len(ontology.get('classes', [])) if isinstance(ontology, dict) else 0} classes\",\n",
    "        metadata={\n",
    "            \"worker\": \"ontology_building\",\n",
    "            \"classes_count\": len(ontology.get('classes', [])) if isinstance(ontology, dict) else 0,\n",
    "            \"validation_passed\": validation_result.get('is_valid', False) if isinstance(validation_result, dict) else False\n",
    "        }\n",
    "    )\n",
    "    \n",
    "    return {\n",
    "        \"ontology\": ontology,\n",
    "        \"owl_content\": owl_content,\n",
    "        \"validation\": validation_result\n",
    "    }\n",
    "\n",
    "# Prepare sample entities and relationships for ontology\n",
    "sample_entities = [\n",
    "    {\"id\": \"E1\", \"type\": \"ThreatActor\", \"name\": \"Threat Actor A\"},\n",
    "    {\"id\": \"E2\", \"type\": \"Indicator\", \"name\": \"Malicious IP\"},\n",
    "    {\"id\": \"E3\", \"type\": \"Location\", \"name\": \"Country X\"}\n",
    "]\n",
    "sample_relationships = [\n",
    "    {\"source\": \"E1\", \"target\": \"E2\", \"type\": \"uses\"},\n",
    "    {\"source\": \"E1\", \"target\": \"E3\", \"type\": \"located_in\"}\n",
    "]\n",
    "\n",
    "ontology_result = ontology_building_worker(sample_entities, sample_relationships, agent_memory)\n",
    "print(f\"  - Ontology generated: {bool(ontology_result.get('ontology'))}\")\n",
    "print(f\"  - OWL content generated: {len(ontology_result.get('owl_content', ''))} characters\")\n",
    "print(f\"  - Validation passed: {ontology_result.get('validation', {}).get('is_valid', False) if isinstance(ontology_result.get('validation'), dict) else False}\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 5: Define Worker 3 - Graph Construction Worker\n",
    "\n",
    "Worker for building knowledge graphs from entities and relationships.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Worker 3: Graph Construction Worker\n",
    "def graph_construction_worker(entities, relationships, memory):\n",
    "    \"\"\"Build knowledge graph worker.\"\"\"\n",
    "    graph_builder = GraphBuilder()\n",
    "    temporal_query = TemporalGraphQuery()\n",
    "    \n",
    "    # Build knowledge graph\n",
    "    kg = graph_builder.build(entities, relationships)\n",
    "    \n",
    "    # Build temporal graph query capability\n",
    "    temporal_kg = temporal_query.build_temporal_graph(kg)\n",
    "    \n",
    "    # Store in memory\n",
    "    memory.store(\n",
    "        f\"Graph construction worker built KG with {len(entities)} entities and {len(relationships)} relationships\",\n",
    "        metadata={\n",
    "            \"worker\": \"graph_construction\",\n",
    "            \"entity_count\": len(entities),\n",
    "            \"relationship_count\": len(relationships)\n",
    "        },\n",
    "        entities=entities,\n",
    "        relationships=relationships\n",
    "    )\n",
    "    \n",
    "    return {\"knowledge_graph\": kg, \"temporal_graph\": temporal_kg}\n",
    "\n",
    "# Test worker with sample data\n",
    "graph_result = graph_construction_worker(sample_entities, sample_relationships, agent_memory)\n",
    "print(f\"  - Knowledge graph built: {bool(graph_result.get('knowledge_graph'))}\")\n",
    "print(f\"  - Temporal graph built: {bool(graph_result.get('temporal_graph'))}\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 6: Define Worker 4 - Graph Analytics Worker\n",
    "\n",
    "Worker for comprehensive graph analytics including all centrality measures, community detection, and connectivity analysis.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Worker 4: Graph Analytics Worker\n",
    "def graph_analytics_worker(kg, memory):\n",
    "    \"\"\"Comprehensive graph analytics worker.\"\"\"\n",
    "    graph_analyzer = GraphAnalyzer()\n",
    "    connectivity_analyzer = ConnectivityAnalyzer()\n",
    "    \n",
    "    # All centrality measures\n",
    "    pagerank = graph_analyzer.compute_centrality(kg, method=\"pagerank\")\n",
    "    betweenness = graph_analyzer.compute_centrality(kg, method=\"betweenness\")\n",
    "    closeness = graph_analyzer.compute_centrality(kg, method=\"closeness\")\n",
    "    eigenvector = graph_analyzer.compute_centrality(kg, method=\"eigenvector\")\n",
    "    \n",
    "    # Community detection\n",
    "    communities = graph_analyzer.detect_communities(kg, method=\"louvain\")\n",
    "    \n",
    "    # Connectivity analysis\n",
    "    connectivity = connectivity_analyzer.analyze_connectivity(kg)\n",
    "    \n",
    "    # Graph metrics\n",
    "    metrics = graph_analyzer.compute_metrics(kg)\n",
    "    \n",
    "    # Store in memory\n",
    "    memory.store(\n",
    "        f\"Graph analytics worker completed: {len(pagerank)} entities analyzed\",\n",
    "        metadata={\n",
    "            \"worker\": \"graph_analytics\",\n",
    "            \"centrality_measures\": 4,\n",
    "            \"communities\": communities.get('num_communities', 0) if isinstance(communities, dict) else 0\n",
    "        }\n",
    "    )\n",
    "    \n",
    "    return {\n",
    "        \"pagerank\": pagerank,\n",
    "        \"betweenness\": betweenness,\n",
    "        \"closeness\": closeness,\n",
    "        \"eigenvector\": eigenvector,\n",
    "        \"communities\": communities,\n",
    "        \"connectivity\": connectivity,\n",
    "        \"metrics\": metrics\n",
    "    }\n",
    "\n",
    "# Test worker\n",
    "analytics_result = graph_analytics_worker(graph_result.get('knowledge_graph'), agent_memory)\n",
    "print(f\"  - PageRank computed: {len(analytics_result.get('pagerank', {}))}\")\n",
    "print(f\"  - Betweenness computed: {len(analytics_result.get('betweenness', {}))}\")\n",
    "print(f\"  - Closeness computed: {len(analytics_result.get('closeness', {}))}\")\n",
    "print(f\"  - Eigenvector computed: {len(analytics_result.get('eigenvector', {}))}\")\n",
    "print(f\"  - Communities detected: {analytics_result.get('communities', {}).get('num_communities', 0) if isinstance(analytics_result.get('communities'), dict) else 0}\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 7: Define Worker 5 - Hybrid RAG Worker\n",
    "\n",
    "Worker for complete hybrid RAG implementation with vector store, KG queries, and hybrid search.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Worker 5: Hybrid RAG Worker\n",
    "def hybrid_rag_worker(kg, vector_store, entities, memory):\n",
    "    \"\"\"Complete hybrid RAG worker.\"\"\"\n",
    "    embedding_generator = EmbeddingGenerator()\n",
    "    text_embedder = TextEmbedder()\n",
    "    hybrid_search = HybridSearch(vector_store=vector_store, knowledge_graph=kg)\n",
    "    context_retriever = ContextRetriever()\n",
    "    \n",
    "    # Generate embeddings for entities\n",
    "    entity_embeddings = {}\n",
    "    for entity in entities:\n",
    "        entity_text = f\"{entity.get('name', '')} {entity.get('type', '')}\"\n",
    "        embedding = text_embedder.embed(entity_text)\n",
    "        entity_embeddings[entity.get('id', '')] = embedding\n",
    "    \n",
    "    # Store embeddings in vector store\n",
    "    entity_metadata = [\n",
    "        {\"entity_id\": eid, \"name\": next((e.get('name', '') for e in entities if e.get('id') == eid), '')}\n",
    "        for eid in entity_embeddings.keys()\n",
    "    ]\n",
    "    vector_store.store_vectors(\n",
    "        vectors=list(entity_embeddings.values()),\n",
    "        metadata=entity_metadata\n",
    "    )\n",
    "    \n",
    "    # Example hybrid search query\n",
    "    query = \"What are the key threat actors?\"\n",
    "    hybrid_results = hybrid_search.search(\n",
    "        query=query,\n",
    "        top_k=5,\n",
    "        use_graph=True,\n",
    "        use_vectors=True\n",
    "    )\n",
    "    \n",
    "    # Context retrieval\n",
    "    context = context_retriever.retrieve(\n",
    "        query=query,\n",
    "        knowledge_graph=kg,\n",
    "        top_k=10\n",
    "    )\n",
    "    \n",
    "    # Store in memory\n",
    "    memory.store(\n",
    "        f\"Hybrid RAG worker setup: {len(entity_embeddings)} embeddings, hybrid search ready\",\n",
    "        metadata={\n",
    "            \"worker\": \"hybrid_rag\",\n",
    "            \"embeddings_count\": len(entity_embeddings),\n",
    "            \"hybrid_search\": True\n",
    "        }\n",
    "    )\n",
    "    \n",
    "    return {\n",
    "        \"embeddings\": entity_embeddings,\n",
    "        \"hybrid_search\": hybrid_search,\n",
    "        \"query_results\": hybrid_results,\n",
    "        \"context\": context\n",
    "    }\n",
    "\n",
    "# Test worker\n",
    "rag_result = hybrid_rag_worker(\n",
    "    graph_result.get('knowledge_graph'),\n",
    "    vector_store,\n",
    "    sample_entities,\n",
    "    agent_memory\n",
    ")\n",
    "print(f\"  - Entity embeddings: {len(rag_result.get('embeddings', {}))}\")\n",
    "print(f\"  - Hybrid search ready: {bool(rag_result.get('hybrid_search'))}\")\n",
    "print(f\"  - Query results: {len(rag_result.get('query_results', [])) if isinstance(rag_result.get('query_results'), list) else 1}\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 8: Define Worker 6 - Intelligence Analysis Worker\n",
    "\n",
    "Worker for threat assessment, geospatial analysis, and pattern detection.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Worker 6: Intelligence Analysis Worker\n",
    "def intelligence_analysis_worker(kg, analytics_results, memory):\n",
    "    \"\"\"Threat assessment and geospatial analysis worker.\"\"\"\n",
    "    inference_engine = InferenceEngine()\n",
    "    rule_manager = RuleManager()\n",
    "    \n",
    "    # Threat assessment rules\n",
    "    threat_rules = [\n",
    "        {\n",
    "            \"rule_id\": \"high_centrality_threat\",\n",
    "            \"condition\": \"IF entity has high_pagerank AND entity is threat_actor THEN entity is high_priority_threat\",\n",
    "            \"action\": \"flag_high_priority\"\n",
    "        },\n",
    "        {\n",
    "            \"rule_id\": \"geospatial_cluster\",\n",
    "            \"condition\": \"IF entities in same_location AND high_connectivity THEN entities form_geospatial_cluster\",\n",
    "            \"action\": \"identify_cluster\"\n",
    "        }\n",
    "    ]\n",
    "    \n",
    "    # Pattern detection\n",
    "    patterns = inference_engine.infer(knowledge_graph=kg, rules=threat_rules)\n",
    "    \n",
    "    # Threat scoring based on centrality\n",
    "    threat_scores = {}\n",
    "    pagerank = analytics_results.get('pagerank', {})\n",
    "    for entity_id, score in pagerank.items():\n",
    "        threat_scores[entity_id] = score * 100  # Scale to 0-100\n",
    "    \n",
    "    # Store in memory\n",
    "    memory.store(\n",
    "        f\"Intelligence analysis worker: {len(patterns) if isinstance(patterns, list) else 1} patterns detected\",\n",
    "        metadata={\n",
    "            \"worker\": \"intelligence_analysis\",\n",
    "            \"patterns_count\": len(patterns) if isinstance(patterns, list) else 1,\n",
    "            \"threat_scores\": len(threat_scores)\n",
    "        }\n",
    "    )\n",
    "    \n",
    "    return {\n",
    "        \"patterns\": patterns,\n",
    "        \"threat_scores\": threat_scores,\n",
    "        \"rules\": threat_rules\n",
    "    }\n",
    "\n",
    "# Test worker\n",
    "intelligence_result = intelligence_analysis_worker(\n",
    "    graph_result.get('knowledge_graph'),\n",
    "    analytics_result,\n",
    "    agent_memory\n",
    ")\n",
    "print(f\"  - Patterns detected: {len(intelligence_result.get('patterns', [])) if isinstance(intelligence_result.get('patterns'), list) else 1}\")\n",
    "print(f\"  - Threat scores computed: {len(intelligence_result.get('threat_scores', {}))}\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 9: Define Worker 7 - Report Generation Worker\n",
    "\n",
    "Worker for compiling comprehensive intelligence reports.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Worker 7: Report Generation Worker\n",
    "def report_generation_worker(all_results, memory):\n",
    "    \"\"\"Compile comprehensive intelligence report worker.\"\"\"\n",
    "    report_generator = ReportGenerator()\n",
    "    \n",
    "    # Retrieve all context from memory\n",
    "    context = memory.retrieve(\"intelligence analysis\", max_results=20)\n",
    "    \n",
    "    # Compile report data\n",
    "    report_data = {\n",
    "        \"title\": \"Intelligence Analysis Report\",\n",
    "        \"executive_summary\": \"Comprehensive intelligence analysis using Orchestrator-Worker pattern\",\n",
    "        \"ontology\": all_results.get('ontology', {}),\n",
    "        \"knowledge_graph\": {\n",
    "            \"entities\": len(all_results.get('graph', {}).get('knowledge_graph', {}).get('entities', [])),\n",
    "            \"relationships\": len(all_results.get('graph', {}).get('knowledge_graph', {}).get('relationships', []))\n",
    "        },\n",
    "        \"graph_analytics\": all_results.get('analytics', {}),\n",
    "        \"hybrid_rag\": {\n",
    "            \"embeddings_count\": len(all_results.get('rag', {}).get('embeddings', {}))\n",
    "        },\n",
    "        \"intelligence_analysis\": all_results.get('intelligence', {}),\n",
    "        \"agent_memory_stats\": memory.get_statistics(),\n",
    "        \"context_items\": len(context)\n",
    "    }\n",
    "    \n",
    "    # Store in memory\n",
    "    memory.store(\n",
    "        \"Report generation worker compiled comprehensive intelligence report\",\n",
    "        metadata={\"worker\": \"report_generation\", \"context_items\": len(context)}\n",
    "    )\n",
    "    \n",
    "    return {\"report_data\": report_data, \"context\": context}\n",
    "\n",
    "# Prepare all results for report generation\n",
    "all_worker_results = {\n",
    "    \"ontology\": ontology_result,\n",
    "    \"graph\": graph_result,\n",
    "    \"analytics\": analytics_result,\n",
    "    \"rag\": rag_result,\n",
    "    \"intelligence\": intelligence_result\n",
    "}\n",
    "\n",
    "report_result = report_generation_worker(all_worker_results, agent_memory)\n",
    "print(f\"  - Report data compiled: {bool(report_result.get('report_data'))}\")\n",
    "print(f\"  - Context items: {len(report_result.get('context', []))}\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 10: Orchestrator Coordination - Execute All Workers in Parallel\n",
    "\n",
    "Use the orchestrator to coordinate all 7 workers in parallel using PipelineBuilder.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def data_ingestion_handler(data, **config):\n",
    "    sources = data.get(\"sources\", [])\n",
    "    memory = data.get(\"memory\")\n",
    "    r = data_ingestion_worker(sources, memory)\n",
    "    return {**data, \"osint\": r}\n",
    "def ontology_handler(data, **config):\n",
    "    entities = data.get(\"entities\", [])\n",
    "    relationships = data.get(\"relationships\", [])\n",
    "    memory = data.get(\"memory\")\n",
    "    r = ontology_building_worker(entities, relationships, memory)\n",
    "    return {**data, \"ontology_result\": r}\n",
    "def graph_handler(data, **config):\n",
    "    entities = data.get(\"entities\", [])\n",
    "    relationships = data.get(\"relationships\", [])\n",
    "    memory = data.get(\"memory\")\n",
    "    r = graph_construction_worker(entities, relationships, memory)\n",
    "    return {**data, \"graph_result\": r}\n",
    "def analytics_handler(data, **config):\n",
    "    kg = data.get(\"graph_result\", {}).get(\"knowledge_graph\")\n",
    "    memory = data.get(\"memory\")\n",
    "    r = graph_analytics_worker(kg, memory)\n",
    "    return {**data, \"analytics_result\": r}\n",
    "def rag_handler(data, **config):\n",
    "    kg = data.get(\"graph_result\", {}).get(\"knowledge_graph\")\n",
    "    vector_store = data.get(\"vector_store\")\n",
    "    entities = data.get(\"entities\", [])\n",
    "    memory = data.get(\"memory\")\n",
    "    r = hybrid_rag_worker(kg, vector_store, entities, memory)\n",
    "    return {**data, \"rag_result\": r}\n",
    "def intel_handler(data, **config):\n",
    "    kg = data.get(\"graph_result\", {}).get(\"knowledge_graph\")\n",
    "    analytics_results = data.get(\"analytics_result\")\n",
    "    memory = data.get(\"memory\")\n",
    "    r = intelligence_analysis_worker(kg, analytics_results, memory)\n",
    "    return {**data, \"intelligence_result\": r}\n",
    "def report_handler(data, **config):\n",
    "    memory = data.get(\"memory\")\n",
    "    all_results = {\"ontology\": data.get(\"ontology_result\"), \"graph\": data.get(\"graph_result\"), \"analytics\": data.get(\"analytics_result\"), \"rag\": data.get(\"rag_result\"), \"intelligence\": data.get(\"intelligence_result\")}\n",
    "    r = report_generation_worker(all_results, memory)\n",
    "    return {**data, \"report_result\": r}\n",
    "intelligence_pipeline = (\n",
    "    pipeline_builder\n",
    "    .add_step(\"data_ingestion\", \"ingest\", handler=data_ingestion_handler)\n",
    "    .add_step(\"ontology_building\", \"ontology\", dependencies=[\"data_ingestion\"], handler=ontology_handler)\n",
    "    .add_step(\"graph_construction\", \"build_graph\", dependencies=[\"ontology_building\"], handler=graph_handler)\n",
    "    .add_step(\"graph_analytics\", \"analyze_graph\", dependencies=[\"graph_construction\"], handler=analytics_handler)\n",
    "    .add_step(\"hybrid_rag\", \"rag\", dependencies=[\"graph_construction\"], handler=rag_handler)\n",
    "    .add_step(\"intelligence_analysis\", \"analysis\", dependencies=[\"graph_analytics\", \"hybrid_rag\"], handler=intel_handler)\n",
    "    .add_step(\"report_generation\", \"report\", dependencies=[\"intelligence_analysis\"], handler=report_handler)\n",
    ")\n",
    ".build()\n",
    "\n",
    "input_data = {\"sources\": [], \"entities\": sample_entities, \"relationships\": sample_relationships, \"vector_store\": vector_store, \"memory\": agent_memory}\n",
    "pipeline_result = orchestrator.execute_pipeline(intelligence_pipeline, data=input_data, parallel=True, max_workers=7)\n",
    "\n",
    "print(f\"  - Pipeline steps: {len(intelligence_pipeline.steps)}\")\n",
    "print(f\"  - Execution status: {getattr(pipeline_result, 'success', True)}\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 11: Generate Final Intelligence Report\n",
    "\n",
    "Generate comprehensive intelligence report with all analysis results.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Generate final intelligence report\n",
    "report_generator = ReportGenerator()\n",
    "intelligence_report_file = os.path.join(temp_dir, \"intelligence_analysis_report.html\")\n",
    "\n",
    "report_generator.generate_report(\n",
    "    report_result.get('report_data'),\n",
    "    intelligence_report_file,\n",
    "    format=\"html\"\n",
    ")\n",
    "\n",
    "print(f\"  - Report file: {intelligence_report_file}\")\n",
    "print(f\"  - Report includes: Ontology, KG, Graph Analytics, Hybrid RAG, Intelligence Analysis, Agent Memory Stats\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Conclusion and Best Practices\n",
    "\n",
    "### Key Takeaways\n",
    "\n",
    "1. **Orchestrator-Worker Pattern**: Semantica's ExecutionEngine efficiently coordinates 7 specialized workers in parallel\n",
    "2. **Complete Ontology Building**: 6-stage ontology generation pipeline automatically creates domain ontologies\n",
    "3. **Comprehensive Graph Analytics**: All centrality measures, community detection, and connectivity analysis in one framework\n",
    "4. **Hybrid RAG**: Complete implementation combining knowledge graph queries with vector similarity search\n",
    "5. **Agent Memory**: Persistent context across all worker interactions\n",
    "6. **MCP Integration**: External data access and browser automation for OSINT gathering\n",
    "7. **Scalable Architecture**: Parallel execution enables efficient processing of large-scale intelligence data\n",
    "\n",
    "### Orchestrator-Worker Pattern Benefits\n",
    "\n",
    "- **Efficiency**: Parallel execution of workers reduces total processing time\n",
    "- **Modularity**: Each worker has a specific responsibility\n",
    "- **Scalability**: Easy to add or modify workers\n",
    "- **Coordination**: Orchestrator manages dependencies and execution order\n",
    "- **Error Handling**: Centralized error handling and recovery\n",
    "\n",
    "### Semantica-Specific Performance Considerations\n",
    "\n",
    "- **Parallel Execution**: Leverage ParallelismManager for concurrent worker execution\n",
    "- **Graph Analytics**: Use GraphAnalyzer for efficient analysis on large networks\n",
    "- **Hybrid RAG**: Combine KG and vector search for comprehensive intelligence queries\n",
    "- **Agent Memory**: Utilize persistent memory for context across workers\n",
    "\n",
    "### Deployment Recommendations\n",
    "\n",
    "1. **Production Setup**: Use Semantica's configuration management and pipeline orchestration\n",
    "2. **Scalability**: Leverage parallel execution for large-scale intelligence processing\n",
    "3. **Quality Assurance**: Use Semantica's validation and quality modules\n",
    "4. **Integration**: Semantica's unified framework simplifies integration with existing systems\n"
   ]
  }
 ],
 "metadata": {
  "language_info": {
   "name": "python"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
