py-idp

Languages: English · 繁體中文 · 简体中文 · Docs site · PyPI · GitHub

General-purpose, AI-enabled Intelligent Document Processing for Python. Six-stage pipeline (parse → classify → extract → assess → validate → HITL). 12+ LLM backends. Pydantic-schema-driven. Built-in eval harness. Auto-chunking for oversized documents. Self-hosted OCR via Nanonets-OCR2-3B. AI-driven schema discovery.

Install

pip install py-idp                # core (pydantic + typer + rich + httpx + pdfplumber + tiktoken)
pip install py-idp[docling]      # IBM Docling — best PDF table extraction
pip install py-idp[ocr]           # tesseract fallback for noisy scans
pip install py-idp[openai]       # OpenAI SDK (also used for 8 China LLMs)
pip install py-idp[anthropic]    # Anthropic SDK
pip install py-idp[ollama]       # Ollama client
pip install py-idp[china]        # 8 China LLM providers via OpenAI-compatible protocol
pip install py-idp[hf-vlm]        # Self-hosted Nanonets-OCR2-3B (Apple Silicon / CUDA)
pip install py-idp[api]           # FastAPI server (idp.api:app — production-ready)
pip install py-idp[pdf-render]    # PDF → image rendering for multimodal backends
pip install py-idp[eval]          # datasets + pandas for `idp eval`
pip install py-idp[dev]           # pytest + ruff + mypy + hypothesis + pytest-benchmark
pip install py-idp[docs]          # mkdocs + mkdocstrings for building the docs site locally

Combine: pip install py-idp[docling,anthropic,eval,dev].

No API key needed to install or run the test suite — MockBackend ships in-tree. tiktoken is installed automatically by the core package (used for token-budget chunking).

See CHANGELOG.md for release notes.

30-second tour

import idp
from idp.pipeline import Pipeline

result = Pipeline(
    backend="mock",                # or "ollama", "openai", "anthropic", "china:qwen" ...
    schema="Invoice",
    business_rules=[...],
).run(idp.Document.from_path("invoice.pdf"))

print(result.extraction)    # dict — validated against your Pydantic schema
print(result.confidence)    # dict — per-field 0..1, <0.6 flagged for review
print(result.validation)    # dict — schema + business-rule outcomes

A faithful end-to-end run on the in-tree sample invoice, measured live:

metric	value
classification	`invoice` (conf 0.99)
extraction shape	12 fields, 2 line items
validation	PASS
exact-match fields vs gold	7 / 9 = 78 % (single doc)
low-confidence flags (HITL)	2 (subtotal, tax_amount — small-model arithmetic)
latency (mocked LLM)	< 2 ms

Full eval harness, 3 invoices, real local Ollama (qwen2.5:0.5b, 397 MB):

metric	value
schema-valid rate	100 % (3 / 3)
field F1	0.96 (precision 1.00, recall 0.93)
latency	2.05 s / doc on Apple Silicon
per-doc exact match	inv-001 7/9 · inv-002 9/9 · inv-003 9/9

The framework is honest about what small models get wrong: arithmetic on tiny models (subtotal/tax_amount) is flagged with conf 0.10 and routed to HITL review, not silently passed.

Why py-idp vs. a hand-rolled LLM script?

hand-rolled py-idp

PDF parsing + table extraction your problem Docling, pdfplumber, plain — pick one

Multi-page LLM calls write a chunker built-in, with per-chunk failure handling

Confidence scoring your problem heuristic + optional LLM self-rate

Schema validation your problem Pydantic-typed, errors surfaced in doc.errors

HITL review your problem Streamlit app, persists corrections

Reliability your problem retry, cache, checkpoint built-in

Schema discovery your problem discover_schema() from a hint + a sample doc

Production eval your problem idp eval with field F1 + schema-valid rate

RL on corrections your problem idp rl-update — deterministic, inspectable

12+ LLM backends your problem one Pipeline(backend="...") line

Dual license (AGPL + commercial) your problem revenue path without losing the open-source core

The pipeline

flowchart LR
    A[INGEST<br/>file path / URL / bytes] --> B[PARSE<br/>Docling · pdfplumber · plain]
    B --> C[CLASSIFY<br/>rule-first, LLM fallback]
    C --> D{ROUTE<br/>multimodal or<br/>OCR+LLM?}
    D -->|multimodal| E1[EXTRACT<br/>images → VLM]
    D -->|OCR+LLM| E2[EXTRACT<br/>text → LLM]
    E1 --> F[ASSESS<br/>per-field confidence]
    E2 --> F
    F --> G[VALIDATE<br/>Pydantic + business rules]
    G --> H{any low<br/>confidence?}
    H -->|yes| I[HITL<br/>Streamlit review]
    H -->|no| J[(PipelineResult)]
    I --> K[save correction<br/>+ update policy]
    K --> J

    style A fill:#1f6feb,stroke:#1f6feb,color:#fff
    style B fill:#2da44e,stroke:#2da44e,color:#fff
    style C fill:#2da44e,stroke:#2da44e,color:#fff
    style D fill:#bf8700,stroke:#bf8700,color:#fff
    style E1 fill:#8250df,stroke:#8250df,color:#fff
    style E2 fill:#8250df,stroke:#8250df,color:#fff
    style F fill:#cf222e,stroke:#cf222e,color:#fff
    style G fill:#cf222e,stroke:#cf222e,color:#fff
    style I fill:#6e7781,stroke:#6e7781,color:#fff
    style J fill:#0a3069,stroke:#0a3069,color:#fff

Each stage is a pure function over a Document. They run independently, are unit-testable in isolation, and any one can be swapped.

stage	module	default	what it does
INGEST	`idp.core.document`	`Document.from_path`	Bring bytes into a `Document` (path, URL, or in-memory)
parse	`idp.parse`	Docling (PDF) · pdfplumber (fallback) · plain text	Extracts text + tables + page images
classify	`idp.classify`	rule-first, LLM fallback	Detects doc type: invoice, contract, bank_statement, …
route	`idp.parse.router`	auto	Chooses multimodal VLM vs OCR+LLM based on doc features
extract	`idp.extract`	Pydantic-schema-driven	Validated structured extraction from text or images
assess	`idp.assess`	heuristic + optional LLM self-rate	Per-field confidence 0..1
validate	`idp.validate`	Pydantic + user predicates	Schema check + business rules
HITL	`idp.hitl`	Streamlit UI	Review low-confidence fields, save corrections
pipeline	`idp.pipeline.pipeline`	orchestrator	Composes the above, returns `PipelineResult`

Legend — blue: entry, green: parser/structure, yellow: decision, purple: extraction, red: confidence/validation, gray: human-in-the-loop, dark blue: result

Architecture at a glance

flowchart TB
    subgraph CLI["idp CLI (idp.pipeline.cli)"]
        run[run / schemas /<br/>providers / eval /<br/>rl-update / serve]
    end
    subgraph API["FastAPI server (idp.api)"]
        rest["/extract / /templates<br/>/healthz / /metrics"]
    end
    subgraph Core["idp.core + idp.parse + idp.chunker"]
        doc[Document /<br/>Page / Block]
    end
    subgraph LLM["idp.llm"]
        backends[Backend Protocol<br/>+ Reliability wrappers]
    end
    subgraph Eval["idp.eval + idp.rl"]
        harness[eval runner /<br/>policy update /<br/>calibration]
    end
    subgraph HITL["idp.hitl"]
        streamlit[Streamlit<br/>review UI]
    end
    subgraph Storage["idp.storage"]
        json[JsonFileStorage /<br/>SqlStorage]
    end
    subgraph Templates["idp.templates"]
        reg["*.md →<br/>TemplateRegistry"]
    end

    CLI --> Core
    API --> Core
    Core --> LLM
    Core --> Templates
    LLM --> Core
    Core --> Eval
    Core --> HITL
    HITL --> Storage
    Eval --> Storage
    Storage --> Core

    style CLI fill:#1f6feb,stroke:#1f6feb,color:#fff
    style API fill:#1f6feb,stroke:#1f6feb,color:#fff
    style Core fill:#8250df,stroke:#8250df,color:#fff
    style LLM fill:#bf8700,stroke:#bf8700,color:#fff
    style Eval fill:#2da44e,stroke:#2da44e,color:#fff
    style HITL fill:#6e7781,stroke:#6e7781,color:#fff
    style Storage fill:#cf222e,stroke:#cf222e,color:#fff
    style Templates fill:#0a3069,stroke:#0a3069,color:#fff

The color coding matches the pipeline diagram above. Each box is a subpackage you can import idp.<name> to use independently — idp.core needs nothing from the rest, idp.llm only needs idp.errors, etc.

LLM backends

py-idp has a single Backend protocol and every provider implements it. The pipeline never sees provider-specific code — it always calls backend.complete(CompletionRequest).

flowchart LR
    subgraph pyidp[py-idp Pipeline]
        EX[extract / assess /<br/>validate / rl-eval]
    end
    EX --> Backend[Backend<br/>Protocol]
    Backend --> Wrap{RetryingBackend<br/>+ ExtractionCache}
    Wrap --> Call[complete<br/>request]
    Call --> P[openai / anthropic /<br/>ollama / china:qwen /<br/>compat: any OpenAI-style]
    Call --> N[Nanonets-OCR2-3B<br/>self-hosted VLM]
    Call --> M[mock / mock-random /<br/>mock-omits]

International (5 providers, any OpenAI-compat endpoint)

name	notes
`openai`	GPT-4o (vision), GPT-4.1, o1
`anthropic`	Claude 3.5/4 Sonnet, Claude Haiku (vision)
`ollama`	local `llama3.2-vision`, `qwen2.5-vl` — default base URL `http://localhost:11434/v1`
`vllm` / `lm-studio` / `compat`	any OpenAI-compatible chat-completions endpoint
`mock`	offline / CI baseline (`mock`, `mock-random`, `mock-omits`)

export OPENAI_API_KEY=...
idp run invoice.pdf --schema Invoice --backend openai

China (8 providers — all speak the OpenAI Chat-Completions protocol)

Run idp providers to print the full table. Highlights:

provider	env var	default	vision model
`deepseek`	`DEEPSEEK_API_KEY`	deepseek-chat	— (text-only)
`qwen`	`DASHSCOPE_API_KEY`	qwen-plus	qwen2.5-vl-72b-instruct
`zhipu`	`ZHIPUAI_API_KEY`	glm-4-plus	glm-4v-plus
`moonshot`	`MOONSHOT_API_KEY`	moonshot-v1-128k	moonshot-v1-128k-vision-preview
`yi`	`YI_API_KEY`	yi-large	yi-vision
`doubao`	`ARK_API_KEY`	doubao-pro-32k	doubao-1-5-vision-pro-32k
`hunyuan`	`HUNYUAN_API_KEY`	hunyuan-pro	hunyuan-vision
`baichuan`	`BAICHUAN_API_KEY`	baichuan4	— (text-only)

from idp.llm import get_china_backend

backend = get_china_backend("qwen", multimodal=True)
# backend.model == "qwen2.5-vl-72b-instruct"

Self-hosted (Nanonets-OCR2-3B on Apple Silicon / CUDA)

For documents you can't send to a third party. No API key, no cloud egress, fully offline after the first download (~7 GB cached to ~/.cache/huggingface/hub/).

pip install py-idp[hf-vlm]    # adds torch + transformers + accelerate + safetensors
export IDP_ENABLE_NANONETS=1  # explicit opt-in (avoids surprise downloads)
export IDP_BACKEND=nanonets
idp run scan.pdf --backend nanonets

Memory budget on Apple M4 16 GB (float16, 448×448 image):

weights + vision encoder + KV cache: ~8.3 GB
OS + apps: ~3.5 GB
headroom: ~4 GB (comfortable)

Speed: ~5-15 sec per page on M4. First call: 5-10 min to download the model. Subsequent calls: ~10 s to load from cache.

Why Nanonets-OCR2-3B: open weights, no auth, Apache-2.0 (Qwen2.5-VL base) — verify the Nanonets fine-tune license before commercial use. Outperforms Tesseract on noisy scans and handles multilingual docs.

Why gated: model download is large and slow. We refuse to auto-trigger it; you must explicitly set IDP_ENABLE_NANONETS=1.

Platform support (verified at construction time):

Platform	Status
macOS arm64 (M1/M2/M3/M4, 16+ GB)	✅ tested target, MPS
macOS arm64 (8 GB)	❌ OOM (use Docling instead)
macOS x86_64 (Intel)	❌ no MPS, eGPU CUDA flaky — fails loud
Linux x86_64 + CUDA	✅ best (1-5s per page)
Linux x86_64 CPU-only	⚠️ works, 30-60s per page
Linux arm64	⚠️ works, CPU only
Windows x86_64 + CUDA	✅ same as Linux CUDA
Windows arm64	❌ PyTorch has no Windows-arm64 wheels — fails loud

from idp.llm.nanonets import NanonetsVLBackend
backend = NanonetsVLBackend(
    device="mps",          # or "cuda", "cpu", "auto"
    max_image_side=448,    # 4x less vision memory than 1024, ~95% acc
    load_in_4bit=False,    # True if you OOM at float16
)

# End-to-end with PdfPagesParser (renders pages to images)
from idp import Document, Pipeline
from idp.parse.parser import parse_document
from idp.core.schemas import Invoice

doc = Document.from_path("scan.pdf")
parse_document(doc, parser="pdf-pages")   # renders pages to base64 PNG
result = Pipeline(backend=backend, schema=Invoice).run(doc)
print(result.document.extraction)

Auto-chunking for oversized documents

Nanonets-OCR2-3B has a 16k token context. A 50-page invoice PDF won't fit in one call. extract() detects this and automatically splits the input, runs the model once per chunk, and merges the per-chunk extractions. No glue code required — it's invisible to the caller.

Two chunkers ship:

chunker	when	default config
`PageChunker`	multimodal (NanonetsVLBackend + page images)	4 pages per chunk, 1-page overlap
`TokenChunker`	text extractors (OCR + LLM)	4000 tokens per chunk, 200-token overlap (tiktoken)

from idp.chunker import PageChunker, TokenChunker

# Tighter memory budget on a small M-series Mac
chunker = PageChunker(max_pages=2, overlap_pages=1)

# Or pass directly to the pipeline
from idp.pipeline import Pipeline
pipe = Pipeline(backend=backend, schema=Invoice, chunker=chunker)

# End-to-end: chunks, calls, merges, validates — one call
result = pipe.run(Document.from_path("huge-50-page-scan.pdf"))

The merged extraction includes a _chunk_count marker so you can attribute cost and observability per chunk run.

Per-chunk failure resilience: if one chunk's LLM call fails, the error is logged (extract_chunk_failed[i]) but other chunks' data is still merged in. You get partial results + a clear error trail, not a hard crash.

See src/idp/chunker.py for the implementation and tests/test_chunker.py for the 34 tests.

CLI

idp run path/to/invoice.pdf --schema Invoice --backend ollama --output out.json
idp batch path/to/invoices/ --backend ollama --schema Invoice \    # many docs in one shot
              --output out.jsonl --report report.json --dlq dlq.jsonl
idp providers                                          # full provider table
idp schemas                                            # built-in Pydantic schemas
idp discover-schema scan.pdf --hint "extract vendor_name, total_amount" --output schema.json
idp eval --dataset src/idp/eval/datasets/invoices \
          --strategy mock,mock-omits --output results.json
idp serve                                              # launch Streamlit HITL UI on :8501

For copy-pasteable scripts that show each backend / pipeline pattern end-to-end, see examples/ — every numbered example is runnable offline with python examples/NN_*.py and falls back to MockBackend if no API key is set.

Batch processing (100s–1000s of documents)

For runs over many documents (Databricks jobs, cron sweeps, S3 prefix replay), use the idp batch CLI instead of looping idp run:

idp batch /mnt/inbox/ \                         # directory: recursive scan for *.pdf/*.png/*.jpg/*.tiff/*.txt
          --backend ollama --schema Invoice \
          --output out.jsonl \                  # per-doc JSONL: path, ok, extraction, classification, confidence, ...
          --report report.json \                # aggregate: throughput, p50/p95 latency, error histogram
          --dlq dlq.jsonl \                     # failed docs only (for a 2nd-pass retry)
          --checkpoint ledger.jsonl \           # resume ledger: rerun picks up where it stopped
          --progress-every 50                   # log every 50 docs; 0 to silence

Sources can also be a glob, an explicit list (@/mnt/inbox.lst), or multiple paths mixed with directories. Files are deduplicated by absolute path. Mermaid of the flow:

flowchart LR
    A["sources<br/>(paths / dirs / @file)"] --> B["collect_paths"]
    B --> C["process_batch<br/>(per doc: pipeline.run)"]
    C --> D["out.jsonl<br/>(per-doc JSONL)"]
    C --> E["dlq.jsonl<br/>(failed only)"]
    C --> F["checkpoint.jsonl<br/>(resume ledger)"]
    D --> G["report.json<br/>(throughput, latency,<br/>error histogram)"]
    style C fill:#f9f,stroke:#333

For programmatic use, idp.process_batch and idp.BatchItemResult are exported at the top level — from idp import process_batch works directly. (The legacy import from idp.llm.nanonets_batch import process_batch is kept as a 12-line re-export shim.)

The pipeline is serial in v0.3.x — concurrency is a future add. For now, expect roughly the same throughput as a single idp run per doc.

Bring your own schema

The built-in Invoice, Contract, BankStatement schemas are convenience references — pass any Pydantic model:

classDiagram
    class Document {
        +str source_path
        +str raw_text
        +list~Page~ pages
        +dict extraction
        +dict confidence
        +str schema_name
        +str template_name
    }
    class Page {
        +int page_num
        +str text
        +list~str~ images_b64
        +list~dict~ tables
    }
    class PipelineResult {
        +Document document
        +str schema_name
        +str backend_name
        +str mode
        +list~StageTiming~ timings
    }
    class BaseModel {
        <<interface>>
    }
    Document "1" --> "*" Page : contains
    PipelineResult "1" --> "1" Document : wraps
    Document ..> BaseModel : validated against

from pydantic import BaseModel
from idp import Document
from idp.pipeline import Pipeline

class Receipt(BaseModel):
    merchant: str
    total: float
    currency: str
    date: str

result = Pipeline(backend="ollama", schema=Receipt).run(
    Document.from_path("receipt.jpg")
)

Auto-schema discovery

Try it: python -m examples.discover_schema_sample Runs 6 end-to-end scenarios on a real PDF (generates a 2-page invoice, discovers schema, runs extraction, exercises edge cases). No API key or poppler required.

You have a scanned PDF and a vague sense of "I want fields X, Y, Z" — but no Pydantic class yet. discover_schema() asks the multimodal LLM (NanonetsVLBackend by default) to propose a JSON Schema, then compiles it to a Pydantic class you can pass straight into Pipeline(schema=...).

flowchart LR
    A[scan.pdf +<br/>hint text] --> B[discover_schema]
    B --> C[Nanonets VLM<br/>multimodal LLM]
    C --> D[JSON Schema<br/>proposal]
    D --> E[parse +<br/>validate]
    E -->|invalid| F[retry with<br/>error feedback]
    F --> C
    E -->|valid| G[compile to<br/>Pydantic class]
    G --> H[hint grounding<br/>check]
    H -->|score < 0.5| I[log warning:<br/>LLM ignored hint]
    H --> J[Pipeline.run<br/>schema=Schema]
    I --> J
    J --> K[(extraction)]

import idp

Schema, schema_dict = idp.discover_schema(
    "scan.pdf",
    hint="extract vendor_name, invoice_number, total_amount, and line items",
)
# Schema is a Pydantic BaseModel subclass — pass it directly:
result = idp.Pipeline(backend="nanonets", schema=Schema).run(
    idp.Document.from_path("scan.pdf")
)
print(result.document.extraction)

The returned DiscoveryResult exposes both the compiled Pydantic class and the raw JSON Schema dict:

result = idp.discover_schema("scan.pdf", hint="...")
result.schema_class    # the Pydantic class
result.json_schema     # the raw JSON Schema dict
result.raw_response    # raw LLM output (debug aid)
result.backend_name    # "NanonetsVLBackend"
result.doc             # the parsed Document (reuse for extraction)

CLI equivalent:

export IDP_ENABLE_NANONETS=1
idp discover-schema scan.pdf \
    --hint "extract vendor_name, total_amount, and line items" \
    --output schema.json

Defaults: pages capped at 4 (fits most 16k-context VLMs), Nanonets backend (must set IDP_ENABLE_NANONETS=1), fallback to Mock for tests.

Hint grounding: when you provide a hint, discover_schema() extracts candidate field-name tokens from it and checks how many of them appear in the discovered schema (exact match, plus fuzzy match with SequenceMatcher ratio > 0.8). The result is on DiscoveryResult.hint_grounding as a dict with hint_tokens, schema_fields, grounded, ungrounded, and a grounding_score (0.0 = none of your hint tokens appear, 1.0 = perfect match). If the score is below 0.5, a warning is logged telling you which hint tokens the LLM ignored. Doesn't fix wrong names — makes the wrongness observable so you know to verify.

result = idp.discover_schema("scan.pdf", hint="...")
if result.hint_grounding and result.hint_grounding["grounding_score"] < 0.5:
    print("LLM largely ignored your hint!")
    print("missing:", result.hint_grounding["ungrounded"])

Honest limits:

LLM-proposed field names are sometimes wrong — the user hint steers this but doesn't guarantee it. The hint_grounding field above makes this observable. Always review the resulting schema against a few real extractions before using in production.
Field types are inferred from the JSON Schema (string / number / integer / boolean / array / nested object). Required-vs-optional is preserved.
LLMs sometimes emit ```json fences or wrap output in prose; the parser strips both. Pure garbage raises ValueError with the first 200 chars for debugging.
This is schema discovery — it tells you what fields exist and what they're called. It is not schema validation — pass the discovered schema into Pipeline(schema=...) and use HITL review for the validation step.

See src/idp/discover.py for the implementation, tests/test_discover.py for the 45 tests, and examples/discover_schema_sample.py for a runnable end-to-end demo.

Templates per PDF type (LLM context via Markdown)

For each document type you process, write a templates/invoice.md / templates/contract.md file. The Markdown body becomes LLM context on every extraction call — field descriptions, worked examples, common-mistakes notes. JSON Schema alone can't express "subtotal and total_amount are NOT the same — subtract tax to get subtotal", but a template can.

flowchart LR
    A[templates/<br/>invoice.md] -->|load at startup| B[TemplateRegistry]
    B -->|filename/MIME match| C[Template]
    C -->|body + name + version| D[Pipeline.run]
    D -->|prepend to prompt| E[LLM call<br/>text + template body<br/>+ JSON Schema]
    E -->|extraction| F[PipelineResult<br/>+ doc.template_name<br/>+ doc.template_version]
    F -.audit.-> G[which template<br/>version produced<br/>this row?]

# templates/invoice.md
---
name: invoice
schema: Invoice
version: 1
mime_types: [application/pdf, image/jpeg]
filename_patterns: ["*invoice*", "*receipt*"]
---

# Invoice

## Fields

* **vendor_name** — the company issuing the invoice. Look for the
  largest bold text near the top.
* **invoice_number** — usually near the top, format varies
  (`INV-2024-001`, `2024-09-15-001`).
* **subtotal** — sum of line items, BEFORE tax.
* **total_amount** — final amount including tax. **NOT** the subtotal.

## Common mistakes

* **subtotal vs total_amount**: subtract `tax_amount` from
  `total_amount` to get `subtotal`. If the invoice only shows one,
  leave the other as `null`.
* **date_due vs date_issued**: due date is usually later than issued.

from idp import Pipeline
from idp.templates import TemplateRegistry

registry = TemplateRegistry.load("./templates")
pipe = Pipeline(backend="nanonets", schema="Invoice", template="invoice")
pipe.set_template_registry(registry)
result = pipe.run(scan_path)  # template body prepended to LLM prompt
print(result.template_name, result.template_version)

When you POST /extract without pinning schema_name, the server auto-picks the right template by filename + MIME and uses its body as LLM context. The matched template is recorded in PipelineResult.template_name + template_version for audit.

Three sample templates ship in templates/: invoice.md, contract.md, bank_statement.md. See src/idp/templates.py and tests/test_template_to_llm.py for the full API.

Chunking for oversized documents

Most LLMs cap context at 6k-200k tokens. A long invoice, contract, or multi-page scan may exceed that. py-idp auto-detects oversized input, chunks it, runs the LLM once per chunk, and merges the per-chunk extractions — all without glue code.

flowchart TD
    A[Document] --> B{multimodal<br/>backend?}
    B -->|yes| C[count pages]
    B -->|no| D[count tokens<br/>via tiktoken]
    C --> E{pages > 4?}
    D --> F{tokens > 4000?}
    E -->|yes| G[PageChunker<br/>4 pages, 1 overlap]
    E -->|no| H[single call]
    F -->|yes| I[TokenChunker<br/>4000 tok, 200 overlap]
    F -->|no| H
    G --> J[LLM per chunk]
    I --> J
    J --> K[merge_extractions]
    K --> L[Pydantic validation<br/>on the merged result]
    L --> M[(PipelineResult)]
    H --> M

chunker	when used	default config
`PageChunker`	multimodal backends (NanonetsVLBackend, GPT-4o, etc.)	4 pages per chunk, 1-page overlap
`TokenChunker`	text extractors (OCR + LLM)	4000 tokens per chunk, 200-token overlap (tiktoken)

Defaults are tuned for the most common models:

4 pages @ 200dpi ≈ 3000 image tokens → fits Nanonets-OCR2-3B (16k context)
4000 text tokens → fits qwen2.5:0.5b (6k context) and llama3.2 (8k)

Per-chunk failure resilience: if one chunk's LLM call fails, the error is logged (extract_chunk_failed[i]) but other chunks' data is still merged. Partial results > no results.

from idp.chunker import PageChunker

# Tight memory budget (M-series Mac with 16 GB unified)
chunker = PageChunker(max_pages=2, overlap_pages=1)
result = Pipeline(backend="nanonets", schema="Invoice", chunker=chunker).run(
    Document.from_path("huge-50-page-scan.pdf")
)
print(result.document.extraction.get("_chunk_count"))  # ~25

The merged extraction is schema-validated as a whole after merging, so you still get a Pydantic-typed result even though it was built from many small extractions.

See src/idp/chunker.py for the implementation and tests/test_chunker.py for the 34 tests.

Add business rules

flowchart LR
    Ext[extract] --> Schema[schema validation<br/>Pydantic types]
    Schema --> Rules[business_rules<br/>list of predicates]
    Rules --> R1{required_<br/>fields_rule}
    Rules --> R2{numeric_<br/>range_rule}
    Rules --> R3[user predicate<br/>e.g. cross-field check]
    R1 -->|fail| V["validation =<br/>{passed: false, errors}"]
    R2 -->|fail| V
    R3 -->|raise| V
    R1 -->|pass| R2
    R2 -->|pass| R3
    R3 -->|pass| V2["validation =<br/>{passed: true}"]
    V --> Conf[confidence<br/>stays as-is]
    V2 --> Conf

from idp.validate import required_fields_rule, numeric_range_rule

pipe = Pipeline(
    backend="ollama",
    schema="Invoice",
    business_rules=[
        required_fields_rule("invoice_number", "vendor_name", "total_amount"),
        numeric_range_rule("total_amount", min_v=0.0, max_v=10_000_000.0),
    ],
)

Two built-ins ship; define your own by writing a (dict) -> (bool, str | None) predicate. Rules that raise are caught — they don't crash the pipeline.

Demo

A live end-to-end run on the in-tree sample invoice (MockBackend — no API key):

The same extraction viewed through the Streamlit HITL review UI:

(The SVGs above are illustrative mockups. For real screen recordings, run idp run path/to/your-invoice.pdf --backend ollama and idp serve.)

Eval harness

Honest extraction claims need labeled data and side-by-side backend comparison. py-idp ships both.

flowchart LR
    A[datasets/<br/>cases.jsonl + docs/] -->|load| B[Eval Runner]
    B -->|for each strategy| C[Pipeline.run<br/>+ extract + assess]
    C --> D{schema<br/>valid?}
    D -->|yes| E[field_match<br/>vs gold]
    D -->|no| F[mark invalid]
    E --> G[aggregate<br/>P/R/F1]
    F --> G
    G --> H[results.json<br/>per-strategy rows]
    H -->|ci or report| I[publish or<br/>block merge]

idp eval --dataset src/idp/eval/datasets/invoices \
         --strategy mock,mock-omits,ollama --output results.json

Reports per-strategy: schema-valid rate, field-level F1, $/doc, latency. The in-tree fixtures (3 invoices, 2 contracts, 5 CORD-style receipts) are hand-labeled so you can publish numbers you actually verified.

CORD-style receipt benchmark

A 5-receipt hand-curated subset modeled on the CORD: Consolidated Receipt Dataset lives at src/idp/eval/datasets/cord_subset/. Run it with:

python examples/benchmark_cord.py

This runs the in-tree MockBackend against all 5 receipts and prints per-field precision / recall / F1 plus latency. No API key needed — the numbers are reproducible by anyone with pip install py-idp[eval]. To benchmark a real backend, swap "mock" for "ollama" / "openai" / "anthropic" / "china:qwen" in examples/benchmark_cord.py.

Reliability: retries, cache, checkpoint

Three opt-in features for production workloads. They compose cleanly — retry wraps the backend, cache wraps the retry, and checkpoint tracks per-document completion independently.

flowchart LR
    User[Pipeline.run] --> Check{cached?}
    Check -->|yes| CacheHit[return cached<br/>extraction]
    Check -->|no| Retry{retryable<br/>error?}
    Retry -->|yes| Wait[wait 1s→2s→4s→8s<br/>±20% jitter]
    Wait --> Call[backend.complete]
    Retry -->|no| Fail[raise IDPError]
    Call --> Retry
    Call -->|success| Store[cache.put<br/>+ checkpoint.record]
    Store --> Done[(PipelineResult)]
    CacheHit --> Done

RetryingBackend — automatic retries with backoff

Wrap any Backend with exponential-backoff retries on transient errors (rate limits, timeouts, connection errors). Auth and bad-request errors fail fast — no point retrying those.

from idp import Pipeline
from idp.reliability import RetryConfig

pipe = Pipeline(
    backend="openai",
    schema="Invoice",
    retry=RetryConfig(max_retries=5, initial_delay_sec=2.0, max_delay_sec=60.0),
)

Defaults: 4 attempts, 1s → 2s → 4s → 8s with ±20% jitter, capped at 30s. On non-retryable errors (AuthError, BadRequestError) the wrapped backend raises immediately. Errors are classified via message pattern matching — see idp.reliability.classify_exception for the taxonomy.

ExtractionCache — disk-backed dedup

Same input → no LLM call. Cache key = sha256 of (schema_name, backend_name, request payload). Hits are tracked per schema for observability.

from idp import Pipeline
from idp.reliability import ExtractionCache

pipe = Pipeline(
    backend="nanonets",
    schema="Invoice",
    cache=ExtractionCache("/dbfs/mnt/idp/extract.db"),  # default: ~/.cache/idp/extract.db
)

Default location is ~/.cache/idp/extract.db — survives across process restarts. Stats via cache.stats() return entries, total_hits, per-schema breakdown.

CheckpointStore — batch resume

For process_batch() over hundreds/thousands of docs, an interrupted run (server restart, network blip) loses no work on retry. Idempotent by default — just pass the same checkpoint path on retry:

from idp.llm.nanonets_batch import process_batch

# First run: processes docs 1-1000, dies at doc 500
results = process_batch(paths, pipeline, checkpoint="/dbfs/.../cp.jsonl")
# Second run: docs 1-499 skipped (in ledger), resumes from doc 500
results = process_batch(paths, pipeline, checkpoint="/dbfs/.../cp.jsonl")

Set archive_at_start=True to rotate the ledger between runs (one file per run, history preserved). Use CheckpointStore.clear() to force re-processing.

Both retry=True and cache=True compose: cache is applied AFTER retry so cached hits skip the retry loop entirely.

See src/idp/reliability.py, src/idp/checkpoint.py, and tests/test_reliability.py / tests/test_checkpoint.py for the full API.

Production scaffolding (built in, optional)

concern	ships with	swap for production
Async job queue	`idp.queue.InProcessQueue`	ARQ / Celery / SQS
Persistent storage	`idp.storage.JsonFileStorage`	Postgres + S3
API key auth	`idp.auth.keys`	wire into FastAPI dep
HTTP API	`idp.api:app` (production, FastAPI, auth+rate-limit+metrics)	your own service
HITL UI	`idp.hitl.app` (Streamlit)	React / FastAPI
Docker	`Dockerfile`, `docker-compose.yml`	your infra
RL from HITL corrections	`idp.rl` + `idp rl-update`	online per-review update (`PolicyCache`)
Document chunking	`idp.chunker` (auto for oversized input)	custom `PageChunker` / `TokenChunker`
Schema discovery	`idp.discover_schema` + `idp discover-schema`	custom multimodal backend

Not in 0.3.x (deliberately)

Multi-tenant isolation, SSO/SAML/RBAC, audit-grade storage — needed for SaaS but premature for a single-tenant self-host. Open an issue to request.

Learning from HITL corrections (RL)

Every human review in idp.storage becomes a training signal. The framework ships an offline batch policy update that turns "fields humans keep correcting" into higher-confidence-floor + lower-confidence-penalty for those fields — so they reliably surface to HITL review in the next run.

sequenceDiagram
    participant User
    participant Pipeline
    participant LLM
    participant HITL as HITL UI
    participant Storage
    participant Policy as policy.json

    User->>Pipeline: run(doc)
    Pipeline->>LLM: extract(doc, schema)
    LLM-->>Pipeline: extraction + confidence
    Pipeline->>Pipeline: assess(per-field)
    Pipeline->>HITL: low-confidence fields
    HITL->>User: show field for review
    User->>HITL: edit (or accept)
    HITL->>Storage: mark_reviewed(result_id, edited, reviewer)
    Note over Storage: append-only,<br/>schema: reviews +<br/>review_edits

    User->>Policy: idp rl-update --storage ... --output policy.json
    Policy->>Storage: aggregate rewards
    Policy->>Policy: apply min_reviews=10 guard
    Policy-->>User: new policy with field_floors

    User->>Pipeline: run(doc, policy_path=policy.json)
    Pipeline->>Pipeline: assess adjusts confidence<br/>by field_penalties
    Note over Pipeline: next run routes<br/>"vendor_name" to HITL<br/>more reliably

# Offline batch: derive rewards from accumulated reviews, write policy.json
idp rl-update --storage idp_data/results.jsonl \
               --output policy.json

# Apply policy in the pipeline:
result = Pipeline(
    backend="ollama",
    schema="Invoice",
    policy_path="policy.json",
).run(Document.from_path("invoice.pdf"))

# Or hand-craft reviews if you don't have storage yet:
idp rl-update --reviews reviews.jsonl --output policy.json

What this is: a deterministic, inspectable, version-controllable rule update. It is not a learned reward model, not a fine-tuned LLM. We're learning the post-hoc confidence adjustment that decides what to flag for HITL — not the model itself.

Why this approach: real-world ROI is highest at this layer. Training an LLM with RLHF/DPO gives ~2-3% F1 gain for weeks of work; a 7B model would beat that for less. Learning which fields to send to HITL more reliably compounds every review.

Measured (real Ollama, qwen2.5:0.5b, in-tree fixture):

field	without policy	with policy (after 5 human corrections)	delta
`vendor_name`	0.75 (would pass HITL)	0.55 (now flagged)	−0.20
`subtotal`	0.10 (already flagged)	0.0 (urgent)	−0.10
`invoice_number`	0.75	0.75 (no override)	0.0

Online (per-review) update ships in v0.2 via PolicyCache; the offline batch is fully wired today.

Lifecycle of a single document

stateDiagram-v2
    [*] --> Queued: Pipeline.run()
    Queued --> Extracted: low confidence
    Queued --> Accepted: high confidence
    Extracted --> Queued: re-run with new policy
    Extracted --> Reviewed: human edits field
    Reviewed --> Accepted: human accepts
    Reviewed --> Edited: human saves correction
    Edited --> Queued: idp rl-update
    Accepted --> [*]
    Edited --> [*]: contributes to policy

The Queued → Reviewed → Edited → policy loop is what makes this "RL" — every human edit is a training signal that updates which fields get routed to review next time.

Calibration eval — does the policy actually do what it claims?

# Generate synthetic reviews from gold truth, derive a policy, evaluate it
idp rl-update --reviews reviews.jsonl --output policy.json
idp rl-eval   --policy policy.json --fixtures src/idp/eval/datasets/invoices \
              --injection-rate 0.30 --output calibration.json

Reports hit rate when policy fires (did humans correct what we flagged?) and true-accept rate when policy silent (did humans accept what we didn't flag?), with explicit n= and a synthetic=true flag — synthetic reviews are biased optimistic (gold truth IS the human's correction), so real HITL data will be noisier.

Honest measured results (synthetic reviews from 3 in-tree invoices, qwen2.5:0.5b real Ollama run, fields × docs = 27 pairs):

metric	value	what it means
policy caught (flag → human corrected)	21	without the policy, these errors would have escaped HITL
policy silenced (was flagged, no longer flagged)	0	no regressions
already flagged by both	2	no change
model was right, not flagged	4	correct accepts — model was actually right

Honest call-out: with qwen2.5:0.5b specifically, the base confidence heuristic is so pessimistic that almost every error was already escaping HITL — so the policy's gain looks dramatic. A larger model with cleaner confidence calibration would benefit less. The honest sample size here is 27 (field, doc) pairs; do not extrapolate beyond this.

Online policy update (per-review, in-process)

The PolicyCache watches storage.mark_reviewed() and incrementally folds each new review into the in-memory policy, with debounced atomic disk flushes. The very next Pipeline.run() sees the updated override — no restart, no separate CLI invocation.

from idp.storage import make_storage
from idp.rl import PolicyCache

storage = make_storage("sql", db_url="sqlite:///./idp.db")
cache = PolicyCache(policy_path="policy.json", flush_interval_sec=1.0)
cache.attach_to_storage(storage)   # patches mark_reviewed to fire on_review

# From now on, every human review edits the policy in the background.

Defaults: flush_interval_sec=1.0 (debounce window), min_reviews=10 (the small-sample guard — fields with fewer than 10 total observations get no override regardless of fail rate, because fail_rate estimates are too noisy at n<10).

Multi-process: only one process should hold the cache (e.g. the FastAPI server). Other processes (CLI tools, the Streamlit reviewer UI) read policy.json from disk. The cache uses os.replace for atomic writes, so a crash mid-flush leaves the previous policy intact.

Real HITL data collection

The SqlStorage backend persists everything JsonFileStorage does plus per-field edit history in a real relational database. SQLite works out-of-the-box (zero extra deps); Postgres is opt-in via pip install py-idp[sql].

# SQLite, single-file
export IDP_DB_URL="sqlite:///./idp.db"
idp serve                                  # Streamlit UI now reads/writes this DB
idp rl-update --db-url "sqlite:///./idp.db" --output policy.json
idp rl-eval  --db-url "sqlite:///./idp.db" --policy policy.json \
             --output calibration.json

Schema (4 tables): reviewers, stored_results (denormalised cache of latest review state), reviews (one row per review session), review_edits (one row per field-level diff). The split lets you compute per-reviewer agreement, per-field edit rate over time, and "did the policy flag this and the human agreed it was wrong" without scanning full result blobs.

Why the split matters: review_edits is the granular signal the RL layer consumes (one row per corrected field). Without it, you can't tell which field in a multi-field review the human changed.

Development

git clone https://github.com/rollroyces/py-idp
cd py-idp
python -m venv .venv && source .venv/bin/activate
pip install -e ".[dev]"

pytest -v                       # 654 tests, no API key needed
ruff check src tests examples   # lint
mypy src/idp                    # type-check (clean across 61 files)

python -m examples.invoice      # end-to-end demo (no API key needed)
python -m examples.nanonets_ocr2  # NanonetsVLBackend end-to-end (needs IDP_ENABLE_NANONETS=1)
python -m examples.batch        # process_batch() helper for Databricks-style batches
python -m examples.discover_schema_sample  # AI-driven schema discovery (6 scenarios, generates a real PDF)

import idp; idp.__version__ → 0.3.8.

Security

Found a vulnerability? See docs/SECURITY.md — please do not file it as a public issue.

Citing

If py-idp helped your research or product, the academic citation lives in CITATION.cff. The BibTeX export is one click on the GitHub sidebar ("Cite this repository").

Contributing

Issues, PRs, and Discussions are welcome. The full guide — including how to add a new LLM backend or schema, commit-message conventions, and the release flow — lives in CONTRIBUTING.md. Bug reports do best with a minimal reproduction script and your py-idp version. CI runs ruff + mypy + 654 tests across Python 3.10 / 3.11 / 3.12 on every PR.

License

py-idp is dual-licensed:

AGPL-3.0-or-later — for open-source use. You may use, modify, and run py-idp freely. Modifications deployed as a network-accessible service must also be published under AGPL. This is the copyleft that prevents competitors from cloning the work into a SaaS without contributing back. See LICENSE-AGPL.
Commercial License — for organisations that need to embed py-idp in proprietary products or hosted SaaS without the AGPL copyleft. See LICENSE-COMMERCIAL.

This mirrors the MariaDB / Sentry / MinIO model: pay for the convenience of running in a closed product; get the full source for free if you keep your changes open.

Indicative commercial pricing:

tier	use case	pricing
Solo	single developer, single legal entity	$300 / yr
Team	up to 10 developers, single entity	$1,500 / yr
Enterprise	unlimited developers + SLA + support	contact
SaaS-OEM	embed in a hosted SaaS, per active user	per-seat

Contact Royce Lam (roycelam@umich.edu) for a signed agreement.

Named-entity carve-out. Chinachem Group Holdings Limited (HK) and its covered subsidiaries hold a pre-paid commercial license on Enterprise terms (fee waived, revocable on material breach or change of control). See §9 of LICENSE-COMMERCIAL. This is a one-off business arrangement — it does not extend to any other entity.

Acknowledgments

Pipeline shape, HITL, confidence design — extended from aws-solutions-library-samples/accelerated-intelligent-document-processing-on-aws
PDF parsing / table extraction — wraps IBM Docling (arXiv 2408.09869)
Pydantic-schema-driven extraction API — inspired by run-llama/llama_cloud_services
Multi-format chunking patterns — from Unstructured-IO/unstructured

If you cite py-idp in research, please cite this repo and Docling.

Maintainer

Royce Lam · @rollroyces · roycelam@umich.edu

Name		Name	Last commit message	Last commit date
.githooks		.githooks
.github		.github
docs		docs
examples		examples
idp_data		idp_data
scripts		scripts
src/idp		src/idp
templates		templates
tests		tests
.gitignore		.gitignore
.pre-commit-config.yaml		.pre-commit-config.yaml
CHANGELOG.md		CHANGELOG.md
CITATION.cff		CITATION.cff
CONTRIBUTING.md		CONTRIBUTING.md
Dockerfile		Dockerfile
LICENSE		LICENSE
LICENSE-AGPL		LICENSE-AGPL
LICENSE-COMMERCIAL		LICENSE-COMMERCIAL
README.md		README.md
README.zh-CN.md		README.zh-CN.md
README.zh-TW.md		README.zh-TW.md
docker-compose.yml		docker-compose.yml
mkdocs.yml		mkdocs.yml
pyproject.toml		pyproject.toml
requirements.txt.lock		requirements.txt.lock

	hand-rolled	py-idp
PDF parsing + table extraction	your problem	Docling, pdfplumber, plain — pick one
Multi-page LLM calls	write a chunker	built-in, with per-chunk failure handling
Confidence scoring	your problem	heuristic + optional LLM self-rate
Schema validation	your problem	Pydantic-typed, errors surfaced in `doc.errors`
HITL review	your problem	Streamlit app, persists corrections
Reliability	your problem	retry, cache, checkpoint built-in
Schema discovery	your problem	`discover_schema()` from a hint + a sample doc
Production eval	your problem	`idp eval` with field F1 + schema-valid rate
RL on corrections	your problem	`idp rl-update` — deterministic, inspectable
12+ LLM backends	your problem	one `Pipeline(backend="...")` line
Dual license (AGPL + commercial)	your problem	revenue path without losing the open-source core

Latest commit

History

Folders and files

Repository files navigation

py-idp

Contents

Install

30-second tour

The pipeline

Architecture at a glance

LLM backends

International (5 providers, any OpenAI-compat endpoint)

China (8 providers — all speak the OpenAI Chat-Completions protocol)

Self-hosted (Nanonets-OCR2-3B on Apple Silicon / CUDA)

Auto-chunking for oversized documents

CLI

Batch processing (100s–1000s of documents)

Bring your own schema

Auto-schema discovery

Templates per PDF type (LLM context via Markdown)

Chunking for oversized documents

Add business rules

Demo

Eval harness

CORD-style receipt benchmark

Reliability: retries, cache, checkpoint

RetryingBackend — automatic retries with backoff

ExtractionCache — disk-backed dedup

CheckpointStore — batch resume

Production scaffolding (built in, optional)

Not in 0.3.x (deliberately)

Learning from HITL corrections (RL)

Lifecycle of a single document

Calibration eval — does the policy actually do what it claims?

Online policy update (per-review, in-process)

Real HITL data collection

Development

Security

Citing

Contributing

License

Acknowledgments

Maintainer

About

Topics

Resources

Contributing

Security policy

Stars

Watchers

Forks

Releases

Packages

Contributors

Languages