End-to-End Enterprise RAG Pipeline with IBM Docling Pipelines and OpenSearch

Implementing a RAG pipeline with docling-pipelines and OpenSearch

Introduction

In enterprise Retrieval-Augmented Generation (RAG) applications, moving from a simple proof-of-concept to a robust production system requires high-fidelity document conversion, flexible topology orchestration, and low-latency vector retrieval. Unstructured enterprise documents—such as complex multi-page PDFs, technical whitepapers, and dynamic web pages—often break naive text splitters by scattering table structures, disrupting read order, or discarding critical metadata.

To solve these challenges at scale, this article demonstrates an end-to-end, production-ready RAG pipeline built using IBM Docling Pipelines (docling-pipeline) and OpenSearch. Docling Pipelines provides a Directed Acyclic Graph (DAG) framework for document curation—enabling declarative flows that handle document ingestion, high-accuracy text and structure extraction, language detection, readability analysis, hybrid chunking, embedding generation, and vector indexing. Combined with OpenSearch's k-Nearest Neighbors (k-NN) search capabilities and Ollama's local LLM/embedding inference, this stack provides a privacy-conscious, local-first RAG solution.

TL;DR-What is Docling-Pipelines?

Excerpt from docling-pipelines github repository.

Docling pipelines is an enterprise-grade document curation pipeline for Retrieval Augmented Generation (RAG) applications. It ingests data from unstructured sources, curates documents, and writes entities and vector embeddings to targets — enabling AI-ready pipelines at scale.

It connects to cloud document sources (S3, OneDrive, SharePoint, Google Drive, Box, and more) and extracts content and entities from PDF, DOCX, HTML, images, and other formats using Docling. Extracted content is curated for LLMs, converted into chunks and embeddings, and stored in a vector database such as Milvus or OpenSearch.

Features

• 📥 Multi-source ingestion — local filesystem, Amazon S3, IBM COS, SharePoint, OneDrive, Google Drive, Box, and web pages
• 📄 Document extraction — PDF, DOCX, HTML, images, and more via Docling, with optional VLM and ASR pipelines
• 🧠 Entity extraction — LLM-based extraction via LiteLLM (100+ providers), IBM watsonx.ai, or Docling templates
• ✂️ Chunking — Docling-native and semantic chunking strategies
• 🔢 Embeddings — vector embedding generation for any downstream vector store
• 🔍 Quality operators — language detection, readability scoring, PII/HAP detection, deduplication, redaction, SQL filtering, document classification, and ML enrichment
• 🗄️ Vector storage — write to OpenSearch or Milvus
• 🔀 DAG-based flows — define pipelines as JSON with automatic dependency resolution and parallel execution
• 🔌 Extensible — load custom operators from Python packages, local paths, or S3 without modifying core code
• 🖥️ Multiple interfaces — CLI, Python API (DocpipeFlowManager), and REST API (FastAPI)

---

Architecture Overview of the Solution

The pipeline processes unstructured documents through two execution paths: a real execution engine backed by docling-pipelines CLI/Python APIs, and a fallback simulation manager designed for zero-dependency test environments.

┌─────────────────────────────────────────────────────────────────────────────┐
│                           Document Ingestion Layer                          │
│          (Filesystem, Web URLs, S3/COS, SharePoint, Uploaded Files)          │
└──────────────────────────────────────┬──────────────────────────────────────┘
                                       │
                                       ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                           IBM Docling Curation DAG                          │
│                                                                             │
│  ┌────────────────┐     ┌──────────────────┐     ┌───────────────────────┐  │
│  │ ingest_source  ├────►│ extract_operator ├────►│ lang_detect/readability│ │
│  └────────────────┘     └────────┬─────────┘     └───────────┬───────────┘  │
│                                  │                           │              │
│                                  ▼                           ▼              │
│                         ┌────────────────────────────────────────┐          │
│                         │        docling_hybrid Chunker          │          │
│                         └────────────────┬───────────────────────┘          │
│                                          │                                  │
│                                          ▼                                  │
│                         ┌────────────────────────────────────────┐          │
│                         │    embeddings (Ollama/LiteLLM)         │          │
│                         └────────────────┬───────────────────────┘          │
└──────────────────────────────────────────┼──────────────────────────────────┘
                                           │
                                           ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                        OpenSearch Vector Store Index                        │
│                   (k-NN HNSW Indexing & Semantic Search)                    │
└──────────────────────────────────────┬──────────────────────────────────────┘
                                       │
                                       ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                             RAG Engine & Chat                               │
│              (Context Retrieval → Grounded LLM Prompting)                   │
└─────────────────────────────────────────────────────────────────────────────┘

The system is organized into modular components:

📄 Documents  →  📥 Ingest  →  📝 Extract  →  ✂️ Chunk  →  🔢 Embed  →  🗄️ VectorDB

• UI & Control Center (app.py): Interactively configures pipeline parameters, template topologies, document sources, and operator settings.

• Flow Definition Builder (flow_builder.py): Programmatically generates valid JSON DAG specifications following the official Docling Flow Authoring Format.

• Pipeline Manager (pipeline_manager.py): Orchestrates document processing, handling document parsing, step-by-step operator execution, schema evolution, and artifact export.

• Vector Database & RAG Engine (opensearch_client.py & rag_engine.py): Manages OpenSearch index creation, dense vector embeddings, k-NN retrieval, and LLM chat generation.

---

Implementation

Components Stack

The code and project uses the following stacks;

• Docling-pipelines library
• OpenSearch as RGA implementation
• Podman (for provisionning OpenSearch locally)
• Ollama (with granite-embedding:278m for embedding and granite4.2:latest for local chat enablement)
• Python Streamlit as the application's framework

---

Programmatic Pipeline Authoring (flow_builder.py)

Docling Pipelines relies on declarative flow representations that enforce dependency graphs between operators. The FlowBuilder class transforms user settings into standard flow JSON objects.

Below is an excerpt showing how the Full RAG Pipeline topology is constructed, complete with LiteLLM-compatible embedding parameters and OpenSearch vector database configs:

# flow_builder.py
from __future__ import annotations
import json
from pathlib import Path
from config import AppConfig

class FlowBuilder:
    """Builds a Docling Pipelines flow JSON from a UI pipeline configuration dict."""

    def __init__(self, config: AppConfig) -> None:
        self.config = config

    def _build_full_rag_pipeline(self, pipeline_cfg: dict) -> dict:
        """Complete RAG pipeline: Ingest → Extract → Chunk → Embed → VectorDB."""
        return {
            "flow_name": "full-rag-pipeline",
            "description": (
                "Complete RAG pipeline: ingest documents, extract text via Docling, "
                "chunk for retrieval, generate embeddings with Ollama, and store in OpenSearch."
            ),
            "flow": [
                self._ingest_operator(pipeline_cfg),
                self._extract_operator(["ingest"]),
                self._chunker_operator(["extract"], pipeline_cfg),
                self._embeddings_operator(["chunker"], pipeline_cfg),
                self._vectordb_operator(["embeddings"], pipeline_cfg),
            ],
            "global_config": {
                "doc_column": "content",
                "disable_validation": False,
            },
        }

    def _embeddings_operator(self, depends_on: list[str], pipeline_cfg: dict) -> dict:
        model = pipeline_cfg.get("ollama_model", self.config.ollama_embedding_model)
        if not model.startswith("openai/"):
            litellm_model = f"openai/{model}"
        else:
            litellm_model = model

        base = self.config.ollama_base_url.rstrip("/")
        api_base = base if base.endswith("/v1") else f"{base}/v1"

        return {
            "name": "embeddings",
            "type": "embeddings",
            "depends_on": depends_on,
            "config": {
                "provider": "litellm",
                "provider_config": {
                    "model_id": litellm_model,
                    "api_base": api_base,
                    "api_key": self.config.ollama_api_key,
                },
            },
        }

    def _vectordb_operator(self, depends_on: list[str], pipeline_cfg: dict) -> dict:
        return {
            "name": "vectordb",
            "type": "vectordb",
            "depends_on": depends_on,
            "config": {
                "provider": "opensearch",
                "provider_config": {
                    "index_name": pipeline_cfg.get(
                        "opensearch_index", self.config.opensearch_index
                    ),
                    "host": self.config.opensearch_host,
                    "port": self.config.opensearch_port,
                    "user": self.config.opensearch_user,
                    "password": self.config.opensearch_password,
                    "use_ssl": self.config.opensearch_use_ssl,
                },
            },
        }

Pipeline Execution Engine (pipeline_manager.py)

The PipelineManager resolves execution requests by dynamically choosing between the system binary docling-pipelines and a simulation mode. During document extraction, binary document types (such as PDFs or Word documents) are parsed using docling.

# pipeline_manager.py
from dataclasses import dataclass, field
import json
import logging
import os
import subprocess
import time
from pathlib import Path
from config import AppConfig

logger = logging.getLogger(__name__)

@dataclass
class PipelineResult:
    """Uniform result object returned by any backend."""
    success: bool
    duration_seconds: float
    docs_total: int
    docs_completed: int
    docs_failed: int
    error_message: str = ""
    operator_results: list[dict] = field(default_factory=list)
    output_documents: list[str] = field(default_factory=list)
    chunks: list[dict] = field(default_factory=list)
    schema_changes: dict[str, list[str]] = field(default_factory=dict)
    metadata: dict = field(default_factory=dict)

class PipelineManager:
    def __init__(self, config: AppConfig) -> None:
        self.config = config

    def run(self, flow_def: dict, pipeline_cfg: dict, progress_callback=None) -> PipelineResult:
        simulate = pipeline_cfg.get("simulate", not self.config.docling_pipelines_enabled)
        if not simulate and not self._cli_available():
            logger.warning("docling-pipelines CLI not found; falling back to simulation mode.")
            simulate = True

        if simulate:
            return self._run_simulation(flow_def, pipeline_cfg, progress_callback)
        else:
            return self._run_real(flow_def, pipeline_cfg, progress_callback)

    def _run_real(self, flow_def: dict, pipeline_cfg: dict, progress_callback=None) -> PipelineResult:
        ts = time.strftime("%Y%m%dT%H%M%S")
        flow_path = Path(self.config.output_dir) / f"flow_{ts}.json"
        flow_path.parent.mkdir(parents=True, exist_ok=True)
        flow_path.write_text(json.dumps(flow_def, indent=2), encoding="utf-8")

        cli = self._cli_path() or "docling-pipelines"
        cmd = [cli, "--flow-file", str(flow_path)]
        env = os.environ.copy()
        env["PYTHONPATH"] = "src:" + env.get("PYTHONPATH", "")

        start = time.time()
        completed = subprocess.run(cmd, capture_output=True, text=True, timeout=600, env=env)
        duration = time.time() - start

        success = (completed.returncode == 0)
        error_msg = "" if success else (completed.stdout or completed.stderr)

        return PipelineResult(
            success=success,
            duration_seconds=duration,
            docs_total=len(pipeline_cfg.get("saved_paths", [])) or 1,
            docs_completed=1 if success else 0,
            docs_failed=0 if success else 1,
            error_message=error_msg,
            output_documents=self._collect_output_files(),
        )

Dense Vector Ingestion & RAG Orchestration (rag_engine.py)

The RAGEngine links document processing outputs to OpenSearch vector stores. It generates text embeddings via Ollama models (e.g., granite-embedding:278m), performs k-NN retrieval, and queries chat models (e.g., granite4.2:latest).

To prevent hallucinated answers when no context matches a user query, the RAGEngine short-circuits execution and returns a deterministic fallback message without incurring unnecessary LLM generation costs.

# rag_engine.py
import logging
import requests
from config import AppConfig
from opensearch_client import OpenSearchClient

logger = logging.getLogger(__name__)

class RAGEngine:
    _NO_CONTEXT_MSG = (
        "No information about this topic was found in the knowledge base. "
        "Please ingest relevant documents before querying."
    )

    def __init__(self, *, config: AppConfig, os_client: OpenSearchClient | None = None) -> None:
        self.config = config
        self._os = os_client or OpenSearchClient(config=config)

    def ingest_chunks(self, *, chunks: list[dict], progress_callback=None) -> dict:
        if not chunks:
            return {"indexed": 0, "failed": 0, "message": "No chunks provided."}

        self._os.ensure_index()
        embedded = []
        for chunk in chunks:
            text = chunk.get("text", "").strip()
            if not text:
                continue
            vec = self._embed(text)
            embedded.append({**chunk, "embedding": vec})

        index_result = self._os.index_chunks(chunks=embedded)
        return {
            "indexed": index_result["indexed"],
            "failed": index_result.get("failed", 0),
            "message": f"Ingested {index_result['indexed']} chunks.",
        }

    def chat(self, *, query: str, top_k: int = 5, source_filter: str | None = None) -> dict:
        context = self.retrieve(query=query, top_k=top_k, source_filter=source_filter)

        # Deterministic check when retrieval returns zero chunks
        if not context:
            return {
                "answer": self._NO_CONTEXT_MSG,
                "context": [],
                "query": query,
                "sources": [],
                "retrieved_count": 0,
                "fallback": True,
            }

        ctx_text = "\n\n".join(
            f"[Source: {c['source_document']}, chunk {c['chunk_index']}]\n{c['text']}"
            for c in context
        )
        system = (
            "You are a helpful assistant. "
            "Answer the user's question using ONLY the provided context. "
            "If the context does not contain enough information, say so clearly."
        )
        prompt = f"Context:\n{ctx_text}\n\nQuestion: {query}\n\nAnswer:"
        answer = self._chat(prompt, system)
        sources = list(dict.fromkeys(c["source_document"] for c in context))

        return {
            "answer": answer,
            "context": context,
            "query": query,
            "sources": sources,
            "retrieved_count": len(context),
            "fallback": False,
        }

    def _embed(self, text: str) -> list[float]:
        url = f"{self.config.ollama_base_url.rstrip('/')}/api/embed"
        resp = requests.post(url, json={"model": self.config.ollama_embedding_model, "input": text}, timeout=60)
        data = resp.json()
        embeddings = data.get("embeddings") or data.get("embedding")
        return embeddings[0] if isinstance(embeddings[0], list) else embeddings

    def _chat(self, prompt: str, system: str = "") -> str:
        url = f"{self.config.ollama_base_url.rstrip('/')}/api/chat"
        messages = []
        if system:
            messages.append({"role": "system", "content": system})
        messages.append({"role": "user", "content": prompt})

        resp = requests.post(url, json={"model": self.config.ollama_llm_model, "messages": messages, "stream": False}, timeout=120)
        return resp.json().get("message", {}).get("content", "").strip()

---

Conclusion

Combining IBM Docling Pipelines with OpenSearch creates a reliable, scalable foundation for enterprise RAG applications. By decoupling document curation into modular operator DAGs—covering ingestion, high-precision layout extraction, structure-aware chunking, dense vector embedding, and indexing—organizations can handle diverse unstructured document formats consistently.

Integrating this pipeline with vector search and local LLM runtimes ensures strict context grounding while reducing hallucination risks. As unstructured data continues to grow in enterprise environments, modular Curation-as-Code frameworks like docling-pipeline will be critical for keeping production RAG systems reliable, observable, and easy to maintain.

Thanks for reading 🦆

Links

• Full code repository for this post: https://github.com/aairom/docling-pipeline-implementaion-tests
• Docling-Pipelines: https://github.com/IBM/docling-pipelines
• OpenSearch: https://opensearch.org/