title: Operator Reference

Operator Reference

Overview

OPERATOR_REFERENCE.md centralizes the public APIs that are visible to pipeline authors, application integrators, and operator users.

This reference is organized around four entry points:

• Operators: flow node implementations under src/docpipe/core/operators
• Programmatic execution: DocpipeFlowManager
• CLI execution: docling-pipelines
• Flow JSON definitions: DAG configuration consumed by the orchestrator

How to use this reference

• Use the operator sections when authoring flow JSON.
• Use the flow manager section when embedding Docling Pipelines in Python code.
• Use the CLI section when running or validating flows from the shell.
• For classification-specific architecture details, see docs/operators/quality/document_classifier_readme.md, which documents the simplified service-based architecture used by DocumentClassifierOperator.

Operator API Reference

Common Operator Contract

All operators ultimately inherit from AbstractOperator.

Shared behavior

• Operators receive a config dictionary during initialization.
• Operators expose metadata through the static method get_metadata(), which can be called on the class without instantiation (e.g., OperatorClass.get_metadata()).
• Input column requirements are expressed with get_required_features().
• Validation hooks are implemented via validate().
• Runtime work is usually performed by transform() or runner() methods depending on the operator.

Common input shape

Most operators consume a pyarrow.Table with some subset of these columns:

Operator Ownership Attribute

All operators must declare an owner class variable to support priority-based resolution when multiple operators share the same short_name.

# Built-in docpipe operators
from docpipe.core.constants.constants import DocpipeConstants
owner: str = DocpipeConstants.OWNER_DOCPIPE  # MUST be set for all built-in operators

# Custom operators
owner: str = "custom"  # MUST be set for all custom operators

For full details on the priority system, override behaviour, and registering custom tiers, see CONTRIBUTING.md — Custom Operator Requirements and External Operator Integration — Operator Priority.

Ingest Operators

IngestSourceOperator

Purpose: Multi-provider ingest abstraction for sources such as object storage, SharePoint, OneDrive, Google Drive, Box, Dropbox, web pages, and filesystem adapters.

Category: Ingest

Class: core.operators.ingest.ingest_source.IngestSourceOperator

Input Schema

• No input table required

Output Schema

Exceptions

• ImportError
• ValueError
• authentication and network failures

Example

Folder Ingestion:

{
  "name": "ingest-s3",
  "type": "ingest_source",
  "config": {
    "provider": "s3",
    "provider_config": {
      "bucket": "example-bucket",
      "prefix": "incoming/",
      "access_key": "${S3_ACCESS_KEY}",
      "secret_key": "${S3_SECRET_KEY}"
    },
    "include_filter": "pdf,docx",
    "max_files": 200
  }
}

Filesystem Ingestion:

{
  "name": "ingest-local",
  "type": "ingest_source",
  "config": {
    "provider": "filesystem",
    "provider_config": {
      "paths": ["/data/documents"],
      "recursive": true
    },
    "include_filter": "pdf,docx,txt"
  }
}

Note: S3 supports file-level ingestion by specifying a full key as the prefix (e.g., "incoming/report.pdf"). For other providers, use folder-level configuration.

Extract Operators

ExtractOperator

Purpose: Unified extraction operator using hexagonal architecture with multiple adapters for text extraction (docling_library, docling_serve) and entity extraction (litellm, watsonx, docling, none).

Category: Extract

Class: core.operators.extract.extract_operator.ExtractOperator

Streaming execution: When entity_extraction is enabled, text and entity extraction run concurrently. Each document is submitted for entity extraction as soon as its text extraction finishes, without waiting for the full batch. When entity extraction is disabled the operator behaves as before.

Input Schema

• name (string)
• binary payload column from an ingest operator
• optional existing content column

Output Schema

• content (or configured via doc_column parameter) - Primary extracted content: Markdown when doc_format=markdown (default), DocLang XML when doc_format=doclang
• content_html - HTML format (if additional_formats includes "html")
• content_json - JSON structured format (if additional_formats includes "json")
• content_text - Plain text format (if additional_formats includes "text")
• content_doctags - Docling's native DocTags format (if additional_formats includes "doctags")
• content_doclang - DocLang format (if additional_formats includes "doclang" and doc_format=markdown)
• content_markdown - Markdown format (if additional_formats includes "markdown" and doc_format=doclang)
• entities (or configured output_column) - Extracted entities as JSON string (if entity extraction enabled)
• doc_id_hash - Document hash identifier
• pages_processed - Number of pages in the document. Obtained from Docling extraction metadata when available; otherwise estimated using 3000 characters = 1 page
• Individual entity columns (if expand_extracted_data=true)

Execution Metadata

The operator provides the following metadata after execution:

• page_type_stats (dict): Aggregate estimated pages grouped by source document format (e.g., {"pdf": 120, "docx": 45})
• total_pages_converted (int): Total estimated pages across all successfully processed documents

Exceptions

• FlowExecutionFailedException
• ValueError for invalid configuration
• Provider-specific exceptions (Ollama, LiteLLM, Docling)

Example: Basic Text Extraction

{
  "id": "extract-node",
  "name": "extract",
  "operator": "extract_operator",
  "config": {
    "text_extraction": {
      "provider": "docling_library",
      "doc_column": "content",
      "provider_config": {}
    },
    "entity_extraction": {
      "provider": "none"
    }
  }
}

Example: VLM Text Extraction

{
  "id": "extract-node",
  "name": "extract",
  "operator": "extract_operator",
  "config": {
    "text_extraction": {
      "provider": "docling_library",
      "provider_config": {
        "vlm_pipeline": {
          "preset": "granite_docling",
          "engine": "api_ollama",
          "engine_options": {
            "api_base": "http://localhost:11434",
            "model_id": "ibm/granite-docling:258m"
          }
        }
      }
    },
    "entity_extraction": {
      "provider": "none"
    },
    "max_workers": 1
  }
}

Example: GPU-Accelerated Text Extraction

{
  "id": "extract-node",
  "name": "extract",
  "operator": "extract_operator",
  "config": {
    "text_extraction": {
      "provider": "docling_library",
      "doc_column": "content",
      "provider_config": {
        "standard_pipeline": {
          "accelerator": {
            "device": "mps",
            "num_threads": 6
          }
        }
      }
    },
    "entity_extraction": {
      "provider": "none"
    },
    "max_workers": 1,
    "use_processes": false
  }
}

Example: Docling Serve with OCR

{
  "id": "extract-node",
  "name": "extract",
  "operator": "extract_operator",
  "config": {
    "text_extraction": {
      "provider": "docling_serve",
      "provider_config": {
        "base_url": "http://localhost:5001",
        "ocr": {
          "enabled": true,
          "engine": "easyocr"
        },
        "pdf_backend": "dlparse_v4"
      }
    },
    "entity_extraction": {
      "provider": "none"
    }
  }
}

Example: Text + LiteLLM Entity Extraction (with Ollama)

{
  "id": "extract-node",
  "name": "extract",
  "operator": "extract_operator",
  "config": {
    "text_extraction": {
      "provider": "docling_library"
    },
    "entity_extraction": {
      "provider": "litellm",
      "provider_config": {
        "model_id": "openai/granite4:latest",
        "temperature": 0.0,
        "max_tokens": 4096,
        "api_base": "http://localhost:11434/v1",
        "api_key": "<ollama_key>"
      },
      "custom_schema": {
        "invoice_number": "string",
        "total_amount": "float"
      }
    }
  }
}

Example: Text + LiteLLM Entity Extraction (with OpenAI)

{
  "id": "extract-node",
  "name": "extract",
  "operator": "extract_operator",
  "config": {
    "text_extraction": {
      "provider": "docling_library"
    },
    "entity_extraction": {
      "provider": "litellm",
      "provider_config": {
        "model_id": "openai/gpt-3.5-turbo",
        "temperature": 0.0,
        "max_tokens": 2000,
        "api_key": "${OPENAI_API_KEY}",
        "api_base": "https://api.openai.com/v1"
      }
    }
  }
}

Example: Multi-Format Output

{
  "id": "extract-node",
  "name": "extract",
  "operator": "extract_operator",
  "config": {
    "text_extraction": {
      "provider": "docling_library",
      "provider_config": {
        "additional_formats": ["html", "json", "text", "doclang"]
      }
    },
    "entity_extraction": {
      "provider": "none"
    }
  }
}

Example: Text + WatsonX Entity Extraction

{
  "id": "extract-node",
  "name": "extract",
  "operator": "extract_operator",
  "config": {
    "text_extraction": {
      "provider": "docling_library"
    },
    "entity_extraction": {
      "provider": "watsonx",
      "provider_config": {
        "model_id": "ibm/granite-13b-chat-v2",
        "temperature": 0.0,
        "max_tokens": 2000,
        "api_key": "${WATSONX_API_KEY}",
        "container_id": "${WATSONX_CONTAINER_ID}",
        "api_base": "https://us-south.ml.cloud.ibm.com",
        "container_kind": "project"
      },
      "custom_schema": {
        "invoice_number": "string",
        "total_amount": "float"
      }
    }
  }
}

Example: Docling Template-Based Entity Extraction

{
  "id": "extract-node",
  "name": "extract",
  "operator": "extract_operator",
  "config": {
    "text_extraction": {
      "provider": "docling_library"
    },
    "entity_extraction": {
      "provider": "docling",
      "custom_schema": {
        "type": "object",
        "properties": {
          "invoice_number": { "type": "string" },
          "total_amount": { "type": "number" }
        }
      }
    }
  }
}

Example: Docling Entity Extraction with Custom Inline Model

{
  "id": "extract-node",
  "name": "extract",
  "operator": "extract_operator",
  "config": {
    "text_extraction": {
      "provider": "docling_library"
    },
    "entity_extraction": {
      "provider": "docling",
      "provider_config": {
        "vlm_pipeline": {
          "model_type": "inline",
          "inline_model": {
            "repo_id": "numind/NuExtract-2.0-2B",
            "inference_framework": "transformers",
            "scale": 2.0,
            "temperature": 0.0,
            "max_new_tokens": 4096,
            "load_in_8bit": true,
            "torch_dtype": "bfloat16"
          }
        }
      },
      "custom_schema": {
        "type": "object",
        "properties": {
          "invoice_number": { "type": "string" },
          "total_amount": { "type": "number" }
        }
      }
    }
  }
}

Custom VLM Configuration for Docling Entity Extraction

The vlm_pipeline parameter enables custom VLM model configuration for the Docling entity extraction adapter. Only inline models (HuggingFace) are supported as DocumentExtractor does not support remote API endpoints.

Note: For API-based entity extraction, use entity_extraction.provider: "litellm" or "watsonx" instead of Docling.

Configuration Structure:

Usage Notes:

• Inline Models Only: Only HuggingFace models loaded directly into memory are supported. DocumentExtractor does not support remote API endpoints.
• API-Based Extraction: For API-based entity extraction (Ollama via LiteLLM, OpenAI, etc.), use entity_extraction.provider: "litellm" or "watsonx" instead.
• Default Behavior: If vlm_pipeline is not provided, Docling uses its default model configuration.
• Performance: Inline models require sufficient GPU memory and are suitable for local deployment with GPU resources.
• Compatibility: Ensure the chosen model supports the inference framework and hardware configuration.

Architecture

The ExtractOperator uses hexagonal architecture (ports and adapters pattern) with clear separation of concerns:

Layers:

• Domain Layer: EntityExtractionService handles business logic (prompt building, schema validation, response parsing)
• Port Layer: TextExtractionPort and EntityExtractionPort define extraction interfaces
• Adapter Layer: Concrete implementations for different extraction strategies
• Text: DoclingAdapter (docling_library), DoclingServeAdapter (docling_serve)
• Entity: LLMEntityAdapter (unified for litellm/watsonx), DoclingEntityAdapter (docling)
• Factory Layer: TextExtractionAdapterFactory and EntityExtractionAdapterFactory create adapters based on provider

Key Benefits:

• Easy addition of new extraction strategies by implementing ports
• Clear separation between business logic, interfaces, and implementations
• Independent text and entity extraction provider selection
• Unified LLM support: Both litellm and watsonx providers use the same LLMEntityAdapter
• Parallel processing with auto-optimized worker counts

Integration Requirements

• Ollama (for litellm entity provider with Ollama): Server at http://localhost:11434, model pulled (e.g., ollama pull llama3.2). Access via litellm provider with openai/ model prefix
• Docling Serve (for docling_serve text provider): Service at configured URL (default http://localhost:5001)
• LiteLLM (for litellm entity provider): API keys for chosen provider (OpenAI, Anthropic, etc.)
• WatsonX (for watsonx entity provider): Environment variables WATSONX_API_KEY, WATSONX_CONTAINER_ID, optional WATSONX_API_BASE_URL, WATSONX_CONTAINER_KIND
• ffmpeg (for audio/video processing): Required for M4A, AAC, OGG, FLAC audio formats and all video formats (MP4, AVI, MOV). Not required for WAV/MP3. Install: brew install ffmpeg (macOS) or sudo apt install ffmpeg (Linux)

Usage Notes

• Dual-provider operation: text and entity extraction in single operator
• Text providers: docling_library (local, optional VLM/ASR) or docling_serve (remote API with OCR)
• Entity providers: litellm (100+ providers including Ollama via openai/ prefix), watsonx (IBM WatsonX.ai), docling (template-based), none (default)
• Entity Extraction Validation: When using any entity extraction provider (not none), you must provide either:
• A custom_schema in the operator configuration, OR
• A document_type column from an upstream classification operator (e.g., DocumentClassifierOperator)
• If neither is provided, a ConfigurationError will be thrown with message: "Entity extraction requires either a custom_schema in operator config OR a document_type column from upstream classification operator"
• VLM Pipeline: Configure via nested text_extraction.provider_config.vlm_pipeline object with preset, engine, and engine_options for enhanced extraction of complex documents
• ASR Pipeline: Configure via nested text_extraction.provider_config.asr_pipeline object with model_id for audio/video transcription
• Docling Serve provider supports OCR for scanned documents and multi-language processing
• Text File Handling: .txt files are automatically processed locally using UTF-8/latin-1 decoding, bypassing Docling Serve even when docling_serve provider is configured
• Extension Detection: Files without extensions are automatically detected using magic byte analysis (supports PDF, DOCX, XLSX, PPTX, images, HTML, and text formats)
• Audio/Video Support: Processes audio (WAV, MP3, M4A, AAC, OGG, FLAC) and video (MP4, AVI, MOV) files using ASR. Requires ffmpeg for M4A, AAC, OGG, FLAC, and all video formats
• Extension Validation: Files with unsupported extensions are automatically skipped and logged. Supported extensions vary by provider:
• docling_library: PDF, DOCX, PPTX, XLSX, images, HTML, Markdown, AsciiDoc, TXT, and audio/video (with ASR)
• docling_serve: Same as docling_library except NO audio/video support
• docling entity extraction: PDF, DOCX, PPTX, HTML, images (excludes XLSX, TXT, MD, WEBP)
• See ExtractOperator for complete documentation including detailed extension support

Quality Operators

DocumentClassifierOperator

Purpose: Classifies documents into predefined types using LLM-based classification with confidence scoring and reasoning. Uses simplified service-based architecture with shared LLM infrastructure supporting multiple providers (LiteLLM, Watsonx).

Category: Quality

Class: core.operators.quality.classification.document_classifier.DocumentClassifierOperator

Provider-Specific Configuration

LiteLLM (100+ providers):

{
  "provider": "litellm",
  "provider_config": {
    "model_id": "openai/gpt-4o-mini",
    "api_key": "${OPENAI_API_KEY}",
    "timeout": 120
  }
}

Supported LiteLLM providers:

• OpenAI: openai/gpt-4o-mini, openai/gpt-4, openai/gpt-3.5-turbo
• Anthropic: anthropic/claude-3-opus, anthropic/claude-3-sonnet, anthropic/claude-3-haiku
• Azure OpenAI: azure/gpt-4
• AWS Bedrock: bedrock/anthropic.claude-3-sonnet
• Google Vertex AI: vertex_ai/gemini-pro
• HuggingFace: huggingface/meta-llama/Llama-3.3-70B-Instruct, huggingface/mistralai/Mistral-7B-Instruct-v0.2
• Ollama via OpenAI-compatible endpoint: openai/llama3.2:latest, openai/granite3.1-dense:8b with api_base: "http://localhost:11434/v1"

Watsonx:

{
  "provider": "watsonx",
  "provider_config": {
    "model_id": "ibm/granite-13b-chat-v2",
    "api_base": "https://us-south.ml.cloud.ibm.com",
    "api_key": "${WATSONX_API_KEY}",
    "container_kind": "project",
    "container_id": "${WATSONX_CONTAINER_ID}",
    "timeout": 120
  }
}

Input Schema

• PyArrow Table with document content (text column or binary content for extraction)
• Optional content column (if not present, will be fetched from binary content)
• File Extension Validation: Only documents with supported file extensions are processed: .pdf, .docx, .pptx, .doc, .ppt
• Unsupported file types are skipped (not classified) but remain in the output table with None classification values

Output Schema

Adds the following columns:

• document_type (string): Classified document type
• document_type_confidence (float): Confidence score 1-10 (if include_confidence=true)
• document_type_reasoning (string): Classification explanation (if include_reasoning=true)
• content (string): Document content (if fetched and not already present)

Metadata

The operator tracks document processing statistics in metadata:

• processed_docs: Number of successfully classified documents
• failed_docs: List of failed document paths with reasons (errors during processing)
• failed_docs_count: Total number of failed documents
• skipped_docs: List of skipped document paths with reasons (includes unsupported file extensions)
• skipped_docs_count: Total number of skipped documents

Document Types Configuration

Simple list format:

{
  "document_types": ["invoice", "receipt", "contract", "report", "letter"]
}

Detailed dictionary format (recommended):

{
  "document_types": {
    "invoice": "Business invoice with line items, totals, and payment terms",
    "receipt": "Payment receipt or transaction confirmation",
    "contract": "Legal contract or agreement document",
    "report": "Business or technical report with analysis and findings",
    "other": "Other document types not fitting above categories"
  }
}

Exceptions

• DocpipeException: Adapter initialization failures, invalid provider configuration
• ValueError: Invalid response format from LLM
• json.JSONDecodeError: Failed to parse LLM response

Example - LiteLLM with OpenAI

{
  "id": "classify-node",
  "name": "classify",
  "operator": "classification_operator",
  "config": {
    "provider": "litellm",
    "provider_config": {
      "model_id": "openai/gpt-4o-mini",
      "api_key": "${OPENAI_API_KEY}"
    },
    "document_types": {
      "invoice": "Business invoice with line items and totals",
      "receipt": "Payment receipt or confirmation",
      "contract": "Legal contract or agreement",
      "report": "Business or technical report"
    },
    "confidence_threshold": 8.0,
    "include_confidence": true,
    "include_reasoning": true,
    "max_content_length": 4000
  }
}

Example - LiteLLM with Ollama OpenAI-Compatible Endpoint

{
  "id": "classify-node",
  "name": "classify",
  "operator": "document_classifier",
  "config": {
    "provider": "litellm",
    "provider_config": {
      "model_id": "openai/llama3.2:latest",
      "api_key": "${api-key}",
      "api_base": "http://localhost:11434/v1"
    },
    "document_types": {
      "invoice": "Business invoice with line items, totals, and payment terms",
      "receipt": "Payment receipt or transaction confirmation",
      "contract": "Legal contract or agreement document",
      "other": "Other document types"
    },
    "confidence_threshold": 7.0,
    "include_confidence": true,
    "include_reasoning": true
  }
}

Architecture

Uses simplified service-based architecture:

• Operator Layer: DocumentClassifierOperator handles PyArrow table processing and orchestration
• Service Layer: ClassificationService contains business logic for document classification
• Domain Layer: Pure domain models (ClassificationRequest, ClassificationResponse) and prompt building
• Infrastructure Layer: Leverages shared LLMAdapterFactory for multi-provider LLM support (LiteLLM, Watsonx)

This simplified design removes the port/adapter overhead while maintaining clean separation of concerns and provider flexibility through the shared LLM infrastructure.

Related Documentation

• Classification Operator Guide
• Extract Operator

LanguageDetect

Purpose: Detect document language and confidence scores using a pluggable adapter.

Category: Quality

Class: core.operators.quality.language_detection.lang_id.LanguageDetect

Output Schema

• content (or configured via doc_column parameter) - Extracted markdown text (always generated)
• content_html - HTML format (if additional_formats includes "html")
• content_json - JSON structured format (if additional_formats includes "json")
• content_text - Plain text format (if additional_formats includes "text")
• content_doctags - Docling's native DocTags format (if additional_formats includes "doctags")
• content_doclang - DocLang format (if additional_formats includes "doclang")
• entities (or configured output_column) - Extracted entities as JSON string (if entity extraction enabled)
• doc_id_hash - Document hash identifier
• pages_processed - Number of pages in the document. Obtained from Docling extraction metadata when available; otherwise estimated using 3000 characters = 1 page
• Individual entity columns (if expand_extracted_data=true)

Execution Metadata

The operator provides the following metadata after execution:

• page_type_stats (dict): Aggregate estimated pages grouped by source document format (e.g., {"pdf": 120, "docx": 45})
• total_pages_converted (int): Total estimated pages across all successfully processed documents

Exceptions

• FlowExecutionFailedException
• ValueError for invalid configuration
• Provider-specific exceptions (Ollama, LiteLLM, Docling)

Example: Basic Text Extraction

{
  "id": "extract-node",
  "name": "extract",
  "operator": "extract_operator",
  "config": {
    "text_extraction": {
      "provider": "docling_library",
      "doc_column": "content",
      "provider_config": {
      }
    },
    "entity_extraction": {
      "provider": "none"
    }
  }
}

Example: VLM Text Extraction

{
  "id": "extract-node",
  "name": "extract",
  "operator": "extract_operator",
  "config": {
    "text_extraction": {
      "provider": "docling_library",
      "provider_config": {
        "vlm_pipeline": {
          "preset": "granite_docling",
          "engine": "api_ollama",
          "engine_options": {
            "api_base": "http://localhost:11434",
            "model_id": "ibm/granite-docling:258m"
          }
        }
      }
    },
    "entity_extraction": {
      "provider": "none"
    },
    "max_workers": 1
  }
}

Example: Docling Serve with OCR

{
  "id": "extract-node",
  "name": "extract",
  "operator": "extract_operator",
  "config": {
    "text_extraction": {
      "provider": "docling_serve",
      "provider_config": {
        "base_url": "http://localhost:5001",
        "ocr": {
          "enabled": true,
          "engine": "easyocr"
        },
        "pdf_backend": "dlparse_v4"
      }
    },
    "entity_extraction": {
      "provider": "none"
    }
  }
}

Example: Text + LiteLLM Entity Extraction (with Ollama)

{
  "id": "extract-node",
  "name": "extract",
  "operator": "extract_operator",
  "config": {
    "text_extraction": {
      "provider": "docling_library"
    },
    "entity_extraction": {
      "provider": "litellm",
      "provider_config": {
        "model_id": "openai/granite4:latest",
        "temperature": 0.0,
        "max_tokens": 4096,
        "api_base": "http://localhost:11434/v1",
        "api_key": "<ollama_key>"
      },
      "custom_schema": {
        "invoice_number": "string",
        "total_amount": "float"
      }
    }
  }
}

Example: Text + LiteLLM Entity Extraction (with OpenAI)

{
  "id": "extract-node",
  "name": "extract",
  "operator": "extract_operator",
  "config": {
    "text_extraction": {
      "provider": "docling_library"
    },
    "entity_extraction": {
      "provider": "litellm",
      "provider_config": {
        "model_id": "openai/gpt-3.5-turbo",
        "temperature": 0.0,
        "max_tokens": 2000,
        "api_key": "${OPENAI_API_KEY}",
        "api_base": "https://api.openai.com/v1"
      }
    }
  }
}

Example: Multi-Format Output

{
  "id": "extract-node",
  "name": "extract",
  "operator": "extract_operator",
  "config": {
    "text_extraction": {
      "provider": "docling_library",
      "provider_config": {
        "additional_formats": ["html", "json", "text", "doclang"],
      }
    },
    "entity_extraction": {
      "provider": "none"
    }
  }
}

Example: Text + WatsonX Entity Extraction

{
  "id": "extract-node",
  "name": "extract",
  "operator": "extract_operator",
  "config": {
    "text_extraction": {
      "provider": "docling_library"
    },
    "entity_extraction": {
      "provider": "watsonx",
      "provider_config": {
        "model_id": "ibm/granite-13b-chat-v2",
        "temperature": 0.0,
        "max_tokens": 2000,
        "api_key": "${WATSONX_API_KEY}",
        "container_id": "${WATSONX_CONTAINER_ID}",
        "api_base": "https://us-south.ml.cloud.ibm.com",
        "container_kind": "project"
      },
      "custom_schema": {
        "invoice_number": "string",
        "total_amount": "float"
      }
    }
  }
}

Example: Docling Template-Based Entity Extraction

{
  "id": "extract-node",
  "name": "extract",
  "operator": "extract_operator",
  "config": {
    "text_extraction": {
      "provider": "docling_library"
    },
    "entity_extraction": {
      "provider": "docling",
      "custom_schema": {
        "type": "object",
        "properties": {
          "invoice_number": { "type": "string" },
          "total_amount": { "type": "number" }
        }
      }
    }
  }
}

Example: Docling Entity Extraction with Custom Inline Model

{
  "id": "extract-node",
  "name": "extract",
  "operator": "extract_operator",
  "config": {
    "text_extraction": {
      "provider": "docling_library"
    },
    "entity_extraction": {
      "provider": "docling",
      "provider_config": {
        "vlm_pipeline": {
          "model_type": "inline",
          "inline_model": {
            "repo_id": "numind/NuExtract-2.0-2B",
            "inference_framework": "transformers",
            "scale": 2.0,
            "temperature": 0.0,
            "max_new_tokens": 4096,
            "load_in_8bit": true,
            "torch_dtype": "bfloat16"
          }
        }
      },
      "custom_schema": {
        "type": "object",
        "properties": {
          "invoice_number": { "type": "string" },
          "total_amount": { "type": "number" }
        }
      }
    }
  }
}

Custom VLM Configuration for Docling Entity Extraction

The vlm_pipeline parameter enables custom VLM model configuration for the Docling entity extraction adapter. Only inline models (HuggingFace) are supported as DocumentExtractor does not support remote API endpoints.

Note: For API-based entity extraction, use entity_extraction.provider: "litellm" or "watsonx" instead of Docling.

Configuration Structure:

Usage Notes:

• Inline Models Only: Only HuggingFace models loaded directly into memory are supported. DocumentExtractor does not support remote API endpoints.
• API-Based Extraction: For API-based entity extraction (Ollama via LiteLLM, OpenAI, etc.), use entity_extraction.provider: "litellm" or "watsonx" instead.
• Default Behavior: If vlm_pipeline is not provided, Docling uses its default model configuration.
• Performance: Inline models require sufficient GPU memory and are suitable for local deployment with GPU resources.
• Compatibility: Ensure the chosen model supports the inference framework and hardware configuration.

Architecture

The ExtractOperator uses hexagonal architecture (ports and adapters pattern) with clear separation of concerns:

Layers:
- Domain Layer: EntityExtractionService handles business logic (prompt building, schema validation, response parsing)
- Port Layer: TextExtractionPort and EntityExtractionPort define extraction interfaces
- Adapter Layer: Concrete implementations for different extraction strategies
- Text: DoclingAdapter (docling_library), DoclingServeAdapter (docling_serve)
- Entity: LLMEntityAdapter (unified for litellm/watsonx), DoclingEntityAdapter (docling)
- Factory Layer: TextExtractionAdapterFactory and EntityExtractionAdapterFactory create adapters based on provider

Key Benefits:
- Easy addition of new extraction strategies by implementing ports
- Clear separation between business logic, interfaces, and implementations
- Independent text and entity extraction provider selection
- Unified LLM support: Both litellm and watsonx providers use the same LLMEntityAdapter
- Parallel processing with auto-optimized worker counts

Integration Requirements

• Ollama (for litellm entity provider with Ollama): Server at http://localhost:11434, model pulled (e.g., ollama pull llama3.2). Access via litellm provider with openai/ model prefix
• Docling Serve (for docling_serve text provider): Service at configured URL (default http://localhost:5001)
• LiteLLM (for litellm entity provider): API keys for chosen provider (OpenAI, Anthropic, etc.)
• WatsonX (for watsonx entity provider): Environment variables WATSONX_API_KEY, WATSONX_CONTAINER_ID, optional WATSONX_API_BASE_URL, WATSONX_CONTAINER_KIND
• ffmpeg (for audio/video processing): Required for M4A, AAC, OGG, FLAC audio formats and all video formats (MP4, AVI, MOV). Not required for WAV/MP3. Install: brew install ffmpeg (macOS) or sudo apt install ffmpeg (Linux)

Usage Notes

• Dual-provider operation: text and entity extraction in single operator
• Text providers: docling_library (local, optional VLM/ASR) or docling_serve (remote API with OCR)
• Entity providers: litellm (100+ providers including Ollama via openai/ prefix), watsonx (IBM WatsonX.ai), docling (template-based), none (default)
• Entity Extraction Validation: When using any entity extraction provider (not none), you must provide either:
• A custom_schema in the operator configuration, OR
• A document_type column from an upstream classification operator (e.g., DocumentClassifierOperator)
• If neither is provided, a ConfigurationError will be thrown with message: "Entity extraction requires either a custom_schema in operator config OR a document_type column from upstream classification operator"
• VLM Pipeline: Configure via nested text_extraction.provider_config.vlm_pipeline object with preset, engine, and engine_options for enhanced extraction of complex documents
• ASR Pipeline: Configure via nested text_extraction.provider_config.asr_pipeline object with model_id for audio/video transcription
• Docling Serve provider supports OCR for scanned documents and multi-language processing
• Text File Handling: .txt files are automatically processed locally using UTF-8/latin-1 decoding, bypassing Docling Serve even when docling_serve provider is configured
• Extension Detection: Files without extensions are automatically detected using magic byte analysis (supports PDF, DOCX, XLSX, PPTX, images, HTML, and text formats)
• Audio/Video Support: Processes audio (WAV, MP3, M4A, AAC, OGG, FLAC) and video (MP4, AVI, MOV) files using ASR. Requires ffmpeg for M4A, AAC, OGG, FLAC, and all video formats
• Extension Validation: Files with unsupported extensions are automatically skipped and logged. Supported extensions vary by provider:
• docling_library: PDF, DOCX, PPTX, XLSX, images, HTML, Markdown, AsciiDoc, TXT, and audio/video (with ASR)
• docling_serve: Same as docling_library except NO audio/video support
• docling entity extraction: PDF, DOCX, PPTX, HTML, images (excludes XLSX, TXT, MD, WEBP)
• See ExtractOperator for complete documentation including detailed extension support

ReadabilityOperator

Purpose: Compute readability metrics using dpk_readability.

Category: Quality

Class: core.operators.quality.readability.ReadabilityOperator

Output Schema

• selected readability columns

RedactionOperator

Purpose: Mask words or regex matches in document content.

Category: Quality

Class: core.operators.quality.redaction.RedactionOperator

Output Schema

• updated content
• redaction stats column

PIIAndHAPAnnotator

Purpose: Detect Personally Identifiable Information (PII) and Hate, Abuse, and Profanity (HAP) content using LLM-based detection, with optional redaction.

Category: Quality

Short Name: pii_and_hap

Class: core.operators.quality.pii_and_hap.pii_and_hap_annotator.PIIAndHAPAnnotator

provider_config — LiteLLM:

provider_config — WatsonX:

Output Schema:

When display_pii: true, additional *_info columns are appended for each PII type. Never enable in production.

See Also: PII and HAP Documentation

EdedupOperator

Purpose: Remove exact duplicate documents using dpk_ededup.

Category: Quality

Class: core.operators.quality.ededup.EdedupOperator

MLEnrichmentOperator

Purpose: Compute text quality features using dpk_enrichment.

Category: Quality

Class: core.operators.quality.ml_enrichment.MLEnrichmentOperator

SQLFilterOperator

Purpose: Filter rows with SQL-like expressions or structured criteria and optionally drop selected columns.

Category: Quality

Class: core.operators.quality.sql_filter.SQLFilterOperator

Functional Operators

ChunkerOperator

Purpose: Split extracted text into chunks using simple, semantic, or hybrid/docling strategies with optional multi-provider LLM summarization.

Category: Functional

Class: core.operators.functional.chunker.ChunkerOperator

Note: Flat configuration (top-level summarization_provider, summarization_provider_config, etc.) is still supported for backward compatibility but the nested summarization object is recommended.

Summarization Providers:

When summarization object is present with a provider specified, the operator uses the common LLM infrastructure to generate summaries for each chunk:

• LiteLLM (default): Unified API for 100+ providers (OpenAI, Azure, Anthropic, Cohere, AWS Bedrock, GCP Vertex AI, etc.)
• Watsonx: IBM watsonx.ai cloud service (enterprise AI)

Provider-Specific Configuration (summarization.provider_config):

Configuration Structure:

The recommended approach is to use the nested summarization object:

{
  "summarization": {
    "provider": "litellm",
    "provider_config": {
      "model_id": "openai/llama3.2:3b",
      "api_base": "http://localhost:11434/v1",
      "api_key": "<ollama>"
    },
    "summary_sentences": 3,
    "summary_max_words": 50
  }
}

Backward Compatibility:

The flat configuration structure is still supported:

{
  "summarization": {
    "provider": "litellm",
    "provider_config": {
      "model_id": "openai/llama3.2:3b",
      "api_base": "http://localhost:11434/v1",
      "api_key": "<ollama>"
    },
    "summary_sentences": 3
  },
  "summary_max_words": 50
}

Input Schema

• configured doc_column, usually content

Output Schema

• chunk_sequence_number
• start_index
• chunked_content (array of objects with chunk, start_index, and optional summary fields when summarization is configured)

Exceptions

• DocpipeException
• validation messages
• Ollama errors for semantic chunking
• LLM provider errors for summarization (handled gracefully)

Examples

Basic chunking without summarization:

{
  "id": "chunk-node",
  "name": "chunk",
  "operator": "chunker",
  "config": {
    "chunk_type": "hybrid",
    "doc_column": "content",
    "chunk_size": 512,
    "chunk_overlap": 128
  }
}

Chunking with LiteLLM summarization (Ollama) - Nested Structure:

{
  "id": "chunk-with-summary",
  "name": "chunk",
  "operator": "chunker",
  "config": {
    "chunk_type": "simple",
    "doc_column": "content",
    "chunk_size": 1000,
    "chunk_overlap": 200,
    "summarization": {
      "provider": "litellm",
      "provider_config": {
        "model_id": "openai/llama3.2:3b",
        "api_base": "http://localhost:11434/v1",
        "api_key": "<ollama>"
      },
      "summary_sentences": 2,
      "summary_max_words": 20
    }
  }
}

Chunking with Watsonx summarization - Nested Structure:

{
  "id": "chunk-watsonx",
  "name": "chunk",
  "operator": "chunker",
  "config": {
    "chunk_type": "simple",
    "doc_column": "content",
    "chunk_size": 1000,
    "summarization": {
      "provider": "watsonx",
      "provider_config": {
        "model_id": "ibm/granite-13b-chat-v2",
        "api_key": "${WATSONX_API_KEY}",
        "api_base": "${WATSONX_API_BASE}",
        "container_id": "${WATSONX_PROJECT_ID}",
        "container_kind": "project"
      }
    }
  }
}

EntityCurationOperator

Purpose: Transform extracted entities into structured, curated data using document class schemas with 4 core transformation functions for currency, date, number, and weight parsing.

Category: Functional

Class: docpipe.core.operators.functional.entity_curation.entity_curation_operator.EntityCurationOperator

Input Schema

• entities (dict): Extracted entity key-value pairs from ExtractOperator
• document_type (string): Document class identifier (e.g., "invoice", "purchase_order")
• Other columns are preserved

Output Schema

• All input columns preserved
• transformed_entities column: JSON string containing nested structure of curated entities organized by target tables

Transformation Functions

The operator includes 4 core transformations:

Document Class Schemas

Schemas are defined with target_tables specifying field mappings and transformations. Supports 40+ document classes including invoice, purchase_order, receipt, insurance_claim, passport, and more.

Exceptions

• ValidationError - Missing required columns
• Transformation errors are logged but don't stop processing (graceful degradation)

Example

{
  "id": "curate-node",
  "name": "entity_curation",
  "operator": "entity_curation",
  "config": {
    "entities_column": "entities",
    "document_type_column": "document_type"
  }
}

Usage Notes

• Should be placed after ExtractOperator in the pipeline when entity extraction is enabled
• Requires document class schemas for transformation (returns empty dict for unknown document types)
• Output is always in JSON format with nested structure matching schema's target tables
• See Entity Curation README for detailed documentation

EmbeddingsOperator

Purpose: Generate dense embeddings using pluggable providers such as Ollama, HuggingFace, LiteLLM-backed vendors, and IBM watsonx.ai.

Category: Functional

Class: core.operators.functional.embeddings.embeddings_operator.EmbeddingsOperator

Supported Providers:

• LiteLLM: Unified API for 100+ providers (OpenAI, Azure, Cohere, AWS, GCP, Ollama, HuggingFace API)
• Watsonx: IBM watsonx.ai cloud service (enterprise AI)
• HuggingFace: Native local or API-based inference with sentence-transformers models

Provider-Specific Configuration (provider_config):

Note: For Ollama models via LiteLLM, use openai/ prefix (e.g., openai/nomic-embed-text). For HuggingFace API via LiteLLM, use huggingface/ prefix (e.g., huggingface/sentence-transformers/all-MiniLM-L6-v2). For native HuggingFace local inference, use provider: "huggingface" with the model name directly.

Input Schema

• content or chunked_content

Output Schema

• embeddings

Exceptions

• DocpipeException
• provider authentication/network failures

Example 1: LiteLLM with Ollama (local)

{
  "id": "embedding-node",
  "name": "embeddings",
  "operator": "embeddings",
  "config": {
    "provider": "litellm",
    "provider_config": {
      "model_id": "openai/nomic-embed-text",
      "api_base": "http://localhost:11434",
      "batch_size": 32,
      "timeout": 120
    },
    "embeddings_column": "embeddings",
    "text_column": "content"
  }
}

Example 2: LiteLLM with HuggingFace API

{
  "id": "embedding-node",
  "name": "embeddings",
  "operator": "embeddings",
  "config": {
    "provider": "litellm",
    "provider_config": {
      "model_id": "huggingface/sentence-transformers/all-MiniLM-L6-v2",
      "api_key": "${HUGGINGFACE_API_KEY}",
      "batch_size": 16
    },
    "embeddings_column": "embeddings",
    "text_column": "content"
  }
}

Example 2b: Native HuggingFace (Local Inference)

{
  "id": "embedding-node",
  "name": "embeddings",
  "operator": "embeddings",
  "config": {
    "provider": "huggingface",
    "provider_config": {
      "model_id": "sentence-transformers/all-MiniLM-L6-v2",
      "use_local": true,
      "device": "cpu",
      "batch_size": 16
    },
    "embeddings_column": "embeddings",
    "text_column": "content"
  }
}

Example 3: WatsonX embeddings

{
  "id": "embedding-node",
  "name": "embeddings",
  "operator": "embeddings",
  "config": {
    "provider": "watsonx",
    "provider_config": {
      "model_id": "ibm/slate-125m-english-rtrvr",
      "api_key": "${WATSONX_API_KEY}",
      "api_base": "https://us-south.ml.cloud.ibm.com",
      "container_id": "${WATSONX_PROJECT_ID}",
      "container_kind": "project",
      "batch_size": 32
    },
    "embeddings_column": "embeddings",
    "text_column": "content"
  }
}

Example 4: LiteLLM with OpenAI

{
  "id": "embedding-node",
  "name": "embeddings",
  "operator": "embeddings",
  "config": {
    "provider": "litellm",
    "provider_config": {
      "model_id": "openai/text-embedding-3-small",
      "api_key": "${OPENAI_API_KEY}",
      "batch_size": 32
    },
    "embeddings_column": "embeddings",
    "text_column": "content"
  }
}

BranchingOperator

Purpose: Split a table into multiple output tables using SQL-like filter conditions.

Category: Functional

Class: core.operators.functional.branching_operator.BranchingOperator

Input Schema

• Any table with columns referenced by branch criteria

Output Schema

• Multiple output tables, one per branch

Exceptions

• validation errors
• propagated SQL filter errors

MergeOperator

Purpose: Combine multiple PyArrow tables from different branches using row concatenation or column joins.

Category: Functional

Class: core.operators.functional.merge.MergeOperator

Input Schema

• Multiple PyArrow tables from different branches, each identified by link_name
• All tables must contain an id column for row merge duplicate detection and column merge joins

Output Schema

• Row Merge (merge_type=rows): Single table with all rows concatenated vertically
• Preserves all columns from all input tables
• Validates no duplicate IDs across tables
• Column Merge (merge_type=columns): Single table with columns joined horizontally
• inner_join: Only rows with matching IDs across all tables
• full_outer: All rows from all tables, with nulls for missing values
• Non-ID columns from subsequent tables get _<link_name> suffix
• Complex types (lists, structs) are remapped without suffix

Exceptions

• FlowExecutionFailedException: Fewer than 2 input links provided
• DocpipeException: Duplicate IDs detected in row merge
• Validation errors for missing or invalid configuration

Usage Notes

• Typically used after BranchingOperator to recombine split data flows
• The order of input_links determines column suffix application in column merge
• Row merge requires unique IDs across all input tables
• Column merge performs ID-based joins, similar to SQL JOIN operations

Example Flow Configuration

{
  "id": "merge-node",
  "name": "merge_branches",
  "operator": "merge",
  "config": {
    "merge_type": "columns",
    "column_option": "inner_join",
    "input_links": [{ "link_name": "branch1" }, { "link_name": "branch2" }]
  },
  "input_edges": [
    { "node_id_ref": "branch1-node", "link_name": "branch1" },
    { "node_id_ref": "branch2-node", "link_name": "branch2" }
  ]
}

NOOPOperator

Purpose: Pass input rows through unchanged, optionally sleeping for a configured interval.

Category: Functional

Class: core.operators.functional.noop.NOOPOperator

Input Schema

• Any pyarrow.Table

Output Schema

• Same as input table

DocIdHashOperator

Purpose: Generate a stable hash identifier from content.

Category: Functional

Class: core.operators.functional.doc_id_hash.DocIdHashOperator

Availability: Internal operator.

Output Schema

• doc_id_hash

VectorDB Operators

VectorDBOperator

Purpose: Index documents and embeddings into a vector database through a provider adapter interface.

Category: VectorDB

Class: core.operators.vectordb.vectordb_operator.VectorDBOperator

Multi-Model Embeddings Support:

The VectorDBOperator supports multiple embedding columns with different dimensions:

• OpenSearch: Full multi-model support
• Automatically detects all vector columns from available_features (columns with type: "vector")
• Auto-detects dimension for each vector column from actual embedding data
• Creates index fields for all vector columns with their respective dimensions
• Example: Store both 384-dim and 768-dim embeddings in same document (e.g., embeddings and embeddings_alt)
• Use with BranchingOperator + multiple EmbeddingsOperators + MergeOperator for multi-model pipelines
• See sample_flows/advanced/branching_dual_embeddings_with_ingest_report.json for complete example
• Milvus: Single-model support (uses embeddings column only)
• Dimension auto-detected from embeddings column data
• Additional vector columns in the table are stored as metadata but not indexed as vectors
• Multi-model support planned for future update
• For multi-model scenarios with Milvus, use separate collections per embedding model

Supported Vector Databases:

1. OpenSearch (provider: "opensearch")
2. Multiple KNN engines: NMSLIB, Faiss, Lucene
3. Configurable space types: cosinesimil, l2, innerproduct

4. HTTP/HTTPS connections with optional authentication

5. Milvus (provider: "milvus")

6. Index Types:
   • Dense: HNSW, IVF_FLAT, IVF_SQ8, IVF_PQ, FLAT, DISKANN, AUTOINDEX
   • Sparse: SPARSE_INVERTED_INDEX, SPARSE_WAND
7. Metric Types:
   • Dense: L2 (Euclidean), IP (Inner Product), COSINE
   • Sparse: BM25 (required for sparse mode)
8. Vector Modes:
   • Dense vectors only (default)
   • Sparse + Dense vectors (dual storage with BM25 function)
9. Deployment Options:
   • Standalone Milvus (local or remote)
   • Watsonx.data Milvus (IBM Cloud, SSL-enabled)
10. Features:
    • Auto-detection of vector dimensions from embeddings
    • BM25 sparse vector generation from content
    • Configurable index parameters (M, efConstruction for HNSW)
    • Batch processing with configurable batch sizes
    • SSL/TLS support for secure connections

OpenSearch Configuration Example:

{
  "operator": "vectordb",
  "config": {
    "provider": "opensearch",
    "create_index": true,
    "provider_config": {
      "index_name": "my_documents",
      "host": "localhost",
      "port": 9200,
      "engine": "nmslib",
      "space_type": "cosinesimil",
      "username": "<value>",
      "password": "<value>"
    }
  }
}

Milvus Standalone Configuration Example:

{
  "operator": "vectordb",
  "config": {
    "provider": "milvus",
    "create_index": true,
    "add_sparse_vector": false,
    "provider_config": {
      "collection_name": "my_collection",
      "auth_type": "standalone",
      "host": "localhost",
      "port": 19530,
      "uri": null,
      "token": null,
      "username": "root",
      "password": "<your-milvus-password>",
      "database": "default",
      "secure": false,
      "index_type": "HNSW",
      "metric_type": "L2",
      "index_parameters": {
        "M": 16,
        "efConstruction": 256
      },
      "batch_size": 100,
      "primary_key_field": "pk"
    },
    "available_features": {
      "doc_id_hash": {
        "name": "Document ID",
        "available_for_vector_db": true,
        "mandatory_for_vector_db": true,
        "type": "string",
        "is_primary": true
      },
      "embeddings": {
        "name": "Embeddings",
        "available_for_vector_db": true,
        "mandatory_for_vector_db": true,
        "type": "vector"
      }
    },
    "feature_mappings": {
      "doc_id_hash": "pk",
      "embeddings": "vector_embeddings"
    }
  }
}

Milvus Watsonx.data with gRPC Configuration Example:

{
  "operator": "vectordb",
  "config": {
    "provider": "milvus",
    "add_sparse_vector": false,
    "create_index": true,
    "provider_config": {
      "collection_name": "wxdata_collection",
      "auth_type": "grpc",
      "host": "YOUR_WXDATA_HOST.lakehouse.ibmappdomain.cloud",
      "port": 32671,
      "uri": null,
      "token": null,
      "username": "ibmlhapikey_YOUR_USERNAME",
      "password": "<your-api-key>",
      "database": "default",
      "secure": true,
      "index_type": "HNSW",
      "metric_type": "L2",
      "index_parameters": {
        "M": 16,
        "efConstruction": 256
      },
      "batch_size": 100,
      "primary_key_field": "pk"
    }
  }
}

Milvus Watsonx.data with IAM Token Configuration Example:

{
  "operator": "vectordb",
  "config": {
    "provider": "milvus",
    "create_index": true,
    "add_sparse_vector": false,
    "provider_config": {
      "collection_name": "wxdata_token_collection",
      "auth_type": "token",
      "host": "YOUR_WXDATA_HOST.lakehouse.ibmappdomain.cloud",
      "port": 32671,
      "uri": null,
      "token": "YOUR_IAM_TOKEN",
      "username": "YOUR_USERNAME",
      "password": null,
      "database": "default",
      "secure": true,
      "index_type": "HNSW",
      "metric_type": "L2",
      "index_parameters": {
        "M": 16,
        "efConstruction": 256
      },
      "batch_size": 100,
      "primary_key_field": "pk"
    }
  }
}

Provider Config Parameters:

OpenSearch:

• host: OpenSearch server hostname
• port: OpenSearch server port (default: 9200)
• engine: KNN engine (nmslib, faiss, lucene)
• space_type: Distance metric (cosinesimil, l2, innerproduct)
• username: Optional authentication username
• password: Optional authentication password

Milvus:

• auth_type: Required - Authentication type (standalone, grpc, uri, or token)
• host: Milvus server hostname (required for standalone, grpc, token)
• port: Milvus server port (default: 19530, required for standalone, grpc, token)
• uri: Full connection URI (required for uri auth_type)
• token: IAM token (required for token auth_type)
• username: Authentication username (optional for standalone, required for grpc and token)
• password: Authentication password/API key (optional for standalone, required for grpc - should be API key)
• database: Database name (default: "default")
• secure : Required for IBM watsonx.data MilvusDB for https connection and is set to true in this case.
• index_type: Index algorithm (HNSW, IVF_FLAT, IVF_SQ8, IVF_PQ, FLAT, DISKANN, AUTOINDEX for dense; SPARSE_INVERTED_INDEX, SPARSE_WAND for sparse)
• metric_type: Distance metric (L2, IP, COSINE for dense; BM25 required for sparse mode)
• index_parameters: Index-specific parameters (e.g., M and efConstruction for HNSW)
• batch_size: Batch size for bulk operations (default: 100)
• primary_key_field: Name of the primary key field in Milvus collection (default: "pk")

Milvus Authentication Types:

• standalone: Local Milvus with optional username/password
• grpc: IBM wx.data with gRPC (username must have ibmlhapikey_ prefix, password is API key)
• uri: Pre-constructed URI with embedded API key (format: https://ibmlhapikey_<username>:<api-key>@<host>:<port>)
• token: IAM token-based (constructs URI internally: https://ibmlhtoken_<username>:<token>@<host>:<port>)

Sparse Vector Mode:

When add_sparse_vector: true is set:

• Requires metric_type: "BM25" (validated)
• Creates BM25 function for automatic sparse vector generation from text content
• Feature mappings must include:
• embeddings → vector (dense embeddings)
• sparse_embeddings → sparse_vector (BM25-generated)
• content → text (source text for BM25)
• Index type auto-set to SPARSE_INVERTED_INDEX if not specified
• See sample_flows/vectordb/milvus_integration.json for complete example

Notes:

• Vector dimensions are auto-detected from actual embedding data for each vector column
• OpenSearch supports multiple vector columns with different dimensions in a single index
• Milvus currently supports single vector column (multi-model support planned for future update)
• See docs/integrations/milvus/README.md for detailed Milvus configuration
• See VectorDB Operator README for provider configuration patterns

Provider Config (OpenSearch)

The provider_config object supports the following parameters for OpenSearch:

Schema Templates

Schema templates provide reusable index configurations with placeholder-based dynamic values:

• Built-in Templates:
• schemas/default_schema.v1.json: Basic schema with standard field types

• schemas/template_with_content_analyzer.v1.json: Template with custom content analyzer for text processing

• Placeholders: Templates support the following placeholders that are replaced at runtime:

• __VECTOR_DIMENSION__: Vector embedding dimension
• __ENGINE__: KNN engine name
• __ALGORITHM__: KNN algorithm name
• __SPACE_TYPE__: Similarity metric

• __ENGINE_PARAMETERS__: Engine-specific parameters object

• Fallback Behavior: If template is not found or invalid, the system falls back to dynamic schema generation with a warning

Metadata Normalization

The VectorDBOperator automatically normalizes and aggregates metadata columns:

• Column Aliases: Automatically maps common column name variations:
• path → source
• pages_processed → page_count

• mime_type, content_type → mimetype

• Field Derivation: Automatically derives missing metadata fields:

• extension: Derived from name or source if missing

• mimetype: Derived from extension using standard MIME type mappings

• Predefined Metadata Fields: The following fields are automatically collected into a metadata object:

• name, size, created_time, modified_time, source, mimetype, extension, page_count

Input Schema

• doc_id_column
• embeddings_column

Output Schema

• input table unchanged
• side effect: indexed vector records

Exceptions

• DocpipeException

Example Configuration

Basic usage with default schema:

{
  "operator_type": "docpipe.core.operators.vectordb.vectordb_operator.VectorDBOperator",
  "operator_params": {
    "provider": "opensearch",
    "provider_config": {
      "index_name": "my_documents",
      "host": "localhost",
      "port": 9200,
      "engine": "faiss",
      "algorithm": "hnsw",
      "vector_dimension": 384
    }
  }
}

Using a schema template:

{
  "operator_type": "docpipe.core.operators.vectordb.vectordb_operator.VectorDBOperator",
  "operator_params": {
    "provider": "opensearch",
    "provider_config": {
      "index_name": "document_chunks",
      "schema_template_path": "schemas/template_with_content_analyzer.v1.json",
      "host": "localhost",
      "port": 9200,
      "engine": "faiss",
      "algorithm": "hnsw",
      "vector_dimension": 768
    }
  }
}

Storage Operators

DocumentSetOperator

Purpose: Store PyArrow table data and document-set metadata using a hexagonal architecture with pluggable metadata and data adapters.

Category: Storage

Class: core.operators.document_sets.document_set_operator.DocumentSetOperator

Description

The Document Set operator persists table rows and document-set metadata through separate port interfaces:

• DocumentSetMetadataRepository
• DocumentSetStorage

The operator creates concrete adapters through:

• MetadataRepositoryFactory
• DataStoreFactory

The metadata backend is controlled by global_config.metadata_storage_type (default: duckdb). The data backend is controlled per-operator via data_backend. Current production support is DuckDB for both. The operator returns the input table unchanged, so it can be placed mid-pipeline without breaking downstream processing.

Architecture

• Hexagonal architecture with ports and adapters
• Separate metadata and data storage abstractions
• Factory-based backend creation
• Shared DuckDB storage reuse for current adapters

Input Schema

• id (required): UUID document identifier

Common upstream fields from the sample flow:

• name
• binary content from ingest
• content

Output Schema

• Input table unchanged (pass-through design)
• Side effect: data and metadata persisted through configured adapters

Metadata Output

• document_set_id: UUID of the document set
• document_set_name: Name of the document set
• database_path: Database path used for storage
• stored_documents: Number of rows written in the operation
• total_size_bytes: Total size in bytes of all stored rows
• total_pages: Total pages across all stored documents
• table_name: Backend-specific data location identifier populated by the storage adapter
• metadata_storage_type: Metadata adapter used (from global_config.metadata_storage_type)
• data_storage_type: Data adapter used (from data_backend)
• error: Error message when persistence fails

Features

• Persistent storage of document collections
• Hexagonal architecture for backend extensibility
• Separate metadata and data persistence layers
• Automatic schema handling through data-store adapters
• Automatic metric recomputation through DocumentSetService
• Pass-through design allows chaining with downstream operators

Exceptions

• FlowValidationException: Invalid operator configuration
• FlowExecutionFailedException: Storage execution failed
• DocpipeException: Adapter, validation, or persistence error

Sample Flow Configuration

{
  "type": "document_set",
  "name": "documents",
  "config": {
    "document_set_name": "documents_collection",
    "description": "Persistent storage of extracted document content with metadata tracking",
    "database_path": "./data/document_sets/extracted_docs.duckdb",
    "data_backend": "duckdb",
    "metadata": {
      "pipeline_version": "1.0",
      "extraction_method": "docling",
      "created_by": "docpipe_pipeline",
      "purpose": "demonstration_flow"
    }
  },
  "depends_on": ["extract_content"]
}

Configuration Notes:

• global_config.metadata_storage_type controls the metadata storage backend (default: duckdb)
• database_path in the operator config specifies the DuckDB file path for both metadata and data
• data_backend in the operator config controls the PyArrow table data adapter (default: duckdb)

Flow Pattern

IngestSourceOperator -> ExtractOperator -> DocumentSetOperator

End-to-End Flow Example

{
  "flow_name": "ingest-extract-documentset",
  "global_config": {
    "doc_column": "content",
    "storage": "in-memory",
    "execute_type": "local"
  },
  "flow": [
    {
      "type": "ingest_source",
      "name": "ingest_documents",
      "config": {
        "provider": "filesystem",
        "connection_params": {"paths": ["./documents"]},
        "include_filter": "pdf,txt"
      }
    },
    {
      "type": "extract_operator",
      "name": "extract_content",
      "config": {
        "text_extraction": { "provider": "docling_library" }
      },
      "depends_on": ["ingest_documents"]
    },
    {
      "type": "document_set",
      "name": "store_documents",
      "config": {
        "document_set_name": "integration_test_documents",
        "description": "Integration test for hexagonal architecture",
        "database_path": "data/integration_test.db",
        "data_backend": "duckdb"
      },
      "depends_on": ["extract_content"]
    }
  ]
}

API Integration

Document sets are also exposed through /api/v1/document-sets:

• POST /api/v1/document-sets
• GET /api/v1/document-sets
• GET /api/v1/document-sets/{id}
• PATCH /api/v1/document-sets/{id}
• DELETE /api/v1/document-sets/{id}
• GET /api/v1/document-sets/{id}/preview

Error Codes

• document_set_not_found
• document_set_invalid_data
• document_set_storage_error
• document_set_already_exists
• document_set_data_store_error
• document_set_table_not_found

Testing

source .venv/bin/activate
export PYTHONPATH="$(pwd)/src:${PYTHONPATH}"
uv run pytest tests/integration/api/test_document_sets_api.py -v

Extension

For new backends, implement the DocumentSetStorage and DocumentSetMetadataRepository ports, register the adapters with DataStoreFactory and MetadataRepositoryFactory respectively. Configure data_backend in the operator config to select the data adapter, and global_config.metadata_storage_type to select the metadata adapter. See ARCHITECTURE.md.

StorageOutputOperator

Purpose: Write pipeline documents to a pluggable storage destination, supporting three modes: processed_content, refetch_original, and comprehensive_export.

Category: Storage

Class: core.operators.storage.storage_output_operator.StorageOutputOperator

destination_config fields

output_format fields

output_structure fields

Path template variables: {doc_id}, {name}, {ext}, {year}, {month}, {day}, {relative_dir}

Output Schema

All input columns are passed through unchanged. The following columns are appended:

Sample Flow Configuration

{
  "type": "storage_output",
  "name": "write_output",
  "config": {
    "mode": "processed_content",
    "destination_config": {
      "provider": "filesystem",
      "provider_config": {
        "root_path": "/output/docs",
        "create_dirs": true
      },
      "credentials": {}
    },
    "output_format": { "content_format": "md" },
    "output_structure": { "path_template": "{year}/{month}/{doc_id}.{ext}" }
  },
  "depends_on": ["extract"]
}

Flow Pattern

IngestSourceOperator -> ExtractOperator -> StorageOutputOperator

For full provider reference, operating mode details, and per-provider examples see docs/operators/storage/storage_output_readme.md.

DocpipeFlowManager API

Class: DocpipeFlowManager

Constructor

• DocpipeFlowManager(flow_file=None, flow_def=None, job_id=None, job_run_id=None, flow_id=None, enable_custom_operators=None)

Exactly one of flow_file or flow_def must be provided.

validate()

Defined at validate().

Returns:

{
  "valid": bool,
  "errors": list,
  "warnings": list
}

execute()

Defined at execute().

Returns the result of flow execution from the executor.

get_execution_metadata()

Defined at get_execution_metadata().

Returns job and flow metadata.

get_execution_logs()

Defined at get_execution_logs().

Returns list[str].

list_operators(verbose=False)

Defined at list_operators().

Returns a formatted operator listing via docpipe.utils.operators.display.list_operators().

Display Modes:

• Default (verbose=False): Summary table with Owner, Attributes (count), Features (count) columns
• Verbose (verbose=True): Detailed view with full operator parameters and descriptions

Category Sorting Order:
Operators are sorted by category: Ingest, Extract, Quality, Functional, VectorDB, Storage

Method name note

The current class does not expose execute_flow() or validate_flow() methods. Use:

• execute()
• validate()

CLI API Reference

Entry point: main()

Command forms

docling-pipelines --flow-file ./path/to/flow.json
docling-pipelines --flow-file ./path/to/flow.json --validate
docling-pipelines validate-flow ./path/to/flow.json
docling-pipelines --list-operators              # Summary table
docling-pipelines --list-operators --verbose    # Detailed view

Global arguments

Exit codes

Flow Configuration API

Docling Pipelines uses a simplified authoring format for creating flows. See sample_flows/quickstart/complete_pipeline_ollama.json for a complete example.

Root structure

{
  "flow_name": "invoice processing flow",
  "description": "description",
  "flow": [],
  "global_config": {
    "storage": "in-memory",
    "execute_type": "local"
  }
}

DocpipeFlowManager also accepts root-level flow definitions without a wrapping flow key.

Flow fields

Operator structure

{
  "name": "operator_name",
  "type": "operator_type",
  "depends_on": ["upstream_operator"],
  "config": {}
}

Operator fields

Dependency declaration

Operators declare dependencies using the depends_on field:

{
  "name": "extract",
  "type": "extract_operator",
  "depends_on": ["ingest"],
  "config": {}
}

The system automatically generates the execution DAG from these dependencies.

Validation rules

Validation is performed by FlowValidator and CLI validation helpers.

Practical rules from the reviewed code:

• operator short names must resolve to registered operators
• required operator config must validate
• required input features must be present
• branch and filter references must use valid columns
• flow file must be valid JSON

Exception Reference

All custom exception types reviewed here come from docpipe_exceptions.py.

DocpipeException

Base exception for application-level failures.

Use when:

• a caller needs an HTTP status code
• an error code should propagate through middleware
• a general Docling Pipelines runtime/configuration failure occurs

FlowExecutionFailedException

Raised for flow execution failures.

FlowValidationException

Raised when a flow definition is invalid.

Carries:

• errors
• warnings

PrefectFlowFailed

Raised when Prefect-backed orchestration fails for a task.

ValidationException

Generic validation failure outside full flow validation.

ConfigurationError

Raised for invalid or missing configuration.

DependencyError

Raised when an optional dependency is missing.

ExternalServiceError

Raised when external services fail.

Typical cases:

• API calls
• auth problems
• rate limits
• network failures

FlowNotFoundException

Raised when requested flow storage entries do not exist.

FlowAlreadyExistsException

Raised when creating a duplicate flow.

FlowInvalidDataException

Raised when flow payload or fields are malformed.

FlowStorageException

Raised when storage operations fail, such as file I/O problems.

RepositoryConfigurationException

Raised when repository type or repository config is invalid.

ValidationAlert

Dictionary-like validation payload used in warnings/errors.

ValidationAlertEncoder

JSON encoder for validation alerts.

Utilities API

PyArrow handler utilities

Defined in pyarrow_handler.py

BaseParquetTableHandler

Abstract contract for parquet read/write/delete operations.

Key methods:

• read_table() - Read PyArrow table from file
• save_table() - Save PyArrow table to file
• delete_rows() - Delete rows from table
• delete_file() - Delete file from storage

CpdParquetTableHandler

Concrete local-file implementation.

get_parquet_table_handler()

Defined at get_parquet_table_handler() in docpipe utilities

Returns the default parquet handler implementation.

Schema utilities

Defined in schema_utils.py

align_table_schema(table, all_cols)

Defined at align_table_schema() in schema utilities

Adds missing columns with null values and aligns ordering.

_combine_tables(tables, table_type)

Defined at _combine_tables() in schema utilities

Safely concatenates tables and warns on duplicate IDs.

_total_rows(tables)

Defined at _total_rows() in schema utilities

Computes total row counts across a table, list, dict, or None.

Document class utilities

Defined in document_class_utils.py

DocumentClassUtils.normalize_filename(name)

Defined at normalize_filename() in document class utilities

Normalizes human labels into stable filenames.

DocumentClassUtils.load_document_class(doc_class_path)

Defined at load_document_class() in document class utilities

Loads a document class JSON definition.

DocumentClassUtils.generate_docling_template(doc_class_path, include_nested=True, max_fields=None)

Defined at generate_docling_template() in document class utilities

Builds a Docling extraction template from a document class schema.

Raises

• FileNotFoundError
• json.JSONDecodeError

Operator display utility

Defined in display.py

list_operators(verbose=False)

Defined at list_operators()

Generates the same operator catalog used by the CLI and DocpipeFlowManager.list_operators().

Parameters:

• verbose (bool): If True, shows detailed operator information with all parameters. If False (default), shows summary table with Owner, Attributes, Features columns.

Output Format:

• Summary mode: Table with columns: Owner (operator name), Attributes (parameter count), Features (capability count)
• Verbose mode: Detailed listing with full parameter descriptions, types, and default values
• Category order: Ingest, Extract, Quality, Functional, VectorDB, Storage