{
 "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/advanced/07_Pipeline_Orchestration.ipynb)\n",
    "\n",
    "# Pipeline Orchestration\n",
    "\n",
    "## Overview\n",
    "\n",
    "Build complex pipelines, execute them, handle failures, enable parallel processing, and monitor execution.\n",
    "\n",
    "\n",
    "**Documentation**: [API Reference](https://semantica.readthedocs.io/reference/pipeline/)\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",
    "## Workflow: Build Pipelines → Execute → Handle Failures → Parallel Processing → Monitor\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from semantica.pipeline import (\n",
    "    PipelineBuilder,\n",
    "    ExecutionEngine,\n",
    "    FailureHandler,\n",
    "    ParallelismManager,\n",
    "    RetryPolicy,\n",
    "    RetryStrategy\n",
    ")\n",
    "from semantica.ingest import FileIngestor\n",
    "from semantica.parse import DocumentParser\n",
    "from semantica.semantic_extract import NERExtractor\n",
    "from semantica.kg import GraphBuilder\n",
    "import time\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 1: Build Complex Pipelines\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "builder = PipelineBuilder()\n",
    "\n",
    "file_ingestor = FileIngestor()\n",
    "document_parser = DocumentParser()\n",
    "ner_extractor = NERExtractor()\n",
    "graph_builder = GraphBuilder()\n",
    "\n",
    "# Define handlers for each pipeline step\n",
    "def ingest_handler(data, **config):\n",
    "    files = data.get(\"files\", [])\n",
    "    if files:\n",
    "        # Ingest first file as example\n",
    "        file_obj = file_ingestor.ingest_file(files[0], read_content=True)\n",
    "        return {**data, \"file\": file_obj}\n",
    "    return data\n",
    "\n",
    "def parse_handler(data, **config):\n",
    "    # If a file was ingested, try parsing; otherwise pass text through\n",
    "    file_obj = data.get(\"file\")\n",
    "    if file_obj and getattr(file_obj, \"path\", None):\n",
    "        parsed = document_parser.parse_document(file_obj.path)\n",
    "        text = parsed.get(\"text\") if isinstance(parsed, dict) else None\n",
    "        return {**data, \"text\": text or data.get(\"text\")}\n",
    "    return data\n",
    "\n",
    "def extract_handler(data, **config):\n",
    "    text = data.get(\"text\", \"\")\n",
    "    entities = ner_extractor.extract_entities(text)\n",
    "    # Normalize to dict list for graph builder\n",
    "    entity_dicts = [\n",
    "        {\"id\": f\"e{i}\", \"name\": e.text, \"type\": e.label} for i, e in enumerate(entities)\n",
    "    ]\n",
    "    return {**data, \"entities\": entity_dicts}\n",
    "\n",
    "def build_graph_handler(data, **config):\n",
    "    entities = data.get(\"entities\", [])\n",
    "    graph = graph_builder.build({\"entities\": entities})\n",
    "    return {**data, \"graph\": graph}\n",
    "\n",
    "# Build pipeline with proper handlers and dependencies\n",
    "pipeline = (\n",
    "    builder\n",
    "    .add_step(\"ingest\", \"ingest\", handler=ingest_handler)\n",
    "    .add_step(\"parse\", \"parse\", dependencies=[\"ingest\"], handler=parse_handler)\n",
    "    .add_step(\"extract\", \"extract\", dependencies=[\"parse\"], handler=extract_handler)\n",
    "    .add_step(\"build_graph\", \"build_graph\", dependencies=[\"extract\"], handler=build_graph_handler)\n",
    ").build()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 2: Execute Pipeline\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "engine = ExecutionEngine()\n",
    "\n",
    "input_data = {\n",
    "    \"text\": \"Alice works at Tech Corp. Bob is a friend of Alice.\",\n",
    "    \"files\": []\n",
    "}\n",
    "\n",
    "start_time = time.time()\n",
    "result = engine.execute_pipeline(pipeline, input_data)\n",
    "execution_time = result.metrics.get(\"execution_time\", time.time() - start_time)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 3: Handle Failures\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Configure retry policy for the 'extract' step type\n",
    "engine.failure_handler.set_retry_policy(\n",
    "    \"extract\",\n",
    "    RetryPolicy(max_retries=3, backoff_factor=2.0, strategy=RetryStrategy.EXPONENTIAL)\n",
    ")\n",
    "\n",
    "result = engine.execute_pipeline(pipeline, input_data)\n",
    "print(\"Pipeline executed with retry policy configured\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 4: Parallel Processing\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "parallelism = ParallelismManager(max_workers=4)\n",
    "\n",
    "# Identify groups of steps that can run in parallel\n",
    "groups = parallelism.identify_parallelizable_steps(pipeline)\n",
    "\n",
    "# Execute first parallelizable group as a demonstration\n",
    "start_time = time.time()\n",
    "parallel_results = []\n",
    "for group in groups:\n",
    "    parallel_results.extend(parallelism.execute_pipeline_steps_parallel(group, input_data, max_workers=4))\n",
    "parallel_time = time.time() - start_time\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 5: Monitor Pipeline Execution\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Metrics from execution engine\n",
    "metrics = result.metrics\n",
    "progress = engine.get_progress(pipeline.name)\n",
    "\n",
    "print(f\"Duration: {metrics.get('execution_time', 0):.2f} seconds\")\n",
    "print(f\"Steps Executed: {metrics.get('steps_executed', 0)}\")\n",
    "print(f\"Steps Failed: {metrics.get('steps_failed', 0)}\")\n",
    "print(f\"Progress: {progress.get('progress_percentage', 0):.1f}% (status: {progress.get('status')})\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Summary\n",
    "\n",
    "Pipeline orchestration workflow:\n",
    "- Complex Pipeline Built\n",
    "- Pipeline Executed\n",
    "- Failure Handling Configured\n",
    "- Parallel Processing Enabled\n",
    "- Full Monitoring and Observability\n"
   ]
  }
 ],
 "metadata": {
  "language_info": {
   "name": "python"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
