API reference¶

This page is auto-generated from the source code's docstrings via mkdocstrings. If you're reading the rendered site, the symbols below expand on click.

Top-level¶

idp ¶

py-idp: General-purpose, AI-enabled Intelligent Document Processing framework.

A six-stage pipeline: parse -> classify -> extract -> assess -> validate -> HITL. Each stage is a pure function over a Document, pluggable, and independently testable.

Design draws from

aws-solutions-library-samples/accelerated-intelligent-document-processing-on-aws (pipeline shape, HITL, confidence assessment)
docling-project/docling (parser: PDF, tables, reading order)
run-llama/llama_cloud_services (Pydantic-schema-driven extraction API)
Unstructured-IO/unstructured (chunking + multi-format ingest)

Document `dataclass` ¶

A document flowing through the IDP pipeline.

Stages mutate this in-place by setting the corresponding attribute (parsed_pages, classification, extraction, confidence, validation).

Source code in src/idp/core/document.py

@dataclass
class Document:
    """A document flowing through the IDP pipeline.

    Stages mutate this in-place by setting the corresponding attribute
    (parsed_pages, classification, extraction, confidence, validation).
    """

    source_path: str
    doc_id: str
    pages: list[Page] = field(default_factory=list)
    raw_text: str = ""
    # populated by stages
    parsed_pages: list[dict[str, Any]] | None = None
    classification: str | None = None
    classification_confidence: float | None = None
    extraction: dict[str, Any] | None = None
    extraction_schema: str | None = None
    confidence: dict[str, float] | None = None
    validation: dict[str, Any] | None = None
    mode: str | None = None  # 'multimodal' | 'ocr_llm' (chosen by router)
    # Template provenance: which template (if any) was used to produce
    # this extraction. Recorded so audit / batch re-runs can answer
    # "which template version produced this row?" without re-running.
    template_name: str | None = None
    template_version: int | None = None
    metadata: dict[str, Any] = field(default_factory=dict)
    errors: list[str] = field(default_factory=list)

    @classmethod
    def from_path(cls, path: str | Path) -> Document:
        p = Path(path)
        if not p.exists():
            raise FileNotFoundError(f"No such file: {path}")
        if p.is_dir():
            # Reject directories instead of producing a silent empty document.
            # Callers that batch a list of paths would otherwise get 0 results
            # with no error to debug.
            raise IsADirectoryError(
                f"from_path() expects a file, got a directory: {path}"
            )
        if not p.is_file():
            raise OSError(f"Not a regular file: {path}")
        mt, _ = mimetypes.guess_type(str(p))
        h = hashlib.sha1(str(p.resolve()).encode()).hexdigest()[:16]
        return cls(
            source_path=str(p),
            doc_id=f"{p.stem}-{h}",
            metadata={"size": p.stat().st_size, "mime": mt or "application/octet-stream"},
        )

    @property
    def page_count(self) -> int:
        return len(self.pages) or len(self.parsed_pages or [])

    @property
    def extension(self) -> str:
        return Path(self.source_path).suffix.lower().lstrip(".")

    def parser_used(self) -> str:
        """Convenience: which parser wrote this Document (or '?' if not yet parsed)."""
        return (self.metadata or {}).get("parser", "?")

parser_used ¶

parser_used() -> str

Convenience: which parser wrote this Document (or '?' if not yet parsed).

Source code in src/idp/core/document.py

def parser_used(self) -> str:
    """Convenience: which parser wrote this Document (or '?' if not yet parsed)."""
    return (self.metadata or {}).get("parser", "?")

Pipeline ¶

Compose the six stages. Each stage can be skipped via flags.

Template support: pass a :class:idp.templates.Template (or a string name resolved against a registry) and the template's Markdown body is prepended to every LLM extraction call as document-type-specific guidance. See idp/templates.py.

Source code in src/idp/pipeline/pipeline.py

class Pipeline:
    """Compose the six stages. Each stage can be skipped via flags.

    Template support: pass a :class:`idp.templates.Template` (or a
    string name resolved against a registry) and the template's
    Markdown body is prepended to every LLM extraction call as
    document-type-specific guidance. See ``idp/templates.py``.
    """

    def __init__(
        self,
        backend: Backend | str = "auto",
        schema: str | type[BaseModel] = "Invoice",
        parser: str | Parser | None = "auto",
        use_llm_confidence: bool = False,
        business_rules: list[BusinessRule] | None = None,
        policy: PolicyConfig | None = None,
        policy_path: str | None = None,
        retry: RetryConfig | bool = False,
        cache: ExtractionCache | bool = False,
        template: Any = None,  # idp.templates.Template | str | None
    ):
        # Resolve backend to a real object first, then optionally wrap it
        # with retry + cache. We wrap lazily (not at construction) so the
        # caller can still see the original backend for debugging, but the
        # wrapped one is what gets called inside the pipeline.
        if isinstance(backend, str):
            backend = get_backend(backend)
        # Save the "logical" name (before wrapping) for cache keying
        _logical_backend_name = getattr(backend, "name", "unknown")
        # Retry wrapper
        if retry is True:
            retry = RetryConfig()
        if isinstance(retry, RetryConfig):
            backend = RetryingBackend(backend, retry)  # type: ignore[assignment]
        # Cache wrapper (applied AFTER retry so cached hits skip retries)
        if cache is True:
            cache = ExtractionCache(
                str(Path.home() / ".cache" / "idp" / "extract.db")
            )
        if isinstance(cache, ExtractionCache):
            # Need schema_name; resolve it first by getting the schema
            pass  # done after self.schema is set
        self.backend: Backend = backend  # type: ignore[assignment]
        self.backend_name = getattr(backend, "name", _logical_backend_name)
        if isinstance(schema, str):
            from idp.core.schemas import get_schema

            self.schema = get_schema(schema)
            self.schema_name = schema
        else:
            self.schema = schema
            self.schema_name = schema.__name__
        # Now that schema_name is known, wrap with cache if requested.
        # We re-wrap because CachingBackend requires schema_name.
        if isinstance(cache, ExtractionCache):
            self.backend = CachingBackend(
                backend, cache, schema_name=self.schema_name
            )  # type: ignore[assignment]
            self.backend_name = self.backend.name
        self.cache = cache if isinstance(cache, ExtractionCache) else None
        self.parser = parser  # resolved lazily inside run() once we have a Document
        self.parser_name = parser if isinstance(parser, str) else None
        self.use_llm_confidence = use_llm_confidence
        self.business_rules = business_rules or []
        # Template: accept either a Template object, a string name
        # (loaded from the in-process registry if set via
        # ``Pipeline.set_template_registry``), or None for no template.
        # The body is read at run() time, not at construction, so the
        # template can be edited and hot-reloaded between calls.
        self._template_obj: Any = None
        self._template_name: str | None = None
        self._template_body: str = ""
        self._template_version: int | None = None
        self._template_registry: Any = None
        if template is not None:
            if isinstance(template, str):
                self._template_name = template
            else:
                # Assume it's a Template-shaped object
                self._template_obj = template
                self._template_name = template.name
                self._template_body = template.body
                self._template_version = template.version
        # Policy: load from path if given, else use the passed object
        self.policy = None
        if policy is not None:
            self.policy = policy
        elif policy_path is not None:
            try:
                from idp.rl.policy import PolicyConfig
                self.policy = PolicyConfig.load(policy_path)
            except Exception as e:  # noqa: BLE001
                log.warning("failed to load policy from %s: %s", policy_path, e)

    def _resolve_parser(self, doc: Document) -> Parser:
        # explicit parser object? take it as-is
        if self.parser is not None and not isinstance(self.parser, str):
            return self.parser
        # explicit name ('pdfplumber', 'docling', 'plain')? use legacy factory
        if isinstance(self.parser, str) and self.parser != "auto":
            return get_parser(self.parser)
        # None or "auto": pick by extension (the recommended path)
        from idp.parse.parser import _auto_pick

        return _auto_pick(doc)

    def run(self, doc: Document) -> PipelineResult:
        timings: list[StageTiming] = []

        # Resolve template body if a name was provided (not a Template obj).
        # The body is read at run() time, not at construction, so the
        # template registry can hot-reload changes between calls.
        template_name = self._template_name
        template_version = self._template_version
        template_body = self._template_body
        if template_name and not template_body and self._template_registry is not None:
            try:
                t = self._template_registry.get(template_name)
                template_body = t.body
                template_version = t.version
            except Exception:  # noqa: BLE001
                log.warning("template %r not found in registry; running without it", template_name)
                template_name = None

        # PARSE ------------------------------------------------------------
        t = time.perf_counter()
        parse_document(doc, parser=self._resolve_parser(doc))
        timings.append(StageTiming("parse", time.perf_counter() - t))

        # CLASSIFY ---------------------------------------------------------
        t = time.perf_counter()
        classify_document(doc, self.backend)
        timings.append(StageTiming("classify", time.perf_counter() - t))

        # ROUTE → EXTRACT --------------------------------------------------
        t = time.perf_counter()
        mode = choose_mode(doc, backend_is_multimodal=self.backend.is_multimodal)
        timings.append(StageTiming("route", time.perf_counter() - t, {"chosen": mode.value}))

        t = time.perf_counter()
        extract(doc, self.schema, self.backend, mode=mode,
                template_body=template_body,
                template_name=template_name,
                template_version=template_version)
        timings.append(StageTiming("extract", time.perf_counter() - t, {"mode": mode.value}))

        # ASSESS -----------------------------------------------------------
        t = time.perf_counter()
        assess_confidence(
            doc, backend=self.backend, use_llm=self.use_llm_confidence, policy=self.policy,
        )
        timings.append(StageTiming("assess", time.perf_counter() - t))

        # VALIDATE ---------------------------------------------------------
        t = time.perf_counter()
        validate(doc, rules=self.business_rules)
        timings.append(StageTiming("validate", time.perf_counter() - t))

        return PipelineResult(
            document=doc,
            schema_name=self.schema_name,
            timings=timings,
            backend_name=self.backend_name,
            mode=mode.value,
            classification=doc.classification,
            confidence=doc.confidence,
            validation_passed=bool((doc.validation or {}).get("passed", False)),
            template_name=template_name,
            template_version=template_version,
        )

    def set_template_registry(self, registry: Any) -> None:
        """Attach a TemplateRegistry for resolving string template names.

        Usage::

            from idp.templates import TemplateRegistry
            registry = TemplateRegistry.load("./templates")
            pipeline = Pipeline(backend="mock", template="invoice")
            pipeline.set_template_registry(registry)
            result = pipeline.run(doc)

        The registry is consulted at run() time, so calls benefit from
        hot-reload if ``watch=True`` was passed when loading.
        """
        self._template_registry = registry

set_template_registry ¶

set_template_registry(registry: Any) -> None

Attach a TemplateRegistry for resolving string template names.

Usage::

from idp.templates import TemplateRegistry
registry = TemplateRegistry.load("./templates")
pipeline = Pipeline(backend="mock", template="invoice")
pipeline.set_template_registry(registry)
result = pipeline.run(doc)

The registry is consulted at run() time, so calls benefit from hot-reload if watch=True was passed when loading.

Source code in src/idp/pipeline/pipeline.py

def set_template_registry(self, registry: Any) -> None:
    """Attach a TemplateRegistry for resolving string template names.

    Usage::

        from idp.templates import TemplateRegistry
        registry = TemplateRegistry.load("./templates")
        pipeline = Pipeline(backend="mock", template="invoice")
        pipeline.set_template_registry(registry)
        result = pipeline.run(doc)

    The registry is consulted at run() time, so calls benefit from
    hot-reload if ``watch=True`` was passed when loading.
    """
    self._template_registry = registry

discover_schema ¶

discover_schema(source: str | Path | Document, *, hint: str = _DEFAULT_HINT, backend: Backend | None = None, page_limit: int = 4) -> DiscoveryResult

Infer a Pydantic schema from a PDF + natural-language hint.

Parameters:

Name	Type	Description	Default
`source`	`str \| Path \| Document`	PDF path or a pre-parsed `Document`. If a Document without `pages[i].images_b64` is passed, this will run `PdfPagesParser` on it first.	required
`hint`	`str`	Natural-language description of fields to extract. Example: "extract vendor_name, total_amount, and line items". Pass an empty string to let the LLM infer fields from the document alone.	`_DEFAULT_HINT`
`backend`	`Backend \| None`	Multimodal backend to use. Defaults to NanonetsVLBackend (gated by `IDP_ENABLE_NANONETS=1`). Any backend with `is_multimodal == True` works.	`None`
`page_limit`	`int`	Max pages to render for the prompt. Default 4 keeps the prompt under most models' context budgets.	`4`

Returns:

Type	Description
`DiscoveryResult`	`DiscoveryResult` with the inferred `schema_class` and the
`DiscoveryResult`	raw `json_schema`. Pass `result.schema_class` to
`DiscoveryResult`	`Pipeline(schema=...)`.

Raises:

Type	Description
`FileNotFoundError`	if the source path doesn't exist.
`ValueError`	if the LLM's output isn't valid JSON Schema.
`RuntimeError`	if the backend fails.

Source code in src/idp/discover.py

def discover_schema(
    source: str | Path | Document,
    *,
    hint: str = _DEFAULT_HINT,
    backend: Backend | None = None,
    page_limit: int = 4,
) -> DiscoveryResult:
    """Infer a Pydantic schema from a PDF + natural-language hint.

    Args:
        source:    PDF path or a pre-parsed ``Document``. If a Document
                   without ``pages[i].images_b64`` is passed, this will
                   run ``PdfPagesParser`` on it first.
        hint:      Natural-language description of fields to extract.
                   Example: "extract vendor_name, total_amount, and line
                   items". Pass an empty string to let the LLM infer
                   fields from the document alone.
        backend:   Multimodal backend to use. Defaults to NanonetsVLBackend
                   (gated by ``IDP_ENABLE_NANONETS=1``). Any backend with
                   ``is_multimodal == True`` works.
        page_limit: Max pages to render for the prompt. Default 4 keeps
                    the prompt under most models' context budgets.

    Returns:
        ``DiscoveryResult`` with the inferred ``schema_class`` and the
        raw ``json_schema``. Pass ``result.schema_class`` to
        ``Pipeline(schema=...)``.

    Raises:
        FileNotFoundError: if the source path doesn't exist.
        ValueError:        if the LLM's output isn't valid JSON Schema.
        RuntimeError:      if the backend fails.
    """
    if backend is None:
        backend = make_backend("nanonets")

    if not backend.is_multimodal:
        raise ValueError(
            f"discover_schema needs a multimodal backend (got {type(backend).__name__} "
            f"with is_multimodal={backend.is_multimodal}). Use NanonetsVLBackend or "
            "another VLM-capable backend."
        )

    # 1. Get a Document with images_b64 populated on pages
    doc = _ensure_parsed(source, page_limit=page_limit)

    # 2. Build the prompt
    images = [img for p in doc.pages for img in (p.images_b64 or [])][:page_limit]
    source_path = str(doc.source_path)
    user_content = _USER_PROMPT_TEMPLATE.format(path=source_path, hint=hint or "(none)")

    messages = [
        Message(role="system", content=_SYSTEM_PROMPT),
        Message(role="user", content=user_content, images_b64=images),
    ]
    req = CompletionRequest(messages=messages, json_mode=True, temperature=0.0)

    # 3. Call the backend
    raw = backend.complete(req)
    log.info("discover_schema: backend returned %d chars", len(raw))

    # 4. Parse the JSON Schema
    schema_dict = _parse_json_schema(raw)

    # 5. Compile to Pydantic
    schema_class = _json_schema_to_pydantic(schema_dict, fallback_name="InferredDocument")

    # 6. Ground against user hint (mitigates the "LLM ignores your hint"
    #    limitation). When hint is empty, skip — no grounding possible.
    grounding = _compute_hint_grounding(hint, schema_dict)

    # Surface low-grounding as a warning, not an error. The schema
    # might still be correct (LLM chose better names); the user just
    # needs to know to verify.
    if grounding is not None and grounding["grounding_score"] < 0.5:
        log.warning(
            "discover_schema: only %d of %d hint tokens appear in the "
            "discovered schema. The LLM may have ignored your hint. "
            "Hint tokens not found: %s",
            len(grounding["grounded"]),
            len(grounding["hint_tokens"]),
            grounding["ungrounded"],
        )

    return DiscoveryResult(
        schema_class=schema_class,
        json_schema=schema_dict,
        raw_response=raw,
        backend_name=getattr(backend, "name", type(backend).__name__),
        doc=doc,
        hint_grounding=grounding,
    )

Pipeline¶

pipeline ¶

Pipeline orchestrator.

Composes the six stages: parse -> classify -> extract -> assess -> validate.

Each stage is a pure function that mutates a Document. The pipeline adds structured logging and timing, and returns a PipelineResult with both the document and a per-stage timing/cost breakdown.

Pipeline ¶

Compose the six stages. Each stage can be skipped via flags.

Template support: pass a :class:idp.templates.Template (or a string name resolved against a registry) and the template's Markdown body is prepended to every LLM extraction call as document-type-specific guidance. See idp/templates.py.

Source code in src/idp/pipeline/pipeline.py

class Pipeline:
    """Compose the six stages. Each stage can be skipped via flags.

    Template support: pass a :class:`idp.templates.Template` (or a
    string name resolved against a registry) and the template's
    Markdown body is prepended to every LLM extraction call as
    document-type-specific guidance. See ``idp/templates.py``.
    """

    def __init__(
        self,
        backend: Backend | str = "auto",
        schema: str | type[BaseModel] = "Invoice",
        parser: str | Parser | None = "auto",
        use_llm_confidence: bool = False,
        business_rules: list[BusinessRule] | None = None,
        policy: PolicyConfig | None = None,
        policy_path: str | None = None,
        retry: RetryConfig | bool = False,
        cache: ExtractionCache | bool = False,
        template: Any = None,  # idp.templates.Template | str | None
    ):
        # Resolve backend to a real object first, then optionally wrap it
        # with retry + cache. We wrap lazily (not at construction) so the
        # caller can still see the original backend for debugging, but the
        # wrapped one is what gets called inside the pipeline.
        if isinstance(backend, str):
            backend = get_backend(backend)
        # Save the "logical" name (before wrapping) for cache keying
        _logical_backend_name = getattr(backend, "name", "unknown")
        # Retry wrapper
        if retry is True:
            retry = RetryConfig()
        if isinstance(retry, RetryConfig):
            backend = RetryingBackend(backend, retry)  # type: ignore[assignment]
        # Cache wrapper (applied AFTER retry so cached hits skip retries)
        if cache is True:
            cache = ExtractionCache(
                str(Path.home() / ".cache" / "idp" / "extract.db")
            )
        if isinstance(cache, ExtractionCache):
            # Need schema_name; resolve it first by getting the schema
            pass  # done after self.schema is set
        self.backend: Backend = backend  # type: ignore[assignment]
        self.backend_name = getattr(backend, "name", _logical_backend_name)
        if isinstance(schema, str):
            from idp.core.schemas import get_schema

            self.schema = get_schema(schema)
            self.schema_name = schema
        else:
            self.schema = schema
            self.schema_name = schema.__name__
        # Now that schema_name is known, wrap with cache if requested.
        # We re-wrap because CachingBackend requires schema_name.
        if isinstance(cache, ExtractionCache):
            self.backend = CachingBackend(
                backend, cache, schema_name=self.schema_name
            )  # type: ignore[assignment]
            self.backend_name = self.backend.name
        self.cache = cache if isinstance(cache, ExtractionCache) else None
        self.parser = parser  # resolved lazily inside run() once we have a Document
        self.parser_name = parser if isinstance(parser, str) else None
        self.use_llm_confidence = use_llm_confidence
        self.business_rules = business_rules or []
        # Template: accept either a Template object, a string name
        # (loaded from the in-process registry if set via
        # ``Pipeline.set_template_registry``), or None for no template.
        # The body is read at run() time, not at construction, so the
        # template can be edited and hot-reloaded between calls.
        self._template_obj: Any = None
        self._template_name: str | None = None
        self._template_body: str = ""
        self._template_version: int | None = None
        self._template_registry: Any = None
        if template is not None:
            if isinstance(template, str):
                self._template_name = template
            else:
                # Assume it's a Template-shaped object
                self._template_obj = template
                self._template_name = template.name
                self._template_body = template.body
                self._template_version = template.version
        # Policy: load from path if given, else use the passed object
        self.policy = None
        if policy is not None:
            self.policy = policy
        elif policy_path is not None:
            try:
                from idp.rl.policy import PolicyConfig
                self.policy = PolicyConfig.load(policy_path)
            except Exception as e:  # noqa: BLE001
                log.warning("failed to load policy from %s: %s", policy_path, e)

    def _resolve_parser(self, doc: Document) -> Parser:
        # explicit parser object? take it as-is
        if self.parser is not None and not isinstance(self.parser, str):
            return self.parser
        # explicit name ('pdfplumber', 'docling', 'plain')? use legacy factory
        if isinstance(self.parser, str) and self.parser != "auto":
            return get_parser(self.parser)
        # None or "auto": pick by extension (the recommended path)
        from idp.parse.parser import _auto_pick

        return _auto_pick(doc)

    def run(self, doc: Document) -> PipelineResult:
        timings: list[StageTiming] = []

        # Resolve template body if a name was provided (not a Template obj).
        # The body is read at run() time, not at construction, so the
        # template registry can hot-reload changes between calls.
        template_name = self._template_name
        template_version = self._template_version
        template_body = self._template_body
        if template_name and not template_body and self._template_registry is not None:
            try:
                t = self._template_registry.get(template_name)
                template_body = t.body
                template_version = t.version
            except Exception:  # noqa: BLE001
                log.warning("template %r not found in registry; running without it", template_name)
                template_name = None

        # PARSE ------------------------------------------------------------
        t = time.perf_counter()
        parse_document(doc, parser=self._resolve_parser(doc))
        timings.append(StageTiming("parse", time.perf_counter() - t))

        # CLASSIFY ---------------------------------------------------------
        t = time.perf_counter()
        classify_document(doc, self.backend)
        timings.append(StageTiming("classify", time.perf_counter() - t))

        # ROUTE → EXTRACT --------------------------------------------------
        t = time.perf_counter()
        mode = choose_mode(doc, backend_is_multimodal=self.backend.is_multimodal)
        timings.append(StageTiming("route", time.perf_counter() - t, {"chosen": mode.value}))

        t = time.perf_counter()
        extract(doc, self.schema, self.backend, mode=mode,
                template_body=template_body,
                template_name=template_name,
                template_version=template_version)
        timings.append(StageTiming("extract", time.perf_counter() - t, {"mode": mode.value}))

        # ASSESS -----------------------------------------------------------
        t = time.perf_counter()
        assess_confidence(
            doc, backend=self.backend, use_llm=self.use_llm_confidence, policy=self.policy,
        )
        timings.append(StageTiming("assess", time.perf_counter() - t))

        # VALIDATE ---------------------------------------------------------
        t = time.perf_counter()
        validate(doc, rules=self.business_rules)
        timings.append(StageTiming("validate", time.perf_counter() - t))

        return PipelineResult(
            document=doc,
            schema_name=self.schema_name,
            timings=timings,
            backend_name=self.backend_name,
            mode=mode.value,
            classification=doc.classification,
            confidence=doc.confidence,
            validation_passed=bool((doc.validation or {}).get("passed", False)),
            template_name=template_name,
            template_version=template_version,
        )

    def set_template_registry(self, registry: Any) -> None:
        """Attach a TemplateRegistry for resolving string template names.

        Usage::

            from idp.templates import TemplateRegistry
            registry = TemplateRegistry.load("./templates")
            pipeline = Pipeline(backend="mock", template="invoice")
            pipeline.set_template_registry(registry)
            result = pipeline.run(doc)

        The registry is consulted at run() time, so calls benefit from
        hot-reload if ``watch=True`` was passed when loading.
        """
        self._template_registry = registry

set_template_registry ¶

set_template_registry(registry: Any) -> None

Attach a TemplateRegistry for resolving string template names.

Usage::

from idp.templates import TemplateRegistry
registry = TemplateRegistry.load("./templates")
pipeline = Pipeline(backend="mock", template="invoice")
pipeline.set_template_registry(registry)
result = pipeline.run(doc)

The registry is consulted at run() time, so calls benefit from hot-reload if watch=True was passed when loading.

Source code in src/idp/pipeline/pipeline.py

def set_template_registry(self, registry: Any) -> None:
    """Attach a TemplateRegistry for resolving string template names.

    Usage::

        from idp.templates import TemplateRegistry
        registry = TemplateRegistry.load("./templates")
        pipeline = Pipeline(backend="mock", template="invoice")
        pipeline.set_template_registry(registry)
        result = pipeline.run(doc)

    The registry is consulted at run() time, so calls benefit from
    hot-reload if ``watch=True`` was passed when loading.
    """
    self._template_registry = registry

PipelineResult `dataclass` ¶

Source code in src/idp/pipeline/pipeline.py

@dataclass
class PipelineResult:
    document: Document
    schema_name: str
    timings: list[StageTiming]
    backend_name: str
    mode: str | None
    classification: str | None
    confidence: dict[str, float] | None
    validation_passed: bool
    template_name: str | None = None
    template_version: int | None = None

    def to_dict(self) -> dict[str, Any]:
        return {
            "doc_id": self.document.doc_id,
            "source_path": self.document.source_path,
            "schema": self.schema_name,
            "backend": self.backend_name,
            "mode": self.mode,
            "classification": self.classification,
            "extraction": self.document.extraction,
            "confidence": self.confidence,
            "validation": self.document.validation,
            "template_name": self.template_name,
            "template_version": self.template_version,
            "errors": self.document.errors,
            "timings": [{"name": t.name, "seconds": t.seconds, **t.extra} for t in self.timings],
        }

save_result ¶

save_result(result: PipelineResult, output_path: str | Path) -> None

Source code in src/idp/pipeline/pipeline.py

def save_result(result: PipelineResult, output_path: str | Path) -> None:
    Path(output_path).write_text(json.dumps(result.to_dict(), indent=2, default=str))

Backends¶

backend ¶

LLM backend abstraction.

All stages call Backend.complete() and get back a string. Backends handle their own deps lazily (openai/anthropic/ollama) so the framework runs without any of them installed.

A MockBackend ships for tests + reproducible eval, when the user has no API keys / hardware.

Backend ¶

Bases: ABC

Pluggable LLM backend.

Source code in src/idp/llm/backend.py

class Backend(abc.ABC):
    """Pluggable LLM backend."""

    name: str = "base"

    @abc.abstractmethod
    def complete(self, req: CompletionRequest) -> str: ...

    # Convenience helpers ------------------------------------------------
    def json_complete(self, messages: list[Message], **kw: Any) -> dict[str, Any]:
        """Complete + parse JSON. Never raises.

        On parse failure returns `{"_error": ..., "_raw": ...}` so the
        caller can decide whether the empty dict is acceptable or whether
        it should route to an error path.
        """
        req = CompletionRequest(messages=messages, json_mode=True, **kw)
        out = self.complete(req)
        return _safe_json(out)

    @property
    def is_multimodal(self) -> bool:
        return False  # overridden by multimodal-capable backends

json_complete ¶

json_complete(messages: list[Message], **kw: Any) -> dict[str, Any]

Complete + parse JSON. Never raises.

On parse failure returns {"_error": ..., "_raw": ...} so the caller can decide whether the empty dict is acceptable or whether it should route to an error path.

Source code in src/idp/llm/backend.py

def json_complete(self, messages: list[Message], **kw: Any) -> dict[str, Any]:
    """Complete + parse JSON. Never raises.

    On parse failure returns `{"_error": ..., "_raw": ...}` so the
    caller can decide whether the empty dict is acceptable or whether
    it should route to an error path.
    """
    req = CompletionRequest(messages=messages, json_mode=True, **kw)
    out = self.complete(req)
    return _safe_json(out)

get_backend ¶

get_backend(name: str = 'auto', **kwargs: Any) -> Backend

Resolve a backend by name.

'auto' picks in this order: env override -> anthropic -> openai -> ollama -> mock.

Recognized names

For China providers you can pass multimodal=True to switch to the provider's vision model if it has one. Pass api_key= or set the provider's env var.

The slowmock backend is only registered when IDP_ENABLE_SLOWMOCK=1 is set in the environment. Production deployments never set this; it's intended for load-testing only.

Source code in src/idp/llm/backend.py

def get_backend(name: str = "auto", **kwargs: Any) -> Backend:
    """Resolve a backend by name.

    'auto' picks in this order: env override -> anthropic -> openai -> ollama -> mock.

    Recognized names:
        mock | mock-ideal | mock-random | mock-omits   (no API key needed)
        openai | ollama | vllm | compat | anthropic     (international)
        china:deepseek | china:qwen | china:zhipu |     (China providers)
        china:moonshot | china:yi | china:doubao |
        china:hunyuan | china:baichuan
        slowmock                                            (load-test only)

    For China providers you can pass `multimodal=True` to switch to the
    provider's vision model if it has one. Pass `api_key=` or set the
    provider's env var.

    The ``slowmock`` backend is only registered when ``IDP_ENABLE_SLOWMOCK=1``
    is set in the environment. Production deployments never set this; it's
    intended for load-testing only.
    """
    if not name:
        name = "auto"
    if name == "auto":
        name = os.environ.get("IDP_BACKEND", "auto")
    if name == "auto":
        # Priority order:
        #   1. Explicitly configured via env var (handled above)
        #   2. anthropic (if ANTHROPIC_API_KEY set)
        #   3. openai (if OPENAI_API_KEY set)
        #   4. ollama (if running locally)
        #   5. mock (last resort: no API key, no ollama — must work offline)
        #
        # The mock fallback ensures first-time users can run the pipeline
        # without any API keys. To force a real backend, set IDP_BACKEND
        # explicitly or the matching *_API_KEY env var.
        if os.environ.get("ANTHROPIC_API_KEY"):
            name = "anthropic"
        elif os.environ.get("OPENAI_API_KEY"):
            name = "openai"
        elif os.environ.get("OLLAMA_HOST") or os.environ.get("IDP_OLLAMA"):
            name = "ollama"
        else:
            name = "mock"
            import logging
            logging.getLogger("idp.llm.backend").info(
                "no LLM credentials found; falling back to 'mock' backend. "
                "Set IDP_BACKEND=openai|anthropic|ollama|... to use a real LLM."
            )
    # Mock variants
    if name in ("mock", "mock-ideal", "mock-random", "mock-omits"):
        mode = name.split("-", 1)[1] if "-" in name else "ideal"
        return MockBackend(mode=mode)
    # SlowMock for load testing — only enabled via env var
    if name == "slowmock":
        if os.environ.get("IDP_ENABLE_SLOWMOCK") != "1":
            raise ValueError(
                "slowmock backend is disabled in production. "
                "Set IDP_ENABLE_SLOWMOCK=1 to enable it (load-test only)."
            )
        from idp._testing_backends import SlowMockBackend as _SlowMockBackend
        return _SlowMockBackend(
            latency_ms=int(os.environ.get("LOAD_LATENCY_MS", "1500")),
            jitter_ms=int(os.environ.get("LOAD_JITTER_MS", "500")),
            **kwargs,
        )
    if name == "nanonets":
        # Gated like slowmock: requires explicit opt-in. Reason: the model
        # download is ~7 GB and the first call is slow. Production deployments
        # shouldn't accidentally trigger that.
        if os.environ.get("IDP_ENABLE_NANONETS") != "1":
            raise ValueError(
                "nanonets backend is gated. To enable:\n"
                "  pip install py-idp[hf-vlm]   # installs torch + transformers\n"
                "  export IDP_ENABLE_NANONETS=1\n"
                "  IDP_BACKEND=nanonets  # or: backend = NanonetsVLBackend()"
            )
        from idp.llm.nanonets import NanonetsVLBackend as _NanonetsVLBackend
        return _NanonetsVLBackend(**kwargs)
    # China providers: prefix with china:
    if name.startswith("china:"):
        from idp.llm.china import get_china_backend

        provider = name.split(":", 1)[1]
        multimodal = kwargs.pop("multimodal", False)
        return get_china_backend(
            provider,
            model=kwargs.pop("model", None),
            multimodal=multimodal,
            api_key=kwargs.pop("api_key", None),
            timeout=kwargs.pop("timeout", 120.0),
        )
    # International OpenAI-compat
    if name in ("openai", "compat", "ollama", "vllm", "lm-studio"):
        return OpenAICompatBackend(**kwargs)
    if name == "anthropic":
        return AnthropicBackend(**kwargs)
    raise ValueError(f"Unknown backend: {name}")

Schemas¶

schemas ¶

Built-in example schemas that ship with py-idp.

Users can supply their own Pydantic models; these are reference implementations.

SCHEMA_REGISTRY `module-attribute` ¶

SCHEMA_REGISTRY: dict[str, type[BaseModel]] = {'Invoice': Invoice, 'Contract': Contract, 'BankStatement': BankStatement, 'Receipt': Receipt}

Invoice ¶

Bases: BaseModel

Standard invoice extraction schema.

Minimal required fields chosen from real-world IDP requirements: an invoice that lacks an ID, vendor identity, or a total isn't an invoice we can route or pay. Everything else is optional because partial extractions from weaker LLMs are still useful.