1. Authoring Flows

A flow is a JSON document (or Python dict) that describes a document-processing pipeline as a directed acyclic graph (DAG) of operator nodes. This chapter covers the complete format, global_config options, dependency syntax, and annotated examples.

1.1. Flow JSON Structure

{
  "flow_name": "my-pipeline",
  "description": "Optional human-readable description",
  "global_config": { ... },
  "flow": [
    { "type": "ingest_source", "name": "ingest", "config": { ... } },
    { "type": "extract_operator", "name": "extract", "config": { ... }, "depends_on": ["ingest"] }
  ]
}
Field Required Description

flow_name

Yes

Human-readable identifier; used to derive the deterministic job UUID.

description

No

Prose description; not used by the engine.

global_config

No

Pipeline-wide settings (see Global Configuration).

flow

Yes

Array of operator node objects.

1.2. Operator Node Fields

Field Required Description

type

Yes

The operator’s short_name (e.g. ingest_source, chunker). See the catalog in [_operator_catalog].

name

Yes

Unique node identifier within the flow. Used by other nodes in depends_on.

config

Yes

Operator-specific parameters (can be an empty object {}).

depends_on

No (omit for root)

Array of name values this node depends on. Omit for the root ingest node.

1.3. Dependency Syntax

1.3.1. Simple chain

{ "name": "b", "depends_on": ["a"] }

1.3.2. Multiple inputs (fan-in)

{ "name": "merge", "type": "merge", "depends_on": ["branch_a", "branch_b"] }

1.3.3. Branching (fan-out)

Multiple nodes may declare the same upstream node in depends_on. The engine will fork the data stream and run both branches concurrently.

{ "name": "branch_a", "type": "chunker",   "depends_on": ["extract"] },
{ "name": "branch_b", "type": "doc_quality", "depends_on": ["extract"] }

1.3.4. DAG validation rules

  • Exactly one root node (no depends_on); it must be an ingest operator.

  • All names in depends_on must refer to nodes defined in the same flow.

  • No cycles — the engine detects and rejects circular dependencies.

  • Every node declared in flow must be reachable from the root.

1.4. Global Configuration

The global_config block applies pipeline-wide defaults. All fields are optional.

Field Default Description

doc_column

"content"

Name of the column that holds document text. Passed to all operators.

storage

"in-memory"

Intermediate storage strategy. "in-memory" or "filesystem".

execute_type

"local"

Execution mode. "local" for in-process; "prefect" for distributed.

disable_validation

false

Skip flow DAG validation. Not recommended for production.

skip_custom_op_validation

false

Skip validation checks for custom operators.

force_ingest

false

Re-ingest all documents even if previously processed (disable incremental mode).

retain_deleted_docs

false

Keep documents in the output even if the source file is no longer present.

output_folder

—

Directory for filesystem-backed intermediate results.

data_storage_type

—

PyArrow data store type for intermediate tables.

enable_micro_batching

true

Split large document sets into parallel sub-batches.

micro_batch_size

100

Documents per micro-batch.

max_concurrent_batches

10

Maximum micro-batches in flight concurrently.

prefect

—

Prefect-specific configuration block (see [_production_deployment_and_observability]).

1.4.1. Operator-level overrides

global_config also accepts operator-specific override blocks. Use the operator name or type as the key:

"global_config": {
  "doc_column": "content",
  "my_extract_node": {
    "timeout": 300
  }
}

1.5. Annotated Complete Example: Basic Extraction Pipeline

{
  "flow_name": "basic-extraction",
  "global_config": {
    "doc_column": "content",
    "storage": "in-memory",
    "execute_type": "local",
    "enable_micro_batching": false
  },
  "flow": [
    {
      "type": "ingest_source",
      "name": "ingest",                        (1)
      "config": {
        "provider": "filesystem",
        "connection_params": {
          "paths": ["./docs"]
        },
        "include_filter": "pdf,docx"           (2)
      }
    },
    {
      "type": "extract_operator",
      "name": "extract",
      "depends_on": ["ingest"],               (3)
      "config": {
        "text_extraction": {
          "provider": "docling_library"
        },
        "entity_extraction": {
          "provider": "none"
        }
      }
    },
    {
      "type": "lang_detect",
      "name": "language",
      "depends_on": ["extract"],
      "config": {
        "language_provider": "fasttext"
      }
    },
    {
      "type": "sql_filter",
      "name": "filter_english",
      "depends_on": ["language"],
      "config": {
        "criteria_list": ["lang_name = 'en'"]  (4)
      }
    },
    {
      "type": "chunker",
      "name": "chunk",
      "depends_on": ["filter_english"],
      "config": {
        "chunk_type": "hybrid",
        "chunk_size": 512,
        "chunk_overlap": 50
      }
    }
  ]
}
1 Root node — no depends_on. Must be an ingest operator.
2 Only ingest PDF and DOCX files; ignore others.
3 Chain the extract node to the ingest node.
4 sql_filter uses the lang_name column written by lang_detect upstream.

1.6. Annotated Complete Example: RAG Pipeline with Branching

This example runs quality checks in a separate branch while also producing embeddings.

{
  "flow_name": "rag-with-quality-branch",
  "global_config": {
    "enable_micro_batching": true,
    "micro_batch_size": 50
  },
  "flow": [
    {
      "type": "ingest_source",
      "name": "ingest",
      "config": {
        "provider": "filesystem",
        "connection_params": { "paths": ["./corpus"] },
        "include_filter": "pdf"
      }
    },
    {
      "type": "extract_operator",
      "name": "extract",
      "depends_on": ["ingest"],
      "config": {
        "text_extraction": { "provider": "docling_library" },
        "entity_extraction": { "provider": "none" }
      }
    },
    {
      "type": "ededup",
      "name": "dedup",
      "depends_on": ["extract"],
      "config": {}
    },
    {
      "type": "branching",
      "name": "branch",
      "depends_on": ["dedup"],
      "config": {}
    },
    {
      "type": "doc_quality",
      "name": "quality_metrics",
      "depends_on": ["branch"],
      "config": { "text_lang": "en" }
    },
    {
      "type": "sql_filter",
      "name": "quality_gate",
      "depends_on": ["quality_metrics"],
      "config": {
        "criteria_list": [
          "docq_total_words >= 50",
          "docq_contain_bad_word = false"
        ]
      }
    },
    {
      "type": "chunker",
      "name": "chunk",
      "depends_on": ["quality_gate"],
      "config": { "chunk_type": "hybrid", "chunk_size": 512 }
    },
    {
      "type": "embeddings",
      "name": "embed",
      "depends_on": ["chunk"],
      "config": {
        "provider": "litellm",
        "provider_config": {
          "model_id": "openai/nomic-embed-text",
          "api_base": "http://localhost:11434"
        }
      }
    },
    {
      "type": "vectordb",
      "name": "index",
      "depends_on": ["embed"],
      "config": {
        "provider": "opensearch",
        "vector_dimension": 768,
        "provider_config": {
          "index_name": "rag_corpus",
          "host": "localhost",
          "port": 9200,
          "use_ssl": false,
          "verify_certs": false
        }
      }
    }
  ]
}

1.7. Flow Identification and Job IDs

The flow engine derives a deterministic job UUID from flow_name. A new job_run_id is generated for each execution.

You can supply explicit IDs when using the Python API:

from docpipe.lib.docpipe_flow_manager import DocpipeFlowManager

manager = DocpipeFlowManager(
    flow_file="my_flow.json",
    job_id="my-custom-job-id",
    job_run_id="run-001"
)
manager.execute()

See [_python_api_and_programmatic_usage] for the full Python API.

1.8. Running and Validating Flows

# Execute a flow
docling-pipelines --flow-file my_flow.json

# Validate only (no execution)
docling-pipelines --flow-file my_flow.json --validate

# Override log level
DS_LOG_LEVEL=DEBUG docling-pipelines --flow-file my_flow.json

1.9. Best Practices

  • Keep flow_name short, lowercase, and hyphenated — it becomes part of the job ID.

  • Always place the ingest node first with no depends_on.

  • Set enable_micro_batching: false during development for simpler logs; enable it in production.

  • Use validate mode to catch configuration errors before committing to a full run.

  • Store API keys and passwords in environment variables, never hardcoded in flow JSON.

  • Reference sample_flows/ in the repository for working, annotated pipeline examples.