HowdenPipeline

A DAG-based async pipeline for processing PDF documents through configurable steps (parsing, LLM extraction, etc.) with built-in MLflow observability.

Overview

HowdenPipeline orchestrates multi-step document processing pipelines as directed acyclic graphs (DAGs). Each PDF is processed concurrently, with steps executing in topological order per file. Results are cached on disk, so unchanged steps are skipped on re-runs.

PDF files
   │
   ├── [Parser] ──► result.md
   │       │
   │       ├── [LLM: payment_date] ──► result.json
   │       └── [LLM: payment_info] ──► result.json
   │
   └── ... (all files processed concurrently)

Requirements

• Python >= 3.12, < 3.14
• uv (recommended)

Installation

uv sync

Usage

import asyncio
from pathlib import Path
from HowdenPipeline.flow.graph_pipeline import GraphPipeline
from HowdenPipeline.manager.tracker import Tracker

async def main():
    pipeline = GraphPipeline(
        pdf=Path("data/pdfs"),   # directory of subfolders, each containing one PDF
        delete_folder=True,       # clean step output folders before each run
        tracker=tracker,          # optional MLflow tracker
        parameter=parameter       # HowdenConfig parameter object
    )

    pipeline.add_step(parser, output_filetype="md")
    pipeline.add_step(payment_date, dependencies=[parser], output_filetype="json", track=True)
    pipeline.add_step(payment_info, dependencies=[parser], output_filetype="json", track=True)

    matches = await pipeline.execute()

asyncio.run(main())

add_step parameters

Input folder structure

Each subfolder under the pdf root should contain one PDF and optionally a GT.json for accuracy evaluation:

data/pdfs/
├── claim_001/
│   ├── document.pdf
│   └── GT.json          # optional ground truth
├── claim_002/
│   └── document.pdf

Step output caching

Step results are written to disk alongside the input PDF. If a result file already exists, the step is skipped. To force re-execution, set delete_folder=True or delete the output folders manually.

data/pdfs/claim_001/
├── document.pdf
├── Parser/
│   ├── result.md
│   └── parameter.json
└── Parser/payment_info/
    ├── result.json
    └── parameter.json

MLflow Observability

When a Tracker is provided, the pipeline logs:

• All pipeline parameters
• Per-step timing (avg, min, max, total)
• Step result artifacts
• LLM token usage and model metadata
• Accuracy against ground truth JSON files
• Prompt templates with accuracy annotations

Start the MLflow UI:

mlflow ui --backend-store-uri sqlite:///mlflow.db --port 5000

To disable tracking, pass tracker=None (default).

Running Tests

uv run pytest
uv run pytest -v        # verbose
uv run pytest tests/test_async_script.py::test_files_run_concurrently  # single test

Project Structure

HowdenPipeline/
├── flow/
│   ├── graph_pipeline.py        # Main orchestrator — GraphPipeline class
│   ├── file_pipeline_runner.py  # Single-file execution with step traversal
│   ├── pipeline_graph_manager.py # DAG management with NetworkX
│   ├── match.py                 # Result dataclass (path, ground truth, file path)
│   └── parameter_serializer.py  # Serialization for step parameter logging
└── manager/
    ├── tracker.py               # MLflow / LangSmith logging abstraction
    └── jsonMatcher.py           # Accuracy comparison against ground truth
tests/
├── test_async_script.py         # Async concurrency tests
└── file_pipeline_runner_tests.py

Architecture Notes

• Concurrency: All PDFs are processed in parallel via asyncio.gather. Steps within a single file are sequential (topological order).
• Steps: Any callable — sync or async — that accepts a Path input and returns a str. The runner detects asyncio.iscoroutinefunction and awaits accordingly.
• Graph copy: Each file gets its own copy of the DAG so step state (result paths on edges) does not bleed between files.
• Git guard: When MLflow tracking is enabled, the pipeline will warn if there are uncommitted changes — ensuring experiment runs are tied to a clean git state.