You don’t need to fork docling-pipelines to add your own logic

The extension model is cleaner than I expected. Here’s the whole thing in one worked example — commands, code, and real output.

Aravindnk

7 min read

·

Aug 31, 2026

--

Listen

Share

A few weeks ago I was building a document ingestion pipeline for financial filings. The standard docling-pipelines quality operators — language detection, PII filtering, deduplication — were all fine, but I needed one more step: filter out anything too dense to be useful for downstream RAG. A simple Flesch readability gate.

My first instinct was to fork the repo and add it. I spent about twenty minutes reading source code before I realised that was completely unnecessary. docling-pipelines has a proper extension mechanism. Your operator loads the same way as the 21 built-in ones. The flow validator, the orchestrator, the CLI’s --list-operators flag — everything treats it identically.

This post walks through how that works, using the readability gate as the example. Every command below has been run against main and the output is real.

How the pipeline sees your operator

Every node in a docling-pipelines flow has a "type" field. At runtime the engine resolves that string to an operator class, instantiates it with the node's "config" block, and calls transform(). That's the whole contract. Your operator just needs to be registered before the flow runs so the resolver can find it.

The data moving between operators is a PyArrow table. transform() receives one, returns a list of tables plus a metadata dict. Returning a list makes branching natural — you can split one table into two and have downstream nodes pick up each path independently.

The framework doesn’t know or care whether your operator lives in the core package or in a file on your laptop. The resolution is purely by short_name.

Writing the operator

Your class inherits from AbstractOperator and needs three things declared at the class level: a short_name (what you'll put in the flow JSON), a category, and an owner. The owner should always be DocpipeConstants.OWNER_CUSTOM — it's what distinguishes your operators from built-ins in the registry output.

Beyond that, three methods. transform() for your logic. get_metadata() to declare your config schema. get_required_features() to tell the validator which upstream columns you depend on.

Here’s the readability gate in full:

from typing import Any
import pyarrow as pa
import pyarrow.compute as pc

from docpipe.core.operators.abstract_operator import AbstractOperator, OperatorCategory
from docpipe.core.constants.constants import DocpipeConstants, AttributeDataTypes
from docpipe.core.constants.operator_constants import OperatorConstants
from docpipe.utils.infrastructure.logging import get_logger

logger = get_logger()


class ReadabilityGateOperator(AbstractOperator):
    """Filters documents whose readability score falls below a threshold."""

    short_name: str = "readability_gate"
    category: OperatorCategory = OperatorCategory.Quality
    owner: str | None = DocpipeConstants.OWNER_CUSTOM

    def __init__(self, config: dict[str, Any]) -> None:
        super().__init__(config)

    def transform(
        self, table: pa.Table, file_name: str | None = None
    ) -> tuple[list[pa.Table], dict[str, Any]]:
        # Always use getattr() with a default — never self.threshold directly
        threshold = getattr(self, "threshold", 50.0)
        score_col = getattr(self, "score_column", "readability_score")
        metadata = self.create_base_metadata(total_docs_count=table.num_rows)

        if score_col not in table.column_names:
            logger.warning("Score column '%s' not found — passing table through", score_col)
            return [table], metadata

        mask = pc.greater_equal(table[score_col], threshold)
        passed = table.filter(mask)
        metadata["docs_filtered_out"] = table.num_rows - passed.num_rows

        logger.info(
            "Readability gate: kept %s, filtered %s docs",
            passed.num_rows,
            metadata["docs_filtered_out"],
        )
        return [passed], metadata

    @staticmethod
    def get_metadata() -> dict[str, Any]:
        return {
            OperatorConstants.Misc.CATEGORY: ReadabilityGateOperator.category.value,
            OperatorConstants.Misc.IS_OPERATOR_AVAILABLE: True,
            OperatorConstants.Misc.LABEL: "Readability Gate",
            OperatorConstants.Config.FEATURES: {},
            OperatorConstants.Config.ATTRIBUTES: {
                "threshold": {
                    OperatorConstants.Misc.NAME: "Readability Threshold",
                    OperatorConstants.Config.DESCRIPTION: "Minimum Flesch score to pass (0-100)",
                    OperatorConstants.Config.REQUIRED: False,
                    OperatorConstants.Config.DEFAULT: 50.0,
                    OperatorConstants.Filtering.MIN_VALUE: 0.0,
                    OperatorConstants.Filtering.MAX_VALUE: 100.0,
                    OperatorConstants.Misc.TYPE: AttributeDataTypes.FLOAT,
                },
                "score_column": {
                    OperatorConstants.Misc.NAME: "Score Column",
                    OperatorConstants.Config.DESCRIPTION: "Column containing the readability score",
                    OperatorConstants.Config.REQUIRED: False,
                    OperatorConstants.Config.DEFAULT: "readability_score",
                    OperatorConstants.Misc.TYPE: AttributeDataTypes.STRING,
                },
            },
        }

    @staticmethod
    def get_required_features() -> list[str]:
        return ["content"]

Two things changed from what you might write instinctively. First, the logger: the project has its own get_logger() in docpipe.utils.infrastructure.logging that wires into the structured logging pipeline. Using logging.getLogger(__name__) instead won't break anything, but your operator's logs won't appear under the docpipe namespace and won't carry job/run context. Second, the __init__ signature takes config positionally, not as a keyword argument — that's what AbstractOperator expects and what the framework will call.

Two things that will silently fail

get_metadata() must be a @staticmethod. It's called before the class is instantiated — during operator discovery and flow validation — so self doesn't exist yet. If you write it as an instance method it'll silently not work in those contexts. Add the decorator. Every time.

Never access config attributes as self.threshold. Config values from your flow JSON are injected onto the instance at runtime, which means they might not be there if the key wasn't provided. The pattern is always getattr(self, "threshold", 50.0). This is the thing I see trip up every engineer who's new to the framework — it doesn't raise during class definition, only when transform() runs on a partially-configured node.

Registering it

Point the DOCPIPE_CUSTOM_OPERATORS environment variable at the directory containing your file. docling-pipelines scans it recursively for .py files at startup.

export DOCPIPE_CUSTOM_OPERATORS="$HOME/my_operators"
docling-pipelines --list-operators

You should see the operator loaded and listed in the summary:

Run with --verbose to confirm your config schema was picked up correctly:

docling-pipelines --list-operators --verbose

If your operator doesn’t appear in this list, check for a Python syntax error in your file first — the loader skips files it can’t import without surfacing a loud error by default. Run DS_LOG_LEVEL=DEBUG docling-pipelines --list-operators to see the full import trace.

Plugging it into a flow

Reference it by short_name in the "type" field, exactly like any built-in. Config params declared in ATTRIBUTES go directly in "config":

{
  "flow_name": "financial-filings-pipeline",
  "global_config": { "doc_column": "content" },
  "flow": [
    {
      "name": "ingest_1",  "type": "ingest_source",
      "config": { "provider": "filesystem", "connection_params": { "paths": ["./filings"] } }
    },
    {
      "name": "extract_1",  "type": "extract_operator",
      "config": {},  "depends_on": ["ingest_1"]
    },
    {
      "name": "readability_1",  "type": "readability_gate",
      "config": { "threshold": 60.0, "score_column": "flesch_score" },
      "depends_on": ["extract_1"]
    },
    {
      "name": "vectordb_1",  "type": "vectordb",
      "config": {},  "depends_on": ["readability_1"]
    }
  ]
}

Validate before running to catch any issues without touching real data:

export DOCPIPE_CUSTOM_OPERATORS="$HOME/my_operators"
docling-pipelines --flow-file quality-pipeline.json --validate

The readability_1 task completed cleanly — which is the point. The three errors are from the other operators in the example flow needing their own connection config. Your operator has no opinion about file paths or database connections, so it passes immediately.

Distributing it to a team

The env var approach is fine for local development. If you need to share operators across a team or deploy them reproducibly, the better path is a pip-installable package with a single entry point declaration.

Create a standard Python package. The only docling-pipelines specific thing is the entry point group name, which must be exactly "docpipe.operators" — not "docpipe.operator", not "docpipe_operators". Get one character wrong and it's silently ignored. Also make sure the entry point value starts with your package name — the adapter filters by module path prefix at discovery time.

[project]
name = "my-docpipe-operators"
version = "1.0.0"
requires-python = ">=3.12"
dependencies = ["docling-pipelines>=0.1.0"]

# Group name must be EXACTLY "docpipe.operators"
# Entry point value must start with your package name
[project.entry-points."docpipe.operators"]
readability_gate = "my_docpipe_operators.readability_gate:ReadabilityGateOperator"

# Install in editable mode during development
pip install -e .

# Operators are auto-discovered — no env var needed
docling-pipelines --list-operators | grep readability_gate

That’s the pattern we settled on for the financial filings project — one internal package, all custom operators in it, versioned and pinned like any other dependency.

Adding early validation

The last thing worth adding is a validate() method. It runs before any document is processed and gets access to the list of columns that will be available from upstream operators — so you can check that your score column actually exists in the pipeline, rather than finding out at row one of the first batch.

def validate(
    self,
    errors: list[str],
    warnings: list[str],
    available_features: list[str],
) -> None:
    # Call super() to run the built-in required-features check
    super().validate(errors, warnings, available_features)

    score_col = getattr(self, "score_column", "readability_score")
    if score_col not in available_features:
        warnings.append(
            f"'{score_col}' not in upstream features — operator will pass through unchanged"
        )

    threshold = getattr(self, "threshold", 50.0)
    if threshold > 90.0:
        warnings.append("threshold > 90 is aggressive — expect most documents to be filtered")

Appending to errors blocks execution. Appending to warnings lets it proceed with a log notice. The super() call runs the parent's required-features check against whatever you declared in get_required_features() — skip it and that validation silently disappears.

The registration options and the metadata schema have more surface area than I’ve covered here — there’s S3 URI support for centralised operator distribution, a full attribute type system with range validation, and a get_static_required_features() companion for operators that use instance state in get_required_features(). The docs/guides/CUSTOM_OPERATORS_GUIDE.md in the repo covers all of it.

But the core is what’s above. A class, three methods, an env var. Most custom operators I’ve written since that financial filings project have stayed close to this shape — the framework doesn’t push you toward complexity, which is the thing I appreciate most about it.