DEV Community: Alain Airom (Ayrom)

Inside Docling Factory: Building a Multimodal RAG Powerhouse

Alain Airom (Ayrom) — Sat, 02 May 2026 15:25:23 +0000

The vO.1a release of a project on which I’m working on for a while…📍

Intro-TL;DR

We’ve all been there: staring at a pile of messy PDFs, complex XBRL financial reports, and scattered CSVs, wondering how to turn that “dark data” into something an LLM can actually understand. Today, I’m pulling back the curtain on Docling Factory, a project I’ve been refining to solve exactly that.

The purpose and final expectation of Docling Factory is to design itas a multimodal document processing pipeline. It transforms unstructured documents (PDFs, Docx, etc.) into structured formats and provides an interactive chat interface to query those documents using a local or remote LLM.

The aim is to have a set of tools which are more than just a parser; it’s a full-stack document intelligence pipeline that bridges the gap between raw files and meaningful conversation.

🔩 Key Components

Web Interface: Built with Gradio, providing a multi-tabbed UI for uploading, parsing, and chatting.
Parsing Engine (DoclingParser): Leverages the docling library for layout-aware document conversion.
RAG Engine (RAGEngine): Orchestrates semantic search using OpenSearch as a vector database and Ollama or LiteLLM for embeddings and text generation.
Observability: Integrated with OpenLLMetry and a custom MetricsCollector to track latency, token usage, and system health.
LLM Engine: The LLM orchestration layer follows a hybrid approach: local inference is handled by Ollama, while a LiteLLM gateway provides a unified API interface for external model integration. This setup provides the best of both worlds — local execution for standard tasks and an easy ‘escape hatch’ to commercial LLMs via a standardized gateway when higher reasoning capabilities are required.

The Core Engine: Docling & Layout Awareness

At the heart of the system is the DoclingParser. Unlike traditional parsers that treat a page as a flat bag of words, this engine is layout-aware. It identifies headers, tables, and even those tricky images. Using the docling library, the user can convert a complex PDF into clean Markdown while keeping the structure intact.

I’ve baked in support for RapidOCR, EasyOCR, and even macOS Vision. Whether you have a digital-native DOCX or a grainy scan of a 1990s invoice, the parser adapts. It even handles XBRL and CSV files natively, transforming structured data into LLM-friendly formats.

# docling_parser.py
"""
Docling Parser Module - Enhanced Version
Handles document parsing using the Docling library with support for:
- Batch and individual processing
- Multiple output formats (Markdown, HTML, JSON, Multimodal)
- Figure extraction
- Full page OCR with multiple OCR engines
- XBRL document conversion
- CSV file conversion
"""

import os
import logging
from pathlib import Path
from datetime import datetime
from typing import List, Dict, Optional, Union, Callable
from docling.document_converter import DocumentConverter, PdfFormatOption
from docling.datamodel.base_models import InputFormat
from docling.datamodel.pipeline_options import (
    PdfPipelineOptions,
    EasyOcrOptions,
    RapidOcrOptions,
    TesseractOcrOptions,
    OcrMacOptions
)
from docling_core.types.doc.base import ImageRefMode
from docling_core.types.doc.document import PictureItem, TableItem
import json

#####
#####
#####

def main():
    """Example usage of the enhanced DoclingParser class."""
    parser = DoclingParser(use_gpu=False, output_dir="output")

    # Example: Parse with OCR and figure extraction
    def progress_update(message):
        print(f"Progress: {message}")

    # Batch processing with all features
    def batch_progress(message, current, total):
        percentage = (current / total * 100) if total > 0 else 0
        print(f"[{percentage:.1f}%] {message}")

    results = parser.parse_batch(
        "input",
        output_formats=['markdown', 'html', 'json'],
        export_figures=True,
        export_multimodal=False,
        ocr_engine='easyocr',
        force_ocr=False,
        progress_callback=batch_progress
    )

    # Print summary
    print("\n" + "="*50)
    print("BATCH PROCESSING SUMMARY")
    print("="*50)
    for result in results:
        status_icon = "✓" if result["status"] == "success" else "✗"
        file_name = Path(result['input_file']).name
        if result["status"] == "success":
            formats = ", ".join(result.get('formats', []))
            figures = result.get('figure_count', 0)
            print(f"{status_icon} {file_name}: {formats} ({figures} figures)")
        else:
            print(f"{status_icon} {file_name}: {result.get('error', 'Unknown error')}")
    print("="*50)


if __name__ == "__main__":
    main()

# Made with Bob

Why Multimodal Matters: Documents aren’t just text. They have charts, figures, and diagrams. Docling Factory detects PictureItem elements and extracts them as separate PNGs, or embeds them directly into the Markdown. This means your RAG system doesn't just "read"—it "sees."

RAG & Vector Storage (`rag_engine.py`)

How do the user(s) chat with these documents? I built the RAGEngine to be as flexible as possible. It uses OpenSearch with k-NN (k-Nearest Neighbors) search for lightning-fast vector retrieval. The RAG system enables “Chat with Documents” by indexing parsed text into a vector space.

Vector Database: Uses OpenSearch with k-NN (k-Nearest Neighbors) search enabled. It uses the lucene engine and cosinesimil space type for vector comparisons.

Hybrid Backend

The real magic is the Dual Backend Support. You can run everything locally using Ollama (perfect for privacy-sensitive data) or flip a switch to LiteLLM. LiteLLM acts as a gateway to over 100 providers, letting you use GPT-4, Claude, or Gemini without changing a single line of business logic.

Ollama: For local execution of models like llama3.2 and granite-embedding.

LiteLLM: Acts as an AI Gateway to connect to remote providers (OpenAI, Anthropic) or proxied local models.

# LiteLLM Configuration File
# This file configures the LiteLLM AI Gateway for unified access to multiple LLM providers
# Documentation: https://docs.litellm.ai/docs/proxy/configs

model_list:
  # OpenAI Models (requires OPENAI_API_KEY environment variable)
  - model_name: gpt-4
    litellm_params:
      model: gpt-4
      api_key: os.environ/OPENAI_API_KEY  # Never hardcode API keys

  - model_name: gpt-3.5-turbo
    litellm_params:
      model: gpt-3.5-turbo
      api_key: os.environ/OPENAI_API_KEY  # Never hardcode API keys

  # OpenAI Embeddings
  - model_name: text-embedding-ada-002
    litellm_params:
      model: text-embedding-ada-002
      api_key: os.environ/OPENAI_API_KEY  # Never hardcode API keys

  # Anthropic Claude (requires ANTHROPIC_API_KEY environment variable)
  - model_name: claude-3-sonnet
    litellm_params:
      model: claude-3-sonnet-20240229
      api_key: os.environ/ANTHROPIC_API_KEY  # Never hardcode API keys

  - model_name: claude-3-opus
    litellm_params:
      model: claude-3-opus-20240229
      api_key: os.environ/ANTHROPIC_API_KEY  # Never hardcode API keys

  # Local Ollama models (connect to host Ollama)
  - model_name: ollama/llama3.2
    litellm_params:
      model: ollama/llama3.2:latest
      api_base: http://host.docker.internal:11434

  - model_name: ollama/granite-embedding
    litellm_params:
      model: ollama/granite-embedding:30m
      api_base: http://host.docker.internal:11434

  # Azure OpenAI (requires AZURE_API_KEY, AZURE_API_BASE, AZURE_API_VERSION)
  # - model_name: azure-gpt-4
  #   litellm_params:
  #     model: azure/gpt-4
  #     api_key: os.environ/AZURE_API_KEY  # Never hardcode API keys
  #     api_base: os.environ/AZURE_API_BASE
  #     api_version: os.environ/AZURE_API_VERSION

  # Google Vertex AI (requires GOOGLE_APPLICATION_CREDENTIALS)
  # - model_name: gemini-pro
  #   litellm_params:
  #     model: vertex_ai/gemini-pro
  #     vertex_project: os.environ/VERTEX_PROJECT
  #     vertex_location: os.environ/VERTEX_LOCATION

  # AWS Bedrock (requires AWS credentials)
  # - model_name: bedrock-claude
  #   litellm_params:
  #     model: bedrock/anthropic.claude-v2
  #     aws_region_name: us-east-1

# General settings
general_settings:
  master_key: os.environ/LITELLM_MASTER_KEY  # REQUIRED: Set via environment variable, never hardcode
  database_url: os.environ/DATABASE_URL  # PostgreSQL connection string

  # Logging
  success_callback: ["langfuse"]  # Optional: integrate with Langfuse for observability

  # Rate limiting (optional)
  # max_parallel_requests: 100
  # max_budget: 100  # USD

  # Caching (optional)
  # cache: true
  # cache_params:
  #   type: "redis"
  #   host: "redis"
  #   port: 6379

# Router settings for load balancing and fallbacks
router_settings:
  routing_strategy: simple-shuffle  # Options: simple-shuffle, least-busy, usage-based-routing
  model_group_alias:
    gpt-4-group:
      - gpt-4
      - azure-gpt-4  # Fallback to Azure if OpenAI fails

  # Retry settings
  num_retries: 2
  timeout: 600  # seconds

  # Fallback models
  fallbacks:
    - gpt-4: ["gpt-3.5-turbo"]
    - claude-3-opus: ["claude-3-sonnet"]

# Environment variables required:
# - LITELLM_MASTER_KEY: Master key for API authentication
# - DATABASE_URL: PostgreSQL connection string
# - OPENAI_API_KEY: For OpenAI models (optional)
# - ANTHROPIC_API_KEY: For Claude models (optional)
# - AZURE_API_KEY, AZURE_API_BASE, AZURE_API_VERSION: For Azure OpenAI (optional)
# - GOOGLE_APPLICATION_CREDENTIALS: For Google Vertex AI (optional)
# - AWS credentials: For AWS Bedrock (optional)

# Made with Bob

Chunking: Employs RecursiveCharacterTextSplitter with a chunk size of 500 characters and 100-character overlap to ensure context is preserved without exceeding model limits.

# rag_engine.py
"""
RAG Engine Module
Implements Retrieval-Augmented Generation using OpenSearch and Ollama.
Supports document indexing, semantic search, and chat with documents.
"""

import os
import logging
from pathlib import Path
from typing import List, Dict, Optional, Tuple
import json
from datetime import datetime

# OpenSearch
from opensearchpy import OpenSearch, helpers
from opensearchpy.exceptions import NotFoundError

# LLM Clients
import ollama
from litellm import completion, embedding
import litellm

# LangChain for RAG orchestration
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.documents import Document

# OpenLLMetry for observability
from traceloop.sdk import Traceloop
from traceloop.sdk.decorators import workflow, task

# OpenTelemetry for metrics
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor

# Metrics collector
from metrics_collector import MetricsCollector, initialize_metrics_collector, get_metrics_collector

#####
#####
#####

# Initialize LLM components based on configuration
        if use_litellm:
            logger.info("Using LiteLLM for embeddings and LLM")
            self.embeddings = LiteLLMEmbeddings(
                model=embedding_model,
                api_base=litellm_api_base,
                api_key=litellm_api_key
            )
            self.llm = LiteLLMLLM(
                model=llm_model,
                api_base=litellm_api_base,
                api_key=litellm_api_key
            )
        else:
            logger.info("Using Ollama for embeddings and LLM")
            self.embeddings = OllamaEmbeddings(model=embedding_model, base_url=ollama_base_url)
            self.llm = OllamaLLM(model=llm_model, base_url=ollama_base_url)

        # Text splitter for chunking
        # Reduced chunk size to avoid exceeding embedding model context length
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=500,  # Reduced from 1000 to fit within model context
            chunk_overlap=100,  # Reduced overlap proportionally
            length_function=len,
        )

A Word on The RAG Database: OpenSeach RAG

The OpenSearch implementation serves as the high-performance backbone of the Docling Factory’s RAG engine, specifically configured as a vector database to handle complex semantic retrieval. Within the RAGEngine class, the system initializes an OpenSearch client that utilizes k-NN (k-Nearest Neighbors) search, leveraging the lucene engine and cosinesimil space type to measure the similarity between document chunks and user queries. During the indexing phase, the engine converts Markdown chunks into vectors—typically using a 384-dimensional model—and performs bulk uploads into a dedicated index where index.knn is set to true. For retrieval, the search method transforms the user’s prompt into an embedding and queries OpenSearch to return the top k most relevant context fragments based on their vector score, which are then used to augment the LLM's final response.

Observability: Seeing the Unseen

“Black box” AI is a no-go for production. That’s why I integrated OpenLLMetry. Through our MetricsCollector, the application tracks everything: latency, token usage, and even cost estimation. If a model starts acting up or a prompt is getting too expensive, you’ll see it on the Plotly-powered dashboard before it becomes a problem.

The MetricsCollector uses OpenTelemetry to intercept spans from LLM calls. The standalone_dashboard.py script can then generate a visual representation of this data using Plotly, showing error rates, token costs, and P95/P99 latency percentiles.

# metrics_collector.py
"""
OpenLLMetry Metrics Collector
Collects and aggregates metrics from OpenTelemetry traces for dashboard display.
"""

import logging
from typing import Dict, List, Optional
from datetime import datetime, timedelta
from collections import defaultdict
import threading
import time

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider, ReadableSpan
from opentelemetry.sdk.trace.export import SpanExporter, SpanExportResult
from opentelemetry.sdk.resources import Resource

logger = logging.getLogger(__name__)


class MetricsCollector(SpanExporter):
    """
    Custom span exporter that collects metrics from OpenTelemetry spans.
    Stores metrics in memory for dashboard display.
    """

    def __init__(self, max_history: int = 1000):
        """
        Initialize metrics collector.

        Args:
            max_history: Maximum number of spans to keep in history
        """
        self.max_history = max_history
        self.spans: List[Dict] = []
        self.metrics: Dict = {
            "total_requests": 0,
            "total_tokens": 0,
            "total_latency_ms": 0,
            "error_count": 0,
            "operations": defaultdict(int),
            "models_used": defaultdict(int),
            "hourly_requests": defaultdict(int),
            "latency_by_operation": defaultdict(list),
        }
        self.lock = threading.Lock()
        logger.info("MetricsCollector initialized")

######
######
######

Ensuring Reliability: The Docling Factory Test Suite

Building a complex document processing pipeline requires more than just functional code; it requires a rigorous testing framework to ensure that every component — from OCR to vector search — performs predictably under pressure. I’ve implemented a comprehensive suite of unit tests to validate the core logic of the factory.
Document Parsing Validation
The reliability of our ingestion starts with test_docling_parser.py. This test suite ensures that the DoclingParser correctly identifies and processes various file formats and OCR requirements:

Initialization & Formats: Verifies that the parser correctly initializes with or without GPU support and correctly identifies supported input extensions like .pdf, .docx, and .txt.
OCR Logic: Tests the validation of different OCR engines (RapidOCR, EasyOCR, etc.) and ensures that the system fails over to a default engine if an invalid one is specified.
Functional Success: Uses mocking of the DocumentConverter to simulate successful document exports to Markdown and JSON without requiring a full local setup.
Error Handling: Validates that the system gracefully handles “file not found” scenarios and unsupported file types.

RAG & Embedding Accuracy

To ensure our “chat with documents” feature doesn’t hallucinate or fail, test_rag_engine.py provides targeted testing for the retrieval pipeline:

Connectivity Checks: Includes health checks to verify that both the OpenSearch cluster and the Ollama model server are reachable.
Indexing Pipeline: Tests the workflow of splitting text into chunks and generating 384-dimensional embeddings (standard for models like all-minilm).
Semantic Search: Simulates search queries to verify that OpenSearch returns the expected hits and scores.
Embedding & LLM Classes: Provides isolated unit tests for OllamaEmbeddings and OllamaLLM to confirm they correctly format prompts and return responses with the right temperature settings.

# test_rag_engine.py
"""
Unit tests for rag_engine.py module
"""

import unittest
import os
import tempfile
from unittest.mock import Mock, patch, MagicMock
import sys

# Add parent directory to path
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))


class TestRAGEngine(unittest.TestCase):
    """Test cases for RAGEngine class"""

    def setUp(self):
        """Set up test fixtures"""
        self.test_dir = tempfile.mkdtemp()

    def tearDown(self):
        """Clean up test fixtures"""
        import shutil
        if os.path.exists(self.test_dir):
            shutil.rmtree(self.test_dir)

    @patch('rag_engine.OpenSearch')
    @patch('rag_engine.ollama')
    def test_initialization(self, mock_ollama, mock_opensearch):
        """Test RAG engine initialization"""
        from rag_engine import RAGEngine

        rag = RAGEngine(
            opensearch_host="localhost",
            opensearch_port=9200,
            embedding_model="test-embedding",
            llm_model="test-llm",
            enable_tracing=False
        )

        self.assertIsNotNone(rag)
        self.assertEqual(rag.embedding_model, "test-embedding")
        self.assertEqual(rag.llm_model, "test-llm")

    @patch('rag_engine.OpenSearch')
    @patch('rag_engine.ollama')
    def test_health_check(self, mock_ollama, mock_opensearch):
        """Test health check functionality"""
        from rag_engine import RAGEngine

        # Mock OpenSearch client
        mock_client = MagicMock()
        mock_client.ping.return_value = True
        mock_opensearch.return_value = mock_client

        # Mock Ollama client
        mock_ollama_client = MagicMock()
        mock_response = MagicMock()
        mock_response.models = []
        mock_ollama_client.list.return_value = mock_response
        mock_ollama.Client.return_value = mock_ollama_client

        rag = RAGEngine(enable_tracing=False)
        health = rag.health_check()

        self.assertIsInstance(health, dict)
        self.assertIn('opensearch', health)

    @patch('rag_engine.OpenSearch')
    @patch('rag_engine.ollama')
    def test_index_document(self, mock_ollama, mock_opensearch):
        """Test document indexing"""
        from rag_engine import RAGEngine

        # Mock OpenSearch
        mock_client = MagicMock()
        mock_client.indices.exists.return_value = True
        mock_opensearch.return_value = mock_client

        # Mock Ollama embeddings
        mock_ollama.embeddings.return_value = {
            'embedding': [0.1] * 384
        }

        rag = RAGEngine(enable_tracing=False)

        result = rag.index_document(
            file_path="test.pdf",
            content="Test content for indexing",
            metadata={"source": "test"}
        )

        self.assertIsInstance(result, dict)
        self.assertIn('chunks_indexed', result)

    @patch('rag_engine.OpenSearch')
    @patch('rag_engine.ollama')
    def test_search(self, mock_ollama, mock_opensearch):
        """Test semantic search"""
        from rag_engine import RAGEngine

        # Mock OpenSearch search response
        mock_client = MagicMock()
        mock_client.search.return_value = {
            'hits': {
                'hits': [
                    {
                        '_source': {
                            'content': 'Test result',
                            'file_path': 'test.pdf'
                        },
                        '_score': 0.9
                    }
                ]
            }
        }
        mock_opensearch.return_value = mock_client

        # Mock Ollama embeddings
        mock_ollama.embeddings.return_value = {
            'embedding': [0.1] * 384
        }

        rag = RAGEngine(enable_tracing=False)
        results = rag.search("test query", top_k=5)

        self.assertIsInstance(results, list)

    @patch('rag_engine.OpenSearch')
    @patch('rag_engine.ollama')
    def test_get_stats(self, mock_ollama, mock_opensearch):
        """Test getting index statistics"""
        from rag_engine import RAGEngine

        # Mock OpenSearch stats
        mock_client = MagicMock()
        mock_client.count.return_value = {'count': 100}
        mock_client.indices.stats.return_value = {
            'indices': {
                'documents': {
                    'primaries': {
                        'store': {'size_in_bytes': 1024000}
                    }
                }
            }
        }
        mock_opensearch.return_value = mock_client

        rag = RAGEngine(enable_tracing=False)
        stats = rag.get_stats()

        self.assertIsInstance(stats, dict)
        self.assertIn('total_chunks', stats)


class TestOllamaEmbeddings(unittest.TestCase):
    """Test cases for OllamaEmbeddings class"""

    @patch('rag_engine.ollama')
    def test_embed_documents(self, mock_ollama):
        """Test embedding multiple documents"""
        from rag_engine import OllamaEmbeddings

        mock_ollama.embeddings.return_value = {
            'embedding': [0.1] * 384
        }

        embedder = OllamaEmbeddings(model="test-model")
        texts = ["text1", "text2", "text3"]
        embeddings = embedder.embed_documents(texts)

        self.assertIsInstance(embeddings, list)
        self.assertEqual(len(embeddings), 3)

    @patch('rag_engine.ollama')
    def test_embed_query(self, mock_ollama):
        """Test embedding a single query"""
        from rag_engine import OllamaEmbeddings

        mock_ollama.embeddings.return_value = {
            'embedding': [0.1] * 384
        }

        embedder = OllamaEmbeddings(model="test-model")
        embedding = embedder.embed_query("test query")

        self.assertIsInstance(embedding, list)
        self.assertEqual(len(embedding), 384)


class TestOllamaLLM(unittest.TestCase):
    """Test cases for OllamaLLM class"""

    @patch('rag_engine.ollama')
    def test_generate(self, mock_ollama):
        """Test LLM text generation"""
        from rag_engine import OllamaLLM

        mock_ollama.generate.return_value = {
            'response': 'Generated response'
        }

        llm = OllamaLLM(model="test-model")
        response = llm.generate("test prompt")

        self.assertIsInstance(response, str)
        self.assertEqual(response, 'Generated response')

    @patch('rag_engine.ollama')
    def test_generate_with_temperature(self, mock_ollama):
        """Test LLM generation with temperature parameter"""
        from rag_engine import OllamaLLM

        mock_ollama.generate.return_value = {
            'response': 'Generated response'
        }

        llm = OllamaLLM(model="test-model", temperature=0.7)
        response = llm.generate("test prompt")

        self.assertIsInstance(response, str)


if __name__ == '__main__':
    unittest.main()

# Made with Bob

Metrics and Observability

Finally, the test_metrics_collector.py suite ensures that our observability dashboard isn't just showing pretty numbers, but accurate data:

Span Management: Validates the export of OpenTelemetry spans, ensuring that operation names, start times, and end times are captured correctly.
Aggregation Logic: Tests the collector’s ability to sum up token usage (prompt vs. completion) and calculate error counts across different operations.
Data Structures: Verifies that time-series data for hourly requests and latency percentiles are generated in the correct format for the Plotly dashboard.

By maintaining these tests, the application ensure that by adding new features — like new OCR backends or remote LLM providers via LiteLLM — the core foundation of the Docling Factory remains rock solid.

Flexible Processing: From Single Files to Batch Operations

The Docling Factory is engineered for maximum versatility, allowing users to choose between processing individual documents or executing high-volume batch operations. For targeted tasks, the parse_single_file function handles specific uploads via the Gradio interface, while the parse_batch method in the DoclingParser class enables the system to traverse entire directories and convert multiple files in a single pass. Whether you are indexing a single technical manual or a massive archive of financial reports, the architecture scales to meet the demand, ensuring consistent output across all supported formats like Markdown, JSON, and HTML.

CPU or GPU?

To accommodate varying hardware environments, the application is designed to operate seamlessly with or without GPU acceleration. The DoclingParser class includes a use_gpu initialization parameter that toggles the processing backend, and the deployment infrastructure provides dedicated Docker images for both CPU-only and GPU-enabled configurations. This flexibility ensures that the factory can be deployed on standard local machines using CPU-based OCR engines like RapidOCR, or on high-performance servers leveraging NVIDIA GPUs to significantly accelerate the layout analysis and document conversion pipeline.

Cloud Deployment

The Docling Factory features a production-ready cloud deployment architecture built on Kubernetes, supporting highly available and scalable environments. The infrastructure is designed with a multi-container strategy, utilizing Docker for both CPU-optimized and GPU-accelerated versions of the application. To ensure operational stability, the deployment includes:

Orchestration and Scaling: Kubernetes deployments manage multiple replicas of the application, with specific resource limits (up to 8Gi memory and 4000m CPU for GPU nodes) and health checks for both liveness and readiness.
Infrastructure Components: Integrated services include OpenSearch for vector storage, a dedicated LiteLLM gateway **with a PostgreSQL backend for managed LLM access, and an **Ingress controller for secure external traffic management via TLS.

apiVersion: apps/v1
kind: Deployment
metadata:
  name: opensearch
  namespace: docling-factory
  labels:
    app: opensearch
spec:
  replicas: 1
  selector:
    matchLabels:
      app: opensearch
  template:
    metadata:
      labels:
        app: opensearch
    spec:
      containers:
      - name: opensearch
        image: opensearchproject/opensearch:2.11.0
        ports:
        - containerPort: 9200
          name: http
        - containerPort: 9600
          name: performance
        env:
        - name: discovery.type
          value: "single-node"
        - name: OPENSEARCH_JAVA_OPTS
          value: "-Xms512m -Xmx512m"
        - name: DISABLE_SECURITY_PLUGIN
          value: "true"
        - name: DISABLE_INSTALL_DEMO_CONFIG
          value: "true"
        resources:
          requests:
            memory: "1Gi"
            cpu: "500m"
          limits:
            memory: "2Gi"
            cpu: "1000m"
        volumeMounts:
        - name: opensearch-data
          mountPath: /usr/share/opensearch/data
        livenessProbe:
          httpGet:
            path: /_cluster/health
            port: 9200
          initialDelaySeconds: 60
          periodSeconds: 30
          timeoutSeconds: 10
        readinessProbe:
          httpGet:
            path: /_cluster/health
            port: 9200
          initialDelaySeconds: 30
          periodSeconds: 10
          timeoutSeconds: 5
      volumes:
      - name: opensearch-data
        persistentVolumeClaim:
          claimName: opensearch-data-pvc
---
apiVersion: v1
kind: Service
metadata:
  name: opensearch-service
  namespace: docling-factory
  labels:
    app: opensearch
spec:
  type: ClusterIP
  ports:
  - port: 9200
    targetPort: 9200
    protocol: TCP
    name: http
  - port: 9600
    targetPort: 9600
    protocol: TCP
    name: performance
  selector:
    app: opensearch
---
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
  name: opensearch-data-pvc
  namespace: docling-factory
spec:
  accessModes:
    - ReadWriteOnce
  resources:
    requests:
      storage: 10Gi
  storageClassName: standard

# Made with Bob

Persistent Storage: Robust data management is handled through several Persistent Volume Claims (PVCs), providing up to 50Gi for document output and separate storage for input and logs.

apiVersion: v1
kind: PersistentVolumeClaim
metadata:
  name: docling-factory-input-pvc
  namespace: docling-factory
  labels:
    app: docling-factory
spec:
  accessModes:
    - ReadWriteMany
  resources:
    requests:
      storage: 10Gi
  storageClassName: standard
---
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
  name: docling-factory-output-pvc
  namespace: docling-factory
  labels:
    app: docling-factory
spec:
  accessModes:
    - ReadWriteMany
  resources:
    requests:
      storage: 50Gi
  storageClassName: standard
---
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
  name: docling-factory-logs-pvc
  namespace: docling-factory
  labels:
    app: docling-factory
spec:
  accessModes:
    - ReadWriteMany
  resources:
    requests:
      storage: 5Gi
  storageClassName: standard

# Made with Bob

GPU Integration: Specialized Kubernetes manifests utilize nodeSelector and https://nvidia.com/gpu resource requests to target nodes with NVIDIA hardware, ensuring high-performance document parsing and layout analysis in cloud environments.
CPU Deployment

# Docling Factory - CPU Version (Optimized Build)
# Multi-stage build with better caching and faster builds

FROM python:3.11-slim as builder

# Set working directory
WORKDIR /app

# Install system dependencies in one layer
RUN apt-get update && apt-get install -y --no-install-recommends \
    build-essential \
    git \
    curl \
    && rm -rf /var/lib/apt/lists/*

# Copy only requirements first for better caching
COPY requirements.txt .

# Install dependencies in stages for better caching
# Stage 1: Install lightweight dependencies first
RUN pip install --no-cache-dir --user \
    numpy>=1.24.0,<2.0.0 \
    pandas>=2.0.0 \
    pillow>=10.0.0 \
    python-docx>=1.1.0 \
    PyPDF2>=3.0.0 \
    python-dateutil>=2.8.2 \
    pathlib>=1.0.1 \
    lxml>=4.9.0 \
    openpyxl>=3.1.0

# Stage 2: Install medium-weight dependencies
RUN pip install --no-cache-dir --user \
    gradio>=4.0.0 \
    streamlit>=1.31.0 \
    plotly>=5.18.0 \
    httpx>=0.24.0 \
    tiktoken>=0.5.0

# Stage 3: Install OpenSearch and LangChain
RUN pip install --no-cache-dir --user \
    opensearch-py>=2.4.0 \
    langchain>=0.1.0 \
    langchain-community>=0.0.20

# Stage 4: Install LLM clients (lightweight)
RUN pip install --no-cache-dir --user \
    ollama>=0.1.0 \
    litellm>=1.0.0

# Stage 5: Install OpenTelemetry (lightweight)
RUN pip install --no-cache-dir --user \
    opentelemetry-api>=1.20.0 \
    opentelemetry-sdk>=1.20.0 \
    opentelemetry-instrumentation>=0.41b0 \
    traceloop-sdk>=0.30.0

# Stage 6: Install Docling (can be slow)
RUN pip install --no-cache-dir --user \
    docling>=2.0.0 \
    docling-core>=2.0.0 \
    docling-parse>=2.0.0 \
    python-xbrl>=1.1.1

# Stage 7: Install CV dependencies (slowest - do last)
RUN pip install --no-cache-dir --user \
    opencv-python-headless>=4.8.0,<5.0.0 \
    scikit-image>=0.21.0 \
    pytesseract>=0.3.10

# Stage 8: Install PyTorch (very slow - separate for better error handling)
RUN pip install --no-cache-dir --user \
    torch>=2.0.0 \
    torchvision>=0.15.0 \
    || echo "PyTorch installation failed, continuing..."

# Stage 9: Install ML dependencies (slow)
RUN pip install --no-cache-dir --user \
    sentence-transformers>=2.2.0 \
    faiss-cpu>=1.7.4 \
    chromadb>=0.4.0 \
    || echo "ML dependencies installation failed, continuing..."

# Stage 10: Install EasyOCR last (slowest)
RUN pip install --no-cache-dir --user \
    easyocr>=1.7.0 \
    || echo "EasyOCR installation failed, continuing..."

# Final stage
FROM python:3.11-slim

# Set working directory
WORKDIR /app

# Install runtime dependencies
RUN apt-get update && apt-get install -y --no-install-recommends \
    libgomp1 \
    libglib2.0-0 \
    libsm6 \
    libxext6 \
    libxrender-dev \
    libgl1-mesa-glx \
    && rm -rf /var/lib/apt/lists/*

# Copy Python packages from builder
COPY --from=builder /root/.local /root/.local

# Make sure scripts in .local are usable
ENV PATH=/root/.local/bin:$PATH

# Copy application files
COPY docling_parser.py .
COPY app_enhanced.py .
COPY rag_engine.py .
COPY metrics_collector.py .
COPY standalone_dashboard.py .
COPY metrics_dashboard.py .

# Create necessary directories
RUN mkdir -p input output output/figures logs

# Expose Gradio port
EXPOSE 7860

# Set environment variables
ENV GRADIO_SERVER_NAME=0.0.0.0
ENV GRADIO_SERVER_PORT=7860
ENV PYTHONUNBUFFERED=1
ENV OPENSEARCH_HOST=opensearch
ENV OPENSEARCH_PORT=9200
ENV OLLAMA_BASE_URL=http://host.docker.internal:11434
ENV LITELLM_API_BASE=http://litellm:4000
ENV LITELLM_API_KEY=sk-1234

# Health check
HEALTHCHECK --interval=30s --timeout=10s --start-period=40s --retries=3 \
    CMD python -c "import requests; requests.get('http://localhost:7860')" || exit 1

# Run the application
CMD ["python", "app_enhanced.py"]

GPU Deployment

# Docling Factory - GPU Version with LiteLLM Support
# Multi-stage build for optimized image size with CUDA support

FROM nvidia/cuda:12.1.0-runtime-ubuntu22.04 as builder

# Set working directory
WORKDIR /app

# Install Python and system dependencies
RUN apt-get update && apt-get install -y --no-install-recommends \
    python3.11 \
    python3.11-dev \
    python3-pip \
    build-essential \
    git \
    && rm -rf /var/lib/apt/lists/*

# Create symbolic links for python
RUN ln -sf /usr/bin/python3.11 /usr/bin/python && \
    ln -sf /usr/bin/python3.11 /usr/bin/python3

# Copy requirements file
COPY requirements-gpu.txt .

# Install Python dependencies
RUN pip install --no-cache-dir --user -r requirements-gpu.txt

# Final stage
FROM nvidia/cuda:12.1.0-runtime-ubuntu22.04

# Set working directory
WORKDIR /app

# Install Python runtime
RUN apt-get update && apt-get install -y --no-install-recommends \
    python3.11 \
    python3-pip \
    libgomp1 \
    && rm -rf /var/lib/apt/lists/*

# Create symbolic links for python
RUN ln -sf /usr/bin/python3.11 /usr/bin/python && \
    ln -sf /usr/bin/python3.11 /usr/bin/python3

# Copy Python packages from builder
COPY --from=builder /root/.local /root/.local

# Make sure scripts in .local are usable
ENV PATH=/root/.local/bin:$PATH

# Copy application files
COPY docling_parser.py .
COPY app_enhanced.py .
COPY rag_engine.py .
COPY metrics_collector.py .
COPY standalone_dashboard.py .
COPY metrics_dashboard.py .

# Create necessary directories
RUN mkdir -p input output output/figures logs

# Expose Gradio port
EXPOSE 7860

# Set environment variables
ENV GRADIO_SERVER_NAME=0.0.0.0
ENV GRADIO_SERVER_PORT=7860
ENV PYTHONUNBUFFERED=1
ENV CUDA_VISIBLE_DEVICES=0
ENV OPENSEARCH_HOST=opensearch
ENV OPENSEARCH_PORT=9200
ENV OLLAMA_BASE_URL=http://host.docker.internal:11434
ENV LITELLM_API_BASE=http://litellm:4000
ENV LITELLM_API_KEY=sk-1234

# Health check
HEALTHCHECK --interval=30s --timeout=10s --start-period=40s --retries=3 \
    CMD python -c "import requests; requests.get('http://localhost:7860')" || exit 1

# Run the application
CMD ["python", "app_enhanced.py"]

apiVersion: apps/v1
kind: Deployment
metadata:
  name: docling-factory-gpu
  namespace: docling-factory
  labels:
    app: docling-factory
    version: gpu
spec:
  replicas: 1
  selector:
    matchLabels:
      app: docling-factory
      version: gpu
  template:
    metadata:
      labels:
        app: docling-factory
        version: gpu
    spec:
      containers:
      - name: docling-factory
        image: docling-factory:gpu-latest
        imagePullPolicy: IfNotPresent
        ports:
        - containerPort: 7860
          name: http
          protocol: TCP
        envFrom:
        - configMapRef:
            name: docling-factory-config
        env:
        - name: CUDA_VISIBLE_DEVICES
          value: "0"
        resources:
          requests:
            memory: "4Gi"
            cpu: "2000m"
            nvidia.com/gpu: 1
          limits:
            memory: "8Gi"
            cpu: "4000m"
            nvidia.com/gpu: 1
        volumeMounts:
        - name: input-storage
          mountPath: /app/input
        - name: output-storage
          mountPath: /app/output
        - name: logs-storage
          mountPath: /app/logs
        livenessProbe:
          httpGet:
            path: /
            port: 7860
          initialDelaySeconds: 60
          periodSeconds: 30
          timeoutSeconds: 10
          failureThreshold: 3
        readinessProbe:
          httpGet:
            path: /
            port: 7860
          initialDelaySeconds: 30
          periodSeconds: 10
          timeoutSeconds: 5
          failureThreshold: 3
      volumes:
      - name: input-storage
        persistentVolumeClaim:
          claimName: docling-factory-input-pvc
      - name: output-storage
        persistentVolumeClaim:
          claimName: docling-factory-output-pvc
      - name: logs-storage
        persistentVolumeClaim:
          claimName: docling-factory-logs-pvc
      nodeSelector:
        accelerator: nvidia-gpu
      tolerations:
      - key: nvidia.com/gpu
        operator: Exists
        effect: NoSchedule

# Made with Bob

The “End of Level 1” Log

So, here I am — about two months into this rabbit hole, and I’m still standing, mostly thanks to the digital wizardry of IBM Bob. Is the Docling Factory “finished”? In the words of every dev ever: Hell no. I’m currently in that glorious “it works on my machine” phase, but the quest log is still overflowing. I’ve got to stress-test the GPU until the fans scream, feed it files so large they have their own gravity, and ensure the unit tests cover more than just the “happy path”. I’m also dreaming of a universal Cloud Object Storage (COS) implementation to make file repo handling truly elite. This is just a fragment of the “to-do” list currently living rent-free in my brain. The code is out there on GitHub, and frankly, I’d love some co-op players for this raid — carrying this entire multimodal stack on my own two shoulders is getting a bit “Atlas-tier” heavy. Pull requests are open, the coffee is brewing, and the evolution has only just begun!

Thanks for reading 🤠

Mamba vs. Transformers: Architecture Comparison

Alain Airom (Ayrom) — Thu, 30 Apr 2026 11:58:36 +0000

My latest studies and understandings of Mamba versus Transformers!

The Core Architectures: Transformers and Mamba

Transformers operate on the principle of Self-Attention, which allows the model to process all tokens in a sequence simultaneously to understand their relationships. By calculating pairwise interactions between every word in a sentence, the model can capture complex, long-range dependencies with high precision. However, this comprehensive “look-at-everything” approach comes with a quadratic computational cost (O(L2)), meaning that as the input length increases, the memory and processing power required grow exponentially. This architecture is the gold standard for tasks requiring deep reasoning and precise data retrieval, though it often struggles with efficiency when handling extremely long documents or real-time streaming.

Mamba introduces a more efficient alternative based on Selective State Space Models (SSMs), which process data linearly (O(L)) rather than quadratically. Unlike the Transformer’s massive memory cache, Mamba utilizes a fixed-size hidden state that acts as a compressed version of the entire preceding sequence, allowing it to maintain performance even as the context grows. Its “selective” mechanism enables the model to dynamically decide which information is important to keep and what can be discarded, effectively solving the traditional memory bottlenecks of older recurrent models. This makes Mamba particularly powerful for high-throughput applications, long-context analysis, and environments where fast, constant-time inference is a priority.

Mamba vs. Traditional Transformers
The primary difference between Mamba and traditional Transformers lies in how they process sequence data. While Transformers rely on the Self-Attention mechanism to “attend” to all previous tokens simultaneously, Mamba utilizes S*elective State Space Models (SSMs)* to maintain a compressed, evolving internal state (Mukhammadiev, 2026; IBM, 2025).

Key Architectural Differences

| Feature             | Traditional Transformer                                     | Mamba (Selective SSM)                                      |
| ------------------- | ----------------------------------------------------------- | ---------------------------------------------------------- |
| **Core Mechanism**  | Self-Attention (every token looks at every other token)     | Selective State Updates (recurrent internal state)         |
| **Complexity**      | **Quadratic** *O*(*L*2)—cost quadruples if sequence doubles | **Linear** *O*(*L*)—cost scales proportionally with length |
| **Memory**          | High (grows with context due to KV Cache)                   | Low (fixed-size "hidden state")                            |
| **Best For**        | Precise retrieval and few-shot reasoning                    | Long context, real-time streaming, and high throughput     |
| **Inference Speed** | Slower as sequences get longer                              | Extremely fast and constant per-token latency              |

Traditional Transformers

Transformers process an entire sequence at once during training. However, during generation (inference), they must store a KV Cache (Key-Value Cache) of all previous tokens to avoid recomputing them, which leads to massive memory consumption for long documents (Mukhammadiev, 2026; Duhan, 2026).

Conceptual Code (Simplified Self-Attention):

import torch
import torch.nn.functional as F

def transformer_attention(query, key, value):
    # query, key, value shape: [batch, seq_len, dim]
    # 1. Compute scores (Every token vs Every token)
    scores = torch.matmul(query, key.transpose(-2, -1)) / (query.size(-1)**0.5)

    # 2. Apply Softmax (Quadratic memory/compute)
    weights = F.softmax(scores, dim=-1)

    # 3. Weighted sum of values
    return torch.matmul(weights, value)

Mamba (Selective SSM)

Mamba acts more like a “smart” Recurrent Neural Network (RNN). It doesn’t look back at every token; instead, it updates a fixed-size hidden state that acts as a compressed memory (Duhan, 2026; IBM, 2025). Mamba’s “Selection” mechanism allows it to decide which information to keep or discard based on the current input, solving the “forgetting” problem of older RNNs (Mukhammadiev, 2026; OpenReview).

Conceptual Code (Simplified SSM Update):

def mamba_ssm_step(x_t, hidden_state, A, B, C, delta):
    # x_t: current token input
    # 1. Discretization (making continuous math digital)
    # delta is input-dependent, allowing "selection"
    dA = torch.exp(delta * A)
    dB = (delta * B)

    # 2. Update the internal "memory" (Hidden State)
    # This is Linear O(L) because it's a step-by-step update
    new_hidden = dA * hidden_state + dB * x_t

    # 3. Generate output
    y_t = C * new_hidden
    return y_t, new_hidden

Why Mamba Matters?

In 2026, Mamba-3 has emerged as a production-ready alternative to Transformers, offering up to 5x higher throughput for long-document tasks (Duhan, 2026; VentureBeat, 2026). While Transformers still excel at exact data retrieval (like a database), Mamba is superior for “inference-first” applications where speed and memory efficiency are critical (Mukhammadiev, 2026; Duhan, 2026).

IBM and Mamba!

As of April 2026, IBM has integrated Mamba architecture into its flagship Granite family and specialized Biomedical Foundation Models to improve efficiency and performance for long-sequence tasks.

IBM Granite 4.0 & 4.1 Family

IBM introduced a Hybrid Mamba-2/Transformer architecture in its Granite 4.0 and 4.1 models. This design combines the efficient state-tracking of Mamba with the precision of standard Transformers, leading to a 70% reduction in GPU memory needs and 2x faster inference.

> Disclaimer: based on my own research!

Key Granite models using this architecture include:

Granite 4.1 LLMs: Granite 4.1 models use a decoder-only dense transformer architecture. The core design choices include Grouped Query Attention (GQA), Rotary Position Embeddings (RoPE), SwiGLU activations, RMSNorm, and shared input/output embeddings. (https://huggingface.co/blog/ibm-granite/granite-4-1)
Granite-H-Small: A 32B parameter model (9B active) designed for heavy-duty enterprise tasks.
Granite-H-Tiny: A 7B parameter model (1B active) optimized for low-latency, high-volume tasks.
Granite-H-Micro: A 3B parameter dense model tailored for local and edge deployments.
Granite-H-Nano: Compact variants (350M and 1B parameters) designed for on-device applications.

IBM Research Biomedical Models

IBM Research utilizes Mamba blocks in specialized models to handle the linear complexity required for genomics and molecular data.

biomed.omics.bl.sm.ma-ted-458m: A multi-domain, sequence-based model trained on biologics, small molecules, and single-cell RNA-seq data.
Bamba: A research hybrid model developed to fuse SSM efficiency with Transformer-level accuracy for large-scale scientific modeling.
Genomics Foundation Models: IBM has analyzed Mamba’s performance in processing over 1 million tokens in genomic sequences, outperforming traditional baselines

Model Comparison Summary

| Model Series     | Type                     | Primary Benefit                                              |
| ---------------- | ------------------------ | ------------------------------------------------------------ |
| **Granite-H**    | Hybrid Mamba/Transformer | Reduces GPU memory by 70% and doubles inference speed.       |
| **Biomed Omics** | Specialized SSM          | Scales linearly to handle extremely long DNA and protein sequences. |
| **Bamba**        | Research Hybrid          | Combines Transformer accuracy with Mamba's linear scaling for scientific discovery. |

That’s a wrap! 🫔

Links, References and Citations

OpenReview: Mamba — Linear-Time Sequence Modeling with Selective State Spaces: https://openreview.net/forum?id=AL1fq05o7H (by Tri Dao and Albert Gu introducing the Selective SSM architecture.)
What is a Mamba Model: https://www.ibm.com/think/topics/mamba-model
Mamba vs Transformer: The Real Shift in AI Architecture (2026): https://medium.com/@uzbrainai/mamba-vs-transformer-the-real-shift-in-ai-architecture-2026-bf758ba278ec (Mukhammadiev, 2026)
New Mamba-3 AI Model Beats Transformers by 4%, Runs 7x Faster: https://winbuzzer.com/2026/03/18/open-source-mamba-3-arrives-to-surpass-transformer-xcxwbn/
The Architecture Wars Are Back: Mamba-3 Challenges Transformers While Nvidia Fights to Keep Them Alive: https://dev.to/taskconcierge/the-architecture-wars-are-back-mamba-3-challenges-transformers-while-nvidia-fights-to-keep-them-fag
Open source Mamba 3 arrives to surpass Transformer architecture with nearly 4% improved language modeling, reduced latency: https://venturebeat.com/technology/open-source-mamba-3-arrives-to-surpass-transformer-architecture-with-nearly
Mamba-3 vs Transformers: Which Architecture Should You Build On?: https://medium.com/@harshduhan/mamba-3-vs-transformers-which-architecture-should-you-build-on-f24cd52b482f

No More Automation from Scratch: Automate and Build Your Terraform with Draw.io Diagrams

Alain Airom (Ayrom) — Tue, 28 Apr 2026 12:44:46 +0000

Creating Terraform Scripts from Draw.io XML Files!

Introduction

The concept and the idea originated from a realization that drawings from open-source tools, such as Draw.io, are essentially structured XML files. If the diagram explicitly maps out an infrastructure architecture, it would be logical to analyze that XML file and programmatically generate corresponding Terraform scripts.

> Disclaimer: note that this application is currently in its alpha stage; as an early-access release, it requires more rigorous, “bulletproof” testing and further functional enhancements before it reaches full maturity. This tool marks the first in a planned series of automation utilities I am developing to significantly accelerate my personal productivity and streamline technical workflows.

A word on the external tool used: Draw.io

The core of this application’s utility is its deep integration with Draw.io, a premier open-source diagramming tool. Rather than forcing users into a proprietary design environment, the generator leverages Draw.io’s flexible, web-based interface and its extensive library of cloud architecture shapes. By treating the exported Draw.io XML as a structured data source, the application can precisely map visual components and their interconnections to valid Terraform resource blocks. This heavy reliance on an open-source standard ensures that users can design their infrastructure using a familiar, free, and highly accessible tool, while the backend handles the complex translation from a visual drawing to production-ready code.

Terraform Script Generator: A Diagram-to-Code Pipeline

The Terraform Script Generator is a web-based utility designed to streamline the translation of cloud infrastructure diagrams, specifically those created in Draw.io, into deployable Terraform configurations. This automation bridges the gap between architectural planning and infrastructure implementation, particularly for Cloud platforms like IBM (but not only), but adaptable to other platforms as well. The generator employs a client-server architecture, utilizing a user-friendly frontend to receive design inputs and a robust backend to handle the analytical heavy lifting.

terraform-builder/
│
├── backend/                      # Backend Python modules
│   ├── app.py                   # Flask API server
│   ├── drawio_parser.py         # XML parser for Draw.io
│   └── terraform_generator.py   # Terraform code generator
│
├── frontend/                     # Frontend web files
│   ├── index.html               # Main UI
│   ├── styles.css               # Styling
│   └── app.js                   # Client-side logic
│
├── k8s/                         # Kubernetes deployment files
│   ├── deployment.yaml          # Deployment and Service manifests
│   ├── configmap.yaml           # Application configuration
│   └── README.md                # Kubernetes deployment guide
│
├── Docs/                        # Documentation
│   └── Architecture.md          # This file
│
├── scripts/                     # Utility scripts
│   ├── start.sh                # Start application
│   ├── stop.sh                 # Stop application
│   └── validate_terraform.py   # Terraform validation
│
├── _diagrams/                   # Example diagrams
│   └── d1.drawio               # Sample IBM Cloud diagram
│
├── _images/                     # Project images
├── _sources/                    # Reference sources
│
├── Dockerfile                   # Docker image definition
├── requirements.txt             # Python dependencies
├── .gitignore                  # Git ignore rules
└── README.md                   # Project documentation

The Backend: Analytical Engine with Flask and Python

At the heart of the system is a Python backend built with the Flask web framework. This application is responsible for orchestrating the overall process, from serving the main user interface to exposing REST API endpoints for diagram parsing and code generation. It integrates two key internal modules: DrawioParser and TerraformGenerator. The application logic is designed for extensibility, capable of handling complex diagrams and multi-stage processing, supporting features like file uploading, health checks, and state management via JSON for multi-step workflows.

# app.py - Main Flask application for the Terraform Builder
"""
Flask Backend for Terraform Builder Application
Provides API endpoints for parsing Draw.io diagrams and generating Terraform code
"""
from flask import Flask, request, jsonify, send_from_directory
from flask_cors import CORS
import os
from drawio_parser import DrawioParser
from terraform_generator import TerraformGenerator

app = Flask(__name__, static_folder='../frontend', static_url_path='')
CORS(app)

# Initialize parsers and generators
parser = DrawioParser()
generator = TerraformGenerator()


@app.route('/')
def index():
    """Serve the main application page"""
    return send_from_directory(app.static_folder, 'index.html')


@app.route('/api/parse', methods=['POST'])
def parse_diagram():
    """Parse a Draw.io diagram and return components"""
    try:
        if 'file' in request.files:
            # Handle file upload
            file = request.files['file']
            if file.filename == '':
                return jsonify({'error': 'No file selected'}), 400

            # Save temporarily and parse
            temp_path = '/tmp/temp_diagram.drawio'
            file.save(temp_path)

            result = parser.parse_file(temp_path)
            os.remove(temp_path)

        elif 'content' in request.json:
            # Handle XML content
            content = request.json['content']
            result = parser.parse_content(content)
        else:
            return jsonify({'error': 'No diagram data provided'}), 400

        return jsonify({
            'success': True,
            'components': result['components'],
            'connections': result['connections'],
            'summary': parser.get_resource_summary()
        })

    except Exception as e:
        return jsonify({'error': str(e)}), 500


@app.route('/api/generate', methods=['POST'])
def generate_terraform():
    """Generate Terraform code from parsed components"""
    try:
        data = request.json
        components = data.get('components', [])
        provider = data.get('provider', 'ibm')

        if not components:
            return jsonify({'error': 'No components provided'}), 400

        # Create generator with selected provider
        gen = TerraformGenerator(provider=provider)

        # Generate Terraform files
        terraform_files = gen.generate_from_components(components)

        return jsonify({
            'success': True,
            'files': terraform_files
        })

    except Exception as e:
        return jsonify({'error': str(e)}), 500


@app.route('/api/parse-and-generate', methods=['POST'])
def parse_and_generate():
    """Parse diagram and generate Terraform code in one step"""
    try:
        provider = 'ibm'  # Default provider
        custom_provider_url = None

        if 'file' in request.files:
            # Handle file upload
            file = request.files['file']
            if file.filename == '':
                return jsonify({'error': 'No file selected'}), 400

            # Save temporarily and parse
            temp_path = '/tmp/temp_diagram.drawio'
            file.save(temp_path)

            result = parser.parse_file(temp_path)
            os.remove(temp_path)

            # Get provider from form data if available
            provider = request.form.get('provider', 'ibm')
            custom_provider_url = request.form.get('custom_provider_url')

        elif 'content' in request.json:
            # Handle XML content
            content = request.json['content']
            provider = request.json.get('provider', 'ibm')
            custom_provider_url = request.json.get('custom_provider_url')
            result = parser.parse_content(content)
        else:
            return jsonify({'error': 'No diagram data provided'}), 400

        # Create generator with selected provider and custom URL if provided
        gen = TerraformGenerator(provider=provider, custom_provider_url=custom_provider_url)

        # Generate Terraform files
        terraform_files = gen.generate_from_components(result['components'])

        response_data = {
            'success': True,
            'components': result['components'],
            'connections': result['connections'],
            'summary': parser.get_resource_summary(),
            'terraform': terraform_files,
            'provider': provider
        }

        # Add custom provider info if used
        if custom_provider_url and gen.custom_provider_config:
            response_data['custom_provider'] = gen.custom_provider_config

        # Add warning if IBM components detected with non-IBM provider
        if provider != 'ibm' and hasattr(gen, 'skipped_resources') and gen.skipped_resources:
            response_data['warning'] = {
                'message': f'This diagram contains {len(gen.skipped_resources)} IBM Cloud-specific component(s) that were not generated for the {provider.upper()} provider.',
                'suggestion': 'Consider selecting "IBM Cloud" as the provider to generate these resources.',
                'skipped_count': len(gen.skipped_resources)
            }

        return jsonify(response_data)

    except Exception as e:
        return jsonify({'error': str(e)}), 500


@app.route('/api/health', methods=['GET'])
def health_check():
    """Health check endpoint"""
    return jsonify({
        'status': 'healthy',
        'service': 'terraform-builder-api'
    })


@app.route('/api/diagram-preview', methods=['POST'])
def diagram_preview():
    """Generate a preview URL for the diagram"""
    try:
        data = request.json
        content = data.get('content', '')

        if not content:
            return jsonify({'error': 'No diagram content provided'}), 400

        # Return the content for client-side rendering
        return jsonify({
            'success': True,
            'content': content,
            'preview_available': True
        })

    except Exception as e:
        return jsonify({'error': str(e)}), 500


@app.route('/api/save-files', methods=['POST'])
def save_files():
    """Save generated Terraform files to timestamped output folder"""
    try:
        from datetime import datetime
        import shutil

        data = request.json
        files = data.get('files', {})
        diagram_name = data.get('diagram_name', 'diagram')

        if not files:
            return jsonify({'error': 'No files provided'}), 400

        # Remove file extension from diagram name
        diagram_name = diagram_name.replace('.drawio', '').replace('.xml', '')

        # Create timestamp
        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')

        # Create output directory structure
        output_base = 'output'
        output_dir = os.path.join(output_base, f"{diagram_name}_{timestamp}")

        # Create directory if it doesn't exist
        os.makedirs(output_dir, exist_ok=True)

        # Save each file
        saved_files = []
        for filename, content in files.items():
            file_path = os.path.join(output_dir, filename)
            with open(file_path, 'w') as f:
                f.write(content)
            saved_files.append(file_path)

        return jsonify({
            'success': True,
            'output_dir': output_dir,
            'files': saved_files,
            'message': f'Files saved to {output_dir}'
        })

    except Exception as e:
        return jsonify({'error': str(e)}), 500


@app.route('/api/save-and-zip-files', methods=['POST'])
def save_and_zip_files():
    """Save generated Terraform files to timestamped output folder and return as ZIP"""
    try:
        from datetime import datetime
        import shutil
        import zipfile
        from io import BytesIO
        from flask import send_file

        data = request.json
        files = data.get('files', {})
        diagram_name = data.get('diagram_name', 'diagram')

        if not files:
            return jsonify({'error': 'No files provided'}), 400

        # Remove file extension from diagram name
        diagram_name = diagram_name.replace('.drawio', '').replace('.xml', '')

        # Create timestamp
        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
        folder_name = f"{diagram_name}_{timestamp}"

        # Create output directory structure
        output_base = 'output'
        output_dir = os.path.join(output_base, folder_name)

        # Create directory if it doesn't exist
        os.makedirs(output_dir, exist_ok=True)

        # Save each file to disk
        for filename, content in files.items():
            file_path = os.path.join(output_dir, filename)
            with open(file_path, 'w') as f:
                f.write(content)

        # Create ZIP file in memory
        zip_buffer = BytesIO()
        with zipfile.ZipFile(zip_buffer, 'w', zipfile.ZIP_DEFLATED) as zip_file:
            for filename, content in files.items():
                # Add files to ZIP with folder structure
                zip_file.writestr(f"{folder_name}/{filename}", content)

        zip_buffer.seek(0)
        zip_name = f"{folder_name}.zip"

        # Create response with ZIP file
        return send_file(
            zip_buffer,
            mimetype='application/zip',
            as_attachment=True,
            download_name=zip_name
        )

    except Exception as e:
        return jsonify({'error': str(e)}), 500


if __name__ == '__main__':
    # Run the Flask app
    port = int(os.environ.get('PORT', 5002))
    debug = os.environ.get('DEBUG', 'True').lower() == 'true'

    print(f"Starting Terraform Builder API on port {port}")
    print(f"Debug mode: {debug}")

    app.run(host='0.0.0.0', port=port, debug=debug)

# Made with Bob

Diagram Interpretation: Parsing the Draw.io XML

Draw.io architecture diagrams are essentially saved as structured XML files. The backend utilizes the standard xml.etree.ElementTree library to read this XML and extract the critical components. It parses the nested mxCell elements to identify individual cloud resources based on specific, predefined shape patterns and styles (e.g., IBM Cloud shapes). This interpreter doesn't just list components; it also detects connections between them to understand the relationships and generate resource summaries, laying the foundation for accurate script generation.

# drawio_parser.py - Parser for Draw.io XML files containing IBM Cloud diagrams
"""
Draw.io XML Parser for IBM Cloud Diagrams
Parses Draw.io XML files and extracts IBM Cloud components
"""
import xml.etree.ElementTree as ET
from typing import Dict, List, Any


class DrawioParser:
    """Parser for Draw.io XML files containing cloud diagrams"""

    # Mapping of Draw.io shapes to generic resource types
    # These will be converted to provider-specific resources by the generator
    CLOUD_SHAPES = {
        # Core Cloud resources
        'mxgraph.ibm_cloud.ibm-cloud': 'resource_group',
        'mxgraph.ibm.box;prType=cloud': 'resource_group',

        # VPC and Networking
        'mxgraph.ibm_cloud.ibm-cloud--vpc': 'vpc',
        'mxgraph.ibm.box;prType=vpc': 'vpc',
        'mxgraph.ibm.vpc': 'vpc',
        'mxgraph.ibm_cloud.ibm-cloud--subnets': 'subnet',
        'mxgraph.ibm_cloud.floating-ip': 'floating_ip',
        'mxgraph.ibm_cloud.load-balancer--vpc': 'load_balancer',
        'mxgraph.ibm_cloud.network--public': 'public_gateway',

        # Compute
        'mxgraph.ibm_cloud.server--proxy': 'compute_instance',
        'mxgraph.ibm_cloud.application--web': 'app_service',

        # Storage and Databases
        'mxgraph.ibm_cloud.data--base': 'database',

        # Watson Services (IBM-specific, will be skipped for other providers)
        'mxgraph.ibm_cloud.ibm-watsonx--orchestrate': 'watson_service',
        'mxgraph.ibm_cloud.watsonx-governance': 'watson_service',
        'mxgraph.ibm_cloud.watsonx-ai': 'watson_service',

        # Zones (for organizational purposes)
        'mxgraph.ibm.box;prType=zone': 'zone',
    }

    # Service names for IBM Cloud resources
    SERVICE_NAMES = {
        'watsonx--orchestrate': 'watsonx-orchestrate',
        'watsonx-governance': 'watsonx-governance',
        'watsonx-ai': 'watsonx-ai',
    }

    def __init__(self):
        self.components = []
        self.connections = []

    def parse_file(self, file_path: str) -> Dict[str, Any]:
        """Parse a Draw.io XML file and extract IBM Cloud components"""
        try:
            tree = ET.parse(file_path)
            root = tree.getroot()

            # Find all mxCell elements
            for diagram in root.findall('.//diagram'):
                model = diagram.find('.//mxGraphModel')
                if model is not None:
                    self._parse_model(model)

            return {
                'components': self.components,
                'connections': self.connections
            }
        except Exception as e:
            raise Exception(f"Error parsing Draw.io file: {str(e)}")

    def parse_content(self, xml_content: str) -> Dict[str, Any]:
        """Parse Draw.io XML content string"""
        try:
            # Log the first 200 characters for debugging
            print(f"[DEBUG] Received XML content (first 200 chars): {xml_content[:200]}")
            print(f"[DEBUG] XML content length: {len(xml_content)}")
            print(f"[DEBUG] XML content type: {type(xml_content)}")

            # Validate content
            if not xml_content or not xml_content.strip():
                raise Exception("Empty XML content received")

            if not xml_content.strip().startswith('<'):
                raise Exception(f"Not a diagram file (error on line 1 at column 1: Start tag expected, '<' not found). Received: {xml_content[:100]}")

            root = ET.fromstring(xml_content)

            # Find all mxCell elements
            for diagram in root.findall('.//diagram'):
                model = diagram.find('.//mxGraphModel')
                if model is not None:
                    self._parse_model(model)

            return {
                'components': self.components,
                'connections': self.connections
            }
        except ET.ParseError as e:
            print(f"[ERROR] XML Parse Error: {str(e)}")
            print(f"[ERROR] Content that failed: {xml_content[:500]}")
            raise Exception(f"Not a diagram file (error on line 1 at column 1: Start tag expected, '<' not found)")
        except Exception as e:
            print(f"[ERROR] General parsing error: {str(e)}")
            raise Exception(f"Error parsing Draw.io content: {str(e)}")

    def _parse_model(self, model: ET.Element):
        """Parse the mxGraphModel element"""
        root_element = model.find('.//root')
        if root_element is None:
            return

        for cell in root_element.findall('.//mxCell'):
            self._parse_cell(cell)

    def _parse_cell(self, cell: ET.Element):
        """Parse individual mxCell elements"""
        cell_id = cell.get('id')
        if not cell_id:
            return

        value = cell.get('value', '')
        style = cell.get('style', '')

        # Check if this is a cloud component
        for shape_key, resource_type in self.CLOUD_SHAPES.items():
            if shape_key in style:
                component = self._extract_component(cell_id, value, style, shape_key, resource_type)
                if component:
                    self.components.append(component)
                break

        # Check for connections (edges)
        source = cell.get('source')
        target = cell.get('target')
        if source and target:
            self.connections.append({
                'source': source,
                'target': target,
                'id': cell_id
            })

    def _extract_component(self, cell_id: str, value: str, style: str, shape_key: str, resource_type: str) -> Dict[str, Any]:
        """Extract component information from a cell"""
        # Determine the service name
        service_name = None
        for key, service in self.SERVICE_NAMES.items():
            if key in value.lower() or key in shape_key:
                service_name = service
                break

        # Extract geometry if available
        geometry = self._extract_geometry(style)

        component = {
            'id': cell_id,
            'name': value.strip() if value else self._generate_name(shape_key),
            'type': resource_type,
            'shape': shape_key,
            'service': service_name,
            'geometry': geometry
        }

        return component

    def _extract_geometry(self, style: str) -> Dict[str, Any]:
        """Extract geometry information from style string"""
        geometry = {}
        # This is a simplified extraction - can be enhanced
        return geometry

    def _generate_name(self, shape_key: str) -> str:
        """Generate a default name based on the shape type"""
        # Extract the last part of the shape key
        parts = shape_key.split('.')
        if len(parts) > 0:
            name = parts[-1].replace('-', '_').replace('__', '_')
            return name
        return 'unnamed_resource'

    def get_resource_summary(self) -> Dict[str, int]:
        """Get a summary of resources by type"""
        summary = {}
        for component in self.components:
            resource_type = component['type']
            summary[resource_type] = summary.get(resource_type, 0) + 1
        return summary

# Made with Bob

Code Generation: Turning Components into Terraform

Once the diagrams are decoded into component and connection lists, the backend initiates the script generation phase. It programmatically creates three core Terraform configuration files: main.tf for main resources, variables.tf for parameterization, and providers.tf for provider definitions. This process involves sophisticated logic to map Draw.io shapes to corresponding Terraform resource blocks, sanitize names, manage variables, and format the output according to the HashiCorp Configuration Language (HCL). The system is built for modularity, allowing it to generate standardized code for various resource types and cloud providers.

# terraform_generator.py - Generates Terraform code from parsed diagrams
"""
Terraform Code Generator for IBM Cloud Resources
Generates Terraform configuration files from parsed Draw.io diagrams
"""
from typing import Dict, List, Any
import json


class TerraformGenerator:
    """Generates Terraform configuration for multiple cloud providers"""

    # Resource type mappings: generic type -> provider-specific resource type
    RESOURCE_MAPPINGS = {
        'vpc': {
            'ibm': 'ibm_is_vpc',
            'aws': 'aws_vpc',
            'google': 'google_compute_network',
            'azure': 'azurerm_virtual_network',
            'kubernetes': None  # K8s doesn't have VPC concept
        },
        'subnet': {
            'ibm': 'ibm_is_subnet',
            'aws': 'aws_subnet',
            'google': 'google_compute_subnetwork',
            'azure': 'azurerm_subnet',
            'kubernetes': None
        },
        'compute_instance': {
            'ibm': 'ibm_is_instance',
            'aws': 'aws_instance',
            'google': 'google_compute_instance',
            'azure': 'azurerm_virtual_machine',
            'kubernetes': 'kubernetes_deployment'
        },
        'load_balancer': {
            'ibm': 'ibm_is_lb',
            'aws': 'aws_lb',
            'google': 'google_compute_forwarding_rule',
            'azure': 'azurerm_lb',
            'kubernetes': 'kubernetes_service'
        },
        'database': {
            'ibm': 'ibm_database',
            'aws': 'aws_db_instance',
            'google': 'google_sql_database_instance',
            'azure': 'azurerm_postgresql_server',
            'kubernetes': 'kubernetes_stateful_set'
        },
        'resource_group': {
            'ibm': 'ibm_resource_group',
            'aws': None,  # AWS uses tags instead
            'google': 'google_project',
            'azure': 'azurerm_resource_group',
            'kubernetes': 'kubernetes_namespace'
        },
        'floating_ip': {
            'ibm': 'ibm_is_floating_ip',
            'aws': 'aws_eip',
            'google': 'google_compute_address',
            'azure': 'azurerm_public_ip',
            'kubernetes': None
        },
        'public_gateway': {
            'ibm': 'ibm_is_public_gateway',
            'aws': 'aws_nat_gateway',
            'google': 'google_compute_router_nat',
            'azure': 'azurerm_nat_gateway',
            'kubernetes': None
        },
        'app_service': {
            'ibm': 'ibm_resource_instance',
            'aws': 'aws_elastic_beanstalk_application',
            'google': 'google_app_engine_application',
            'azure': 'azurerm_app_service',
            'kubernetes': 'kubernetes_deployment'
        },
        'watson_service': {
            'ibm': 'ibm_resource_instance',
            'aws': None,  # AWS doesn't have Watson equivalent
            'google': None,  # Google doesn't have Watson equivalent
            'azure': None,  # Azure doesn't have Watson equivalent
            'kubernetes': None
        },
        'zone': {
            'ibm': None,  # Organizational only
            'aws': None,
            'google': 'google_compute_zone',
            'azure': None,
            'kubernetes': None
        }
    }

    # Provider configurations
    PROVIDER_CONFIGS = {
        'ibm': {
            'source': 'IBM-Cloud/ibm',
            'version': '>= 1.60.0',
            'docs': 'https://registry.terraform.io/providers/IBM-Cloud/ibm/latest/docs'
        },
        'azure': {
            'source': 'hashicorp/azurerm',
            'version': '>= 3.0.0',
            'docs': 'https://registry.terraform.io/providers/hashicorp/azurerm/latest/docs'
        },
        'google': {
            'source': 'hashicorp/google',
            'version': '>= 5.0.0',
            'docs': 'https://registry.terraform.io/providers/hashicorp/google/latest/docs'
        },
        'aws': {
            'source': 'hashicorp/aws',
            'version': '>= 5.0.0',
            'docs': 'https://registry.terraform.io/providers/hashicorp/aws/latest/docs'
        },
        'kubernetes': {
            'source': 'hashicorp/kubernetes',
            'version': '>= 2.0.0',
            'docs': 'https://registry.terraform.io/providers/hashicorp/kubernetes/latest/docs'
        },
        'terraform': {
            'source': 'hashicorp/tfe',
            'version': '>= 0.50.0',
            'docs': 'https://registry.terraform.io/providers/hashicorp/tfe/latest/docs'
        },
        'instana': {
            'source': 'instana/instana',
            'version': '>= 3.0.0',
            'docs': 'https://registry.terraform.io/providers/instana/instana/latest/docs'
        },
        'vault': {
            'source': 'hashicorp/vault',
            'version': '>= 4.0.0',
            'docs': 'https://registry.terraform.io/providers/hashicorp/vault/latest/docs'
        },
        'turbonomic': {
            'source': 'IBM/turbonomic',
            'version': '>= 1.0.0',
            'docs': 'https://registry.terraform.io/providers/IBM/turbonomic/latest/docs'
        }
    }

    def __init__(self, provider='ibm', custom_provider_url=None):
        self.provider = provider
        self.custom_provider_url = custom_provider_url
        self.custom_provider_config = None

        # If custom provider URL is provided, parse it
        if custom_provider_url and provider == 'other':
            self.custom_provider_config = self._parse_custom_provider_url(custom_provider_url)
            if self.custom_provider_config:
                self.provider = self.custom_provider_config['name']

        self.resources = []
        self.variables = {}
        self.providers = {}

    def _parse_custom_provider_url(self, url: str) -> Dict[str, str] | None:
        """Parse a GitHub provider URL and extract provider information

        Expected format: https://github.com/{org}/{repo}
        Where repo is typically: terraform-provider-{name}

        Returns:
            Dict with 'name', 'source', 'version', and 'docs' keys
        """
        import re

        # Match GitHub URL pattern
        pattern = r'https://github\.com/([\w-]+)/(terraform-provider-([\w-]+))'
        match = re.match(pattern, url)

        if not match:
            return None

        org = match.group(1)
        repo = match.group(2)
        provider_name = match.group(3)

        # Build provider configuration
        return {
            'name': provider_name,
            'source': f'{org}/{provider_name}',
            'version': '>= 1.0.0',  # Default version constraint
            'docs': f'https://registry.terraform.io/providers/{org}/{provider_name}/latest/docs',
            'github_url': url
        }

    def generate_from_components(self, components: List[Dict[str, Any]]) -> Dict[str, str]:
        """Generate Terraform files from parsed components"""
        self.resources = []
        self.variables = {}

        # Process each component
        for idx, component in enumerate(components):
            resource = self._generate_resource(component, idx)
            if resource:
                self.resources.append(resource)

        # Generate the three required files
        main_tf = self._generate_main_tf()
        variables_tf = self._generate_variables_tf()
        providers_tf = self._generate_providers_tf()

        return {
            'main.tf': main_tf,
            'variables.tf': variables_tf,
            'providers.tf': providers_tf
        }

    def _generate_resource(self, component: Dict[str, Any], idx: int) -> Dict[str, Any] | None:
        """Generate a Terraform resource from a component"""
        generic_type = component.get('type')  # Generic type from parser
        name = component.get('name', f'resource_{idx}')
        service = component.get('service')

        # Sanitize name for Terraform
        tf_name = self._sanitize_name(name)

        # Get provider-specific resource type from mapping
        if generic_type not in self.RESOURCE_MAPPINGS:
            return None

        provider_resource_type = self.RESOURCE_MAPPINGS[generic_type].get(self.provider)

        # If this resource type is not supported by the provider, skip it
        if provider_resource_type is None:
            if not hasattr(self, 'skipped_resources'):
                self.skipped_resources = []
            self.skipped_resources.append({
                'name': name,
                'type': generic_type,
                'reason': f'Not supported by {self.provider} provider'
            })
            return None

        # Generate provider-specific resource based on generic type
        if generic_type == 'vpc':
            return self._generate_vpc_resource_multi(tf_name, name, provider_resource_type)
        elif generic_type == 'subnet':
            return self._generate_subnet_resource(tf_name, name, provider_resource_type)
        elif generic_type == 'compute_instance':
            return self._generate_compute_resource(tf_name, name, provider_resource_type)
        elif generic_type == 'load_balancer':
            return self._generate_load_balancer_resource(tf_name, name, provider_resource_type)
        elif generic_type == 'database':
            return self._generate_database_resource(tf_name, name, provider_resource_type)
        elif generic_type == 'resource_group':
            return self._generate_resource_group_multi(tf_name, name, provider_resource_type)
        elif generic_type == 'floating_ip':
            return self._generate_floating_ip_resource(tf_name, name, provider_resource_type)
        elif generic_type == 'public_gateway':
            return self._generate_gateway_resource(tf_name, name, provider_resource_type)
        elif generic_type == 'app_service':
            return self._generate_app_service_resource(tf_name, name, provider_resource_type)
        elif generic_type == 'watson_service':
            return self._generate_watson_service_resource(tf_name, name, service or '', provider_resource_type)

        return None

    def _generate_vpc_resource(self, tf_name: str, display_name: str) -> Dict[str, Any]:
        """Generate VPC resource configuration"""
        self.variables[f'{tf_name}_name'] = {
            'description': f'Name for the {display_name} VPC',
            'type': 'string',
            'default': display_name
        }

        return {
            'type': 'ibm_is_vpc',
            'name': tf_name,
            'config': {
                'name': f'var.{tf_name}_name',
                'resource_group': 'var.resource_group_id',
                'tags': ['var.tags']
            }
        }

    def _generate_resource_instance(self, tf_name: str, display_name: str, service: str) -> Dict[str, Any]:
        """Generate resource instance configuration for Watson services"""
        if not service:
            service = 'watsonx-ai'

        # Map service names to IBM Cloud service names
        service_map = {
            'watsonx-orchestrate': 'watsonx-orchestrate',
            'watsonx-governance': 'watsonx-governance',
            'watsonx-ai': 'watsonx-ai'
        }

        service_name = service_map.get(service, service)

        self.variables[f'{tf_name}_name'] = {
            'description': f'Name for the {display_name} instance',
            'type': 'string',
            'default': display_name
        }

        self.variables[f'{tf_name}_plan'] = {
            'description': f'Service plan for {display_name}',
            'type': 'string',
            'default': 'lite'
        }

        service_map = {
            'watsonx-orchestrate': 'watsonx-orchestrate',
            'watsonx-governance': 'watsonx-governance',
            'watsonx-ai': 'watsonx-ai'
        }

        service_name = service_map.get(service, service)

        return {
            'type': 'ibm_resource_instance',
            'name': tf_name,
            'config': {
                'name': f'var.{tf_name}_name',
                'service': f'"{service_name}"',
                'plan': f'var.{tf_name}_plan',
                'location': 'var.region',
                'resource_group_id': 'var.resource_group_id',
                'tags': ['var.tags']
            }
        }

    # Multi-provider resource generators

    def _generate_vpc_resource_multi(self, tf_name: str, display_name: str, resource_type: str) -> Dict[str, Any]:
        """Generate VPC/Network resource for any provider"""
        self.variables[f'{tf_name}_name'] = {
            'description': f'Name for the {display_name} network',
            'type': 'string',
            'default': display_name
        }

        config = {'name': f'var.{tf_name}_name'}

        # Provider-specific configurations
        if self.provider == 'ibm':
            config['resource_group'] = 'var.resource_group_id'
            config['tags'] = 'var.tags'
        elif self.provider == 'aws':
            config['cidr_block'] = '"10.0.0.0/16"'
            config['tags'] = 'var.tags'
        elif self.provider == 'google':
            config['project'] = 'var.project_id'
            config['auto_create_subnetworks'] = 'false'
        elif self.provider == 'azure':
            config['location'] = 'var.location'
            config['resource_group_name'] = 'azurerm_resource_group.main.name'
            config['address_space'] = '["10.0.0.0/16"]'
            config['tags'] = 'var.tags'

        return {'type': resource_type, 'name': tf_name, 'config': config}

    def _generate_subnet_resource(self, tf_name: str, display_name: str, resource_type: str) -> Dict[str, Any]:
        """Generate subnet resource for any provider"""
        self.variables[f'{tf_name}_name'] = {
            'description': f'Name for the {display_name} subnet',
            'type': 'string',
            'default': display_name
        }

        config = {'name': f'var.{tf_name}_name'}

        if self.provider == 'ibm':
            config['vpc'] = 'ibm_is_vpc.main.id'
            config['zone'] = '"us-south-1"'
            config['ipv4_cidr_block'] = '"10.0.1.0/24"'
        elif self.provider == 'aws':
            config['vpc_id'] = 'aws_vpc.main.id'
            config['cidr_block'] = '"10.0.1.0/24"'
            config['availability_zone'] = '"us-east-1a"'
            config['tags'] = 'var.tags'
        elif self.provider == 'google':
            config['network'] = 'google_compute_network.main.id'
            config['ip_cidr_range'] = '"10.0.1.0/24"'
            config['region'] = 'var.region'
        elif self.provider == 'azure':
            config['resource_group_name'] = 'azurerm_resource_group.main.name'
            config['virtual_network_name'] = 'azurerm_virtual_network.main.name'
            config['address_prefixes'] = '["10.0.1.0/24"]'

        return {'type': resource_type, 'name': tf_name, 'config': config}

    def _generate_compute_resource(self, tf_name: str, display_name: str, resource_type: str) -> Dict[str, Any]:
        """Generate compute instance resource for any provider"""
        self.variables[f'{tf_name}_name'] = {
            'description': f'Name for the {display_name} instance',
            'type': 'string',
            'default': display_name
        }

        config = {'name': f'var.{tf_name}_name'}

        if self.provider == 'ibm':
            config['profile'] = '"cx2-2x4"'
            config['image'] = '"r006-14140f94-fcc4-11e9-96e7-a72723715315"'
            config['zone'] = '"us-south-1"'
            config['vpc'] = 'ibm_is_vpc.main.id'
            config['primary_network_interface'] = '{subnet = ibm_is_subnet.main.id}'
        elif self.provider == 'aws':
            config['ami'] = '"ami-0c55b159cbfafe1f0"'
            config['instance_type'] = '"t2.micro"'
            config['subnet_id'] = 'aws_subnet.main.id'
            config['tags'] = 'var.tags'
        elif self.provider == 'google':
            config['machine_type'] = '"e2-medium"'
            config['zone'] = '"us-central1-a"'
            config['boot_disk'] = '{initialize_params {image = "debian-cloud/debian-11"}}'
            config['network_interface'] = '{network = google_compute_network.main.id, subnetwork = google_compute_subnetwork.main.id}'
        elif self.provider == 'azure':
            config['location'] = 'var.location'
            config['resource_group_name'] = 'azurerm_resource_group.main.name'
            config['network_interface_ids'] = '[azurerm_network_interface.main.id]'
            config['vm_size'] = '"Standard_B2s"'
            config['os_disk'] = '{caching = "ReadWrite", storage_account_type = "Standard_LRS"}'
        elif self.provider == 'kubernetes':
            config['metadata'] = '{name = var.%s_name, namespace = var.namespace}' % tf_name
            config['spec'] = '{replicas = 2, selector {match_labels = {app = var.%s_name}}, template {metadata {labels = {app = var.%s_name}}, spec {container {name = var.%s_name, image = "nginx:latest"}}}}' % (tf_name, tf_name, tf_name)

        return {'type': resource_type, 'name': tf_name, 'config': config}

    def _generate_load_balancer_resource(self, tf_name: str, display_name: str, resource_type: str) -> Dict[str, Any]:
        """Generate load balancer resource for any provider"""
        self.variables[f'{tf_name}_name'] = {
            'description': f'Name for the {display_name} load balancer',
            'type': 'string',
            'default': display_name
        }

        config = {'name': f'var.{tf_name}_name'}

        if self.provider == 'ibm':
            config['subnets'] = '[ibm_is_subnet.main.id]'
        elif self.provider == 'aws':
            config['load_balancer_type'] = '"application"'
            config['subnets'] = '[aws_subnet.main.id]'
            config['tags'] = 'var.tags'
        elif self.provider == 'google':
            config['ip_protocol'] = '"TCP"'
            config['port_range'] = '"80"'
            config['target'] = 'google_compute_target_pool.main.self_link'
        elif self.provider == 'azure':
            config['location'] = 'var.location'
            config['resource_group_name'] = 'azurerm_resource_group.main.name'
        elif self.provider == 'kubernetes':
            config['metadata'] = '{name = var.%s_name, namespace = var.namespace}' % tf_name
            config['spec'] = '{type = "LoadBalancer", selector = {app = "myapp"}, port {port = 80, target_port = 8080}}'

        return {'type': resource_type, 'name': tf_name, 'config': config}

    def _generate_database_resource(self, tf_name: str, display_name: str, resource_type: str) -> Dict[str, Any]:
        """Generate database resource for any provider"""
        self.variables[f'{tf_name}_name'] = {
            'description': f'Name for the {display_name} database',
            'type': 'string',
            'default': display_name
        }

        config = {'name': f'var.{tf_name}_name'}

        if self.provider == 'ibm':
            config['service'] = '"databases-for-postgresql"'
            config['plan'] = '"standard"'
            config['location'] = 'var.region'
            config['resource_group_id'] = 'var.resource_group_id'
        elif self.provider == 'aws':
            config['engine'] = '"postgres"'
            config['engine_version'] = '"14.7"'
            config['instance_class'] = '"db.t3.micro"'
            config['allocated_storage'] = '20'
            config['username'] = '"admin"'
            config['password'] = '"changeme"'
            config['skip_final_snapshot'] = 'true'
            config['tags'] = 'var.tags'
        elif self.provider == 'google':
            config['database_version'] = '"POSTGRES_14"'
            config['region'] = 'var.region'
            config['settings'] = '{tier = "db-f1-micro"}'
        elif self.provider == 'azure':
            config['location'] = 'var.location'
            config['resource_group_name'] = 'azurerm_resource_group.main.name'
            config['sku_name'] = '"B_Gen5_1"'
            config['version'] = '"11"'
            config['administrator_login'] = '"psqladmin"'
            config['administrator_login_password'] = '"H@Sh1CoR3!"'
        elif self.provider == 'kubernetes':
            config['metadata'] = '{name = var.%s_name, namespace = var.namespace}' % tf_name
            config['spec'] = '{replicas = 1, selector {match_labels = {app = "database"}}, template {metadata {labels = {app = "database"}}, spec {container {name = "postgres", image = "postgres:14"}}}}'

        return {'type': resource_type, 'name': tf_name, 'config': config}

    def _generate_resource_group_multi(self, tf_name: str, display_name: str, resource_type: str) -> Dict[str, Any]:
        """Generate resource group/namespace for any provider"""
        self.variables[f'{tf_name}_name'] = {
            'description': f'Name for the {display_name} resource group',
            'type': 'string',
            'default': display_name
        }

        config = {'name': f'var.{tf_name}_name'}

        if self.provider == 'ibm':
            pass  # IBM resource group only needs name
        elif self.provider == 'google':
            config['project_id'] = f'var.{tf_name}_name'
        elif self.provider == 'azure':
            config['location'] = 'var.location'
        elif self.provider == 'kubernetes':
            config['metadata'] = '{name = var.%s_name}' % tf_name

        return {'type': resource_type, 'name': tf_name, 'config': config}

    def _generate_floating_ip_resource(self, tf_name: str, display_name: str, resource_type: str) -> Dict[str, Any]:
        """Generate floating/elastic IP resource for any provider"""
        self.variables[f'{tf_name}_name'] = {
            'description': f'Name for the {display_name} IP',
            'type': 'string',
            'default': display_name
        }

        config = {'name': f'var.{tf_name}_name'}

        if self.provider == 'ibm':
            config['zone'] = '"us-south-1"'
        elif self.provider == 'aws':
            config['vpc'] = 'true'
            config['tags'] = 'var.tags'
        elif self.provider == 'google':
            config['region'] = 'var.region'
        elif self.provider == 'azure':
            config['location'] = 'var.location'
            config['resource_group_name'] = 'azurerm_resource_group.main.name'
            config['allocation_method'] = '"Static"'

        return {'type': resource_type, 'name': tf_name, 'config': config}

    def _generate_gateway_resource(self, tf_name: str, display_name: str, resource_type: str) -> Dict[str, Any]:
        """Generate NAT gateway resource for any provider"""
        self.variables[f'{tf_name}_name'] = {
            'description': f'Name for the {display_name} gateway',
            'type': 'string',
            'default': display_name
        }

        config = {'name': f'var.{tf_name}_name'}

        if self.provider == 'ibm':
            config['vpc'] = 'ibm_is_vpc.main.id'
            config['zone'] = '"us-south-1"'
        elif self.provider == 'aws':
            config['subnet_id'] = 'aws_subnet.main.id'
            config['allocation_id'] = 'aws_eip.main.id'
            config['tags'] = 'var.tags'
        elif self.provider == 'google':
            config['router'] = 'google_compute_router.main.name'
            config['region'] = 'var.region'
            config['nat_ip_allocate_option'] = '"AUTO_ONLY"'
            config['source_subnetwork_ip_ranges_to_nat'] = '"ALL_SUBNETWORKS_ALL_IP_RANGES"'
        elif self.provider == 'azure':
            config['location'] = 'var.location'
            config['resource_group_name'] = 'azurerm_resource_group.main.name'

        return {'type': resource_type, 'name': tf_name, 'config': config}

    def _generate_app_service_resource(self, tf_name: str, display_name: str, resource_type: str) -> Dict[str, Any]:
        """Generate application service resource for any provider"""
        self.variables[f'{tf_name}_name'] = {
            'description': f'Name for the {display_name} application',
            'type': 'string',
            'default': display_name
        }

        config = {'name': f'var.{tf_name}_name'}

        if self.provider == 'ibm':
            config['service'] = '"cloud-object-storage"'
            config['plan'] = '"lite"'
            config['location'] = 'var.region'
        elif self.provider == 'aws':
            config['description'] = f'"{display_name} application"'
        elif self.provider == 'google':
            config['location_id'] = 'var.region'
        elif self.provider == 'azure':
            config['location'] = 'var.location'
            config['resource_group_name'] = 'azurerm_resource_group.main.name'
            config['app_service_plan_id'] = 'azurerm_app_service_plan.main.id'
        elif self.provider == 'kubernetes':
            config['metadata'] = '{name = var.%s_name, namespace = var.namespace}' % tf_name
            config['spec'] = '{replicas = 3, selector {match_labels = {app = var.%s_name}}, template {metadata {labels = {app = var.%s_name}}, spec {container {name = var.%s_name, image = "nginx:latest", port {container_port = 80}}}}}' % (tf_name, tf_name, tf_name)

        return {'type': resource_type, 'name': tf_name, 'config': config}

    def _generate_watson_service_resource(self, tf_name: str, display_name: str, service: str, resource_type: str) -> Dict[str, Any] | None:
        """Generate Watson service resource (IBM only)"""
        if self.provider != 'ibm':
            return None

        return self._generate_resource_instance(tf_name, display_name, service or 'watsonx-ai')

    def _generate_resource_group(self, tf_name: str, display_name: str) -> Dict[str, Any]:
        """Generate resource group configuration"""
        self.variables[f'{tf_name}_name'] = {
            'description': f'Name for the {display_name} resource group',
            'type': 'string',
            'default': display_name
        }

        return {
            'type': 'ibm_resource_group',
            'name': tf_name,
            'config': {
                'name': f'var.{tf_name}_name'
            }
        }

    def _generate_main_tf(self) -> str:
        """Generate main.tf content"""
        # Use custom provider config if available, otherwise use predefined
        if self.custom_provider_config:
            provider_config = self.custom_provider_config
            provider_name = self.custom_provider_config['name']
        else:
            provider_config = self.PROVIDER_CONFIGS.get(self.provider, self.PROVIDER_CONFIGS['ibm'])
            provider_name = self.provider

        lines = [
            f'# {provider_name.upper()} Terraform Configuration',
            '# Generated from Draw.io diagram',
            f'# Provider: {provider_name}',
            f'# Documentation: {provider_config["docs"]}',
            '',
            '# Terraform configuration',
            'terraform {',
            '  required_version = ">= 1.0"',
            '  required_providers {',
            f'    {provider_name} = {{',
            f'      source  = "{provider_config["source"]}"',
            f'      version = "{provider_config["version"]}"',
            '    }',
            '  }',
            '}',
            ''
        ]

        # Add resources
        if self.resources:
            for resource in self.resources:
                lines.extend(self._format_resource(resource))
                lines.append('')
        else:
            # No resources generated - add helpful message
            lines.append('# No resources generated.')
            if hasattr(self, 'skipped_resources') and self.skipped_resources:
                lines.append('# This diagram contains IBM Cloud-specific components that are not')
                lines.append(f'# supported by the {provider_name} provider.')
                lines.append('#')
                lines.append('# Skipped IBM Cloud resources:')
                for skipped in self.skipped_resources:
                    lines.append(f'#   - {skipped["name"]} ({skipped["type"]})')
                lines.append('#')
                lines.append('# To generate IBM Cloud resources, please select "IBM Cloud" as the provider.')
            lines.append('')

        return '\n'.join(lines)

    def _format_resource(self, resource: Dict[str, Any]) -> List[str]:
        """Format a resource as Terraform HCL"""
        lines = [
            f'resource "{resource["type"]}" "{resource["name"]}" {{'
        ]

        for key, value in resource['config'].items():
            if isinstance(value, list):
                lines.append(f'  {key} = {value[0]}')
            else:
                lines.append(f'  {key} = {value}')

        lines.append('}')
        return lines

    def _generate_variables_tf(self) -> str:
        """Generate variables.tf content"""
        provider_vars = self._get_provider_variables()

        lines = [
            f'# Variables for {self.provider.upper()} resources',
            '# Generated from Draw.io diagram',
            ''
        ]

        # Add provider-specific variables
        lines.extend(provider_vars)
        lines.append('')

        # Add component-specific variables
        for var_name, var_config in self.variables.items():
            lines.append(f'variable "{var_name}" {{')
            lines.append(f'  description = "{var_config["description"]}"')
            lines.append(f'  type        = {var_config["type"]}')
            if 'default' in var_config:
                if isinstance(var_config['default'], str):
                    lines.append(f'  default     = "{var_config["default"]}"')
                else:
                    lines.append(f'  default     = {var_config["default"]}')
            lines.append('}')
            lines.append('')

        return '\n'.join(lines)

    def _generate_providers_tf(self) -> str:
        """Generate providers.tf content"""
        provider_config = self._get_provider_config()

        lines = [
            f'# {self.provider.upper()} Provider Configuration',
            '# Generated from Draw.io diagram',
            ''
        ]

        lines.extend(provider_config)
        lines.append('')

        return '\n'.join(lines)

    def _get_provider_variables(self) -> List[str]:
        """Get provider-specific variables"""
        # For custom providers, return generic variables
        if self.custom_provider_config:
            return [
                'variable "api_key" {',
                '  description = "API Key for authentication"',
                '  type        = string',
                '  sensitive   = true',
                '}',
                '',
                'variable "region" {',
                '  description = "Region for resources"',
                '  type        = string',
                '  default     = "us-south"',
                '}',
                '',
                'variable "tags" {',
                '  description = "Tags for resources"',
                '  type        = list(string)',
                '  default     = ["terraform", "drawio-generated"]',
                '}'
            ]

        if self.provider == 'ibm':
            return [
                'variable "ibmcloud_api_key" {',
                '  description = "IBM Cloud API Key"',
                '  type        = string',
                '  sensitive   = true',
                '}',
                '',
                'variable "region" {',
                '  description = "IBM Cloud region"',
                '  type        = string',
                '  default     = "us-south"',
                '}',
                '',
                'variable "resource_group_id" {',
                '  description = "Resource group ID"',
                '  type        = string',
                '}',
                '',
                'variable "tags" {',
                '  description = "Tags for resources"',
                '  type        = list(string)',
                '  default     = ["terraform", "drawio-generated"]',
                '}'
            ]
        elif self.provider == 'azure':
            return [
                'variable "subscription_id" {',
                '  description = "Azure Subscription ID"',
                '  type        = string',
                '}',
                '',
                'variable "tenant_id" {',
                '  description = "Azure Tenant ID"',
                '  type        = string',
                '}',
                '',
                'variable "location" {',
                '  description = "Azure region"',
                '  type        = string',
                '  default     = "eastus"',
                '}',
                '',
                'variable "tags" {',
                '  description = "Tags for resources"',
                '  type        = map(string)',
                '  default     = {',
                '    Environment = "Development"',
                '    ManagedBy   = "Terraform"',
                '  }',
                '}'
            ]
        elif self.provider == 'google':
            return [
                'variable "project_id" {',
                '  description = "Google Cloud Project ID"',
                '  type        = string',
                '}',
                '',
                'variable "region" {',
                '  description = "Google Cloud region"',
                '  type        = string',
                '  default     = "us-central1"',
                '}',
                '',
                'variable "labels" {',
                '  description = "Labels for resources"',
                '  type        = map(string)',
                '  default     = {',
                '    environment = "development"',
                '    managed_by  = "terraform"',
                '  }',
                '}'
            ]
        elif self.provider == 'aws':
            return [
                'variable "region" {',
                '  description = "AWS region"',
                '  type        = string',
                '  default     = "us-east-1"',
                '}',
                '',
                'variable "tags" {',
                '  description = "Tags for resources"',
                '  type        = map(string)',
                '  default     = {',
                '    Environment = "Development"',
                '    ManagedBy   = "Terraform"',
                '  }',
                '}'
            ]
        elif self.provider == 'kubernetes':
            return [
                'variable "config_path" {',
                '  description = "Path to kubeconfig file"',
                '  type        = string',
                '  default     = "~/.kube/config"',
                '}',
                '',
                'variable "config_context" {',
                '  description = "Kubernetes context to use"',
                '  type        = string',
                '  default     = ""',
                '}',
                '',
                'variable "namespace" {',
                '  description = "Default namespace"',
                '  type        = string',
                '  default     = "default"',
                '}'
            ]
        elif self.provider == 'terraform':
            return [
                'variable "tfe_token" {',
                '  description = "Terraform Cloud/Enterprise API Token"',
                '  type        = string',
                '  sensitive   = true',
                '}',
                '',
                'variable "tfe_hostname" {',
                '  description = "Terraform Cloud/Enterprise hostname"',
                '  type        = string',
                '  default     = "app.terraform.io"',
                '}',
                '',
                'variable "organization" {',
                '  description = "Terraform Cloud/Enterprise organization name"',
                '  type        = string',
                '}'
            ]

        return []

    def _get_provider_config(self) -> List[str]:
        """Get provider-specific configuration"""
        # For custom providers, return generic configuration
        if self.custom_provider_config:
            provider_name = self.custom_provider_config['name']
            return [
                f'provider "{provider_name}" {{',
                '  # Configure provider-specific settings here',
                '  # Refer to provider documentation for available options',
                f'  # Documentation: {self.custom_provider_config["docs"]}',
                '}'
            ]

        if self.provider == 'ibm':
            return [
                'provider "ibm" {',
                '  ibmcloud_api_key = var.ibmcloud_api_key',
                '  region           = var.region',
                '}'
            ]
        elif self.provider == 'azure':
            return [
                'provider "azurerm" {',
                '  features {}',
                '  subscription_id = var.subscription_id',
                '  tenant_id       = var.tenant_id',
                '}'
            ]
        elif self.provider == 'google':
            return [
                'provider "google" {',
                '  project = var.project_id',
                '  region  = var.region',
                '}'
            ]
        elif self.provider == 'aws':
            return [
                'provider "aws" {',
                '  region = var.region',
                '}'
            ]
        elif self.provider == 'kubernetes':
            return [
                'provider "kubernetes" {',
                '  config_path    = var.config_path',
                '  config_context = var.config_context',
                '}'
            ]
        elif self.provider == 'terraform':
            return [
                'provider "tfe" {',
                '  token    = var.tfe_token',
                '  hostname = var.tfe_hostname',
                '}'
            ]

        return []

    def _sanitize_name(self, name: str) -> str:
        """Sanitize a name for use in Terraform"""
        # Remove special characters and replace spaces with underscores
        sanitized = name.lower()
        sanitized = sanitized.replace(' ', '_')
        sanitized = sanitized.replace('-', '_')
        sanitized = sanitized.replace('.', '_')
        # Remove any non-alphanumeric characters except underscores
        sanitized = ''.join(c for c in sanitized if c.isalnum() or c == '_')
        # Ensure it starts with a letter
        if sanitized and not sanitized[0].isalpha():
            sanitized = 'res_' + sanitized
        return sanitized or 'unnamed_resource'

# Made with Bob

Strategic Versatility: From a Curated List to Custom Templates

Current versions of the application excel at interpreting IBM Cloud-specific shapes and generating highly accurate HCL for that environment (to be tested for sure). However, this is only the beginning of our multi-cloud vision. Future iterations will expand this specialized support to other major providers like AWS and Azure, moving beyond generic detection to deep resource mapping. We are also developing a robust “Generator Template” feature that will allow users to extend the system’s capabilities independently; by simply providing a repository URL for any GitHub-hosted Terraform provider, users will eventually be able to generate bespoke configurations for even the most niche infrastructure requirements.

The Front-end — HTML And Javascript

The frontend of the Terraform Builder is a modern, responsive single-page application built using a standard web stack of HTML5, CSS3, and vanilla JavaScript. The interface, defined in index.html, features a clean two-column layout that guides users through the workflow of either creating diagrams in Draw.io or uploading existing XML files. Styling is managed through styles.css, which implements a professional look-and-feel inspired by the IBM Design System, utilizing the IBM Plex font family. The client-side logic resides in app.js, which orchestrates the user interaction—handling drag-and-drop file uploads, managing the provider selection (including custom GitHub templates), and dynamically updating the UI with generated HCL code. This JavaScript layer also facilitates the multi-file preview and download functionality, ensuring a seamless experience without the need for complex frontend frameworks.

Output Samples

The application streamlines the transition from design to deployment by delivering the generated HCL scripts in a structured, timestamped format, ensuring every architectural iteration is captured and organized. By programmatically mapping visual components from a Draw.io diagram directly into main.tf, variables.tf, and providers.tf, the tool dramatically accelerates the infrastructure provisioning process, eliminating the manual overhead of writing boilerplate code. However, while this visual-to-code automation provides a high-speed starting point that reflects the intended design, it is crucial that these scripts are thoroughly reviewed and tested in a development environment before production use. This workflow allows architects to focus on the high-level logic while the application handles the technical translation, significantly reducing time-to-market without bypassing the essential step of human validation.

# IBM Terraform Configuration
# Generated from Draw.io diagram
# Provider: ibm
# Documentation: https://registry.terraform.io/providers/IBM-Cloud/ibm/latest/docs

# Terraform configuration
terraform {
  required_version = ">= 1.0"
  required_providers {
    ibm = {
      source  = "IBM-Cloud/ibm"
      version = ">= 1.60.0"
    }
  }
}

resource "ibm_resource_group" "ibm_cloud" {
  name = var.ibm_cloud_name
}

resource "ibm_is_vpc" "vpc" {
  name = var.vpc_name
  resource_group = var.resource_group_id
  tags = var.tags
}

resource "ibm_resource_group" "ibm_cloud" {
  name = var.ibm_cloud_name
}

resource "ibm_database" "data_base" {
  name = var.data_base_name
  service = "databases-for-postgresql"
  plan = "standard"
  location = var.region
  resource_group_id = var.resource_group_id
}

resource "ibm_is_public_gateway" "network_public" {
  name = var.network_public_name
  vpc = ibm_is_vpc.main.id
  zone = "us-south-1"
}

resource "ibm_is_instance" "rev_" {
  name = var.rev__name
  profile = "cx2-2x4"
  image = "r006-14140f94-fcc4-11e9-96e7-a72723715315"
  zone = "us-south-1"
  vpc = ibm_is_vpc.main.id
  primary_network_interface = {subnet = ibm_is_subnet.main.id}
}

resource "ibm_resource_instance" "application_web" {
  name = var.application_web_name
  service = "cloud-object-storage"
  plan = "lite"
  location = var.region
}

resource "ibm_resource_group" "ibm_cloud" {
  name = var.ibm_cloud_name
}

resource "ibm_database" "data_base" {
  name = var.data_base_name
  service = "databases-for-postgresql"
  plan = "standard"
  location = var.region
  resource_group_id = var.resource_group_id
}

resource "ibm_is_public_gateway" "network_public" {
  name = var.network_public_name
  vpc = ibm_is_vpc.main.id
  zone = "us-south-1"
}

resource "ibm_is_instance" "rev_" {
  name = var.rev__name
  profile = "cx2-2x4"
  image = "r006-14140f94-fcc4-11e9-96e7-a72723715315"
  zone = "us-south-1"
  vpc = ibm_is_vpc.main.id
  primary_network_interface = {subnet = ibm_is_subnet.main.id}
}

resource "ibm_is_floating_ip" "floating_ip" {
  name = var.floating_ip_name
  zone = "us-south-1"
}

resource "ibm_resource_instance" "application_web" {
  name = var.application_web_name
  service = "cloud-object-storage"
  plan = "lite"
  location = var.region
}

resource "ibm_is_lb" "load_balancer_vpc" {
  name = var.load_balancer_vpc_name
  subnets = [ibm_is_subnet.main.id]
}

resource "ibm_is_floating_ip" "floating_ip" {
  name = var.floating_ip_name
  zone = "us-south-1"
}

# IBM Provider Configuration
# Generated from Draw.io diagram

provider "ibm" {
  ibmcloud_api_key = var.ibmcloud_api_key
  region           = var.region
}

# Variables for IBM resources
# Generated from Draw.io diagram

variable "ibmcloud_api_key" {
  description = "IBM Cloud API Key"
  type        = string
  sensitive   = true
}

variable "region" {
  description = "IBM Cloud region"
  type        = string
  default     = "us-south"
}

variable "resource_group_id" {
  description = "Resource group ID"
  type        = string
}

variable "tags" {
  description = "Tags for resources"
  type        = list(string)
  default     = ["terraform", "drawio-generated"]
}

variable "ibm_cloud_name" {
  description = "Name for the ibm_cloud resource group"
  type        = string
  default     = "ibm_cloud"
}

variable "vpc_name" {
  description = "Name for the VPC network"
  type        = string
  default     = "VPC"
}

variable "data_base_name" {
  description = "Name for the data_base database"
  type        = string
  default     = "data_base"
}

variable "network_public_name" {
  description = "Name for the network_public gateway"
  type        = string
  default     = "network_public"
}

variable "rev__name" {
  description = "Name for the Rev. instance"
  type        = string
  default     = "Rev."
}

variable "application_web_name" {
  description = "Name for the application_web application"
  type        = string
  default     = "application_web"
}

variable "floating_ip_name" {
  description = "Name for the floating_ip IP"
  type        = string
  default     = "floating_ip"
}

variable "load_balancer_vpc_name" {
  description = "Name for the load_balancer_vpc load balancer"
  type        = string
  default     = "load_balancer_vpc"
}

Deployment

Beyond local execution, the Terraform Builder is engineered for seamless cloud-native deployment. The inclusion of a multi-stage Dockerfile ensures the application is fully containerized, providing a consistent environment across different hosting platforms. For teams utilizing orchestration, the project provides a complete set of Kubernetes manifests: deployment.yaml defines a scalable architecture with dual replicas and integrated liveness/readiness probes for high availability, while configmap.yaml centralizes environment-specific settings. This "production-ready" approach allows the generator to be easily hosted on platforms like IBM Cloud Code Engine, Red Hat OpenShift, or any standard Kubernetes cluster, transforming it from a local utility into a shared enterprise service.

Conclusion

In conclusion, the Terraform Builder provides a comprehensive, end-to-end bridge between visual architectural design and automated infrastructure deployment. By combining a sleek, user friendly frontend with a powerful Python-driven backend, the application successfully transforms static Draw.io diagrams into functional, multi-file Terraform configurations. With its production-ready containerization and Kubernetes manifests, the tool is prepared for immediate deployment in cloud-native environments. While currently optimized for IBM Cloud in its alpha release, the foundational architecture is already in place to support a vast ecosystem of providers and custom templates, representing a significant step forward in personal productivity and the democratization of Infrastructure as Code.

>>> Thanks for reading <<<

Markdown Everything: My New Personal Project: HTML/URL to Markdown Converter

Alain Airom (Ayrom) — Sun, 26 Apr 2026 13:05:08 +0000

Simplifying Documentation using IBM Bob to Create My New Personal Project: HTML/URL to Markdown Converter!

Introduction (and spoiler)

I know, I know — take a deep breath and try to remain calm. This project features exactly zero autonomous AI agents, no sophisticated RAG pipelines, and — brace yourselves — not a single “groundbreaking” LLM integration to disrupt the industry. I’m truly sorry to disappoint the hype-train, but as often happens when I’m left to my own devices, I had a burning need for a tool that actually does something simple. So, instead of building a sentient toaster, I decided to create my own URL-to-Markdown converter for Chrome. It’s just a humble extension designed to solve a real problem, though, in a classic display of “do as I say, not as I do,” I naturally just sat back and made Bob do all the actual manual labor to realize it for me. Simple, right?

The Implementation

The application is structured as a modular utility that can handle both local files and live web URLs, converting messy HTML into clean, standardized Markdown. The core logic relies on the Turndown library, enhanced by custom rules for GitHub-Flavored Markdown (GFM).

Core Conversion Engine

At the heart of every script lies the TurndownService. The logic follows a consistent pipeline:

Initialization: Configures headingStyle (ATX vs Setext) and codeBlockStyle (fenced).
Plugin Integration: Uses turndown-plugin-gfm to ensure tables and task lists are preserved.
Custom Rules: Specifically targets elements like <pre> tags to ensure syntax highlighting is maintained in the output.

#!/usr/bin/env node

/**
 * Node.js CLI script to convert HTML files to Markdown
 * Usage: node scripts/convert-file.js <input-file> [output-file]
 */

const fs = require('fs');
const path = require('path');

// Load the converter
const HtmlToMarkdown = require('../src/html-to-markdown.js');

// Verify JSDOM is available for Node.js
try {
  require('jsdom');
} catch (e) {
  console.error('Error: JSDOM is required for Node.js usage.');
  console.error('Install it with: npm install jsdom');
  process.exit(1);
}

// Parse command line arguments
const args = process.argv.slice(2);

if (args.length === 0 || args.includes('--help') || args.includes('-h')) {
  console.log(`
HTML to Markdown Converter - CLI Tool

Usage:
  node scripts/convert-file.js <input-file> [output-file]
  node scripts/convert-file.js --help

Arguments:
  input-file    Path to the HTML file to convert
  output-file   (Optional) Path for the output Markdown file
                If not provided, will create a timestamped file in output/

Options:
  --help, -h    Show this help message

Examples:
  node scripts/convert-file.js input/sample.html
  node scripts/convert-file.js input/sample.html output/result.md
  node scripts/convert-file.js page.html
  `);
  process.exit(0);
}

const inputFile = args[0];
let outputFile = args[1];

// Check if input file exists
if (!fs.existsSync(inputFile)) {
  console.error(`Error: Input file '${inputFile}' not found.`);
  process.exit(1);
}

// Read the HTML file
console.log(`Reading HTML from: ${inputFile}`);
const html = fs.readFileSync(inputFile, 'utf8');

// Create converter instance
const converter = new HtmlToMarkdown({
  headingStyle: 'atx',
  codeBlockStyle: 'fenced',
  bulletListMarker: '-',
  strongDelimiter: '**',
  emDelimiter: '_'
});

// Convert to Markdown
console.log('Converting HTML to Markdown...');
const markdown = converter.convert(html);

// Generate output filename if not provided
if (!outputFile) {
  const timestamp = new Date().toISOString().replace(/[:.]/g, '-').slice(0, -5);
  const inputBasename = path.basename(inputFile, path.extname(inputFile));
  outputFile = path.join('output', `${inputBasename}-${timestamp}.md`);

  // Ensure output directory exists
  const outputDir = path.dirname(outputFile);
  if (!fs.existsSync(outputDir)) {
    fs.mkdirSync(outputDir, { recursive: true });
  }
}

// Write the Markdown file
fs.writeFileSync(outputFile, markdown, 'utf8');

console.log(`✓ Conversion successful!`);
console.log(`Output saved to: ${outputFile}`);
console.log(`File size: ${markdown.length} characters`);

// Made with Bob

Data Acquisition Logic

The application handles three distinct entry points:

| Entry Point     | Logic Flow                                                   |
| --------------- | ------------------------------------------------------------ |
| **Local File**  | Reads raw HTML from the filesystem using `fs.readFileSync`.  |
| **Simple URL**  | Fetches remote HTML via `axios`, then parses the DOM.        |
| **Complex URL** | Uses **JSDOM** to simulate a browser environment, allowing for cleaner extraction of titles and metadata before conversion. |

Image Handling & Asset Management

As detailed in Image-Handling.md, the tool doesn't just convert text; it manages media context:

Absolute URL Resolution: Logic converts relative paths (/img/photo.png) into absolute URLs based on the source domain.
Alt-Text Preservation: Ensures accessibility by mapping alt attributes to Markdown image syntax: ![alt](url).

#!/usr/bin/env node

/**
 * Node.js CLI script to convert HTML from a URL to Markdown
 * Usage: node scripts/convert-url.js <url> [output-file]
 */

const fs = require('fs');
const path = require('path');
const https = require('https');
const http = require('http');

// Load the converter
const HtmlToMarkdown = require('../src/html-to-markdown.js');

// Verify JSDOM is available
try {
  require('jsdom');
} catch (e) {
  console.error('Error: JSDOM is required for Node.js usage.');
  console.error('Install it with: npm install jsdom');
  process.exit(1);
}

// Parse command line arguments
const args = process.argv.slice(2);

if (args.length === 0 || args.includes('--help') || args.includes('-h')) {
  console.log(`
HTML to Markdown Converter - URL Fetcher

Usage:
  node scripts/convert-url.js <url> [output-file]
  node scripts/convert-url.js --help

Arguments:
  url           URL of the webpage to convert
  output-file   (Optional) Path for the output Markdown file
                If not provided, will create a timestamped file in output/

Options:
  --help, -h    Show this help message

Examples:
  node scripts/convert-url.js https://example.com
  node scripts/convert-url.js https://example.com output/example.md
  node scripts/convert-url.js https://github.com/user/repo/blob/main/README.md
  `);
  process.exit(0);
}

const url = args[0];
let outputFile = args[1];

// Validate URL
try {
  new URL(url);
} catch (e) {
  console.error(`Error: Invalid URL '${url}'`);
  console.error('Please provide a valid URL starting with http:// or https://');
  process.exit(1);
}

/**
 * Fetch HTML content from URL
 */
function fetchUrl(url) {
  return new Promise((resolve, reject) => {
    const protocol = url.startsWith('https') ? https : http;

    console.log(`Fetching content from: ${url}`);

    const request = protocol.get(url, {
      headers: {
        'User-Agent': 'Mozilla/5.0 (compatible; HTML-to-Markdown-Converter/1.0)'
      }
    }, (response) => {
      // Handle redirects
      if (response.statusCode >= 300 && response.statusCode < 400 && response.headers.location) {
        console.log(`Following redirect to: ${response.headers.location}`);
        fetchUrl(response.headers.location).then(resolve).catch(reject);
        return;
      }

      if (response.statusCode !== 200) {
        reject(new Error(`HTTP ${response.statusCode}: ${response.statusMessage}`));
        return;
      }

      let data = '';
      response.on('data', chunk => data += chunk);
      response.on('end', () => resolve(data));
    });

    request.on('error', reject);
    request.setTimeout(30000, () => {
      request.destroy();
      reject(new Error('Request timeout after 30 seconds'));
    });
  });
}

/**
 * Extract domain name from URL for filename
 */
function getDomainName(url) {
  try {
    const urlObj = new URL(url);
    return urlObj.hostname.replace(/^www\./, '');
  } catch (e) {
    return 'webpage';
  }
}

// Main execution
(async () => {
  try {
    // Fetch HTML from URL
    const html = await fetchUrl(url);

    if (!html || html.trim().length === 0) {
      console.error('Error: No content received from URL');
      process.exit(1);
    }

    console.log(`✓ Content fetched (${html.length} characters)`);

    // Create converter instance
    const converter = new HtmlToMarkdown({
      headingStyle: 'atx',
      codeBlockStyle: 'fenced',
      bulletListMarker: '-',
      strongDelimiter: '**',
      emDelimiter: '_'
    });

    // Convert to Markdown
    console.log('Converting HTML to Markdown...');
    const markdown = converter.convert(html);

    // Generate output filename if not provided
    if (!outputFile) {
      const timestamp = new Date().toISOString().replace(/[:.]/g, '-').slice(0, -5);
      const domain = getDomainName(url);
      outputFile = path.join('output', `${domain}-${timestamp}.md`);

      // Ensure output directory exists
      const outputDir = path.dirname(outputFile);
      if (!fs.existsSync(outputDir)) {
        fs.mkdirSync(outputDir, { recursive: true });
      }
    }

    // Write the Markdown file
    fs.writeFileSync(outputFile, markdown, 'utf8');

    console.log(`✓ Conversion successful!`);
    console.log(`Output saved to: ${outputFile}`);
    console.log(`File size: ${markdown.length} characters`);
    console.log(`\nPreview (first 500 characters):`);
    console.log('─'.repeat(60));
    console.log(markdown.substring(0, 500) + (markdown.length > 500 ? '...' : ''));
    console.log('─'.repeat(60));

  } catch (error) {
    console.error(`\n✗ Error: ${error.message}`);

    if (error.code === 'ENOTFOUND') {
      console.error('Could not resolve hostname. Please check the URL and your internet connection.');
    } else if (error.code === 'ECONNREFUSED') {
      console.error('Connection refused. The server may be down or unreachable.');
    } else if (error.message.includes('timeout')) {
      console.error('Request timed out. The server may be slow or unresponsive.');
    }

    process.exit(1);
  }
})();

// Made with Bob

Application Workflow

The following logic flow governs the transition from a URL to a downloaded Markdown file:

Request Phase: User provides a URL or file path.
Extraction Phase: The system fetches the content: it isolates the main content body (often stripping headers/footers if logic is applied).
Transformation Phase: HTML tags are mapped to Markdown equivalents (e.g.,
becomes #) : code blocks are wrapped in triple backticks.
Finalization: Metadata (Title, Date, Source URL) is prepended as Front Matter: the resulting string is saved as a .md file.

Last but not Least: Extension for a Browser

The extension follows the standard Chrome “V3” architecture, distributed across three main functional areas:

The Manifest (Orchestrator)

File: manifest.json
Role: Defines the permissions and entry points. It specifically requests activeTab access to read the page content and downloads permissions to save the final .md file to your computer.

The Popup (User Interface)

Files: popup.html, popup.js

/**
 * Chrome Extension Popup Script
 * Handles user interactions and conversion logic
 */

let currentMarkdown = '';

// Initialize when popup opens
document.addEventListener('DOMContentLoaded', () => {
  const convertBtn = document.getElementById('convert-btn');
  const copyBtn = document.getElementById('copy-btn');
  const downloadBtn = document.getElementById('download-btn');
  const markdownOutput = document.getElementById('markdown-output');
  const headingStyle = document.getElementById('heading-style');
  const codeStyle = document.getElementById('code-style');

  // Convert button click
  convertBtn.addEventListener('click', async () => {
    try {
      showStatus('Converting...', 'info');
      convertBtn.disabled = true;

      // Get the active tab
      const [tab] = await chrome.tabs.query({ active: true, currentWindow: true });

      // Execute script in the page to get HTML
      const results = await chrome.scripting.executeScript({
        target: { tabId: tab.id },
        function: getPageHTML
      });

      if (results && results[0] && results[0].result) {
        const html = results[0].result;

        // Convert HTML to Markdown
        const converter = new HtmlToMarkdown({
          headingStyle: headingStyle.value,
          codeBlockStyle: codeStyle.value
        });

        currentMarkdown = converter.convert(html);
        markdownOutput.value = currentMarkdown;

        // Enable buttons
        copyBtn.disabled = false;
        downloadBtn.disabled = false;

        showStatus('✓ Conversion successful!', 'success');
      } else {
        throw new Error('Could not retrieve page content');
      }
    } catch (error) {
      console.error('Conversion error:', error);
      showStatus('✗ Error: ' + error.message, 'error');
    } finally {
      convertBtn.disabled = false;
    }
  });

  // Copy button click
  copyBtn.addEventListener('click', async () => {
    try {
      await navigator.clipboard.writeText(currentMarkdown);
      showStatus('✓ Copied to clipboard!', 'success');

      // Visual feedback
      const originalText = copyBtn.textContent;
      copyBtn.textContent = '✓ Copied!';
      setTimeout(() => {
        copyBtn.textContent = originalText;
      }, 2000);
    } catch (error) {
      console.error('Copy error:', error);
      showStatus('✗ Failed to copy', 'error');
    }
  });

  // Download button click
  downloadBtn.addEventListener('click', () => {
    try {
      const blob = new Blob([currentMarkdown], { type: 'text/markdown' });
      const url = URL.createObjectURL(blob);
      const timestamp = new Date().toISOString().replace(/[:.]/g, '-').slice(0, -5);
      const filename = `converted-${timestamp}.md`;

      const a = document.createElement('a');
      a.href = url;
      a.download = filename;
      document.body.appendChild(a);
      a.click();
      document.body.removeChild(a);
      URL.revokeObjectURL(url);

      showStatus('✓ Downloaded!', 'success');
    } catch (error) {
      console.error('Download error:', error);
      showStatus('✗ Failed to download', 'error');
    }
  });

  // Update conversion when options change
  headingStyle.addEventListener('change', () => {
    if (currentMarkdown) {
      convertBtn.click();
    }
  });

  codeStyle.addEventListener('change', () => {
    if (currentMarkdown) {
      convertBtn.click();
    }
  });
});

/**
 * Function to be executed in the page context
 * Gets the HTML content of the page
 */
function getPageHTML() {
  // Get the main content, preferring article or main tags
  const article = document.querySelector('article');
  const main = document.querySelector('main');
  const body = document.body;

  // Return the most relevant content
  if (article) {
    return article.outerHTML;
  } else if (main) {
    return main.outerHTML;
  } else {
    return body.innerHTML;
  }
}

/**
 * Show status message
 */
function showStatus(message, type) {
  const status = document.getElementById('status');
  status.textContent = message;
  status.className = `status ${type}`;
  status.style.display = 'block';

  // Auto-hide after 3 seconds for success messages
  if (type === 'success') {
    setTimeout(() => {
      status.style.display = 'none';
    }, 3000);
  }
}

// Made with Bob

Logic: This is the “control center.”
Configuration: It allows users to toggle settings like “Heading Style” (ATX vs. Setext) and “Code Block Style” (Fenced vs. Indented) directly from a dropdown.
Communication: When the “Convert” button is clicked, popup.js sends a message to the content script to grab the page data.
Trigger: Once the Markdown is returned, it creates a Blob and triggers a download via the Chrome Downloads API.

The Content Script (The “Brain”)

File: content.js
Logic: This script lives inside the webpage you are viewing.
Extraction: It scrapes the current DOM, targeting the document.body.innerHTML.
Transformation: It bundles the Turndown library logic to convert that HTML string into Markdown on the fly.
Metadata: It automatically grabs the page <title> and URL to create a header for your document.

/**
 * Content Script
 * Runs in the context of web pages
 * Can be used for additional features like context menu conversion
 */

// Listen for messages from the extension
chrome.runtime.onMessage.addListener((request, sender, sendResponse) => {
  if (request.action === 'getHTML') {
    // Get the HTML content
    const html = document.documentElement.outerHTML;
    sendResponse({ html });
  }
  return true;
});

// Optional: Add keyboard shortcut for quick conversion
document.addEventListener('keydown', (e) => {
  // Ctrl+Shift+M or Cmd+Shift+M to trigger conversion
  if ((e.ctrlKey || e.metaKey) && e.shiftKey && e.key === 'M') {
    e.preventDefault();
    // Open the extension popup programmatically
    chrome.runtime.sendMessage({ action: 'openPopup' });
  }
});

console.log('HTML to Markdown extension loaded');

// Made with Bob

Technical Stack Summary

Runtime: Node.js
Parsing: jsdom (for DOM manipulation)
Conversion: turndown + turndown-plugin-gfm
HTTP Client: axios
CLI Assets: create-icons.sh (for generating Chrome Extension icons)

Note from Bob: The logic is designed to be “plug-and-play.” Whether you are running a script in the terminal or clicking the extension button, the underlying conversion rules remain identical to ensure consistent output quality.

Conclusion

To wrap this all up, there is a distinct, borderline-obsessive satisfaction in using a tool where you know every single line of the source code. Sure, there are dozens of converters out there, but this one is my precious code, and that makes it inherently superior.

The real kicker? The entire transition from “I have a burning need” to a fully functional Chrome extension and CLI suite happened in less than 30 minutes. By offloading the heavy lifting to Bob — who, per my strict instructions, ensured everything was backed by unit tests — the development cycle was essentially at warp speed.

>>> Thanks for reading <<<

Image Generation with Ollama is back with Japanese, Korean and Chinese Languages 🇯🇵 Support!

Alain Airom (Ayrom) — Wed, 22 Apr 2026 07:18:39 +0000

An international collaborative work 🎌

Introduction

At the start of the year, I developed an image generation application using Bob to build a custom interface. The backend integrates the ‘x/flux2-klein’ and ‘x/z-image-turbo’ models alongside Ollama. Recently, the application received significant enhancements thanks to contributions from my colleague, Akira Onishi-san, in Japan.

What’s truly fantastic is the way my colleague transformed the app; by leveraging Bob, he made the architecture modular and much more secure. He also went the extra mile by refining the bash scripts, making the automatic startup and shutdown processes more robust than ever.

So let’s dig into the application’s enhancements.

Implementation

The architecture of the application remains the same, but the main enhanced modules are provided hereafter (the change log is accessible on GitHub)👇

app.js: The latest update to app.js introduces significant hardening and performance tracking. On the security front, we’ve eliminated vulnerabilities by replacing innerHTML with safe DOM manipulation and implementing filename sanitization to prevent directory traversal. Additionally, we’ve integrated memory usage tracking within the history logs and refined the IME composition handling to ensure a seamless experience for Japanese input.

// Application state
let generationHistory = [];

// DOM elements
const promptInput = document.getElementById('prompt');
const modelSelect = document.getElementById('model');
const generateBtn = document.getElementById('generateBtn');
const resultContainer = document.getElementById('result');
const statusDiv = document.getElementById('status');
const historyList = document.getElementById('historyList');
const btnText = generateBtn.querySelector('.btn-text');
const loader = generateBtn.querySelector('.loader');

// Check Ollama connection on load
checkOllamaConnection();

// Event listeners
generateBtn.addEventListener('click', generateImage);

// Check if Ollama is running
async function checkOllamaConnection() {
    statusDiv.textContent = 'Checking connection...';
    statusDiv.className = 'status checking';

    try {
        const response = await fetch('/api/health');
        const data = await response.json();

        if (data.status === 'ok') {
            statusDiv.textContent = '✓ Connected to Ollama';
            statusDiv.className = 'status connected';
        } else {
            statusDiv.textContent = '✗ Ollama not connected';
            statusDiv.className = 'status disconnected';
        }
    } catch (error) {
        statusDiv.textContent = '✗ Cannot connect to server';
        statusDiv.className = 'status disconnected';
    }
}

// Generate image function
async function generateImage() {
    const prompt = promptInput.value.trim();
    const model = modelSelect.value;

    if (!prompt) {
        alert('Please enter a prompt');
        return;
    }

    // Disable button and show loading state
    generateBtn.disabled = true;
    btnText.textContent = 'Generating...';
    loader.style.display = 'block';

    // Clear previous result
    resultContainer.innerHTML = '';
    const placeholder = document.createElement('div');
    placeholder.className = 'placeholder';
    const p = document.createElement('p');
    p.textContent = 'Generating your image...';
    placeholder.appendChild(p);
    resultContainer.appendChild(placeholder);

    try {
        const response = await fetch('/api/generate', {
            method: 'POST',
            headers: {
                'Content-Type': 'application/json'
            },
            body: JSON.stringify({ prompt, model })
        });

        const data = await response.json();

        if (data.success) {
            displayResult(data, prompt, model);
            addToHistory(prompt, model, data.result, data.memoryUsage);
        } else {
            displayError(data.error || 'Failed to generate image');
        }
    } catch (error) {
        displayError('Network error: ' + error.message);
    } finally {
        // Re-enable button
        generateBtn.disabled = false;
        btnText.textContent = 'Generate Image';
        loader.style.display = 'none';
    }
}

// Display the result
function displayResult(data, prompt, model) {
    const resultContent = document.createElement('div');
    resultContent.className = 'result-content';

    // Check if result contains image data (base64)
    if (data.result && (data.result.startsWith('data:image') || data.result.includes('base64'))) {
        const img = document.createElement('img');
        img.src = data.result;
        img.alt = 'Generated image';
        img.className = 'result-image';
        resultContent.appendChild(img);

        // Add download button
        const downloadBtn = document.createElement('button');
        downloadBtn.className = 'download-btn';
        downloadBtn.textContent = '⬇️ Download Image';
        downloadBtn.onclick = () => downloadImage(data.result, prompt);
        resultContent.appendChild(downloadBtn);
    } else {
        // Display as text if not an image
        const textDiv = document.createElement('div');
        textDiv.className = 'result-text';
        textDiv.textContent = data.result;
        resultContent.appendChild(textDiv);
    }

    // Add info section
    const infoDiv = document.createElement('div');
    infoDiv.className = 'result-info';

    const modelLabel = document.createElement('strong');
    modelLabel.textContent = 'Model:';
    infoDiv.appendChild(modelLabel);
    infoDiv.appendChild(document.createTextNode(' ' + model));
    infoDiv.appendChild(document.createElement('br'));

    const promptLabel = document.createElement('strong');
    promptLabel.textContent = 'Prompt:';
    infoDiv.appendChild(promptLabel);
    infoDiv.appendChild(document.createTextNode(' ' + prompt));

    resultContent.appendChild(infoDiv);

    resultContainer.innerHTML = '';
    resultContainer.appendChild(resultContent);
}

// Download image with custom filename
function downloadImage(imageDataUrl, defaultPrompt) {
    // Generate a default filename from the prompt (sanitized)
    const sanitizedPrompt = defaultPrompt
        .toLowerCase()
        .replace(/[^a-z0-9\s]/g, '')
        .replace(/\s+/g, '-')
        .substring(0, 50);

    const timestamp = new Date().toISOString().split('T')[0];
    const defaultFilename = `${sanitizedPrompt}-${timestamp}.png`;

    // Prompt user for filename
    const filename = prompt('Enter filename for the image:', defaultFilename);

    // If user cancels, don't download
    if (!filename) return;

    // Remove all path separators to prevent directory traversal
    const sanitizedFilename = filename.replace(/[^a-zA-Z0-9._-]/g, '_');

    // Ensure filename has .png extension
    const finalFilename = sanitizedFilename.endsWith('.png') ? sanitizedFilename : `${sanitizedFilename}.png`;

    // Create a temporary link element and trigger download
    const link = document.createElement('a');
    link.href = imageDataUrl;
    link.download = finalFilename;
    document.body.appendChild(link);
    link.click();
    document.body.removeChild(link);
}

// Display error message
function displayError(message) {
    const errorDiv = document.createElement('div');
    errorDiv.className = 'error-message';

    const strong = document.createElement('strong');
    strong.textContent = 'Error:';

    const messageText = document.createTextNode(' ' + message);

    const br1 = document.createElement('br');
    const br2 = document.createElement('br');

    const small = document.createElement('small');
    small.textContent = 'Make sure Ollama is running and the models are installed.';

    errorDiv.appendChild(strong);
    errorDiv.appendChild(messageText);
    errorDiv.appendChild(br1);
    errorDiv.appendChild(br2);
    errorDiv.appendChild(small);

    resultContainer.innerHTML = '';
    resultContainer.appendChild(errorDiv);
}

// Add to history
function addToHistory(prompt, model, result, memoryUsage) {
    const historyItem = {
        prompt,
        model,
        result,
        timestamp: new Date().toLocaleString(),
        memoryUsage: memoryUsage || null
    };

    generationHistory.unshift(historyItem);

    // Keep only last 10 items
    if (generationHistory.length > 10) {
        generationHistory.pop();
    }

    updateHistoryDisplay();
}

// Update history display
function updateHistoryDisplay() {
    historyList.innerHTML = '';

    if (generationHistory.length === 0) {
        const emptyMessage = document.createElement('p');
        emptyMessage.style.color = '#9ca3af';
        emptyMessage.style.textAlign = 'center';
        emptyMessage.textContent = 'No generation history yet';
        historyList.appendChild(emptyMessage);
        return;
    }

    generationHistory.forEach((item, index) => {
        const historyItemDiv = document.createElement('div');
        historyItemDiv.className = 'history-item';

        const promptDiv = document.createElement('div');
        promptDiv.className = 'history-item-prompt';
        promptDiv.textContent = item.prompt;
        historyItemDiv.appendChild(promptDiv);

        const modelDiv = document.createElement('div');
        modelDiv.className = 'history-item-model';
        modelDiv.textContent = item.model;
        historyItemDiv.appendChild(modelDiv);

        const timeDiv = document.createElement('div');
        timeDiv.className = 'history-item-time';
        timeDiv.textContent = item.timestamp;
        historyItemDiv.appendChild(timeDiv);

        if (item.memoryUsage) {
            const memoryDiv = document.createElement('div');
            memoryDiv.className = 'history-item-memory';
            memoryDiv.textContent = `Memory: RSS +${item.memoryUsage.rssDiff}MB, Heap +${item.memoryUsage.heapUsedDiff}MB`;
            historyItemDiv.appendChild(memoryDiv);
        }

        historyItemDiv.addEventListener('click', () => {
            promptInput.value = item.prompt;
            modelSelect.value = item.model;
            displayResult({ result: item.result }, item.prompt, item.model);
        });

        historyList.appendChild(historyItemDiv);
    });
}

// Allow Enter key to submit (with Shift+Enter for new line)
// Ignore Enter key during Japanese IME composition
promptInput.addEventListener('keydown', (e) => {
    if (e.key === 'Enter' && !e.shiftKey && !e.isComposing) {
        e.preventDefault();
        generateImage();
    }
});

// Made with Bob

server.js: The server.js file has been completely refactored into a modular, well-documented architecture using helper functions. To improve observability, we integrated c*omprehensive memory tracking* (monitoring RSS, heap, and external usage), alongside centralized configuration constants for easier maintenance. Finally, the service is now more resilient and easier to troubleshoot thanks to enhanced error handling, robust response parsing, and the addition of conditional debug logging.

const express = require('express');
const cors = require('cors');
const axios = require('axios');
const { exec } = require('child_process');
const util = require('util');
const path = require('path');

const execPromise = util.promisify(exec);

const app = express();
const PORT = process.env.PORT || 3000;
const OLLAMA_API_URL = 'http://localhost:11434';
const MAX_CONTENT_SIZE_BYTES = 50 * 1024 * 1024; // 50MB
const IMAGE_GENERATION_TIMEOUT_MS = 180000; // 3 minutes timeout for image generation
const OLLAMA_REQUEST_CONFIG = {
    timeout: IMAGE_GENERATION_TIMEOUT_MS,
    maxContentLength: MAX_CONTENT_SIZE_BYTES,
    maxBodyLength: MAX_CONTENT_SIZE_BYTES
};

app.use(cors());
app.use(express.json());
app.use(express.static('public'));

// ============================================================================
// Helper Functions
// ============================================================================

/**
 * Validate request body for image generation
 */
function validateRequest(body) {
    const { prompt, model } = body;
    if (!prompt || !model) {
        return { valid: false, error: 'Prompt and model are required' };
    }
    return { valid: true, prompt, model };
}

/**
 * Measure current memory usage
 */
function measureMemory() {
    return process.memoryUsage();
}

/**
 * Calculate and log memory usage difference
 */
function logMemoryUsage(memBefore, memAfter) {
    const memDiff = {
        rss: ((memAfter.rss - memBefore.rss) / 1024 / 1024).toFixed(2),
        heapTotal: ((memAfter.heapTotal - memBefore.heapTotal) / 1024 / 1024).toFixed(2),
        heapUsed: ((memAfter.heapUsed - memBefore.heapUsed) / 1024 / 1024).toFixed(2),
        external: ((memAfter.external - memBefore.external) / 1024 / 1024).toFixed(2)
    };

    debugLog('=== Memory Usage Before Generation ===');
    debugLog(`RSS: ${(memBefore.rss / 1024 / 1024).toFixed(2)} MB`);
    debugLog(`Heap Total: ${(memBefore.heapTotal / 1024 / 1024).toFixed(2)} MB`);
    debugLog(`Heap Used: ${(memBefore.heapUsed / 1024 / 1024).toFixed(2)} MB`);
    debugLog(`External: ${(memBefore.external / 1024 / 1024).toFixed(2)} MB`);

    debugLog('=== Memory Usage After Generation ===');
    debugLog(`RSS: ${(memAfter.rss / 1024 / 1024).toFixed(2)} MB (Δ ${memDiff.rss} MB)`);
    debugLog(`Heap Total: ${(memAfter.heapTotal / 1024 / 1024).toFixed(2)} MB (Δ ${memDiff.heapTotal} MB)`);
    debugLog(`Heap Used: ${(memAfter.heapUsed / 1024 / 1024).toFixed(2)} MB (Δ ${memDiff.heapUsed} MB)`);
    debugLog(`External: ${(memAfter.external / 1024 / 1024).toFixed(2)} MB (Δ ${memDiff.external} MB)`);

    return memDiff;
}

/**
 * Track memory usage for a generation operation
 */
function trackMemoryUsage(beforeMem) {
    const afterMem = measureMemory();
    return logMemoryUsage(beforeMem, afterMem);
}

/**
 * Conditional debug logging
 */
function debugLog(...args) {
    if (process.env.NODE_ENV === 'development' || process.env.DEBUG === 'true') {
        console.log(...args);
    }
}

/**
 * Extract base64 image data from various response formats
 */
function extractImageData(data) {
    if (data.image) return `data:image/png;base64,${data.image}`;
    if (data.images?.[0]) return `data:image/png;base64,${data.images[0]}`;
    if (Buffer.isBuffer(data)) return `data:image/png;base64,${data.toString('base64')}`;
    return null;
}

/**
 * Parse NDJSON string and return last line as JSON
 */
function parseNDJSON(data) {
    const lines = data.trim().split('\n');
    return JSON.parse(lines[lines.length - 1]);
}

/**
 * Handle string response (NDJSON format)
 */
function parseStringResponse(data, log) {
    log('Response is a string, length:', data.length);

    try {
        const parsed = parseNDJSON(data);
        log('Parsed response keys:', Object.keys(parsed));

        const imageData = extractImageData(parsed);
        if (imageData) {
            log('✓ Found image in parsed response');
            return { imageData, ollamaResponse: '' };
        }

        if (parsed.response) {
            log('Found response field, length:', parsed.response.length);
            return { imageData: null, ollamaResponse: parsed.response };
        }
    } catch (e) {
        log('Failed to parse as JSON:', e.message);
        return { imageData: null, ollamaResponse: data };
    }

    return { imageData: null, ollamaResponse: '' };
}

/**
 * Handle object/Buffer response
 */
function parseObjectResponse(data, log) {
    const imageData = extractImageData(data);
    if (imageData) {
        log('✓ Found image in response');
        return { imageData, ollamaResponse: '' };
    }

    if (data.response) {
        log('Response length:', data.response.length);
        return { imageData: null, ollamaResponse: data.response };
    }

    return { imageData: null, ollamaResponse: '' };
}

/**
 * Parse Ollama API response and extract image/text data
 *
 * CRITICAL: Ollama returns newline-delimited JSON (NDJSON) for image generation
 * Format: Each line is a separate JSON object
 * Example:
 * {"model":"x/flux2-klein:4b","created_at":"...","response":"","done":false}
 * {"model":"x/flux2-klein:4b","created_at":"...","response":"","done":false}
 * {"model":"x/flux2-klein:4b","created_at":"...","done":true,"image":"base64data..."}
 *
 * The LAST line contains the complete image in the 'image' field (singular, not 'images')
 */
function parseOllamaResponse(data) {
    debugLog('Ollama API response received');
    debugLog('Response data type:', typeof data);

    if (typeof data === 'object' && !Buffer.isBuffer(data)) {
        debugLog('Response data keys:', Object.keys(data));
    }

    if (typeof data === 'string') {
        return parseStringResponse(data, debugLog);
    }

    if (typeof data === 'object' || Buffer.isBuffer(data)) {
        return parseObjectResponse(data, debugLog);
    }

    debugLog('✗ No image or response data found');
    return { imageData: null, ollamaResponse: '' };
}

/**
 * Build API response object
 */
function buildResponse({ imageData, ollamaResponse, model, memDiff, responseData }) {
    return {
        success: true,
        result: imageData || ollamaResponse || 'No image generated',
        model,
        hasImage: !!imageData,
        memoryUsage: {
            rssDiff: memDiff.rss,
            heapUsedDiff: memDiff.heapUsed
        },
        debug: {
            responseType: typeof responseData,
            isBuffer: Buffer.isBuffer(responseData),
            responseLength: ollamaResponse.length,
            hasImages: !!(responseData.images)
        }
    };
}

/**
 * Generate image using Ollama API
 */
async function generateImageWithOllama(model, prompt) {
    return axios.post(
        `${OLLAMA_API_URL}/api/generate`,
        { model, prompt, stream: false },
        OLLAMA_REQUEST_CONFIG
    );
}

/**
 * Handle error responses
 */
function handleError(error, res) {
    console.error('Error generating image:', error.message);
    if (error.response?.data) {
        console.error('Error details:', error.response.data);
    }
    if (error.stack) {
        console.error('Stack trace:', error.stack);
    }
    res.status(500).json({
        error: 'Failed to generate image',
        details: error.message,
        response: error.response?.data || ''
    });
}

// ============================================================================
// API Endpoints
// ============================================================================

/**
 * Endpoint to generate image using Ollama HTTP API
 * This endpoint handles image generation requests from the frontend
 */
app.post('/api/generate', async (req, res) => {
    const { valid, error, prompt, model } = validateRequest(req.body);
    if (!valid) {
        return res.status(400).json({ error });
    }

    try {
        const memBefore = measureMemory();
        debugLog(`Generating image with model: ${model}`);

        const response = await generateImageWithOllama(model, prompt);
        const { imageData, ollamaResponse } = parseOllamaResponse(response.data);
        const memDiff = trackMemoryUsage(memBefore);

        res.json(buildResponse({
            imageData,
            ollamaResponse,
            model,
            memDiff,
            responseData: response.data
        }));

    } catch (error) {
        handleError(error, res);
    }
});

// Endpoint to check available models
app.get('/api/models', async (req, res) => {
    try {
        const response = await axios.get('http://localhost:11434/api/tags');
        const models = response.data.models || [];

        res.json({
            models: models.map(m => ({ name: m.name }))
        });
    } catch (error) {
        console.error('Error fetching models:', error.message);
        res.status(500).json({
            error: 'Failed to fetch models',
            details: error.message
        });
    }
});

// Health check endpoint
app.get('/api/health', async (req, res) => {
    try {
        await axios.get('http://localhost:11434/api/tags', { timeout: 5000 });
        res.json({ status: 'ok', ollama: 'connected' });
    } catch (error) {
        res.status(503).json({ status: 'error', ollama: 'disconnected', details: error.message });
    }
});

app.listen(PORT, () => {
    console.log(`Server running on http://localhost:${PORT}`);
    console.log(`Using Ollama API for image generation`);
    console.log(`Supported models: x/flux2-klein:4b, x/z-image-turbo:fp8`);
});

// Made with Bob

Testing the app: For a prompt like “In Japanese: 春の湖、桜と富士山、遠くに小鳥の群れ (or in French: Un lac de source, des cerisiers en fleurs, le mont Fuji 🗻 et une volée de petits oiseaux au loin)” the result is outstanding 👏

That’s a wrap 💯 ㊗️ 🏯

Conclusion

This project serves as a fantastic testament to the power of the open-source community and modern development workflows. By leveraging Bob to modularize and refine a codebase that was originally built with the same framework, we’ve created a cleaner, more recursive architecture that is both secure and highly maintainable. Perhaps most rewarding, however, was the international collaboration via GitHub; seeing a colleague from Japan contribute such high-level technical enhancements perfectly illustrates why GitHub is the premier platform for global participation. It’s a great example of how developers can unite across borders to turn a personal project into a robust, community-driven application.

>>> Thanks for reading <<<

Kafka 101: Why Event Streaming is the Central Nervous System of Modern Data

Alain Airom (Ayrom) — Tue, 21 Apr 2026 09:21:31 +0000

Digging into a new universe of data streaming after IBM’s recent accquisition of Confluent!

Introduction
Data is no longer a static asset — it’s a live pulse. Following IBM’s $11B acquisition of Confluent, the bridge between real-time streaming and enterprise AI has never been shorter. I’m diving into the new universes of data-in-motion to see how this ‘Smart Data’ foundation is changing the game for hybrid cloud and agentic AI. 🚀

In the traditional world of data, systems often talk to each other like a game of “telephone” — one service calls another, which calls another, creating a fragile web of dependencies. If one system goes down or slows down, the entire chain breaks. Event streaming changes this paradigm by turning data into a continuous flow of events.

Apache Kafka acts as the “central nervous system” for this flow. Instead of direct connections, services publish events to a central cluster, and any interested service can “tune in” to listen. This decoupling allows businesses to react to customer actions in real-time — whether that’s processing a payment, updating a live dashboard, or triggering a delivery — without the systems ever needing to know each other exists.

Implementation: Building an E-commerce Stream

Using two basic code samples I found on the net, one as a “producer” and one as a “consumer” to explain the concepts of Kafka, I asked Bob to build a simple comprehensive application.

Simple Producer

from confluent_kafka import Producer
import json

# Configuration for connecting to the Kafka cluster
config = {'bootstrap.servers': 'localhost:9092'}
producer = Producer(config)

def delivery_report(err, msg):
    if err is not None:
        print(f"Message delivery failed: {err}")
    else:
        print(f"Order sent to {msg.topic()} [{msg.partition()}]")

# Simulate an order
order_data = {
    "order_id": 1001,
    "user": "jane_doe",
    "total": 59.99,
    "items": ["Wireless Mouse", "Keyboard"]
}

# Trigger the send (Asynchronous)
producer.produce(
    'orders', 
    key="1001", 
    value=json.dumps(order_data), 
    callback=delivery_report
)

producer.flush() # Wait for any outstanding messages to be delivered

Simple Consumer

from confluent_kafka import Consumer

config = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'shipping-group', # Helps Kafka track which orders this group has seen
    'auto.offset.reset': 'earliest'
}

consumer = Consumer(config)
consumer.subscribe(['orders'])

try:
    while True:
        msg = consumer.poll(1.0) # Check for new messages every 1 second
        if msg is None: continue
        if msg.error():
            print(f"Consumer error: {msg.error()}")
            continue

        print(f"Received Order: {msg.value().decode('utf-8')}")
        print("Action: Initiating packaging and shipping...")
finally:
    consumer.close()

After reading my conceptual samples, Bob provided the application structure which follows.

The Producer: Capturing the Event

The Producer (found in src/producer.py) acts as the storefront. When an order is placed, it creates a JSON-serializable dictionary containing the order ID, user details, and items. It uses the confluent-kafka library to "produce" this message to a topic called orders, using the order_id as a key to ensure all updates for a specific order land in the same partition. A delivery callback function is utilized to provide immediate feedback on whether the message successfully reached the Kafka broker or if a retry is necessary.

"""
Kafka Producer - E-commerce Order Processing System
Simulates an online store sending order events to Kafka
"""

from confluent_kafka import Producer
import json
import time
import random
from datetime import datetime

# Configuration for connecting to the Kafka cluster
config = {
    'bootstrap.servers': 'localhost:9092',
    'client.id': 'ecommerce-producer'
}

producer = Producer(config)

def delivery_report(err, msg):
    """
    Callback function called once for each message produced to indicate delivery result.
    Triggered by poll() or flush().
    """
    if err is not None:
        print(f"❌ Message delivery failed: {err}")
    else:
        print(f"✅ Order sent to topic '{msg.topic()}' [partition {msg.partition()}] at offset {msg.offset()}")

# Sample product catalog
PRODUCTS = [
    {"name": "Wireless Mouse", "price": 29.99},
    {"name": "Mechanical Keyboard", "price": 89.99},
    {"name": "USB-C Cable", "price": 12.99},
    {"name": "Laptop Stand", "price": 45.00},
    {"name": "Webcam HD", "price": 79.99},
    {"name": "Headphones", "price": 149.99},
    {"name": "Monitor 27\"", "price": 299.99},
    {"name": "Desk Lamp", "price": 34.99}
]

USERS = ["alice_smith", "bob_jones", "charlie_brown", "diana_prince", "eve_adams"]

def generate_order(order_id):
    """Generate a random order"""
    num_items = random.randint(1, 4)
    items = random.sample(PRODUCTS, num_items)

    order_data = {
        "order_id": order_id,
        "user": random.choice(USERS),
        "timestamp": datetime.now().isoformat(),
        "items": [item["name"] for item in items],
        "total": round(sum(item["price"] for item in items), 2),
        "status": "pending"
    }
    return order_data

def produce_orders(num_orders=10, delay=2):
    """
    Produce multiple orders to Kafka

    Args:
        num_orders: Number of orders to generate
        delay: Delay in seconds between orders
    """
    print(f"🚀 Starting E-commerce Order Producer")
    print(f"📊 Will produce {num_orders} orders with {delay}s delay between each")
    print(f"🔗 Connected to Kafka at: {config['bootstrap.servers']}")
    print("-" * 70)

    for i in range(1, num_orders + 1):
        try:
            # Generate order
            order_data = generate_order(1000 + i)

            # Print order details
            print(f"\n📦 Order #{order_data['order_id']}:")
            print(f"   User: {order_data['user']}")
            print(f"   Items: {', '.join(order_data['items'])}")
            print(f"   Total: ${order_data['total']}")

            # Produce message to Kafka
            producer.produce(
                topic='orders',
                key=str(order_data['order_id']),
                value=json.dumps(order_data),
                callback=delivery_report
            )

            # Trigger delivery reports by polling
            producer.poll(0)

            # Wait before next order
            if i < num_orders:
                time.sleep(delay)

        except KeyboardInterrupt:
            print("\n⚠️  Producer interrupted by user")
            break
        except Exception as e:
            print(f"❌ Error producing message: {e}")

    # Wait for all messages to be delivered
    print("\n⏳ Flushing remaining messages...")
    producer.flush()
    print("✅ All orders sent successfully!")

if __name__ == "__main__":
    import sys

    # Parse command line arguments
    num_orders = int(sys.argv[1]) if len(sys.argv) > 1 else 10
    delay = float(sys.argv[2]) if len(sys.argv) > 2 else 2

    try:
        produce_orders(num_orders, delay)
    except KeyboardInterrupt:
        print("\n👋 Producer stopped")
    except Exception as e:
        print(f"❌ Fatal error: {e}")

# Made with Bob

The Consumer: Decoupled Processing

The Consumer logic (found in src/consumer.py) demonstrates the power of Consumer Groups. Multiple independent services—Shipping, Email, and Analytics—all subscribe to the same orders topic simultaneously. Because each service belongs to its own group (e.g., shipping-group), Kafka tracks their progress (offsets) individually. This means the Shipping department can process messages at its own pace without affecting the speed of the Email service or the Analytics dashboard.

"""
Kafka Consumer - E-commerce Order Processing System
Simulates different departments (Shipping, Email, Analytics) consuming order events
"""

from confluent_kafka import Consumer, KafkaError
import json
import sys
from datetime import datetime

def create_consumer(group_id, bootstrap_servers='localhost:9092'):
    """
    Create and configure a Kafka consumer

    Args:
        group_id: Consumer group identifier
        bootstrap_servers: Kafka broker addresses

    Returns:
        Configured Consumer instance
    """
    config = {
        'bootstrap.servers': bootstrap_servers,
        'group.id': group_id,
        'auto.offset.reset': 'earliest',  # Start from beginning if no offset exists
        'enable.auto.commit': True,
        'auto.commit.interval.ms': 1000,
        'session.timeout.ms': 6000,
        'client.id': f'{group_id}-client'
    }

    return Consumer(config)

def process_shipping(order):
    """Process order for shipping department"""
    print(f"\n📦 SHIPPING DEPARTMENT")
    print(f"   Order ID: {order['order_id']}")
    print(f"   Customer: {order['user']}")
    print(f"   Items to pack: {', '.join(order['items'])}")
    print(f"   ✅ Initiating packaging and shipping process...")

def process_email(order):
    """Process order for email service"""
    print(f"\n📧 EMAIL SERVICE")
    print(f"   Order ID: {order['order_id']}")
    print(f"   Recipient: {order['user']}")
    print(f"   Total: ${order['total']}")
    print(f"   ✅ Sending order confirmation email...")

def process_analytics(order):
    """Process order for analytics dashboard"""
    print(f"\n📊 ANALYTICS DASHBOARD")
    print(f"   Order ID: {order['order_id']}")
    print(f"   Revenue: ${order['total']}")
    print(f"   Items count: {len(order['items'])}")
    print(f"   Timestamp: {order['timestamp']}")
    print(f"   ✅ Updating live sales dashboard...")

# Department processors mapping
PROCESSORS = {
    'shipping-group': process_shipping,
    'email-group': process_email,
    'analytics-group': process_analytics
}

def consume_orders(group_id, topic='orders'):
    """
    Consume orders from Kafka topic

    Args:
        group_id: Consumer group (shipping-group, email-group, analytics-group)
        topic: Kafka topic to subscribe to
    """
    consumer = create_consumer(group_id)
    consumer.subscribe([topic])

    processor = PROCESSORS.get(group_id, process_shipping)
    department = group_id.replace('-group', '').upper()

    print(f"🚀 Starting {department} Consumer")
    print(f"👥 Consumer Group: {group_id}")
    print(f"📋 Subscribed to topic: {topic}")
    print(f"🔗 Connected to Kafka at: localhost:9092")
    print("-" * 70)
    print("⏳ Waiting for messages... (Press Ctrl+C to stop)\n")

    try:
        message_count = 0
        while True:
            # Poll for messages (timeout in seconds)
            msg = consumer.poll(timeout=1.0)

            if msg is None:
                continue

            if msg.error():
                if msg.error().code() == KafkaError._PARTITION_EOF:
                    # End of partition event
                    print(f"📍 Reached end of partition {msg.partition()}")
                else:
                    print(f"❌ Consumer error: {msg.error()}")
                continue

            # Process the message
            try:
                message_count += 1
                order_data = json.loads(msg.value().decode('utf-8'))

                print(f"\n{'='*70}")
                print(f"📨 Message #{message_count} received from partition {msg.partition()} at offset {msg.offset()}")

                # Call the appropriate processor
                processor(order_data)

                print(f"{'='*70}")

            except json.JSONDecodeError as e:
                print(f"❌ Failed to decode message: {e}")
            except Exception as e:
                print(f"❌ Error processing message: {e}")

    except KeyboardInterrupt:
        print(f"\n\n⚠️  Consumer interrupted by user")
        print(f"📊 Total messages processed: {message_count}")

    finally:
        # Close the consumer to commit final offsets
        print("🔒 Closing consumer...")
        consumer.close()
        print("✅ Consumer closed successfully")

if __name__ == "__main__":
    # Parse command line arguments
    if len(sys.argv) < 2:
        print("Usage: python consumer.py <group_id> [topic]")
        print("\nAvailable consumer groups:")
        print("  - shipping-group   : Processes orders for shipping")
        print("  - email-group      : Sends confirmation emails")
        print("  - analytics-group  : Updates analytics dashboard")
        print("\nExample: python consumer.py shipping-group")
        sys.exit(1)

    group_id = sys.argv[1]
    topic = sys.argv[2] if len(sys.argv) > 2 else 'orders'

    if group_id not in PROCESSORS:
        print(f"⚠️  Warning: Unknown group_id '{group_id}'. Using default processor.")

    try:
        consume_orders(group_id, topic)
    except Exception as e:
        print(f"❌ Fatal error: {e}")
        sys.exit(1)

# Made with Bob

The Cluster: Ensuring Reliability

The heart of the system is the Broker Kafka Cluster (defined in docker-compose.yml and [Architecture.md](https://github.com/aairom/kafka-101/blob/main/Docs/Architecture.md)). By running three brokers coordinated by Zookeeper, the architecture ensures High Availability. Each data partition is replicated across at least two brokers; if one broker fails, another automatically takes over as the "leader," ensuring that no orders are lost and the system remains online. This fault tolerance is what makes Kafka "production-ready" compared to simple message queues.

version: '3.8'

services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.5.0
    hostname: zookeeper
    container_name: zookeeper
    ports:
      - "2181:2181"
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    volumes:
      - zookeeper-data:/var/lib/zookeeper/data
      - zookeeper-logs:/var/lib/zookeeper/log

  kafka-broker-1:
    image: confluentinc/cp-kafka:7.5.0
    hostname: kafka-broker-1
    container_name: kafka-broker-1
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
      - "19092:19092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka-broker-1:29092,PLAINTEXT_HOST://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 2
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 2
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'true'
      KAFKA_JMX_PORT: 9101
      KAFKA_JMX_HOSTNAME: localhost
    volumes:
      - kafka-broker-1-data:/var/lib/kafka/data

  kafka-broker-2:
    image: confluentinc/cp-kafka:7.5.0
    hostname: kafka-broker-2
    container_name: kafka-broker-2
    depends_on:
      - zookeeper
    ports:
      - "9093:9093"
      - "19093:19093"
    environment:
      KAFKA_BROKER_ID: 2
      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka-broker-2:29093,PLAINTEXT_HOST://localhost:9093
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 2
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 2
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'true'
      KAFKA_JMX_PORT: 9102
      KAFKA_JMX_HOSTNAME: localhost
    volumes:
      - kafka-broker-2-data:/var/lib/kafka/data

  kafka-broker-3:
    image: confluentinc/cp-kafka:7.5.0
    hostname: kafka-broker-3
    container_name: kafka-broker-3
    depends_on:
      - zookeeper
    ports:
      - "9094:9094"
      - "19094:19094"
    environment:
      KAFKA_BROKER_ID: 3
      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka-broker-3:29094,PLAINTEXT_HOST://localhost:9094
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 2
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 2
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'true'
      KAFKA_JMX_PORT: 9103
      KAFKA_JMX_HOSTNAME: localhost
    volumes:
      - kafka-broker-3-data:/var/lib/kafka/data

volumes:
  zookeeper-data:
  zookeeper-logs:
  kafka-broker-1-data:
  kafka-broker-2-data:
  kafka-broker-3-data:

# Made with Bob

The Monitor: Real-Time Observability

To manage this distributed flow, the Monitor Runner (monitor_runner.py) provides a unified view of the entire ecosystem. It orchestrates the lifecycle of the producer and all consumer threads while capturing their standard output into a web-based dashboard. This allows developers to see the "path of a message" in real-time: from the moment the Producer sends Order #1001 to the moment all three consumer groups acknowledge and process it.

"""
Kafka Monitor Web Server
Provides a web interface to monitor producer and consumer outputs in real-time
"""

import asyncio
import json
import subprocess
import threading
from datetime import datetime
from pathlib import Path
from http.server import HTTPServer, SimpleHTTPRequestHandler
import socketserver

class TerminalMonitor:
    """Monitor terminal outputs and store them"""

    def __init__(self):
        self.outputs = {
            'producer': [],
            'shipping': [],
            'email': [],
            'analytics': []
        }
        self.max_lines = 1000  # Keep last 1000 lines per terminal

    def add_output(self, terminal_id, line):
        """Add a line to terminal output"""
        timestamp = datetime.now().strftime('%H:%M:%S')
        entry = {
            'timestamp': timestamp,
            'line': line
        }

        if terminal_id in self.outputs:
            self.outputs[terminal_id].append(entry)
            # Keep only last max_lines
            if len(self.outputs[terminal_id]) > self.max_lines:
                self.outputs[terminal_id] = self.outputs[terminal_id][-self.max_lines:]

    def get_outputs(self):
        """Get all terminal outputs"""
        return self.outputs

    def clear_terminal(self, terminal_id):
        """Clear a specific terminal"""
        if terminal_id in self.outputs:
            self.outputs[terminal_id] = []

    def clear_all(self):
        """Clear all terminals"""
        for terminal_id in self.outputs:
            self.outputs[terminal_id] = []

# Global monitor instance
monitor = TerminalMonitor()

class MonitorRequestHandler(SimpleHTTPRequestHandler):
    """Custom HTTP request handler"""

    def do_GET(self):
        """Handle GET requests"""
        if self.path == '/':
            self.path = '/monitor.html'
        elif self.path == '/api/outputs':
            self.send_response(200)
            self.send_header('Content-type', 'application/json')
            self.send_header('Access-Control-Allow-Origin', '*')
            self.end_headers()
            self.wfile.write(json.dumps(monitor.get_outputs()).encode())
            return
        elif self.path == '/api/clear':
            monitor.clear_all()
            self.send_response(200)
            self.send_header('Content-type', 'application/json')
            self.end_headers()
            self.wfile.write(json.dumps({'status': 'cleared'}).encode())
            return

        return SimpleHTTPRequestHandler.do_GET(self)

    def log_message(self, format, *args):
        """Suppress default logging"""
        pass

def run_command_and_monitor(command, terminal_id, cwd=None):
    """Run a command and monitor its output"""
    try:
        process = subprocess.Popen(
            command,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            text=True,
            bufsize=1,
            cwd=cwd,
            shell=True
        )

        monitor.add_output(terminal_id, f"🚀 Starting: {command}")

        for line in iter(process.stdout.readline, ''):
            if line:
                monitor.add_output(terminal_id, line.rstrip())

        process.wait()
        monitor.add_output(terminal_id, f"✅ Process completed with exit code: {process.returncode}")

    except Exception as e:
        monitor.add_output(terminal_id, f"❌ Error: {str(e)}")

def start_web_server(port=8080):
    """Start the web server"""
    handler = MonitorRequestHandler

    with socketserver.TCPServer(("", port), handler) as httpd:
        print(f"🌐 Web Monitor Server running at http://localhost:{port}")
        print(f"📊 Open your browser to view the dashboard")
        print(f"Press Ctrl+C to stop")
        httpd.serve_forever()

def main():
    """Main function"""
    import sys

    if len(sys.argv) > 1 and sys.argv[1] == 'server':
        # Just run the web server
        port = int(sys.argv[2]) if len(sys.argv) > 2 else 8080
        start_web_server(port)
    else:
        print("Usage:")
        print("  python3 src/web_monitor.py server [port]")
        print("\nExample:")
        print("  python3 src/web_monitor.py server 8080")

if __name__ == "__main__":
    main()

# Made with Bob

┌─────────────────────────────────────────────────────────────┐
│  ⚡ Kafka Cluster Monitor              [Live] [🔄] [🗑️]       │
├──────────────────────────────┬──────────────────────────────┤
│  📤 Producer                 │  📦 Shipping Consumer         │
│  [0 lines]                   │  [0 lines]                   │
│                              │                              │
│  [09:30:15] 🚀 Starting...  │  [09:30:17] 🚀 Starting...     │
│  [09:30:16] Order #1001     │  [09:30:18] Received Order    │
│  [09:30:16] ✅ Sent         │  [09:30:18] ✅ Processing      │
│                              │                              │
├──────────────────────────────┼──────────────────────────────┤
│  📧 Email Consumer           │  📊 Analytics Consumer        │
│  [0 lines]                   │  [0 lines]                   │
│                              │                              │
│  [09:30:17] 🚀 Starting...  │  [09:30:17] 🚀 Starting...     │
│  [09:30:18] Received Order   │  [09:30:18] Received Order   │
│  [09:30:18] ✅ Email sent    │  [09:30:18] ✅ Updated        │
│                              │                              │
└──────────────────────────────┴──────────────────────────────┘
│  Last updated: 09:30:20                    Auto-scroll: ON  │
└─────────────────────────────────────────────────────────────┘

Alongside with the monitor, we can also monitor the events on the console output.

./scripts/run-producer.sh 10 2
🚀 Starting Kafka Producer
=========================

📊 Configuration:
   - Number of orders: 10
   - Delay between orders: 2s

🚀 Starting E-commerce Order Producer
📊 Will produce 10 orders with 2.0s delay between each
🔗 Connected to Kafka at: localhost:9092
----------------------------------------------------------------------

📦 Order #1001:
   User: bob_jones
   Items: USB-C Cable, Mechanical Keyboard, Desk Lamp
   Total: $137.97

📦 Order #1002:
   User: diana_prince
   Items: Wireless Mouse
   Total: $29.99
✅ Order sent to topic 'orders' [partition 0] at offset 0

📦 Order #1003:
   User: bob_jones
   Items: USB-C Cable
   Total: $12.99
✅ Order sent to topic 'orders' [partition 0] at offset 1

📦 Order #1004:
   User: bob_jones
   Items: Mechanical Keyboard, Headphones, Webcam HD
   Total: $319.97
✅ Order sent to topic 'orders' [partition 0] at offset 2

📦 Order #1005:
   User: diana_prince
   Items: Laptop Stand, Monitor 27", Wireless Mouse, Webcam HD
   Total: $454.97
✅ Order sent to topic 'orders' [partition 0] at offset 3

📦 Order #1006:
   User: diana_prince
   Items: Headphones, Laptop Stand
   Total: $194.99
✅ Order sent to topic 'orders' [partition 0] at offset 4

📦 Order #1007:
   User: charlie_brown
   Items: Laptop Stand, Headphones, Wireless Mouse
   Total: $224.98
✅ Order sent to topic 'orders' [partition 0] at offset 5

📦 Order #1008:
   User: charlie_brown
   Items: Laptop Stand
   Total: $45.0
✅ Order sent to topic 'orders' [partition 0] at offset 6

📦 Order #1009:
   User: alice_smith
   Items: Mechanical Keyboard, Desk Lamp
   Total: $124.98
✅ Order sent to topic 'orders' [partition 0] at offset 7

📦 Order #1010:
   User: eve_adams
   Items: Wireless Mouse
   Total: $29.99
✅ Order sent to topic 'orders' [partition 0] at offset 8

⏳ Flushing remaining messages...
✅ Order sent to topic 'orders' [partition 0] at offset 9
✅ All orders sent successfully!

✅ Producer finished

./scripts/run-consumer.sh shipping-group
🚀 Starting Kafka Consumer
==========================

📊 Configuration:
   - Consumer Group: shipping-group
   - Topic: orders

🚀 Starting SHIPPING Consumer
👥 Consumer Group: shipping-group
📋 Subscribed to topic: orders
🔗 Connected to Kafka at: localhost:9092
----------------------------------------------------------------------
⏳ Waiting for messages... (Press Ctrl+C to stop)


======================================================================
📨 Message #1 received from partition 0 at offset 0

📦 SHIPPING DEPARTMENT
   Order ID: 1001
   Customer: bob_jones
   Items to pack: USB-C Cable, Mechanical Keyboard, Desk Lamp
   ✅ Initiating packaging and shipping process...
======================================================================

======================================================================
📨 Message #2 received from partition 0 at offset 1

📦 SHIPPING DEPARTMENT
   Order ID: 1002
   Customer: diana_prince
   Items to pack: Wireless Mouse
   ✅ Initiating packaging and shipping process...
======================================================================

======================================================================
📨 Message #3 received from partition 0 at offset 2

📦 SHIPPING DEPARTMENT
   Order ID: 1003
   Customer: bob_jones
   Items to pack: USB-C Cable
   ✅ Initiating packaging and shipping process...
======================================================================

======================================================================
📨 Message #4 received from partition 0 at offset 3

📦 SHIPPING DEPARTMENT
   Order ID: 1004
   Customer: bob_jones
   Items to pack: Mechanical Keyboard, Headphones, Webcam HD
   ✅ Initiating packaging and shipping process...
======================================================================

======================================================================
📨 Message #5 received from partition 0 at offset 4

📦 SHIPPING DEPARTMENT
   Order ID: 1005
   Customer: diana_prince
   Items to pack: Laptop Stand, Monitor 27", Wireless Mouse, Webcam HD
   ✅ Initiating packaging and shipping process...
======================================================================

======================================================================
📨 Message #6 received from partition 0 at offset 5

📦 SHIPPING DEPARTMENT
   Order ID: 1006
   Customer: diana_prince
   Items to pack: Headphones, Laptop Stand
   ✅ Initiating packaging and shipping process...
======================================================================

======================================================================
📨 Message #7 received from partition 0 at offset 6

📦 SHIPPING DEPARTMENT
   Order ID: 1007
   Customer: charlie_brown
   Items to pack: Laptop Stand, Headphones, Wireless Mouse
   ✅ Initiating packaging and shipping process...
======================================================================

======================================================================
📨 Message #8 received from partition 0 at offset 7

📦 SHIPPING DEPARTMENT
   Order ID: 1008
   Customer: charlie_brown
   Items to pack: Laptop Stand
   ✅ Initiating packaging and shipping process...
======================================================================

======================================================================
📨 Message #9 received from partition 0 at offset 8

📦 SHIPPING DEPARTMENT
   Order ID: 1009
   Customer: alice_smith
   Items to pack: Mechanical Keyboard, Desk Lamp
   ✅ Initiating packaging and shipping process...
======================================================================

======================================================================
📨 Message #10 received from partition 0 at offset 9

📦 SHIPPING DEPARTMENT
   Order ID: 1010
   Customer: eve_adams
   Items to pack: Wireless Mouse
   ✅ Initiating packaging and shipping process...
======================================================================

/scripts/run-consumer.sh email-group
zsh: no such file or directory: /scripts/run-consumer.sh
> ./scripts/run-consumer.sh email-group
🚀 Starting Kafka Consumer
==========================

📊 Configuration:
   - Consumer Group: email-group
   - Topic: orders

🚀 Starting EMAIL Consumer
👥 Consumer Group: email-group
📋 Subscribed to topic: orders
🔗 Connected to Kafka at: localhost:9092
----------------------------------------------------------------------
⏳ Waiting for messages... (Press Ctrl+C to stop)


======================================================================
📨 Message #1 received from partition 0 at offset 0

📧 EMAIL SERVICE
   Order ID: 1001
   Recipient: bob_jones
   Total: $137.97
   ✅ Sending order confirmation email...
======================================================================

======================================================================
📨 Message #2 received from partition 0 at offset 1

📧 EMAIL SERVICE
   Order ID: 1002
   Recipient: diana_prince
   Total: $29.99
   ✅ Sending order confirmation email...
======================================================================

======================================================================
📨 Message #3 received from partition 0 at offset 2

📧 EMAIL SERVICE
   Order ID: 1003
   Recipient: bob_jones
   Total: $12.99
   ✅ Sending order confirmation email...
======================================================================

======================================================================
📨 Message #4 received from partition 0 at offset 3

📧 EMAIL SERVICE
   Order ID: 1004
   Recipient: bob_jones
   Total: $319.97
   ✅ Sending order confirmation email...
======================================================================

======================================================================
📨 Message #5 received from partition 0 at offset 4

📧 EMAIL SERVICE
   Order ID: 1005
   Recipient: diana_prince
   Total: $454.97
   ✅ Sending order confirmation email...
======================================================================

======================================================================
📨 Message #6 received from partition 0 at offset 5

📧 EMAIL SERVICE
   Order ID: 1006
   Recipient: diana_prince
   Total: $194.99
   ✅ Sending order confirmation email...
======================================================================

======================================================================
📨 Message #7 received from partition 0 at offset 6

📧 EMAIL SERVICE
   Order ID: 1007
   Recipient: charlie_brown
   Total: $224.98
   ✅ Sending order confirmation email...
======================================================================

======================================================================
📨 Message #8 received from partition 0 at offset 7

📧 EMAIL SERVICE
   Order ID: 1008
   Recipient: charlie_brown
   Total: $45.0
   ✅ Sending order confirmation email...
======================================================================

======================================================================
📨 Message #9 received from partition 0 at offset 8

📧 EMAIL SERVICE
   Order ID: 1009
   Recipient: alice_smith
   Total: $124.98
   ✅ Sending order confirmation email...
======================================================================

======================================================================
📨 Message #10 received from partition 0 at offset 9

📧 EMAIL SERVICE
   Order ID: 1010
   Recipient: eve_adams
   Total: $29.99
   ✅ Sending order confirmation email...
======================================================================

There we go with out Kafka cluster! That’s a wrap 💯

Conclusion
Streaming matters because data is most valuable the moment it is created. By using a Kafka cluster, we move away from “batch processing” and toward real-time intelligence.

The implementation shown here — with a replicated cluster, a robust producer, and independent consumer groups — provides a blueprint for systems that are not only fast but also incredibly resilient. Whether you are scaling to handle thousands of orders per second or simply trying to decouple your microservices, Kafka provides the reliable foundation needed for a modern, event-driven architecture.

>>> Thanks for reading <<<

Mastering Modern Hiring Demonstration: Using Docling and PostgreSQL by Bob to Build a Local Candidate RAG Database

Alain Airom (Ayrom) — Sun, 19 Apr 2026 14:53:57 +0000

This blog post’s idea is to explore how to bridge the gap between traditional relational data and unstructured documents to create a powerful hypothetical recruitment tool.

Introduction

The Candidate RAG System leverages a sophisticated technology stack to turn a folder of resumes into an interactive, searchable knowledge base. By combining Docling’s hierarchical document processing with PostgreSQL’s relational and vector capabilities (via pgvector), the system allows recruiters to query candidate data with natural language while maintaining strict data integrity.

The Architecture: A Hybrid Approach

Bob’s implementation uses a “Relational + Vector” hybrid architecture. While standard candidate details (name, DOB, file paths) reside in traditional SQL tables, the actual content of the resumes is transformed into high-dimensional vectors for semantic search.

Key Components:

Document Processing: Docling is used to extract and chunk resume content from PDF and DOCX formats.
Relational & Vector Storage: A single PostgreSQL instance manages both metadata and embeddings using the pgvector extension.
Local AI Engine: Ollama runs the models locally, using granite-embedding:30m for vectors and granite4:latest for the chat interface.

Technologies Used for this project

## Technologies Used

- **[Docling](https://github.com/docling-project/docling)**: Document parsing and chunking
- **[PostgreSQL](https://www.postgresql.org/)**: Database
- **[pgvector](https://github.com/pgvector/pgvector)**: Vector similarity search
- **[Ollama](https://ollama.ai/)**: Local LLM and embeddings
- **[Streamlit](https://streamlit.io/)**: Web interface
- **[Podman](https://podman.io/)**: Container management

The Data Workflow

The system operates through two primary pipelines: Ingestion and Querying.

1. Document Ingestion Flow

Extraction: Docling converts resumes into hierarchical chunks to preserve document structure.
Vectorization: Ollama generates 384-dimensional embeddings for each text chunk.
Storage: Chunks are saved in the resume_chunks table, linked to the candidates table via a foreign key.

2. RAG Query Flow

When a user asks, “Which candidate has experience with Python?”:

The query is converted into an embedding.
PostgreSQL performs a vector similarity search (embedding<=>query_vector) to find the most relevant resume sections.
The retrieved text is sent to the Ollama LLM as context to generate a natural language answer.

Database Schema Design

Bob utilizes a clean relational schema to ensure that every vector chunk is always tied back to a specific person.

CREATE TABLE candidates (
    id SERIAL PRIMARY KEY,
    firstname VARCHAR(100),
    lastname VARCHAR(100),
    date_of_birth DATE,
    resume_path VARCHAR(255),
    created_at TIMESTAMP
);

CREATE TABLE resume_chunks (
    id SERIAL PRIMARY KEY,
    candidate_id INTEGER REFERENCES candidates(id),
    chunk_text TEXT,
    chunk_index INTEGER,
    embedding vector(384),  -- granite-embedding:30m dimension
    metadata JSONB,
    created_at TIMESTAMP
);

| **Table**         | **Column**                                   | **Type**                       | **Purpose**                               |
| ----------------- | -------------------------------------------- | ------------------------------ | ----------------------------------------- |
| **Candidates**    | `id`, `firstname`, `lastname`, `resume_path` | Serial, Varchar                | Stores primary candidate records.         |
| **Resume Chunks** | `candidate_id`, `chunk_text`, `embedding`    | Integer, Text, **Vector(384)** | Stores the AI-searchable resume segments. |

Project Structure

The following is the implementation and structure of the project;

docling-rag-postgresql/
├── app.py                      # Streamlit GUI application
├── docker-compose.yml          # PostgreSQL setup
├── init.sql                    # Database initialization
├── requirements.txt            # Python dependencies
├── .env.example               # Environment variables template
├── README.md                  # This file
├── Docs/                      # Documentation
│   ├── Architecture.md        # System architecture with Mermaid diagrams
│   └── QUICKSTART.md          # Quick start guide
├── scripts/                   # Utility scripts
│   ├── setup.sh              # Setup script
│   └── run.sh                # Run script
├── resumes/                   # Resume files
│   ├── resume1.docx
│   ├── resume2.docx
│   ├── resume3.docx
│   └── resume4.pdf
└── src/                       # Source code
    ├── __init__.py
    ├── config.py              # Configuration management
    ├── database_service.py    # PostgreSQL operations
    ├── document_processor.py  # Docling document processing
    ├── embedding_service.py   # Ollama embeddings
    ├── rag_service.py         # RAG pipeline
    └── data_loader.py         # Resume loading

Database and LLM

Configuration of the PostgreSQL Database and the local Ollama ⚙️

# Database
DB_HOST=localhost
DB_PORT=5432
DB_NAME=candidates_rag
DB_USER=postgres
DB_PASSWORD=postgres

# Ollama
OLLAMA_HOST=http://localhost:11434
EMBEDDING_MODEL=granite-embedding:30m
LLM_MODEL=granite4:latest

# Application
CHUNK_SIZE=500
CHUNK_OVERLAP=50
TOP_K_RESULTS=5

And the code which goes with is shown below;

"""Configuration management for the RAG application."""
import os
from dotenv import load_dotenv

load_dotenv()


class Config:
    """Application configuration."""

    # Database Configuration
    DB_HOST = os.getenv("DB_HOST", "localhost")
    DB_PORT = int(os.getenv("DB_PORT", "5432"))
    DB_NAME = os.getenv("DB_NAME", "candidates_rag")
    DB_USER = os.getenv("DB_USER", "postgres")
    DB_PASSWORD = os.getenv("DB_PASSWORD", "postgres")

    # Ollama Configuration
    OLLAMA_HOST = os.getenv("OLLAMA_HOST", "http://localhost:11434")
    EMBEDDING_MODEL = os.getenv("EMBEDDING_MODEL", "granite-embedding:30m")
    LLM_MODEL = os.getenv("LLM_MODEL", "granite4:latest")

    # Application Configuration
    CHUNK_SIZE = int(os.getenv("CHUNK_SIZE", "500"))
    CHUNK_OVERLAP = int(os.getenv("CHUNK_OVERLAP", "50"))
    TOP_K_RESULTS = int(os.getenv("TOP_K_RESULTS", "5"))

    @property
    def database_url(self):
        """Get database connection URL."""
        return f"postgresql://{self.DB_USER}:{self.DB_PASSWORD}@{self.DB_HOST}:{self.DB_PORT}/{self.DB_NAME}"


config = Config()

# Made with Bob

Sample Data injection for the database 💉

-- Enable pgvector extension
CREATE EXTENSION IF NOT EXISTS vector;

-- Create candidates table
CREATE TABLE IF NOT EXISTS candidates (
    id SERIAL PRIMARY KEY,
    firstname VARCHAR(100) NOT NULL,
    lastname VARCHAR(100) NOT NULL,
    date_of_birth DATE NOT NULL,
    resume_path VARCHAR(255),
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Create resume_chunks table for storing document chunks with embeddings
CREATE TABLE IF NOT EXISTS resume_chunks (
    id SERIAL PRIMARY KEY,
    candidate_id INTEGER REFERENCES candidates(id) ON DELETE CASCADE,
    chunk_text TEXT NOT NULL,
    chunk_index INTEGER NOT NULL,
    embedding vector(384),  -- granite-embedding:30m produces 384-dimensional vectors
    metadata JSONB,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Create index for vector similarity search
CREATE INDEX IF NOT EXISTS resume_chunks_embedding_idx 
ON resume_chunks USING ivfflat (embedding vector_cosine_ops)
WITH (lists = 100);

-- Create index for candidate lookups
CREATE INDEX IF NOT EXISTS resume_chunks_candidate_id_idx 
ON resume_chunks(candidate_id);

-- Insert sample candidates (will be populated by the application)
INSERT INTO candidates (firstname, lastname, date_of_birth, resume_path) VALUES
    ('John', 'Smith', '1990-05-15', 'resumes/resume1.docx'),
    ('Sarah', 'Johnson', '1988-08-22', 'resumes/resume2.docx'),
    ('Michael', 'Williams', '1992-03-10', 'resumes/resume3.docx'),
    ('Emily', 'Brown', '1995-11-30', 'resumes/resume4.pdf')
ON CONFLICT DO NOTHING;

-- Made with Bob

Information Ingestion

The information ingestion process, detailed below, consists of several critical stages. This integration framework is the core of the application, serving as the essential foundation that enables all advanced RAG and relational operations.

data_loader: This orchestration script manages the high-level workflow of importing resumes from the local file system into the system and it coordinates between the document processor, embedding service, and database service to ensure files are parsed, vectorized, and saved correctly.

# data_loader.py
"""Data loader for processing and loading candidate resumes."""
import logging
from pathlib import Path
from typing import List, Dict, Any

from src.database_service import DatabaseService
from src.document_processor import DocumentProcessor
from src.embedding_service import EmbeddingService

logger = logging.getLogger(__name__)


class DataLoader:
    """Load and process candidate resumes into the database."""

    def __init__(
        self,
        db_service: DatabaseService,
        doc_processor: DocumentProcessor,
        embedding_service: EmbeddingService
    ):
        """Initialize data loader.

        Args:
            db_service: Database service instance
            doc_processor: Document processor instance
            embedding_service: Embedding service instance
        """
        self.db_service = db_service
        self.doc_processor = doc_processor
        self.embedding_service = embedding_service
        logger.info("DataLoader initialized")

    def load_candidate_resume(self, candidate_id: int) -> bool:
        """Load and process a candidate's resume.

        Args:
            candidate_id: Candidate ID

        Returns:
            True if successful, False otherwise
        """
        try:
            # Get candidate info
            candidate = self.db_service.get_candidate_by_id(candidate_id)
            if not candidate:
                logger.error(f"Candidate {candidate_id} not found")
                return False

            resume_path = candidate['resume_path']
            if not resume_path:
                logger.error(f"No resume path for candidate {candidate_id}")
                return False

            # Check if file exists
            if not Path(resume_path).exists():
                logger.error(f"Resume file not found: {resume_path}")
                return False

            logger.info(f"Processing resume for {candidate['firstname']} {candidate['lastname']}")

            # Process document
            chunks = self.doc_processor.process_document(resume_path)

            if not chunks:
                logger.warning(f"No chunks extracted from {resume_path}")
                return False

            # Generate embeddings
            chunk_texts = [chunk['text'] for chunk in chunks]
            embeddings = self.embedding_service.generate_embeddings(chunk_texts)

            # Insert into database
            self.db_service.insert_resume_chunks(
                candidate_id=candidate_id,
                chunks=chunks,
                embeddings=embeddings
            )

            logger.info(f"Successfully loaded resume for candidate {candidate_id}")
            return True

        except Exception as e:
            logger.error(f"Error loading resume for candidate {candidate_id}: {e}")
            return False

    def load_all_resumes(self) -> Dict[int, bool]:
        """Load all candidate resumes.

        Returns:
            Dictionary mapping candidate IDs to success status
        """
        results = {}
        candidates = self.db_service.get_all_candidates()

        logger.info(f"Loading resumes for {len(candidates)} candidates")

        for candidate in candidates:
            candidate_id = candidate['id']
            success = self.load_candidate_resume(candidate_id)
            results[candidate_id] = success

        successful = sum(1 for v in results.values() if v)
        logger.info(f"Successfully loaded {successful}/{len(candidates)} resumes")

        return results

    def reload_candidate_resume(self, candidate_id: int) -> bool:
        """Reload a candidate's resume (delete old chunks and reload).

        Args:
            candidate_id: Candidate ID

        Returns:
            True if successful, False otherwise
        """
        try:
            # Delete existing chunks
            self.db_service.delete_candidate_chunks(candidate_id)

            # Load resume
            return self.load_candidate_resume(candidate_id)

        except Exception as e:
            logger.error(f"Error reloading resume for candidate {candidate_id}: {e}")
            return False

# Made with Bob

document_processor: This code utilizes the Docling library to parse and convert PDF and DOCX resumes into structured text formats, and implements hierarchical chunking to break down long documents into smaller, manageable pieces while preserving the document’s logical structure.

# document_processor.py
"""Document processing with Docling."""
import logging
from pathlib import Path
from typing import List, Dict, Any

from docling.document_converter import DocumentConverter
from docling_core.transforms.chunker import HierarchicalChunker

logger = logging.getLogger(__name__)


class DocumentProcessor:
    """Process documents using Docling."""

    def __init__(self, chunk_size: int = 500, chunk_overlap: int = 50):
        """Initialize document processor.

        Args:
            chunk_size: Maximum size of text chunks
            chunk_overlap: Overlap between chunks
        """
        self.converter = DocumentConverter()
        self.chunker = HierarchicalChunker(
            max_tokens=chunk_size,
            overlap_tokens=chunk_overlap
        )
        logger.info("DocumentProcessor initialized")

    def process_document(self, file_path: str) -> List[Dict[str, Any]]:
        """Process a document and return chunks with metadata.

        Args:
            file_path: Path to the document file

        Returns:
            List of dictionaries containing chunk text and metadata
        """
        try:
            logger.info(f"Processing document: {file_path}")

            # Convert document
            result = self.converter.convert(file_path)
            doc = result.document

            # Chunk the document
            chunks = list(self.chunker.chunk(doc))

            # Extract text and metadata from chunks
            processed_chunks = []
            for idx, chunk in enumerate(chunks):
                chunk_data = {
                    "text": chunk.text,
                    "index": idx,
                    "metadata": {
                        "source": file_path,
                        "chunk_index": idx,
                        "total_chunks": len(chunks)
                    }
                }
                processed_chunks.append(chunk_data)

            logger.info(f"Processed {len(processed_chunks)} chunks from {file_path}")
            return processed_chunks

        except Exception as e:
            logger.error(f"Error processing document {file_path}: {e}")
            raise

    def process_multiple_documents(self, file_paths: List[str]) -> Dict[str, List[Dict[str, Any]]]:
        """Process multiple documents.

        Args:
            file_paths: List of document file paths

        Returns:
            Dictionary mapping file paths to their processed chunks
        """
        results = {}
        for file_path in file_paths:
            try:
                results[file_path] = self.process_document(file_path)
            except Exception as e:
                logger.error(f"Failed to process {file_path}: {e}")
                results[file_path] = []

        return results

# Made with Bob

embedding_service: Managing all interactions with the PostgreSQL database, including storing candidate metadata and resume text chunks. Also, this code leverages the pgvector extension to perform specialized vector similarity searches, allowing the system to find the most relevant resume sections for a given query

# embedding_service.py
"""Embedding service using Ollama."""
import logging
from typing import List, Optional
import ollama

from src.config import config

logger = logging.getLogger(__name__)


class EmbeddingService:
    """Generate embeddings using Ollama."""

    def __init__(self, model: Optional[str] = None, host: Optional[str] = None):
        """Initialize embedding service.

        Args:
            model: Ollama model name for embeddings
            host: Ollama host URL
        """
        self.model = model or config.EMBEDDING_MODEL
        self.host = host or config.OLLAMA_HOST
        self.client = ollama.Client(host=self.host)
        logger.info(f"EmbeddingService initialized with model: {self.model}")

    def generate_embedding(self, text: str) -> List[float]:
        """Generate embedding for a single text.

        Args:
            text: Text to embed

        Returns:
            Embedding vector as list of floats
        """
        try:
            response = self.client.embeddings(
                model=self.model,
                prompt=text
            )
            return response['embedding']
        except Exception as e:
            logger.error(f"Error generating embedding: {e}")
            raise

    def generate_embeddings(self, texts: List[str]) -> List[List[float]]:
        """Generate embeddings for multiple texts.

        Args:
            texts: List of texts to embed

        Returns:
            List of embedding vectors
        """
        embeddings = []
        for text in texts:
            try:
                embedding = self.generate_embedding(text)
                embeddings.append(embedding)
            except Exception as e:
                logger.error(f"Failed to generate embedding for text: {e}")
                # Return zero vector on error
                embeddings.append([0.0] * 384)  # granite-embedding:30m dimension

        return embeddings

    def test_connection(self) -> bool:
        """Test connection to Ollama service.

        Returns:
            True if connection successful, False otherwise
        """
        try:
            # Try to generate a test embedding
            self.generate_embedding("test")
            logger.info("Successfully connected to Ollama")
            return True
        except Exception as e:
            logger.error(f"Failed to connect to Ollama: {e}")
            return False

# Made with Bob

RAG Service

Last but not least, this is the core “brain” of the application that orchestrates the Retrieval-Augmented Generation pipeline by combining retrieved context with user queries. It sends the final enriched prompt to the granite4:latest model via Ollama to generate a natural language response based specifically on the candidate data.

# rag_service.py
"""RAG service for retrieval and generation."""
import logging
from typing import List, Dict, Any, Optional
import ollama

from src.config import config
from src.database_service import DatabaseService
from src.embedding_service import EmbeddingService

logger = logging.getLogger(__name__)


class RAGService:
    """Retrieval-Augmented Generation service."""

    def __init__(
        self,
        db_service: DatabaseService,
        embedding_service: EmbeddingService,
        llm_model: Optional[str] = None,
        host: Optional[str] = None
    ):
        """Initialize RAG service.

        Args:
            db_service: Database service instance
            embedding_service: Embedding service instance
            llm_model: Ollama LLM model name
            host: Ollama host URL
        """
        self.db_service = db_service
        self.embedding_service = embedding_service
        self.llm_model = llm_model or config.LLM_MODEL
        self.host = host or config.OLLAMA_HOST
        self.client = ollama.Client(host=self.host)
        logger.info(f"RAGService initialized with LLM: {self.llm_model}")

    def retrieve_context(
        self,
        query: str,
        top_k: int = None,
        candidate_id: Optional[int] = None
    ) -> List[Dict[str, Any]]:
        """Retrieve relevant context for a query.

        Args:
            query: User query
            top_k: Number of results to retrieve
            candidate_id: Optional candidate ID to filter results

        Returns:
            List of relevant chunks with metadata
        """
        if top_k is None:
            top_k = config.TOP_K_RESULTS

        try:
            # Generate query embedding
            query_embedding = self.embedding_service.generate_embedding(query)

            # Search for similar chunks
            results = self.db_service.search_similar_chunks(
                query_embedding=query_embedding,
                top_k=top_k,
                candidate_id=candidate_id
            )

            logger.info(f"Retrieved {len(results)} chunks for query")
            return results

        except Exception as e:
            logger.error(f"Error retrieving context: {e}")
            raise

    def generate_response(
        self,
        query: str,
        context_chunks: List[Dict[str, Any]],
        system_prompt: Optional[str] = None
    ) -> str:
        """Generate response using LLM with retrieved context.

        Args:
            query: User query
            context_chunks: Retrieved context chunks
            system_prompt: Optional system prompt

        Returns:
            Generated response
        """
        try:
            # Build context from chunks
            context_parts = []
            for i, chunk in enumerate(context_chunks, 1):
                candidate_name = f"{chunk['firstname']} {chunk['lastname']}"
                similarity = chunk.get('similarity', 0)
                context_parts.append(
                    f"[Context {i} - {candidate_name} (Relevance: {similarity:.2f})]\n{chunk['chunk_text']}"
                )

            context = "\n\n".join(context_parts)

            # Default system prompt
            if system_prompt is None:
                system_prompt = """You are a helpful AI assistant that answers questions about candidates based on their resumes.
Use only the provided context to answer questions. If the context doesn't contain enough information, say so.
Be specific and cite which candidate the information comes from."""

            # Build user prompt
            user_prompt = f"""Context from candidate resumes:

{context}

Question: {query}

Please provide a detailed answer based on the context above."""

            # Generate response
            response = self.client.chat(
                model=self.llm_model,
                messages=[
                    {"role": "system", "content": system_prompt},
                    {"role": "user", "content": user_prompt}
                ]
            )

            return response['message']['content']

        except Exception as e:
            logger.error(f"Error generating response: {e}")
            raise

    def chat(
        self,
        query: str,
        candidate_id: Optional[int] = None,
        top_k: int = None
    ) -> Dict[str, Any]:
        """Complete RAG pipeline: retrieve and generate.

        Args:
            query: User query
            candidate_id: Optional candidate ID to filter results
            top_k: Number of context chunks to retrieve

        Returns:
            Dictionary with response and metadata
        """
        try:
            # Retrieve context
            context_chunks = self.retrieve_context(
                query=query,
                top_k=top_k,
                candidate_id=candidate_id
            )

            if not context_chunks:
                return {
                    "response": "I couldn't find any relevant information in the candidate resumes to answer your question.",
                    "context_chunks": [],
                    "query": query
                }

            # Generate response
            response = self.generate_response(
                query=query,
                context_chunks=context_chunks
            )

            return {
                "response": response,
                "context_chunks": context_chunks,
                "query": query
            }

        except Exception as e:
            logger.error(f"Error in chat: {e}")
            raise

    def stream_chat(
        self,
        query: str,
        candidate_id: Optional[int] = None,
        top_k: int = None
    ):
        """Stream RAG response.

        Args:
            query: User query
            candidate_id: Optional candidate ID to filter results
            top_k: Number of context chunks to retrieve

        Yields:
            Response chunks
        """
        try:
            # Retrieve context
            context_chunks = self.retrieve_context(
                query=query,
                top_k=top_k,
                candidate_id=candidate_id
            )

            if not context_chunks:
                yield "I couldn't find any relevant information in the candidate resumes to answer your question."
                return

            # Build context
            context_parts = []
            for i, chunk in enumerate(context_chunks, 1):
                candidate_name = f"{chunk['firstname']} {chunk['lastname']}"
                similarity = chunk.get('similarity', 0)
                context_parts.append(
                    f"[Context {i} - {candidate_name} (Relevance: {similarity:.2f})]\n{chunk['chunk_text']}"
                )

            context = "\n\n".join(context_parts)

            system_prompt = """You are a helpful AI assistant that answers questions about candidates based on their resumes.
Use only the provided context to answer questions. If the context doesn't contain enough information, say so.
Be specific and cite which candidate the information comes from."""

            user_prompt = f"""Context from candidate resumes:

{context}

Question: {query}

Please provide a detailed answer based on the context above."""

            # Stream response
            stream = self.client.chat(
                model=self.llm_model,
                messages=[
                    {"role": "system", "content": system_prompt},
                    {"role": "user", "content": user_prompt}
                ],
                stream=True
            )

            for chunk in stream:
                if 'message' in chunk and 'content' in chunk['message']:
                    yield chunk['message']['content']

        except Exception as e:
            logger.error(f"Error in stream_chat: {e}")
            yield f"Error: {str(e)}"

# Made with Bob

Conclusion

The Candidate RAG System demonstrates a powerful shift in recruitment technology by transforming static, unstructured resumes into a dynamic and queryable knowledge base. At the heart of this transformation is Docling, which provides robust information extraction capabilities to accurately parse and chunk content from both PDF and DOCX formats. By integrating this sophisticated document processing with PostgreSQL and pgvector for hybrid relational and semantic storage, the architecture allows organizations to move beyond keyword matching to a deeper, contextual understanding of their talent pool.

This specific use-case proves that combining Docling’s hierarchical chunking with vector similarity search enables recruiters to ask complex questions and receive evidence-based answers, effectively bridging the gap between massive document repositories and actionable hiring insights.

>>> Thanks for reading <<<

Docling Data Pipeline for AstraDB Serverless RAG

Alain Airom (Ayrom) — Fri, 17 Apr 2026 09:13:44 +0000

Adapting an existing AstraDB “Unstructured Serverless” sample to Docling (with IBM Bob)!

Introduction

While researching Astra DB RAG implementations recently, I came across a guide for integrating Unstructured Serverless with Astra DB. Although I was initially unfamiliar with the Unstructured.io ecosystem, diving into their documentation sparked a different idea. Given our access to Docling — an open-source alternative that requires no API keys and offers unlimited usage — I saw an opportunity to optimize. By leveraging Docling’s ability to integrate seamlessly into IBM Code Engine Serverless fleets, I developed a custom sample application modeled after the original Astra DB workflow. This article breaks down that implementation, showcasing a cost-effective, scalable approach to document processing for RAG.

TL;DR — Astra DB Introduction

Astra DB is a cloud-native, fully managed Database-as-a-Service (DBaaS) built on the powerful Apache Cassandra engine. It is designed to handle massive, globally distributed workloads with high availability and low latency, all while removing the operational complexity of managing a traditional database cluster. For modern developers, its standout feature is its serverless vector search capability, which allows for the storage and querying of high-dimensional embeddings. This makes it a premier choice for Generative AI and Retrieval-Augmented Generation (RAG) applications, as it seamlessly integrates real-time data with AI orchestration tools like LangChain and LlamaIndex.

Core Capabilities at a Glance:

Serverless Efficiency: Scales automatically to zero when not in use and expands elastically to meet spikes in traffic, ensuring you only pay for the resources you consume.
Vector-First Design: Optimized for AI-driven semantic search, enabling applications to “understand” and retrieve data based on context rather than just keywords.
Multi-Cloud Flexibility: Deploys across AWS, Google Cloud, Azure and IBM, providing a consistent API and infrastructure regardless of your cloud provider.
Developer-Friendly APIs: Beyond traditional CQL, it offers the Data API (a JSON-based interface), along with REST and GraphQL, making it accessible for frontend and backend developers alike.

Astra DB and HCD enhance the NoSQL database of IBM® watsonx.data® with vector capabilities, strengthening our retrieval-augmented generation and knowledge embedding capabilities. Built for elastic scalability and predictable performance, these solutions support mission-critical workloads with near-zero latency.
Astra DB is a Forrester Leader delivering NoSQL vector search capabilities on cloud and is built on Apache Cassandra®, providing the speed, reliability and multi-model support needed for modern AI workloads — including tabular, search and graph data. This enables complex, context-sensitive searches across diverse data formats for generative AI applications.

Steps of Implementation

I began with the reference implementation provided in the Astra DB documentation — a Python-based workflow centered on Unstructured.io — to serve as a point of comparison for the Docling approach.

The environmnet variables for Astra DB and Unstructured.io (sample using OPENAI)

UNSTRUCTURED_API_KEY=UNSTRUCTURED_API_KEY
UNSTRUCTURED_API_URL=https://api.unstructuredapp.io/general/v0/general
API_ENDPOINT=API_ENDPOINT
APPLICATION_TOKEN=APPLICATION_TOKEN
OPENAI_API_KEY=OPENAI_API_KEY

The sample code on the documentation site of Astra DB;

import os
import requests

from dotenv import load_dotenv
from langchain_astradb import AstraDBVectorStore
from langchain_core.documents import Document
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnablePassthrough

from langchain_community.document_loaders import (
    unstructured,
    UnstructuredAPIFileLoader,
)

from langchain_openai import (
    ChatOpenAI,
    OpenAIEmbeddings,
)

load_dotenv()

# download pdf
url = "https://raw.githubusercontent.com/datastax/ragstack-ai/48bc55e7dc4de6a8b79fcebcedd242dc1254dd63/examples/notebooks/resources/attention_pages_9_10.pdf"
file_path = "./attention_pages_9_10.pdf"

response = requests.get(url)
if response.status_code == 200:
    with open(file_path, "wb") as file:
        file.write(response.content)
    print("Download complete.")
else:
    print("Error downloading the file.")

# simple parse
loader = UnstructuredAPIFileLoader(
    file_path="./attention_pages_9_10.pdf",
    api_key=os.getenv("UNSTRUCTURED_API_KEY"),
    url = os.getenv("UNSTRUCTURED_API_URL"),
)
simple_docs = loader.load()

print(len(simple_docs))
print(simple_docs[0].page_content[0:400])

# complex parse
elements = unstructured.get_elements_from_api(
    file_path="./attention_pages_9_10.pdf",
    api_key=os.getenv("UNSTRUCTURED_API_KEY"),
    api_url=os.getenv("UNSTRUCTURED_API_URL"),
    strategy="hi_res", # default "auto"
    pdf_infer_table_structure=True,
)

print(len(elements))
tables = [el for el in elements if el.category == "Table"]
print(tables[1].metadata.text_as_html)

# create vector store
astra_db_store = AstraDBVectorStore(
    collection_name="langchain_unstructured",
    embedding=OpenAIEmbeddings(),
    token=os.getenv("APPLICATION_TOKEN"),
    api_endpoint=os.getenv("API_ENDPOINT")
)

# load documents
documents = []
current_doc = None

for el in elements:
    if el.category in ["Header", "Footer"]:
        continue # skip these
    if el.category == "Title":
        if current_doc is not None:
            documents.append(current_doc)
        current_doc = None
    if not current_doc:
        current_doc = Document(page_content="", metadata=el.metadata.to_dict())
    current_doc.page_content += el.metadata.text_as_html if el.category == "Table" else el.text
    if el.category == "Table":
        if current_doc is not None:
            documents.append(current_doc)
        current_doc = None

astra_db_store.add_documents(documents)

# prompt and query
prompt = """
Answer the question based only on the supplied context. If you don't know the answer, say "I don't know".
Context: {context}
Question: {question}
Your answer:
"""

llm = ChatOpenAI(model="gpt-3.5-turbo-16k", streaming=False, temperature=0)

chain = (
    {"context": astra_db_store.as_retriever(), "question": RunnablePassthrough()}
    | PromptTemplate.from_template(prompt)
    | llm
    | StrOutputParser()
)

response_1 = chain.invoke("What does reducing the attention key size do?")
print("\n***********New Unstructured Basic Query Engine***********")
print(response_1)

response_2 = chain.invoke("For the transformer to English constituency results, what was the 'WSJ 23 F1' value for 'Dyer et al. (2016) (5]'?")
print("\n***********New Unstructured Basic Query Engine***********")
print(response_2)

response_3 = chain.invoke("When was George Washington born?")
print("\n***********New Unstructured Basic Query Engine***********")
print(response_3)

With the baseline established, the next step was migration. I worked with Bob to draft a Python implementation that runs directly in the terminal, utilizing Docling for document parsing and chunking.

"""
Console Application: PDF to AstraDB Vector Store using Docling
This application replaces Unstructured-io with Docling for PDF processing
"""

import os
from dotenv import load_dotenv
from pathlib import Path

from docling.document_converter import DocumentConverter
from langchain_astradb import AstraDBVectorStore
from langchain_core.documents import Document
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI, OpenAIEmbeddings

# Load environment variables
load_dotenv()

def process_pdf_with_docling(file_path: str) -> list:
    """
    Process PDF file using Docling and return structured documents

    Args:
        file_path: Path to the PDF file

    Returns:
        List of Document objects
    """
    print(f"\n📄 Processing PDF with Docling: {file_path}")

    # Initialize Docling converter
    converter = DocumentConverter()

    # Convert the PDF
    result = converter.convert(file_path)

    # Extract documents from Docling result
    documents = []
    current_doc = None

    # Process the document structure
    doc = result.document

    # Iterate through document elements
    for element, level in doc.iterate_items():
        element_type = element.label if hasattr(element, 'label') else 'text'

        # Skip headers and footers
        if element_type.lower() in ['header', 'footer']:
            continue

        # Start new document on titles
        if element_type.lower() == 'title':
            if current_doc is not None:
                documents.append(current_doc)
            current_doc = None

        # Initialize document if needed
        if not current_doc:
            metadata = {
                'source': file_path,
                'element_type': element_type,
            }
            current_doc = Document(page_content="", metadata=metadata)

        # Add content to current document
        if hasattr(element, 'text'):
            current_doc.page_content += element.text + "\n"

        # For tables, append and start new document
        if element_type.lower() == 'table':
            if current_doc is not None:
                documents.append(current_doc)
            current_doc = None

    # Add last document if exists
    if current_doc is not None:
        documents.append(current_doc)

    print(f"✅ Extracted {len(documents)} document chunks from PDF")
    return documents


def create_vector_store():
    """Create and return AstraDB vector store"""
    print("\n🗄️  Connecting to AstraDB Vector Store...")

    astra_db_store = AstraDBVectorStore(
        collection_name="langchain_docling",
        embedding=OpenAIEmbeddings(),
        token=os.getenv("APPLICATION_TOKEN"),
        api_endpoint=os.getenv("API_ENDPOINT")
    )

    print("✅ Connected to AstraDB")
    return astra_db_store


def query_vector_store(astra_db_store, question: str) -> str:
    """
    Query the vector store with a question

    Args:
        astra_db_store: AstraDB vector store instance
        question: Question to ask

    Returns:
        Answer string
    """
    prompt = """
Answer the question based only on the supplied context. If you don't know the answer, say "I don't know".
Context: {context}
Question: {question}
Your answer:
"""

    llm = ChatOpenAI(model="gpt-3.5-turbo-16k", streaming=False, temperature=0)

    chain = (
        {"context": astra_db_store.as_retriever(), "question": RunnablePassthrough()}
        | PromptTemplate.from_template(prompt)
        | llm
        | StrOutputParser()
    )

    return chain.invoke(question)


def main():
    """Main console application"""
    print("=" * 80)
    print("PDF to AstraDB Vector Store - Console Application (Docling)")
    print("=" * 80)

    # Check for input PDF
    input_dir = Path("input")
    pdf_files = list(input_dir.glob("*.pdf"))

    if not pdf_files:
        print("❌ No PDF files found in 'input' folder")
        return

    # Use first PDF file found
    pdf_file = pdf_files[0]
    print(f"\n📁 Using PDF file: {pdf_file}")

    # Process PDF with Docling
    documents = process_pdf_with_docling(str(pdf_file))

    if not documents:
        print("❌ No documents extracted from PDF")
        return

    # Display sample content
    print(f"\n📝 Sample content from first document:")
    print("-" * 80)
    print(documents[0].page_content[:400])
    print("-" * 80)

    # Create vector store
    astra_db_store = create_vector_store()

    # Add documents to vector store
    print(f"\n💾 Adding {len(documents)} documents to vector store...")
    astra_db_store.add_documents(documents)
    print("✅ Documents added successfully")

    # Query examples
    questions = [
        "What does reducing the attention key size do?",
        "For the transformer to English constituency results, what was the 'WSJ 23 F1' value for 'Dyer et al. (2016) (5]'?",
        "When was George Washington born?"
    ]

    print("\n" + "=" * 80)
    print("QUERY RESULTS")
    print("=" * 80)

    for i, question in enumerate(questions, 1):
        print(f"\n🔍 Question {i}: {question}")
        print("-" * 80)
        response = query_vector_store(astra_db_store, question)
        print(f"💡 Answer: {response}")
        print("-" * 80)

    print("\n✅ Console application completed successfully!")


if __name__ == "__main__":
    main()

# Made with Bob

Building on the console prototype, the next step was to transition to a cloud-native, serverless environment. I chose IBM Code Engine for its ability to manage serverless fleets, which required evolving the script into a full-scale GUI application. To ensure the app was truly ‘cloud-ready,’ I eliminated all hardcoded paths and developed a Streamlit interface for a dynamic user experience. To complete the deployment package, I’ve included the necessary Dockerfile and Kubernetes manifests, providing everything needed to containerize and launch the solution on any cloud platform.

IBM Code Engine Serverless Fleets is a strategic extension of the Code Engine platform designed to handle massive, compute-intensive workloads that require large-scale parallel execution. Unlike standard serverless functions meant for quick tasks, Fleets can orchestrate thousands of concurrent “run-to-completion” instances across virtual machines or specialized GPUs (like NVIDIA L40s). It automatically manages task queuing, provisions the necessary worker nodes to handle batch processing — such as parsing millions of documents with Docling or running complex AI simulations — and scales everything back to zero once the job is finished, ensuring you only pay for the active compute time.

Key Advantages for your Pipeline:

Massive Parallelism: Perfect for the “fleets” mode you mentioned, where you need to process large document batches simultaneously.
Hardware Flexibility: Allows you to toggle between high-performance CPUs and on-demand GPUs for heavy-duty AI tasks.
Single-Tenant Security: Tasks execute in isolated infrastructure within your VPC, providing enterprise-grade security for sensitive data.
Zero Infrastructure Management: It removes the need for SRE or DevOps teams to manually size or scale clusters, even for workloads with 100,000+ processors.

The requirements for the code;

# Core dependencies
python-dotenv>=1.0.0

# Docling for PDF processing
docling>=1.0.0

# LangChain and related packages
langchain>=0.1.0
langchain-core>=0.1.0
langchain-community>=0.0.20
langchain-astradb>=0.3.0
langchain-openai>=0.0.5

# AstraDB
astrapy>=2.0,<3.0

# OpenAI
openai>=1.0.0

# Streamlit for GUI
streamlit>=1.30.0

# Additional utilities
requests>=2.31.0
pathlib>=1.0.1

The Streamlit code;

"""
GUI Application: PDF to AstraDB Vector Store using Docling
Streamlit-based interface for PDF processing and querying
"""

import os
import streamlit as st
from dotenv import load_dotenv
from pathlib import Path
from datetime import datetime

from docling.document_converter import DocumentConverter
from langchain_astradb import AstraDBVectorStore
from langchain_core.documents import Document
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI, OpenAIEmbeddings

# Load environment variables
load_dotenv()

# Page configuration
st.set_page_config(
    page_title="PDF to AstraDB - Docling",
    page_icon="📄",
    layout="wide"
)

# Initialize session state
if 'vector_store' not in st.session_state:
    st.session_state.vector_store = None
if 'documents_loaded' not in st.session_state:
    st.session_state.documents_loaded = False
if 'processing_log' not in st.session_state:
    st.session_state.processing_log = []


def log_message(message: str):
    """Add message to processing log"""
    timestamp = datetime.now().strftime("%H:%M:%S")
    st.session_state.processing_log.append(f"[{timestamp}] {message}")


def process_pdf_with_docling(file_path: str) -> list:
    """
    Process PDF file using Docling and return structured documents

    Args:
        file_path: Path to the PDF file

    Returns:
        List of Document objects
    """
    log_message(f"📄 Processing PDF with Docling: {file_path}")

    # Initialize Docling converter
    converter = DocumentConverter()

    # Convert the PDF
    result = converter.convert(file_path)

    # Extract documents from Docling result
    documents = []
    current_doc = None

    # Process the document structure
    doc = result.document

    # Iterate through document elements
    for element, level in doc.iterate_items():
        element_type = element.label if hasattr(element, 'label') else 'text'

        # Skip headers and footers
        if element_type.lower() in ['header', 'footer']:
            continue

        # Start new document on titles
        if element_type.lower() == 'title':
            if current_doc is not None:
                documents.append(current_doc)
            current_doc = None

        # Initialize document if needed
        if not current_doc:
            metadata = {
                'source': file_path,
                'element_type': element_type,
            }
            current_doc = Document(page_content="", metadata=metadata)

        # Add content to current document
        if hasattr(element, 'text'):
            current_doc.page_content += element.text + "\n"

        # For tables, append and start new document
        if element_type.lower() == 'table':
            if current_doc is not None:
                documents.append(current_doc)
            current_doc = None

    # Add last document if exists
    if current_doc is not None:
        documents.append(current_doc)

    log_message(f"✅ Extracted {len(documents)} document chunks from PDF")
    return documents


def create_vector_store():
    """Create and return AstraDB vector store"""
    log_message("🗄️  Connecting to AstraDB Vector Store...")

    astra_db_store = AstraDBVectorStore(
        collection_name="langchain_docling_gui",
        embedding=OpenAIEmbeddings(),
        token=os.getenv("APPLICATION_TOKEN"),
        api_endpoint=os.getenv("API_ENDPOINT")
    )

    log_message("✅ Connected to AstraDB")
    return astra_db_store


def query_vector_store(astra_db_store, question: str) -> str:
    """
    Query the vector store with a question

    Args:
        astra_db_store: AstraDB vector store instance
        question: Question to ask

    Returns:
        Answer string
    """
    prompt = """
Answer the question based only on the supplied context. If you don't know the answer, say "I don't know".
Context: {context}
Question: {question}
Your answer:
"""

    llm = ChatOpenAI(model="gpt-3.5-turbo-16k", streaming=False, temperature=0)

    chain = (
        {"context": astra_db_store.as_retriever(), "question": RunnablePassthrough()}
        | PromptTemplate.from_template(prompt)
        | llm
        | StrOutputParser()
    )

    return chain.invoke(question)


def main():
    """Main Streamlit application"""

    # Header
    st.title("📄 PDF to AstraDB Vector Store")
    st.subheader("Powered by Docling & Streamlit")

    # Sidebar
    with st.sidebar:
        st.header("⚙️ Configuration")

        # Check environment variables
        api_endpoint = os.getenv("API_ENDPOINT")
        app_token = os.getenv("APPLICATION_TOKEN")
        openai_key = os.getenv("OPENAI_API_KEY")

        if api_endpoint and app_token and openai_key:
            st.success("✅ Environment variables loaded")
        else:
            st.error("❌ Missing environment variables")
            st.stop()

        st.divider()

        # File selection
        st.header("📁 PDF Selection")
        input_dir = Path("input")
        pdf_files = list(input_dir.glob("*.pdf"))

        if not pdf_files:
            st.error("❌ No PDF files found in 'input' folder")
            st.stop()

        selected_pdf = st.selectbox(
            "Select PDF file:",
            options=[f.name for f in pdf_files],
            index=0
        )

        pdf_path = input_dir / selected_pdf

        st.divider()

        # Process button
        if st.button("🚀 Process PDF", type="primary", use_container_width=True):
            st.session_state.processing_log = []

            with st.spinner("Processing PDF..."):
                try:
                    # Process PDF
                    documents = process_pdf_with_docling(str(pdf_path))

                    if not documents:
                        st.error("❌ No documents extracted from PDF")
                        st.stop()

                    # Create vector store
                    st.session_state.vector_store = create_vector_store()

                    # Add documents
                    log_message(f"💾 Adding {len(documents)} documents to vector store...")
                    st.session_state.vector_store.add_documents(documents)
                    log_message("✅ Documents added successfully")

                    st.session_state.documents_loaded = True
                    st.success("✅ PDF processed successfully!")

                except Exception as e:
                    st.error(f"❌ Error: {str(e)}")
                    log_message(f"❌ Error: {str(e)}")

    # Main content area
    col1, col2 = st.columns([2, 1])

    with col1:
        st.header("💬 Query Interface")

        if st.session_state.documents_loaded:
            # Query input
            question = st.text_input(
                "Enter your question:",
                placeholder="What does reducing the attention key size do?"
            )

            if st.button("🔍 Search", type="primary"):
                if question:
                    with st.spinner("Searching..."):
                        try:
                            response = query_vector_store(
                                st.session_state.vector_store,
                                question
                            )

                            st.subheader("💡 Answer:")
                            st.info(response)

                            # Save to output
                            output_dir = Path("output")
                            output_dir.mkdir(exist_ok=True)
                            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
                            output_file = output_dir / f"query_result_{timestamp}.txt"

                            with open(output_file, "w") as f:
                                f.write(f"Question: {question}\n\n")
                                f.write(f"Answer: {response}\n")

                            st.success(f"✅ Result saved to: {output_file}")

                        except Exception as e:
                            st.error(f"❌ Error: {str(e)}")
                else:
                    st.warning("⚠️ Please enter a question")

            # Example questions
            st.divider()
            st.subheader("📝 Example Questions")

            example_questions = [
                "What does reducing the attention key size do?",
                "For the transformer to English constituency results, what was the 'WSJ 23 F1' value for 'Dyer et al. (2016) (5]'?",
                "When was George Washington born?"
            ]

            for i, eq in enumerate(example_questions, 1):
                if st.button(f"Example {i}: {eq}", key=f"example_{i}"):
                    st.session_state.example_question = eq
                    st.rerun()

            # Use example question if set
            if hasattr(st.session_state, 'example_question'):
                question = st.session_state.example_question
                delattr(st.session_state, 'example_question')
        else:
            st.info("👈 Please process a PDF file first using the sidebar")

    with col2:
        st.header("📋 Processing Log")

        if st.session_state.processing_log:
            log_container = st.container(height=400)
            with log_container:
                for log_entry in st.session_state.processing_log:
                    st.text(log_entry)
        else:
            st.info("No processing activity yet")

    # Footer
    st.divider()
    st.caption("Built with Docling, Streamlit, and AstraDB")


if __name__ == "__main__":
    main()

# Made with Bob

The required Dockerfile;

FROM python:3.11-slim

# Set working directory
WORKDIR /app

# Set environment variables
ENV PYTHONUNBUFFERED=1 \
    PYTHONDONTWRITEBYTECODE=1 \
    PORT=8080

# Install system dependencies
RUN apt-get update && apt-get install -y \
    build-essential \
    curl \
    git \
    && rm -rf /var/lib/apt/lists/*

# Copy requirements first for better caching
COPY requirements.txt .

# Install Python dependencies
RUN pip install --no-cache-dir --upgrade pip && \
    pip install --no-cache-dir -r requirements.txt

# Copy application files
COPY app_gui_docling.py .
COPY .env.example .env

# Create necessary directories
RUN mkdir -p input output logs

# Expose port (Code Engine uses PORT env variable)
EXPOSE 8080

# Health check
HEALTHCHECK --interval=30s --timeout=10s --start-period=40s --retries=3 \
    CMD curl -f http://localhost:8080/_stcore/health || exit 1

# Run Streamlit application
CMD streamlit run app_gui_docling.py \
    --server.port=${PORT} \
    --server.address=0.0.0.0 \
    --server.headless=true \
    --browser.gatherUsageStats=false \
    --server.enableCORS=false \
    --server.enableXsrfProtection=true

All the project files are available on the GitHub repository.

Bonus Part-My 1st Implementation of Skills and Rules using Bob!

As is my standard practice, I initially developed a comprehensive prompt to guide Bob through the implementation. However, to shift toward a more industrialized development workflow, I’ve transitioned from single-use instructions to a library of universal rules and ‘skills.’ This modular approach eliminates repetitive setup and ensures consistent quality across projects. Below, I’ve shared the specific skill set I provided to Bob for this build — a framework I intend to expand upon to further accelerate my development velocity and standardize future implementations.

## Project Development Rule

When working with this projects:

1. **Reference Guide**: Always check the links provided here to access the source code repositories for implementation:
   - Docling: https://github.com/docling-project/docling
   - [unstructured-python-client](https://github.com/Unstructured-IO/unstructured-python-client): [unstructured-python-client](https://github.com/Unstructured-IO/unstructured-python-client)
   - Astra DB Python implementation: https://docs.datastax.com/en/astra-db-serverless/api-reference/dataapiclient.html
2. **Project Structure**: Follow these conventions:
   - The documents of the project should be created in "Docs" folder except readme.md
   - Always provide a Mermaid flow architecture for the project
   - All the BASH scripts if needed, should be written in "scripts" folder
   - All the input documents are to be found in "input" folder
   - All the output documents which are asked to be provided should be writen in timestamped format in "output" folder
   - The result documents should be written in "output" folder, if the "output" folder does not exist, it should be created
   - Always provide README.md with architecture + workflow diagrams as described
   - Always provide a ".gitignore" file which filters/ignores any ".env" files or any folders whichs' names start with "_" (underscore) to be pushed to GitHub (e.g.: _sources/, _images/, _docs/... )
3. **Key Patterns**:
   - Always test the functionnality of the code you provide 
   - When you make updates/enhancements and/or correct the bugs, update the existing documents and scripts, don't create new ones

Conclusion

The integration of Docling with Astra DB Serverless represents a significant shift toward more efficient and cost-effective RAG architectures. By replacing third-party proprietary services with an open-source parsing engine, this implementation provides a high-performance pipeline that extracts complex structures — such as nested tables and diverse document formats — and seamlessly transforms them into OpenAI embeddings for vector storage. The resulting system is not only robust but also highly versatile, offering both a streamlined console interface for batch processing and a dynamic Streamlit GUI for interactive, real-time document querying.

Ultimately, moving this solution to IBM Code Engine transitions the project from a local prototype to a production-ready, industrialized fleet. This serverless deployment model ensures that the application scales elastically with demand, offering “scale-to-zero” efficiency that minimizes costs while maintaining high availability. By utilizing the universal “skills” and modular rules established during development, you now have a repeatable framework for deploying sophisticated AI-driven document intelligence tools that are powerful, scalable, and entirely under your operational control.

Disclaimer: This is a v0.1 release focused on the core concept. More robust enhancements and industrialization are in the pipeline. Expect frequent updates as the project evolves.

>>> Thanks for reading <<<

Protecting the Crown Jewels: Securing Source Code in Containerized Applications

Alain Airom (Ayrom) — Tue, 14 Apr 2026 09:32:11 +0000

Protect your source code if it’s not open-sourced!

Introduction and some background

In a recent discussion with a technical partner, a challenging question arose: “How can we distribute containerized applications while preventing the source code from being easily viewed or reverse-engineered?” While I had encountered articles on this topic before, I wanted to provide something more tangible than just a list of links. Instead of a simple search, I teamed up with Bob, our expert digital architect, to build a comprehensive demonstration.

Bob provided functional code samples in four different programming languages — Go, C++, Python, and Node.js — to showcase how varying levels of protection can be achieved within a container. The result is a deep dive into the practical strategies you can use to protect your intellectual property when it leaves your internal environment.

What are the Technics that Bob Provided?

The Multi-Stage Advantage: How to ensure your source code never even enters the final production image.
Language-Specific Hardening: From static binary compilation in Go to PyArmor obfuscation in Python.
Minimal Footprints: Using “Distroless” and minimal base images to reduce the attack surface.
Verification: Techniques to prove your source code and build tools have been successfully removed.

Why This Matters?

Shipping a container is often mistaken for shipping a “black box,” but without the right precautions, a simple docker save can reveal your proprietary algorithms. By the end of this guide, you’ll have a roadmap for securing your applications, regardless of whether you are working with compiled or interpreted languages.

Let’s dive into the implementations.

Choosing Your Shield: Language-Specific Protection Strategies

While Bob has provided functional samples in Go, C++, Python, and Node.js, it is important to understand that not all “shields” are created equal. The applications themselves are designed to be simple demonstrations of a complex problem: how to prevent your logic from being reverse-engineered once it’s in the wild.

The Spectrum of Security

Security is rarely bulletproof; it is a series of hurdles designed to make unauthorized access as difficult and expensive as possible. As the comparison below shows, the “right path” depends heavily on the language you choose:

Compiled Languages (Go, C++): These offer the highest level of security. By compiling code into a static binary and stripping debug symbols, the original source code is never present in the container.
Interpreted Languages (Python, Node.js): These are inherently more challenging to protect. Because the runtime requires the source (or bytecode) to function, we must rely on obfuscation — scrambling the code to make it unreadable to humans while remaining executable by the machine.

Key Takeaways for Implementation

Architecture Matters: Protection begins at the conception phase. If an algorithm is truly sensitive, you might choose to implement that specific component in a compiled language like C++ or Go, even if the rest of your stack is in Node.js.
The Effort vs. Security Trade-off: There is often a direct correlation between the effort required to write a language and the security of its output. Low-level languages may require more development time, but they reward you with a significantly more robust “black box” for distribution.
A Solid Guideline: Use these samples not as a final destination, but as a framework. By combining multi-stage builds with minimal images and code-hardening, you create a layered defense that protects your intellectual property from the majority of common threats.

Go Sample Application (one of the good choices)

The Go app is provided below as one of the examples, alongside with the related Dockerfile.

package main

import (
 "crypto/sha256"
 "encoding/hex"
 "fmt"
 "log"
 "net/http"
 "os"
 "time"
)

// Secret business logic - this will be compiled and not visible in the container
const (
 secretKey       = "my-super-secret-algorithm-key-2024"
 licenseCheckURL = "internal-license-server"
)

// ProprietaryAlgorithm represents our "secret sauce"
type ProprietaryAlgorithm struct {
 secretSalt string
 iterations int
}

// NewAlgorithm creates a new instance with proprietary settings
func NewAlgorithm() *ProprietaryAlgorithm {
 return &ProprietaryAlgorithm{
  secretSalt: secretKey,
  iterations: 10000,
 }
}

// ProcessData applies our proprietary algorithm
func (pa *ProprietaryAlgorithm) ProcessData(input string) string {
 // This is our "secret" business logic
 result := input + pa.secretSalt
 for i := 0; i < pa.iterations; i++ {
  hash := sha256.Sum256([]byte(result))
  result = hex.EncodeToString(hash[:])
 }
 return result
}

// ValidateLicense checks if the application is properly licensed
func ValidateLicense() bool {
 // Simulated license validation
 licenseKey := os.Getenv("LICENSE_KEY")
 if licenseKey == "" {
  log.Println("Warning: No license key provided")
  return false
 }

 // In production, this would call a license server
 expectedHash := "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
 hash := sha256.Sum256([]byte(licenseKey))
 actualHash := hex.EncodeToString(hash[:])

 return actualHash == expectedHash || licenseKey == "DEMO-LICENSE-KEY"
}

func healthHandler(w http.ResponseWriter, r *http.Request) {
 w.WriteHeader(http.StatusOK)
 fmt.Fprintf(w, `{"status":"healthy","timestamp":"%s"}`, time.Now().Format(time.RFC3339))
}

func processHandler(algo *ProprietaryAlgorithm) http.HandlerFunc {
 return func(w http.ResponseWriter, r *http.Request) {
  if r.Method != http.MethodPost {
   http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
   return
  }

  input := r.URL.Query().Get("data")
  if input == "" {
   http.Error(w, "Missing 'data' parameter", http.StatusBadRequest)
   return
  }

  // Apply our proprietary algorithm
  result := algo.ProcessData(input)

  w.Header().Set("Content-Type", "application/json")
  fmt.Fprintf(w, `{"input":"%s","processed":"%s","algorithm":"proprietary-v1"}`, input, result)
 }
}

func main() {
 // Validate license on startup
 if !ValidateLicense() {
  log.Println("Running in DEMO mode - some features may be limited")
 } else {
  log.Println("License validated successfully")
 }

 // Initialize our proprietary algorithm
 algo := NewAlgorithm()

 // Setup HTTP server
 http.HandleFunc("/health", healthHandler)
 http.HandleFunc("/process", processHandler(algo))

 port := os.Getenv("PORT")
 if port == "" {
  port = "8080"
 }

 log.Printf("Starting Go secure application on port %s", port)
 log.Printf("Endpoints: /health (GET), /process (POST)")

 if err := http.ListenAndServe(":"+port, nil); err != nil {
  log.Fatal(err)
 }
}

// Made with Bob

The Dockerfile

# Multi-stage build for Go application
# Stage 1: Build the application
FROM golang:1.21-alpine AS builder

# Set working directory
WORKDIR /build

# Copy go mod files
COPY go.mod ./

# Download dependencies
RUN go mod download

# Copy source code
COPY main.go ./

# Build the application with optimizations
# -ldflags="-s -w" strips debug information and symbol table
# CGO_ENABLED=0 creates a static binary
RUN CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build \
    -ldflags="-s -w -X main.buildTime=$(date -u +%Y-%m-%dT%H:%M:%SZ)" \
    -trimpath \
    -o secure-app \
    main.go

# Stage 2: Create minimal runtime image
# Using distroless for maximum security - no shell, no package manager
FROM gcr.io/distroless/static-debian11:nonroot

# Copy only the binary from builder
COPY --from=builder /build/secure-app /app/secure-app

# Use non-root user (already set in distroless/static:nonroot)
# UID 65532 is the nonroot user

# Expose port
EXPOSE 8080

# Set working directory
WORKDIR /app

# Run the application
ENTRYPOINT ["/app/secure-app"]

# Benefits of this approach:
# 1. Source code is NOT in the final image
# 2. No build tools in the final image
# 3. No shell access (distroless)
# 4. Minimal attack surface
# 5. Binary is stripped of debug symbols
# 6. Non-root user for security
# 7. Image size is minimal (~2MB for Go binary + ~2MB for distroless base)

Test phase;

Analysis and verdict

Python Sample Application (moderate choice)

The Python code

"""
Secure Python Application with Proprietary Algorithm
This code will be obfuscated before containerization
"""

import hashlib
import os
import time
from flask import Flask, request, jsonify

# Secret proprietary constants - will be obfuscated
SECRET_KEY = "python-proprietary-algorithm-2024"
ITERATIONS = 8000


class ProprietaryAlgorithm:
    """
    Our secret business logic class
    This will be obfuscated to protect intellectual property
    """

    def __init__(self):
        self.secret_salt = SECRET_KEY
        self.iterations = ITERATIONS
        self._internal_state = self._initialize_state()

    def _initialize_state(self):
        """Internal initialization - will be obfuscated"""
        return hashlib.sha256(self.secret_salt.encode()).hexdigest()

    def _apply_transformation(self, data: str, iteration: int) -> str:
        """
        Proprietary transformation logic
        This is our "secret sauce" that will be protected
        """
        combined = f"{data}{self._internal_state}{iteration}"
        return hashlib.sha256(combined.encode()).hexdigest()

    def process_data(self, input_data: str) -> str:
        """
        Main processing method with proprietary algorithm
        Multiple iterations make reverse engineering harder
        """
        result = input_data + self.secret_salt

        # Apply our secret transformation multiple times
        for i in range(self.iterations):
            result = self._apply_transformation(result, i)

        return result


class LicenseValidator:
    """License validation logic - will be obfuscated"""

    @staticmethod
    def validate():
        """Check if application is properly licensed"""
        license_key = os.getenv('LICENSE_KEY', '')

        if not license_key:
            print("Warning: No license key provided - running in DEMO mode")
            return False

        # Simulated license validation
        expected_hash = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
        actual_hash = hashlib.sha256(license_key.encode()).hexdigest()

        is_valid = actual_hash == expected_hash or license_key == "DEMO-LICENSE-KEY"

        if is_valid:
            print("License validated successfully")
        else:
            print("Invalid license - some features may be limited")

        return is_valid


# Initialize Flask app
app = Flask(__name__)

# Initialize our proprietary algorithm
algorithm = ProprietaryAlgorithm()

# Validate license on startup
license_valid = LicenseValidator.validate()


@app.route('/health', methods=['GET'])
def health_check():
    """Health check endpoint"""
    return jsonify({
        'status': 'healthy',
        'timestamp': time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime()),
        'licensed': license_valid
    })


@app.route('/process', methods=['POST'])
def process_data():
    """Process data using our proprietary algorithm"""
    data = request.args.get('data')

    if not data:
        return jsonify({'error': 'Missing "data" parameter'}), 400

    # Apply our proprietary algorithm
    result = algorithm.process_data(data)

    return jsonify({
        'input': data,
        'processed': result,
        'algorithm': 'python-proprietary-v1'
    })


@app.route('/')
def index():
    """Root endpoint"""
    return jsonify({
        'service': 'Secure Python Application',
        'endpoints': {
            '/health': 'GET - Health check',
            '/process': 'POST - Process data (requires ?data=value)'
        }
    })


if __name__ == '__main__':
    port = int(os.getenv('PORT', 8080))
    print(f"Starting Python secure application on port {port}")
    print(f"Endpoints: /health (GET), /process (POST)")

    # Run with production settings
    app.run(host='0.0.0.0', port=port, debug=False)

# Made with Bob

The related Dockfile

FROM python:3.11-slim AS builder

RUN apt-get update && apt-get install -y \
    gcc \
    && rm -rf /var/lib/apt/lists/*

WORKDIR /build

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY app.py .

RUN pyarmor gen -O dist app.py

RUN echo "Flask==3.0.0" > dist/requirements.txt && \
    echo "Werkzeug==3.0.1" >> dist/requirements.txt

FROM python:3.11-slim

WORKDIR /app

COPY --from=builder /build/dist/requirements.txt .

RUN pip install --no-cache-dir -r requirements.txt && \
    rm requirements.txt

COPY --from=builder /build/dist/ .

RUN useradd -r -u 1001 appuser && \
    chown -R appuser:appuser /app

USER appuser

EXPOSE 8080

ENV PYTHONUNBUFFERED=1
ENV PORT=8080

CMD ["python", "app.py"]

Test phase;

The verdict;

Conclusion
The complete source for these applications is available on GitHub alongside detailed technical guides. However, the true centerpiece of this project is the “Security-Analysis” document.

This comprehensive analysis serves as a strategic roadmap, offering a side-by-side comparison of protection levels, reverse-engineering difficulty, and image-size trade-offs. It doesn’t just show you what was done; it explains the why behind every security layer — from symbol stripping in compiled binaries to control-flow flattening in obfuscated JavaScript.

Whether you are a developer looking for a ‘quick win’ or an architect designing a high-security distribution model, this guide provides the clarity needed to choose the right shield for your intellectual property.

>>> Thanks for reading <<<

The “The Architecture Handbook for Milvus Vector Database” Book Review

Alain Airom (Ayrom) — Sun, 12 Apr 2026 12:20:50 +0000

A feedback and my thoughts on the “The Architecture Handbook for Milvus Vector Database” book.

The image from the book cover produced by Packt Publications

Introduction-Mastering the Core of AI Memory

I think I can say that I know Milvus vector DB; I’ve used it many times locally for testing my applications with my Ollama+LLM+Vector Database, and Milvus is also one of the many vector databases proposed in the watsonx Data platform. However, I found this book very useful, and one of the best parts (among others) is Chapter 16: “Implementing Multi-Tenancy in Milvus”. This handbook, written by the core contributors of the Milvus project, systematically unpacks the full technology stack — from basic concepts and deployment to performance tuning and real-world implementation.

> Disclaimer: This is an independent review. I have no professional or financial relationship with the publisher or the authors.

A synthesis of the Book

Setup and Configuration

Chapter 1: Introduction to Milvus — This chapter sets the stage by defining vector databases and explaining how they handle unstructured data through encoding processes like one-hot and TF-IDF to enable similarity search.
Chapter 2: Deploying Milvus in Multiple Ways — It provides a practical guide to various deployment methods, including Docker Compose, Kubernetes operators, and Helm charts, helping users choose the right environment for their needs.

If you’ve already setup Milvus on your end, you can bypass these first chapters in my opinion!

Chapter 3: Interacting with Milvus — It learns the core objects — fields, schemas, collections, and indexes — and how to interact with the database using both the Python SDK and REST APIs for standard data operations.
Chapter 4: Configuring the Milvus System — This chapter focuses on managing system dependencies and service configurations, including setting up monitoring tools like Prometheus, Grafana, and Loki for log analysis.

Data Model — Reading Data, Modification and Compacting Data

Chapter 5: Understanding the Milvus Data Model and Architecture — It dives into the disaggregated four-tier architecture (access, coordination, worker, and storage) and the “lifeline” of the system: the Message Queue (MQ) and Timetick mechanism.

Chapter 6: Data Modification and Maintenance in Milvus — This section details the lifecycle of data “segments” and the internal journey of insert, delete, and upsert requests through the system’s various layers.

Chapter 7: Reading Data in Milvus — It explains the “scatter-gather” query pattern, replica management, and how Milvus ensures multi-level consistency between data freshness and search latency.
Chapter 8: Compaction and Garbage Collection — This chapter explores how background processes like L0 and clustering compaction optimize storage and query performance by managing fragmented segments and obsolete data.

In the context of the Milvus Vector Database, the processes of Compaction and Garbage Collection (GC) are critical because they address the inherent challenges of a high-performance, distributed LSM-tree (Log-Structured Merge-tree) based system.

These mechanisms are vital for maintaining a healthy and fast vector database:

Maintaining Search Speed (Performance Optimization): When we delete or update data in Milvus, the system doesn’t immediately “scrub” the old data from the disk (as this would be computationally expensive). Instead, it marks data as deleted. Over time, these “logical deletes” create fragmentation. So, during a search, the system still has to scan through these fragmented segments and “skipped” records, which slows down query latency. Therefore the ‘compaction’ merges small segments into larger, more optimized ones and physically removes deleted records, ensuring the search engine only processes relevant, “healthy” data.
Storage Efficiency: Without a ‘Garbage Collection’ system, the storage usage would grow indefinitely even if we dele as much data as we add.
Solving the “Small File” Problem (L0 Compaction): Milvus often receives data in small batches. If left unmanaged, this results in thousands of tiny files (Level 0 segments). Searching across 1,000 tiny files is significantly slower than searching across one large file due to the overhead of opening files and managing metadata. So “Compaction” continuously “rolls up” these tiny L0 segments into larger, indexed segments that are optimized for the Knowhere search engine.
Clustering ‘Compaction for Spatial Locality’: standard compaction just merges files, but Clustering Compaction goes a step further by reorganizing data based on the values of a specific field (like a partition key or a scalar field). By Pruning it allows the system to perform “segment pruning.” If our query only needs data from “Tenant A,” the system can skip entire segments that it knows only contain data for “Tenant B,” drastically reducing the I/O required for a search.
System Stability and Reliability: If segments are never compacted, the system’s metadata (stored in ETCD) becomes bloated. This can lead a) slow startup times, b) increased memory consumption by the Query Nodes and c) potential system crashes when the metadata layer becomes a bottleneck.

Chapter 9: Exploring Milvus’ Vector Engine — The focus shifts to the Knowhere engine, the core computational heart that supports various hardware acceleration frameworks and distance metrics like L2 and Cosine.
Chapter 10: Indexing in Detail: Algorithms and Parameters — (Based on the book’s structure) This chapter likely provides a deep dive into the mathematical foundations of indexing algorithms like HNSW, IVF, and Product Quantization (PQ).
Chapter 11: Performance Testing and Benchmarking — It outlines how to use tools like VectorDBBench to establish performance baselines and simulate data growth to test system limits.
Chapter 12: Stability and Reliability Testing — This chapter focuses on ensuring the system can handle failures through chaos engineering, recovery testing, and stress testing.

Scalabilty and Performance

Chapter 13: Scalability and Capacity Planning — It provides guidelines on how to scale the various tiers of Milvus — query nodes, data nodes, and index nodes — based on specific workload requirements.

Chapter 14: Performance Tuning and Best Practices — This part distills years of engineering experience into actionable advice for optimizing indexing and search parameters.

"""
Chapter 14: Configure in-memory replicas for horizontal scaling.

Loads a collection with a specified number of in-memory replicas to distribute
search load across multiple QueryNode instances.

Usage:
    python load_replicas.py [--uri localhost:19530] [--replicas 3]
"""

import argparse

from pymilvus import MilvusClient

COLLECTION_NAME = "test_collection"


def main():
    parser = argparse.ArgumentParser(description="Configure Milvus in-memory replicas")
    parser.add_argument(
        "--uri", default="http://localhost:19530", help="Milvus server URI"
    )
    parser.add_argument("--token", default="", help="Milvus authentication token")
    parser.add_argument("--collection", default=COLLECTION_NAME, help="Collection name")
    parser.add_argument(
        "--replicas", type=int, default=3, help="Number of in-memory replicas"
    )
    args = parser.parse_args()

    client = MilvusClient(uri=args.uri, token=args.token)

    # Release collection first if already loaded
    try:
        client.release_collection(args.collection)
        print(f"Released collection '{args.collection}'")
    except Exception:
        pass

    # Load with specified replica count
    client.load_collection(
        collection_name=args.collection,
        replica_number=args.replicas,
    )
    print(
        f"Collection '{args.collection}' loaded with {args.replicas} in-memory replicas"
    )
    print(
        f"Search requests will be distributed across {args.replicas} QueryNode instances"
    )


if __name__ == "__main__":
    main()

Advanced Topics

Chapter 15: Advanced Search and Querying — It explores techniques like batch search processing and horizontal scaling with multiple replicas to handle thousands of concurrent users.

"""
Chapter 15: Batch Search Optimization

This example demonstrates how batch processing dramatically improves
search throughput by amortizing overhead costs using real Wikipedia embeddings.
"""

import os
import time
from datasets import load_dataset
from pymilvus import MilvusClient, DataType


def load_wikipedia_data(limit=50000, batch_size=1000):
    """
    Load Wikipedia embeddings from Hugging Face dataset.

    Args:
        limit: Number of records to load
        batch_size: Records per batch

    Yields:
        Batch of records formatted for Milvus
    """
    print(f"Loading Wikipedia dataset (total_records={limit})...")
    docs = load_dataset("Cohere/wikipedia-22-12-simple-embeddings", split="train")

    # Use slice to get data efficiently
    for start_idx in range(0, limit, batch_size):
        end_idx = min(start_idx + batch_size, limit)
        batch_docs = docs[start_idx:end_idx]

        # Format batch for Milvus
        batch = []
        for i in range(len(batch_docs["emb"])):
            batch.append(
                {
                    "emb": batch_docs["emb"][i],
                    "title": batch_docs["title"][i][:500],  # Truncate to max length
                    "text": batch_docs["text"][i][:2000],  # Truncate to max length
                    "wiki_id": batch_docs["wiki_id"][i],
                    "views": float(batch_docs["views"][i]),
                }
            )

        yield batch


def get_search_vectors(num_searches=100):
    """Get real search vectors from Wikipedia dataset."""
    print(f"Loading {num_searches} search vectors from dataset...")
    docs = load_dataset("Cohere/wikipedia-22-12-simple-embeddings", split="train")

    search_vectors = []

    for i, doc in enumerate(docs):
        if i >= num_searches:
            break
        search_vectors.append(doc["emb"])

    return search_vectors


def main():
    # Connect to Milvus
    client = MilvusClient(uri=os.getenv("MILVUS_URI", "http://localhost:19530"))

    collection_name = "wiki_articles"
    num_vectors = 50000
    dim = 768

    # ========================================================================
    # Setup: Create Collection and Insert Data
    # ========================================================================
    print("=" * 70)
    print("Setup: Creating collection and inserting data")
    print("=" * 70)

    if client.has_collection(collection_name):
        client.drop_collection(collection_name)

    # Create collection
    schema = client.create_schema(auto_id=True, enable_dynamic_field=False)
    schema.add_field(field_name="id", datatype=DataType.INT64, is_primary=True)
    schema.add_field(field_name="emb", datatype=DataType.FLOAT_VECTOR, dim=dim)
    schema.add_field(field_name="title", datatype=DataType.VARCHAR, max_length=100)
    schema.add_field(field_name="text", datatype=DataType.VARCHAR, max_length=5000)
    schema.add_field(field_name="wiki_id", datatype=DataType.INT64)
    schema.add_field(field_name="views", datatype=DataType.FLOAT)

    client.create_collection(collection_name=collection_name, schema=schema)

    # Create HNSW index
    index_params = client.prepare_index_params()
    index_params.add_index(
        field_name="emb",
        index_type="HNSW",
        metric_type="COSINE",
        params={"M": 16, "efConstruction": 200},
    )
    client.create_index(collection_name=collection_name, index_params=index_params)

    # Insert data in batches
    batch_num = 0
    total_inserted = 0
    for batch in load_wikipedia_data(limit=num_vectors, batch_size=1000):
        client.insert(collection_name=collection_name, data=batch)
        batch_num += 1
        total_inserted += len(batch)
        if batch_num % 10 == 0:
            print(f"Inserted {total_inserted} records...")

    print(f"Total inserted: {total_inserted} records")

    # Load collection
    client.load_collection(collection_name)

    search_params = {"metric_type": "COSINE", "params": {"ef": 128}}

    # ========================================================================
    # Example 1: Single Search
    # ========================================================================
    print("\n" + "=" * 70)
    print("Example 1: Single Search (Sequential)")
    print("=" * 70)

    search_vectors = get_search_vectors(num_searches=100)
    print("Running 100 sequential single searches...")

    start_time = time.time()
    for i, search_vector in enumerate(search_vectors):
        client.search(
            collection_name=collection_name,
            data=[search_vector],
            anns_field="emb",
            search_params=search_params,
            limit=10,
        )
    total_time = time.time() - start_time

    print(f"Total time: {total_time:.2f}s")
    print(f"Average time per search: {total_time / len(search_vectors) * 1000:.2f}ms")
    print(f"Throughput: {len(search_vectors) / total_time:.2f} searches/sec")

    # ========================================================================
    # Example 2: Batch Search
    # ========================================================================
    print("\n" + "=" * 70)
    print("Example 2: Batch Search (All at Once)")
    print("=" * 70)

    print("Running single batch search with 100 searches...")

    start_time = time.time()
    results = client.search(
        collection_name=collection_name,
        data=search_vectors,
        anns_field="emb",
        search_params=search_params,
        limit=10,
    )
    total_time = time.time() - start_time

    print(f"Total time: {total_time:.2f}s")
    print(f"Average time per search: {total_time / len(search_vectors) * 1000:.2f}ms")
    print(f"Throughput: {len(search_vectors) / total_time:.2f} searches/sec")

    print("\nSample results from first search:")
    for i, hit in enumerate(results[0][:5]):
        print(f"  {i + 1}. ID: {hit['id']}, Distance: {hit['distance']:.4f}")

    # ========================================================================
    # Cleanup
    # ========================================================================
    print("\nCleaning up...")
    client.release_collection(collection_name)
    client.drop_collection(collection_name)
    print("Done!")


if __name__ == "__main__":
    main()

Chapter 16: Implementing Multi-Tenancy in Milvus — This standout chapter details different strategies for isolating tenant data, ranging from database-level isolation to collection-level and partition-key-based approaches.

Database-Level Multi-Tenancy in Milvus
This notebook demonstrates how to implement database-level isolation for multiple tenants using Milvus. Each tenant gets their own dedicated database, providing the strongest form of tenant isolation.

Prerequisites
Running Milvus instance (v2.5.x or later)
PyMilvus SDK (v2.5.8 or compatible)
Setup and Configuration
# Install required packages if not already installed
# !pip install pymilvus==2.5.8
from pymilvus import MilvusClient, DataType
import random

# Configuration
MILVUS_URI = "http://localhost:19530"  # Update this to your Milvus instance
client = MilvusClient(uri=MILVUS_URI)

print(f"Connected to Milvus at {MILVUS_URI}")
Define Tenant Configurations
Each tenant will have their own database with specific collections.

# Define tenant configurations
TENANTS = {
    "company_a": {"db_name": "tenant_company_a", "collections": ["products", "users"]},
    "company_b": {
        "db_name": "tenant_company_b",
        "collections": ["inventory", "customers"],
    },
    "company_c": {
        "db_name": "tenant_company_c",
        "collections": ["documents", "analytics"],
    },
}

print(f"Configured {len(TENANTS)} tenants:")
for tenant_id, config in TENANTS.items():
    print(
        f"  - {tenant_id}: {config['db_name']} with collections {config['collections']}"
    )
Tenant Database Setup Functions
def setup_tenant_database(tenant_id, config):
    """Set up a dedicated database for a tenant"""
    db_name = config["db_name"]

    # Create database for tenant
    try:
        client.create_database(db_name)
        print(f"✓ Created database '{db_name}' for {tenant_id}")
    except Exception as e:
        print(f"Database '{db_name}' may already exist: {e}")

    # Switch to tenant database
    client.using_database(db_name)

    # Create collections for tenant
    for coll_name in config["collections"]:
        schema = client.create_schema(auto_id=True, enable_dynamic_fields=True)

        # Add fields (example schema)
        schema.add_field(field_name="id", datatype=DataType.INT64, is_primary=True)
        schema.add_field(field_name="vector", datatype=DataType.FLOAT_VECTOR, dim=768)
        schema.add_field(field_name="text", datatype=DataType.VARCHAR, max_length=1000)
        schema.add_field(
            field_name="category", datatype=DataType.VARCHAR, max_length=100
        )

        # Create collection
        try:
            client.create_collection(collection_name=coll_name, schema=schema)
            print(f"✓ Created collection '{coll_name}' in '{db_name}'")

            # Create index for vector field
            client.create_index(
                collection_name=coll_name,
                field_name="vector",
                index_params={"index_type": "FLAT", "metric_type": "COSINE"},
            )
            print(f"✓ Created index for '{coll_name}'")

        except Exception as e:
            print(f"Collection '{coll_name}' may already exist: {e}")

    # Switch back to default database
    client.using_database("default")
    return True
def insert_sample_data(tenant_id, collection_name, num_records=100):
    """Insert sample data into a tenant's collection"""
    db_name = TENANTS[tenant_id]["db_name"]

    # Switch to tenant database
    client.using_database(db_name)

    # Generate sample data
    data = []
    categories = ["electronics", "books", "clothing", "home", "sports"]

    for i in range(num_records):
        data.append(
            {
                "vector": [random.random() for _ in range(768)],
                "text": f"{tenant_id} item {i + 1}",
                "category": random.choice(categories),
            }
        )

    # Insert data
    client.insert(collection_name=collection_name, data=data)

    # Load collection for searching
    client.load_collection(collection_name)

    print(f"✓ Inserted {num_records} records into {tenant_id}/{collection_name}")

    # Switch back to default database
    client.using_database("default")
    return True
def query_tenant_data(tenant_id, collection_name, query_vector, limit=5):
    """Query data from a specific tenant's database"""
    db_name = TENANTS[tenant_id]["db_name"]

    # Switch to tenant database
    client.using_database(db_name)

    # Perform search - data is completely isolated
    results = client.search(
        collection_name=collection_name,
        data=[query_vector],
        limit=limit,
        output_fields=["text", "category"],
    )

    # Switch back to default database
    client.using_database("default")

    return results
Setup All Tenant Databases
Create separate databases for each tenant with their specific collections.

# Setup databases for all tenants
print("Setting up tenant databases...")
for tenant_id, config in TENANTS.items():
    print(f"\nSetting up {tenant_id}:")
    setup_tenant_database(tenant_id, config)
Insert Sample Data for Each Tenant
# Insert sample data for each tenant
print("Inserting sample data...")
for tenant_id, config in TENANTS.items():
    print(f"\nInserting data for {tenant_id}:")
    for collection_name in config["collections"]:
        insert_sample_data(tenant_id, collection_name, num_records=50)
Demonstrate Tenant Isolation
Show that each tenant can only access their own data.

# Generate a query vector
query_vector = [random.random() for _ in range(768)]

print("Demonstrating tenant isolation:")
print("\nQuerying Company A's products:")
results_a = query_tenant_data("company_a", "products", query_vector, limit=3)
for i, result in enumerate(results_a[0]):
    print(
        f"  {i + 1}. {result['entity']['text']} ({result['entity']['category']}) - Score: {result['distance']:.4f}"
    )

print("\nQuerying Company B's inventory:")
results_b = query_tenant_data("company_b", "inventory", query_vector, limit=3)
for i, result in enumerate(results_b[0]):
    print(
        f"  {i + 1}. {result['entity']['text']} ({result['entity']['category']}) - Score: {result['distance']:.4f}"
    )

print("\nQuerying Company C's documents:")
results_c = query_tenant_data("company_c", "documents", query_vector, limit=3)
for i, result in enumerate(results_c[0]):
    print(
        f"  {i + 1}. {result['entity']['text']} ({result['entity']['category']}) - Score: {result['distance']:.4f}"
    )
Verify Database Isolation
# List all databases to verify isolation
databases = client.list_databases()
print("All databases in Milvus:")
for db in databases:
    print(f"  - {db}")

print(
    f"\nTotal tenant databases created: {len([db for db in databases if db.startswith('tenant_')])}"
)
Test Cross-Tenant Access (Should Fail)
Demonstrate that tenants cannot access each other's data.

# Try to access Company B's data from Company A's database context
print("Testing cross-tenant access restriction:")
try:
    # Switch to Company A's database
    client.using_database(TENANTS["company_a"]["db_name"])

    # Try to access Company B's collection (this should fail)
    results = client.search(
        collection_name="inventory",  # Company B's collection
        data=[query_vector],
        limit=5,
    )
    print("❌ Cross-tenant access succeeded (this shouldn't happen!)")
except Exception as e:
    print(f"✓ Cross-tenant access properly blocked: {str(e)[:100]}...")
finally:
    # Switch back to default database
    client.using_database("default")
Performance and Resource Analysis
# Analyze resource usage per tenant
print("Database-Level Multi-Tenancy Analysis:")
print("\nBenefits:")
print("  ✓ Maximum data isolation (100% secure)")
print("  ✓ Simple tenant lifecycle management")
print("  ✓ Granular RBAC control per database")
print("  ✓ Zero risk of data leakage between tenants")

print("\nLimitations:")
print("  ⚠ Limited to ~64 tenants (Milvus database limit)")
print("  ⚠ 40-60% higher memory usage per tenant")
print("  ⚠ 3-5x more operational complexity")
print("  ⚠ Higher infrastructure costs")

total_collections = sum(len(config["collections"]) for config in TENANTS.values())
print("\nCurrent Setup:")
print(f"  - Total tenants: {len(TENANTS)}")
print(f"  - Total databases: {len(TENANTS)}")
print(f"  - Total collections: {total_collections}")
print(f"  - Average collections per tenant: {total_collections / len(TENANTS):.1f}")
Cleanup (Optional)
Uncomment and run the following cell to clean up the created databases.

# Cleanup - Uncomment to remove all tenant databases
# print("Cleaning up tenant databases...")
# for tenant_id, config in TENANTS.items():
#     try:
#         client.drop_database(config["db_name"])
#         print(f"✓ Dropped database {config['db_name']}")
#     except Exception as e:
#         print(f"Error dropping {config['db_name']}: {e}")
# print("Cleanup completed!")
Summary
This notebook demonstrated database-level multi-tenancy in Milvus, which provides:

Strongest Isolation: Complete data separation at the database level
Simple Management: Each tenant has their own namespace
Security: Zero risk of cross-tenant data access
Scalability Limit: Suitable for up to 64 enterprise tenants
This approach is ideal for enterprise customers requiring strict compliance and data isolation guarantees.

Chapter 17: Real-world Applications and Integrations — The book concludes by showing how Milvus integrates with frameworks like LangChain for RAG and image retrieval systems using pre-trained models like ResNet50.

Conclusion

I found this book incredibly useful. Even though, as mentioned, I consider myself a bit experienced with Milvus, the depth of technical detail helped me significantly strengthen my knowledge. Whether you are just starting with local testing or managing large-scale clusters on platforms like watsonx Data, this handbook provides the architectural clarity needed to build truly high-performance vector search systems.

>>> Thanks for reading <<<

Stop Treating Your Code Security Like a “Check Engine” Light (or part #2 of what we can do with Bob even if we’re not coders…)

Alain Airom (Ayrom) — Thu, 09 Apr 2026 12:13:01 +0000

Going to some of my old code and asking Bob to check for vulnerabilities!

Introduction

We’ve all been there. A little red light pops up on the dashboard, and instead of pulling over, we perform the ancient ritual of “The Great Ignore.” We tell ourselves it’s probably just a sensitive sensor or that the car feels fine, so it must be fine. We do the same thing with our digital infrastructure: we see a CVE notification, shrug, and decide that today isn’t the day to deal with a “theoretical” exploit.

But here’s the cold, hard reality: in the world of cybersecurity, that “Check Engine” light isn’t a glitch — it’s the sound of a digital crowbar prying at your front door.

While it’s easy to joke about the absurdity of a refrigerator getting hacked or a smart-toaster joining a botnet, the punchline stops being funny when it’s your customer data, your intellectual property, or your entire company’s reputation on the line. Security isn’t just a checkbox for the compliance department or a “nice-to-have” for when the budget allows; it’s the foundation that keeps the whole house from sliding down the hill.

In this post, I’m going to stop the nervous laughter and look at why treating vulnerabilities like minor inconveniences is the fastest way to become a cautionary tale. It’s time to take security seriously — before a hacker does it for you.

After the introduction, going to some of my old codes, I checked on an old code which I use by the way (locally) to merge my PDF files instead of going to a web site where I’m not quite sure about the privacy matters (by the way the code(s) are in an old blog post). So I asked Bob to to a security/vulnerability check. I got upset by myself from Bob findings 💀💀💀 😂.

As a refresher, I post again the shortest code which does not have a GUI (the Streamlit version is in the post)👇

# pdf_merger_2.py
import os
from PyPDF2 import PdfMerger, PdfReader

def merge_pdfs_in_directory(input_folder, output_folder, output_filename="merged_output.pdf"):
    """
    Recursively finds all PDF files in the input_folder and merges them
    into a single PDF file in the output_folder.

    Args:
        input_folder (str): The path to the folder containing PDF files.
        output_folder (str): The path to the folder where the merged PDF will be saved.
        output_filename (str): The name of the merged PDF file.
    """
    if not os.path.exists(input_folder):
        print(f"Error: Input folder '{input_folder}' does not exist.")
        return

    # --- This is the part that creates the output directory if it doesn't exist ---
    if not os.path.exists(output_folder):
        os.makedirs(output_folder)
        print(f"Created output folder: '{output_folder}'")
    # -----------------------------------------------------------------------------

    merger = PdfMerger()
    pdf_files_found = []

    # Walk through the input folder recursively
    for root, _, files in os.walk(input_folder):
        for file in files:
            if file.lower().endswith('.pdf'):
                filepath = os.path.join(root, file)
                pdf_files_found.append(filepath)

    if not pdf_files_found:
        print(f"No PDF files found in '{input_folder}' and its subdirectories.")
        return

    # Sort the files for consistent merging order (optional, but good practice)
    pdf_files_found.sort()

    print(f"Found {len(pdf_files_found)} PDF files to merge:")
    for pdf_file in pdf_files_found:
        print(f"- {pdf_file}")
        try:
            with open(pdf_file, 'rb') as f:
                merger.append(PdfReader(f))
        except Exception as e:
            print(f"Error appending {pdf_file}: {e}")

    output_filepath = os.path.join(output_folder, output_filename)

    try:
        with open(output_filepath, 'wb') as output_file:
            merger.write(output_file)
        print(f"\nSuccessfully merged PDF files to: '{output_filepath}'")
    except Exception as e:
        print(f"Error writing merged file: {e}")
    finally:
        merger.close()

if __name__ == "__main__":
    script_dir = os.path.dirname(os.path.abspath(__file__))
    input_folder_name = "input"
    output_folder_name = "output"

    # Construct full paths
    input_path = os.path.join(script_dir, input_folder_name)
    output_path = os.path.join(script_dir, output_folder_name)

    # Example Usage:
    # 1. Create an 'input' folder in the same directory as this script.
    # 2. Place some PDF files (and even subfolders with PDFs) inside the 'input' folder.
    # 3. Run this script.
    # The merged PDF will be created in the 'output' folder.
    merge_pdfs_in_directory(input_path, output_path, "all_merged_documents.pdf")

Bob’s anlysis was astonishing… 🫣

Lessons Learned and Conclusion

The security analysis of these two PDF merger applications serves as a stark reminder of how easily vulnerabilities can hide in even the most “straightforward” scripts. As we’ve seen, ignoring simple or complex security pitfalls — from uncontrolled resource consumption to path traversal — can transform a helpful tool into a significant liability.

While one might not “bother much” with extensive hardening for a quick local script, the standard changes entirely in a real-world software delivery environment. In professional production, these pitfalls must be double-checked with extreme rigor. Utilizing specialized tools like IBM Bob SDLC for security scanning and lifecycle management is an essential strategy for identifying these risks early; it ensures that even those who aren’t career developers can maintain a professional security posture and prevent their software from becoming a cautionary tale.

>>> Thanks for Reading <<<

My first Experience with Google’s Agent Development Kit (ADK-Go)

Alain Airom (Ayrom) — Tue, 07 Apr 2026 08:26:20 +0000

A first short experience with Google’s Go language Agent Development Kit (ADK-Go)

Introduction

The landscape of Artificial Intelligence is shifting from models that simply “chat” to agents that “do.” As developers look for more robust ways to integrate LLMs into real-world workflows, Google has released the Agent Development Kit (ADK) — an open-source framework designed to take the complexity out of building reliable, tool-augmented AI agents. Whether you are looking to automate multi-step research tasks, manage local file systems, or bridge the gap between cloud models and local execution, the ADK provides the standardized scaffolding necessary to move from experimental prompts to production-ready autonomous assistants. By open-sourcing this toolkit, Google is offering the developer community a structured, modular approach to session management, tool integration, and model orchestration, effectively lowering the barrier to entry for the next generation of “agentic” software.

Excerpt from ADK-Go Repository

_Agent Development Kit (ADK) is a flexible and modular framework that applies software development principles to AI agent creation. It is designed to simplify building, deploying, and orchestrating agent workflows, from simple tasks to complex systems. While optimized for Gemini, ADK is model-agnostic, deployment-agnostic, and compatible with other frameworks.

This Go version of ADK is ideal for developers building cloud-native agent applications, leveraging Go’s strengths in concurrency and performance.

Idiomatic Go: Designed to feel natural and leverage the power of Go.
Rich Tool Ecosystem: Utilize pre-built tools, custom functions, or integrate existing tools to give agents diverse capabilities.
Code-First Development: Define agent logic, tools, and orchestration directly in Go for ultimate flexibility, testability, and versioning.
Modular Multi-Agent Systems: Design scalable applications by composing multiple specialized agents.
Deploy Anywhere: Easily containerize and deploy agents, with strong support for cloud-native environments like Google Cloud Run._

Running the Provided Sample Out-of-the-Box!

Creating a basic agent is a matter of minutes using the example provided on the documentation site!

Obtain you Google API key which you should export as is or put in an .env file.

export GOOGLE_API_KEY="abcdefghixxxxxxxxxxxxx"

Copy/paste the code

package main

import (
    "context"
    "log"
    "os"

    "google.golang.org/adk/agent"
    "google.golang.org/adk/agent/llmagent"
    "google.golang.org/adk/cmd/launcher"
    "google.golang.org/adk/cmd/launcher/full"
    "google.golang.org/adk/model/gemini"
    "google.golang.org/adk/tool"
    "google.golang.org/adk/tool/geminitool"
    "google.golang.org/genai"
)

func main() {
    ctx := context.Background()

    model, err := gemini.NewModel(ctx, "gemini-2.5-flash", &genai.ClientConfig{
        APIKey: os.Getenv("GOOGLE_API_KEY"),
    })
    if err != nil {
        log.Fatalf("Failed to create model: %v", err)
    }

    timeAgent, err := llmagent.New(llmagent.Config{
        Name:        "hello_time_agent",
        Model:       model,
        Description: "Tells the current time in a specified city.",
        Instruction: "You are a helpful assistant that tells the current time in a city.",
        Tools: []tool.Tool{
            geminitool.GoogleSearch{},
        },
    })
    if err != nil {
        log.Fatalf("Failed to create agent: %v", err)
    }

    config := &launcher.Config{
        AgentLoader: agent.NewSingleLoader(timeAgent),
    }

    l := full.NewLauncher()
    if err = l.Execute(ctx, config, os.Args[1:]); err != nil {
        log.Fatalf("Run failed: %v\n\n%s", err, l.CommandLineSyntax())
    }
}

Configure the dependencies

go mod init my-agent/main
go mod tidy

You can run the application either in console mode or as a web app!

go run agent.go


User -> what is the time in Paris?

Agent -> The current time in Paris, France is 08:31 AM. Paris is currently observing Central European Summer Time (CEST), which is UTC+2.
User ->

ADK Web is not meant for use in production deployments. You should use ADK Web for development and debugging purposes only.

go run agent.go web api webui

Implementing A local Agent using Ollama and Lessons Learned

After a series of intensive integration tests exhausted my Google API quotas, I pivoted to a local development workflow. By leveraging Ollama to host models locally, I was able to continue refining the agent without API constraints. To speed up the transition, I brought in Bob, our AI construction specialist, to streamline the architecture and get the local environment running in record time.

First Test: Console Mode Application

The first attempt is just a simple Go application in console mode.

package main

import (
 "bufio"
 "context"
 "fmt"
 "iter"
 "log"
 "os"
 "os/exec"
 "strings"

 "google.golang.org/adk/agent"
 "google.golang.org/adk/agent/llmagent"
 "google.golang.org/adk/cmd/launcher"
 "google.golang.org/adk/cmd/launcher/full"
 "google.golang.org/adk/model"
 "google.golang.org/adk/session"
 "google.golang.org/adk/tool"
 "google.golang.org/adk/tool/geminitool"
 "google.golang.org/genai"

 openai "github.com/openai/openai-go/v3"
 "github.com/openai/openai-go/v3/option"
)

// OllamaModel wraps Ollama to implement the adk model.LLM interface
type OllamaModel struct {
 client    openai.Client
 modelName string
}

// NewOllamaModel creates a new Ollama model instance
func NewOllamaModel(modelName string, baseURL string) (*OllamaModel, error) {
 if baseURL == "" {
  baseURL = "http://localhost:11434/v1"
 }

 client := openai.NewClient(
  option.WithBaseURL(baseURL),
  option.WithAPIKey("ollama"), // Ollama doesn't need a real API key
 )

 return &OllamaModel{
  client:    client,
  modelName: modelName,
 }, nil
}

// Name returns the model name
func (o *OllamaModel) Name() string {
 return o.modelName
}

// GenerateContent implements the model.LLM interface
func (o *OllamaModel) GenerateContent(ctx context.Context, req *model.LLMRequest, stream bool) iter.Seq2[*model.LLMResponse, error] {
 return func(yield func(*model.LLMResponse, error) bool) {
  // Convert ADK request to OpenAI format
  messages := []openai.ChatCompletionMessageParamUnion{}

  // Add system instruction from config if present
  if req.Config != nil && req.Config.SystemInstruction != nil && len(req.Config.SystemInstruction.Parts) > 0 {
   var systemText strings.Builder
   for _, part := range req.Config.SystemInstruction.Parts {
    if part.Text != "" {
     systemText.WriteString(part.Text)
    }
   }
   if systemText.Len() > 0 {
    messages = append(messages, openai.SystemMessage(systemText.String()))
   }
  }

  // Add conversation history
  for _, content := range req.Contents {
   role := "user"
   if content.Role == "model" {
    role = "assistant"
   }

   var contentText strings.Builder
   for _, part := range content.Parts {
    if part.Text != "" {
     contentText.WriteString(part.Text)
    }
   }

   if contentText.Len() > 0 {
    if role == "user" {
     messages = append(messages, openai.UserMessage(contentText.String()))
    } else {
     messages = append(messages, openai.AssistantMessage(contentText.String()))
    }
   }
  }

  // Call Ollama via OpenAI-compatible API
  params := openai.ChatCompletionNewParams{
   Messages: messages,
   Model:    openai.ChatModel(o.modelName),
  }

  completion, err := o.client.Chat.Completions.New(ctx, params)
  if err != nil {
   yield(nil, fmt.Errorf("ollama API error: %w", err))
   return
  }

  // Convert response back to ADK format
  if len(completion.Choices) == 0 {
   yield(nil, fmt.Errorf("no response from Ollama"))
   return
  }

  responseText := completion.Choices[0].Message.Content

  response := &model.LLMResponse{
   Content: &genai.Content{
    Parts: []*genai.Part{
     genai.NewPartFromText(responseText),
    },
    Role: "model",
   },
  }

  yield(response, nil)
 }
}

func main() {
 // Load environment variables from .env file
 if data, err := os.ReadFile(".env"); err == nil {
  for _, line := range strings.Split(string(data), "\n") {
   line = strings.TrimSpace(line)
   if strings.HasPrefix(line, "export ") {
    line = strings.TrimPrefix(line, "export ")
   }
   if parts := strings.SplitN(line, "=", 2); len(parts) == 2 {
    key := strings.TrimSpace(parts[0])
    value := strings.Trim(strings.TrimSpace(parts[1]), "\"")
    os.Setenv(key, value)
   }
  }
 }

 // Set default model if not specified
 if os.Getenv("OLLAMA_MODEL") == "" {
  os.Setenv("OLLAMA_MODEL", "mistral:7b")
 }

 // Check if we are the sub-process worker
 if len(os.Args) > 1 && os.Args[1] == "internal-run" {
  runAgentWorker()
  return
 }

 // Interactive Loop
 fmt.Println("--- Ollama Agent (ADK) ---")
 fmt.Println("Type 'exit' to stop.")

 scanner := bufio.NewScanner(os.Stdin)
 for {
  fmt.Print("\nUser -> ")
  if !scanner.Scan() {
   break
  }

  input := strings.TrimSpace(scanner.Text())
  if input == "exit" || input == "quit" {
   fmt.Println("Goodbye!")
   break
  }
  if input == "" {
   continue
  }

  // Execute ourselves with the 'internal-run' flag
  cmd := exec.Command(os.Args[0], "internal-run", input)
  cmd.Stdout = os.Stdout
  cmd.Stderr = os.Stderr
  cmd.Env = os.Environ() // Pass environment variables to subprocess

  _ = cmd.Run()
 }
}

func runAgentWorker() {
 ctx := context.Background()

 // Get the user input from args
 if len(os.Args) < 3 {
  return
 }
 userInput := os.Args[2]

 // Create Ollama model - use model from environment or default
 modelName := os.Getenv("OLLAMA_MODEL")
 if modelName == "" {
  modelName = "mistral:7b"
 }

 ollamaModel, err := NewOllamaModel(modelName, "http://localhost:11434/v1")
 if err != nil {
  log.Fatalf("Failed to create Ollama model: %v", err)
 }
 log.Printf("DEBUG: Using Ollama model: %s", modelName)

 // Get Google API key for search tool
 apiKey := os.Getenv("GOOGLE_API_KEY")
 var tools []tool.Tool
 if apiKey != "" {
  // Add Google Search tool if API key is available
  tools = append(tools, geminitool.GoogleSearch{})
  log.Printf("DEBUG: Google Search tool added (API key present: %d chars)", len(apiKey))
 } else {
  log.Printf("DEBUG: No Google API key found, search tool not available")
 }

 // Create agent with Ollama model
 agentInstance, err := llmagent.New(llmagent.Config{
  Name:        "ollama_assistant",
  Model:       ollamaModel,
  Description: "A helpful assistant powered by IBM Granite via Ollama.",
  Instruction: "You are a helpful assistant. When asked about current information like time, weather, or recent events, you MUST use the Google Search tool to find accurate, up-to-date information. Do not make up answers or say you cannot access real-time data - use the search tool instead.",
  Tools:       tools,
 })
 if err != nil {
  log.Fatalf("Failed to create agent: %v", err)
 }

 // Create session service
 sessionService := session.InMemoryService()

 config := &launcher.Config{
  SessionService: sessionService,
  AgentLoader:    agent.NewSingleLoader(agentInstance),
 }

 l := full.NewLauncher()

 // Run console mode and pipe the user input to it
 // Create a pipe to send input
 r, w, err := os.Pipe()
 if err != nil {
  log.Fatalf("Failed to create pipe: %v", err)
 }
 oldStdin := os.Stdin
 os.Stdin = r

 // Write the input and exit command
 go func() {
  fmt.Fprintln(w, userInput)
  fmt.Fprintln(w, "exit")
  w.Close()
 }()

 // Execute in console mode
 if err := l.Execute(ctx, config, []string{"console"}); err != nil {
  os.Stdin = oldStdin
  log.Fatalf("Run failed: %v", err)
 }

 os.Stdin = oldStdin
}

// Made with Bob

The application runs as expected, even in console mode!

Lessons learned while making this application

What one should be aware of (either reading the documentation or learning by doing 😂);

The ADK is not designed to function as a s*tandalone CLI* or monolithic application. Instead, it is built as a highly modular agentic middleware. It is optimized to run in Interactive Mode, where it is orchestrated by other applications or agents, or in Web Mode, providing a robust backend for browser-based AI interfaces.
Another important fact: Not ALL models are fit for function calling!

While It is entirely possible to architect simple, zero-tool agents that function purely on the model’s pre-existing weights. While useful for general reasoning or creative tasks, these agents lack the real-time grounding provided by the ADK’s orchestration layer, making them less effective for data-sensitive or live environments.

# Ollama Model Configuration
# Models that likely support function calling (my list):
# - llama3.2:latest (recommended, 2GB)
# - llama3:latest (4.7GB)
# - mistral:7b (4.4GB)
# - deepseek-r1:latest (5.2GB)
#
# Models without function calling:
# - ibm/granite4:3b (default, no tools)
# - gemma3:4b (no tools)
# export OLLAMA_MODEL="llama3.2:latest"
export OLLAMA_MODEL="mistral:7b"

Second Test: Web UI Mode Application

Next, I wanted to see this agent in its natural habitat: a Web UI. While I prepped the environment, Bob essentially finished the job in two clicks. The result? A perfectly functional web interface that looks and feels exactly like the live demos on the ADK documentation site — Ollama integration and all.

package main

import (
 "context"
 "fmt"
 "iter"
 "log"
 "os"
 "strings"

 "google.golang.org/adk/agent"
 "google.golang.org/adk/agent/llmagent"
 "google.golang.org/adk/cmd/launcher"
 "google.golang.org/adk/cmd/launcher/full"
 "google.golang.org/adk/model"
 "google.golang.org/adk/tool"
 "google.golang.org/adk/tool/geminitool"
 "google.golang.org/genai"

 openai "github.com/openai/openai-go/v3"
 "github.com/openai/openai-go/v3/option"
)

// OllamaModel wraps Ollama to implement the adk model.LLM interface
type OllamaModel struct {
 client    openai.Client
 modelName string
}

// NewOllamaModel creates a new Ollama model instance
func NewOllamaModel(modelName string, baseURL string) (*OllamaModel, error) {
 if baseURL == "" {
  baseURL = "http://localhost:11434/v1"
 }

 client := openai.NewClient(
  option.WithBaseURL(baseURL),
  option.WithAPIKey("ollama"), // Ollama doesn't need a real API key
 )

 return &OllamaModel{
  client:    client,
  modelName: modelName,
 }, nil
}

// Name returns the model name
func (o *OllamaModel) Name() string {
 return o.modelName
}

// GenerateContent implements the model.LLM interface
func (o *OllamaModel) GenerateContent(ctx context.Context, req *model.LLMRequest, stream bool) iter.Seq2[*model.LLMResponse, error] {
 return func(yield func(*model.LLMResponse, error) bool) {
  // Convert ADK request to OpenAI format
  messages := []openai.ChatCompletionMessageParamUnion{}

  // Add system instruction from config if present
  if req.Config != nil && req.Config.SystemInstruction != nil && len(req.Config.SystemInstruction.Parts) > 0 {
   var systemText strings.Builder
   for _, part := range req.Config.SystemInstruction.Parts {
    if part.Text != "" {
     systemText.WriteString(part.Text)
    }
   }
   if systemText.Len() > 0 {
    messages = append(messages, openai.SystemMessage(systemText.String()))
   }
  }

  // Add conversation history
  for _, content := range req.Contents {
   role := "user"
   if content.Role == "model" {
    role = "assistant"
   }

   var contentText strings.Builder
   for _, part := range content.Parts {
    if part.Text != "" {
     contentText.WriteString(part.Text)
    }
   }

   if contentText.Len() > 0 {
    if role == "user" {
     messages = append(messages, openai.UserMessage(contentText.String()))
    } else {
     messages = append(messages, openai.AssistantMessage(contentText.String()))
    }
   }
  }

  // Call Ollama via OpenAI-compatible API
  params := openai.ChatCompletionNewParams{
   Messages: messages,
   Model:    openai.ChatModel(o.modelName),
  }

  completion, err := o.client.Chat.Completions.New(ctx, params)
  if err != nil {
   yield(nil, fmt.Errorf("ollama API error: %w", err))
   return
  }

  // Convert response back to ADK format
  if len(completion.Choices) == 0 {
   yield(nil, fmt.Errorf("no response from Ollama"))
   return
  }

  responseText := completion.Choices[0].Message.Content

  response := &model.LLMResponse{
   Content: &genai.Content{
    Parts: []*genai.Part{
     genai.NewPartFromText(responseText),
    },
    Role: "model",
   },
  }

  yield(response, nil)
 }
}

func main() {
 // Load environment variables from .env file
 if data, err := os.ReadFile(".env"); err == nil {
  for _, line := range strings.Split(string(data), "\n") {
   line = strings.TrimSpace(line)
   if strings.HasPrefix(line, "export ") {
    line = strings.TrimPrefix(line, "export ")
   }
   if parts := strings.SplitN(line, "=", 2); len(parts) == 2 {
    key := strings.TrimSpace(parts[0])
    value := strings.Trim(strings.TrimSpace(parts[1]), "\"")
    os.Setenv(key, value)
   }
  }
 }

 ctx := context.Background()

 // Set default model if not specified
 modelName := os.Getenv("OLLAMA_MODEL")
 if modelName == "" {
  modelName = "mistral:7b"
 }

 // Create Ollama model
 ollamaModel, err := NewOllamaModel(modelName, "http://localhost:11434/v1")
 if err != nil {
  log.Fatalf("Failed to create Ollama model: %v", err)
 }
 log.Printf("Using Ollama model: %s", modelName)

 // Get Google API key for search tool
 apiKey := os.Getenv("GOOGLE_API_KEY")
 var tools []tool.Tool
 if apiKey != "" {
  // Add Google Search tool if API key is available
  tools = append(tools, geminitool.GoogleSearch{})
  log.Printf("Google Search tool enabled")
 } else {
  log.Printf("No Google API key found, search tool not available")
 }

 // Create agent with Ollama model
 agentInstance, err := llmagent.New(llmagent.Config{
  Name:        "ollama_assistant",
  Model:       ollamaModel,
  Description: "A helpful assistant powered via Ollama and ADK.",
  Instruction: "You are a helpful assistant. When asked about current information like time, weather, or recent events, you MUST use the Google Search tool to find accurate, up-to-date information. Do not make up answers or say you cannot access real-time data - use the search tool instead.",
  Tools:       tools,
 })
 if err != nil {
  log.Fatalf("Failed to create agent: %v", err)
 }

 config := &launcher.Config{
  AgentLoader: agent.NewSingleLoader(agentInstance),
 }

 l := full.NewLauncher()

 // Execute with command line arguments (supports web api webui)
 if err = l.Execute(ctx, config, os.Args[1:]); err != nil {
  log.Fatalf("Run failed: %v\n\n%s", err, l.CommandLineSyntax())
 }
}

// Made with Bob

Conclusion

The Agent Development Kit (ADK) is a framework that prioritizes interaction over isolation, designed to transform static LLMs into dynamic collaborators that can reach out to the world via Google Search or run privately through Ollama. By moving the logic into a Web UI and leveraging Bob’s structural guidance, we can demonstrate that the ADK isn’t just a tool for building chatbots — it’s an engine for creating “agentic” workflows that are as at home in a browser as they are in a complex multi-agent ecosystem. Whether you’re relying on a model’s internal knowledge or empowering it with real-time data, the ADK provides the standardized bridge needed to turn AI potential into practical, integrated reality.

>>> Thanks for reading <<<

DEV Community: Alain Airom (Ayrom)

Inside Docling Factory: Building a Multimodal RAG Powerhouse

Intro-TL;DR

🔩 Key Components

The Core Engine: Docling & Layout Awareness

RAG & Vector Storage (rag_engine.py)

Hybrid Backend

A Word on The RAG Database: OpenSeach RAG

Observability: Seeing the Unseen

Ensuring Reliability: The Docling Factory Test Suite

RAG & Embedding Accuracy

Metrics and Observability

Flexible Processing: From Single Files to Batch Operations

CPU or GPU?

Cloud Deployment

The “End of Level 1” Log

Links

Mamba vs. Transformers: Architecture Comparison

The Core Architectures: Transformers and Mamba

Traditional Transformers

Mamba (Selective SSM)

Why Mamba Matters?

IBM and Mamba!

IBM Granite 4.0 & 4.1 Family

IBM Research Biomedical Models

Links, References and Citations

No More Automation from Scratch: Automate and Build Your Terraform with Draw.io Diagrams

Introduction

A word on the external tool used: Draw.io

Terraform Script Generator: A Diagram-to-Code Pipeline

The Backend: Analytical Engine with Flask and Python

Diagram Interpretation: Parsing the Draw.io XML

Code Generation: Turning Components into Terraform

Strategic Versatility: From a Curated List to Custom Templates

The Front-end — HTML And Javascript

Output Samples

Deployment

Conclusion

Links

Markdown Everything: My New Personal Project: HTML/URL to Markdown Converter

Introduction (and spoiler)

The Implementation

Core Conversion Engine

Data Acquisition Logic

Image Handling & Asset Management

Application Workflow

becomes #) : code blocks are wrapped in triple backticks.

Last but not Least: Extension for a Browser

The Manifest (Orchestrator)

The Popup (User Interface)

The Content Script (The “Brain”)

Technical Stack Summary

Conclusion

Links

Image Generation with Ollama is back with Japanese, Korean and Chinese Languages 🇯🇵 Support!

Introduction

Implementation

Conclusion

Links

Kafka 101: Why Event Streaming is the Central Nervous System of Modern Data

The Producer: Capturing the Event

The Consumer: Decoupled Processing

The Cluster: Ensuring Reliability

The Monitor: Real-Time Observability

Links

Mastering Modern Hiring Demonstration: Using Docling and PostgreSQL by Bob to Build a Local Candidate RAG Database

Introduction

The Architecture: A Hybrid Approach

Technologies Used for this project

The Data Workflow

1. Document Ingestion Flow

2. RAG Query Flow

Database Schema Design

Project Structure

Database and LLM

Information Ingestion

RAG Service

Conclusion

Links

Docling Data Pipeline for AstraDB Serverless RAG

RAG & Vector Storage (`rag_engine.py`)