Document processing pipeline

OCR a folder of scanned documents, extract structured fields, classify by type, and index for retrieval — every step durable, with rollback on failure.

This tutorial builds a document processing pipeline: a folder of scanned PDFs and images arrives, and we want structured records out the other end — invoice numbers, dates, totals, vendor names, plus a classification tag so the indexing system knows where each document belongs.

The pipeline has five stages: OCR, structured extraction, classification, indexing, and the workflow that ties them together. Each stage is a Flux task with retries, timeouts, and rollback behavior shaped by what fails for that particular kind of work.

What you are building

A workflow that takes a directory of document files, OCRs each one, asks an LLM to pull structured fields out of the OCR text, classifies the document type, and writes the result to a SQLite index. The fan-out from OCR to extraction + classification is parallel. The workflow is durable end-to-end, so a crash mid-pipeline resumes from the last completed task.

files/ -> OCR -> { extract fields, classify type } -> index

Tesseract is the simplest open-source OCR that works without a service. Production deployments usually swap in a hosted OCR (Google Document AI, AWS Textract, Azure Document Intelligence) for accuracy on hard documents. The workflow boundaries do not change.

The pipeline is also a useful template for any “ingest plus enrich” shape — receipts to bookkeeping rows, contracts to clause databases, forms to CRM records. The combination of OCR, structured extraction, classification, and rollback-safe indexing recurs in most document-intensive systems, so it is worth understanding the failure modes of each stage in isolation before composing them.

  1. OCR as a durable task

    OCR is slow, occasionally flaky, and produces output that we never want to recompute. That makes it a perfect fit for retries plus output storage.

    import pytesseract
from PIL import Image
from flux.task import task
from flux.output_storage import LocalFileStorage


@task.with_options(
    retry_max_attempts=3,
    retry_delay=2,
    retry_backoff=2,
    timeout=120,
    output_storage=LocalFileStorage(),
)
async def ocr_document(file_path: str) -> str:
    image = Image.open(file_path)
    return pytesseract.image_to_string(image)

    output_storage=LocalFileStorage() tells Flux to write the OCR text to disk and store a reference in the event log instead of inlining the full text. For a single invoice that is fine either way; for a hundred-page contract it keeps the event log small and the worker memory bounded. The downstream tasks receive the resolved value automatically — they do not need to know whether the upstream output was inlined or referenced.

    For S3, swap LocalFileStorage() for S3OutputStorage(bucket="...", prefix="..."). The contract is the same: the storage backend handles persistence and the workflow stays agnostic.

    retry_max_attempts=3 with backoff handles transient PIL or Tesseract failures. If the file is genuinely corrupt the third attempt raises and the task ends in FAILED state — which is what we want, since downstream stages should not run on a broken OCR result.

    timeout=120 caps a single document’s OCR time. Long enough for a multi-page TIFF, short enough that a stuck worker eventually fails over.

  2. Field extraction with an agent

    For structured extraction we want the LLM to return a Pydantic model, not free-form text. The agent() primitive takes a response_format= argument for exactly this.

    from pydantic import BaseModel
from flux.tasks.ai import agent


class LineItem(BaseModel):
    description: str
    quantity: float
    unit_price: float


class InvoiceFields(BaseModel):
    invoice_number: str
    invoice_date: str
    vendor_name: str
    total_amount: float
    line_items: list[LineItem]


async def make_extractor():
    return await agent(
        system_prompt=(
            "You extract structured invoice data from OCR text. "
            "Return all fields. If a field is genuinely missing, "
            "use an empty string for strings and 0 for numbers — "
            "do not fabricate values."
        ),
        model="anthropic/claude-sonnet-4-20250514",
        response_format=InvoiceFields,
        max_tokens=2048,
        stream=False,
    )

    The Anthropic provider enforces response_format at the API by issuing a forced tool call, so the model returns JSON matching the InvoiceFields schema. See Anthropic integration for the full provider notes and tool-calling details.

  3. Classification

    Classification is a separate concern from extraction. We want one tag — invoice, receipt, contract, or other — and the model has no reason to think about line items while choosing it. Two narrowly-scoped agent calls beat one omnibus call.

    class DocumentClass(BaseModel):
    category: str  # "invoice" | "receipt" | "contract" | "other"
    confidence: float


async def make_classifier():
    return await agent(
        system_prompt=(
            "Classify the document into one of: invoice, receipt, contract, other. "
            "Return your confidence between 0 and 1. "
            "Use 'other' if the document does not clearly fit the first three."
        ),
        model="anthropic/claude-sonnet-4-20250514",
        response_format=DocumentClass,
        max_tool_calls=1,
        max_tokens=256,
        stream=False,
    )

    max_tool_calls=1 caps the agent’s tool loop. The classifier does not have tools and should answer in a single turn; setting the cap explicitly is a small belt-and-braces against regressions in upstream agent behavior. It also bounds latency: a classifier that loops would otherwise hold the worker on a per-document call long enough to break the throughput model.

    Two narrow agents are almost always better than one omnibus agent. The extractor’s prompt focuses on field schemas and “do not fabricate”. The classifier’s prompt focuses on category boundaries and confidence. Each prompt is short, each response format is small, and a regression in one prompt does not contaminate the other. The cost of the extra agent call is roughly zero compared to the OCR step that precedes it.

  4. Indexing with rollback

    Writing to the index is the only side effect that escapes the workflow. If anything downstream fails, we want the partial row gone — otherwise our index slowly fills with half-processed documents that look complete to the consumer.

    import sqlite3
import json
from flux.task import task


@task
async def delete_index_row(document_id: str) -> None:
    conn = sqlite3.connect("documents.db")
    conn.execute("DELETE FROM documents WHERE id = ?", (document_id,))
    conn.commit()
    conn.close()


@task.with_options(rollback=delete_index_row)
async def insert_index_row(
    document_id: str,
    category: str,
    fields: dict,
) -> str:
    conn = sqlite3.connect("documents.db")
    conn.execute(
        "INSERT INTO documents (id, category, fields_json) VALUES (?, ?, ?)",
        (document_id, category, json.dumps(fields)),
    )
    conn.commit()
    conn.close()
    return document_id

    The rollback= task fires only if the workflow as a whole fails after this task succeeded. If insert_index_row succeeds and a later task then crashes, Flux invokes delete_index_row(document_id) automatically. The index stays clean even when the workflow does not.

    This is the saga pattern in miniature. Each side-effecting task pairs with its compensating action. You do not have to write the compensation orchestrator — Flux runs the chain in reverse order when a workflow enters the rollback path.

    A few sharp edges. Rollback runs on workflow failure, not on workflow cancellation; cancel a workflow and you keep the side effects (this matches the semantics of “I want to stop the work but I have not necessarily made a mistake”). Rollback is best-effort — if delete_index_row itself raises, you get an event recording the failure and the row stays. For idempotent compensations (and DELETE WHERE id = ? is idempotent), this is rarely a problem. For non-idempotent compensations you want extra care, usually a uniqueness guard inside the compensation task itself.

  5. The workflow

    Now we tie OCR, extraction, classification, and indexing together.

    from flux import ExecutionContext
from flux.tasks import parallel
from flux.workflow import workflow
import uuid


@workflow
async def process_document(ctx: ExecutionContext[dict]):
    params = ctx.input or {}
    file_path = params["file_path"]
    document_id = params.get("document_id") or str(uuid.uuid4())

    text = await ocr_document(file_path)

    extractor = await make_extractor()
    classifier = await make_classifier()

    fields, classification = await parallel(
        extractor(instruction="Extract invoice fields.", context=text),
        classifier(instruction="Classify this document.", context=text),
    )

    await insert_index_row(
        document_id=document_id,
        category=classification.category,
        fields=fields.model_dump(),
    )

    return {
        "document_id": document_id,
        "category": classification.category,
        "confidence": classification.confidence,
        "fields": fields.model_dump(),
    }

    The parallel(...) call runs the two agent invocations concurrently. They depend on the same input (the OCR text) but not on each other, so there is no reason to serialize them. On a typical inference call of one to two seconds each, the fan-out cuts the agent stage roughly in half.

  6. Error handling end-to-end

    Walk through what happens when something fails.

    OCR fails three times. The task ends in FAILED, the workflow ends in FAILED, nothing else runs, no index row is created. There is nothing to roll back.

    OCR succeeds, extraction fails. The OCR result is in the event log (well, in output storage with a reference in the log). The workflow fails before any index write. Nothing to roll back. When you fix the extractor and rerun, Flux replays the OCR from the cached output instead of re-OCRing.

    OCR and extraction both succeed, indexing fails. Same as above. No row was written, no rollback needed.

    Indexing succeeds, the workflow then fails for some unrelated reason (a post-index step we have not shown — say, a Slack notification). Flux invokes delete_index_row(document_id) as the compensation, and the index returns to its pre-workflow state.

    The combination — durable per-task results, output storage for big payloads, and rollback for side effects — is what makes this pipeline production-shaped rather than a script.

What to remember

Durability is per-task, not per-workflow. Every successful task adds a row to the event log; every failed task can retry; every replay skips work that has already happened. You compose those properties by choosing where the task boundaries fall.

Structured output from agents is API-enforced. Passing response_format= constrains the model to return JSON matching your Pydantic schema — Anthropic enforces it through a forced tool call, so the result comes back as a typed object ready to use.

Rollback is how you keep side effects honest. Whenever a task escapes the workflow boundary — a database write, an outgoing email, a payment — pair it with a compensating task and pass it as rollback=. The orchestration is free.

Where this shows up next