Metadata-Version: 2.4
Name: HowdenPipeline
Version: 4.0.4
Summary: A simple configuration manager with Pydantic and JSON export.
Author-email: JesperThoftIllemannJ <jesper.jaeger@howdendanmark.dk>
License: MIT
Requires-Python: <3.14,>=3.12
Requires-Dist: howdenconfig>=1.0.6
Requires-Dist: mlflow<4.0.0,>=3.5.1
Requires-Dist: networkx>=3.6.1
Provides-Extra: dev
Requires-Dist: graphviz<0.22,>=0.21; extra == 'dev'
Requires-Dist: howdenconfig<2.0.0,>=1.0.6; extra == 'dev'
Requires-Dist: howdenllm>1.1.1; extra == 'dev'
Requires-Dist: howdenparser>5.0.0; extra == 'dev'
Requires-Dist: pytest<10.0.0,>=9.0.1; extra == 'dev'
Description-Content-Type: text/markdown

# 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](https://docs.astral.sh/uv/) (recommended)

## Installation

```bash
uv sync
```

## Usage

```python
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

| Parameter | Type | Description |
|-----------|------|-------------|
| `step` | Any | Callable step (sync or async). Receives input path, returns string result. |
| `dependencies` | list | Steps that must complete before this one. |
| `output_filetype` | str | File extension for cached result (`"md"`, `"json"`, etc.). |
| `track` | bool | If `True`, logs the result artifact to MLflow. |
| `name` | str | Override the display name for this step. |
| `input_result` | Any | Pass a specific result from an earlier step as secondary input. |

### 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:

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

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

## Running Tests

```bash
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.
