py-idp

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.

Languages: English · 繁體中文 · 简体中文

Install

pip install py-idp                # core (pydantic + typer + rich + httpx + pdfplumber)
pip install py-idp[docling]      # IBM Docling — best PDF table extraction
pip install py-idp[openai]       # OpenAI SDK (also used for 8 China LLMs)
pip install py-idp[anthropic]    # Anthropic SDK
pip install py-idp[api]           # FastAPI server (idp.api:app — production-ready)
pip install py-idp[hf-vlm]        # Self-hosted Nanonets-OCR2-3B (Apple Silicon / CUDA)
pip install py-idp[dev]          # pytest + ruff + mypy

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).

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:

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

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.

The pipeline

INGEST  →  PARSE  →  CLASSIFY  →  ROUTE  →  EXTRACT  →  ASSESS  →  VALIDATE  →  HITL
                                                                         (Streamlit)

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

LLM backends

International (5 providers, any OpenAI-compat endpoint)

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:

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

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:

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

Bring your own schema

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

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=...).

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.

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.

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

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.

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.

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)

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.

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

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

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

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                       # 508 tests, no API key needed
ruff check src tests examples   # lint
mypy src/idp                    # type-check (clean across 59 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.2.

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 + 508 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:

Contact rollroyces for a signed agreement.

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.