Skip to content

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 
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
@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 
119
120
121
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 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
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,
        )

    async def arun(self, doc: Document) -> PipelineResult:
        """Async variant of ``run()`` — runs the blocking pipeline in a
        worker thread so the FastAPI event loop stays responsive.

        Without this, ``Pipeline.run()`` (which calls sync LLM clients and
        on Nanonets runs torch inference) blocks the event loop for
        5–15 s per request, freezing the entire server. ``arun`` uses
        ``asyncio.to_thread`` to dispatch the blocking work to the default
        thread pool, freeing the event loop to handle health checks,
        other concurrent requests, etc.

        Performance note: this trades parallelism-on-the-loop for one
        blocking operation per request. For sustained throughput you still
        want a separate worker process; ``arun`` is the minimum-viable
        fix to keep the HTTP tier responsive when Nanonets (or any
        synchronous backend) is the chosen LLM.
        """
        import asyncio
        return await asyncio.to_thread(self.run, doc)

    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

arun async 

arun(doc: Document) -> PipelineResult

Async variant of run() — runs the blocking pipeline in a worker thread so the FastAPI event loop stays responsive.

Without this, Pipeline.run() (which calls sync LLM clients and on Nanonets runs torch inference) blocks the event loop for 5–15 s per request, freezing the entire server. arun uses asyncio.to_thread to dispatch the blocking work to the default thread pool, freeing the event loop to handle health checks, other concurrent requests, etc.

Performance note: this trades parallelism-on-the-loop for one blocking operation per request. For sustained throughput you still want a separate worker process; arun is the minimum-viable fix to keep the HTTP tier responsive when Nanonets (or any synchronous backend) is the chosen LLM.

Source code in src/idp/pipeline/pipeline.py 
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
async def arun(self, doc: Document) -> PipelineResult:
    """Async variant of ``run()`` — runs the blocking pipeline in a
    worker thread so the FastAPI event loop stays responsive.

    Without this, ``Pipeline.run()`` (which calls sync LLM clients and
    on Nanonets runs torch inference) blocks the event loop for
    5–15 s per request, freezing the entire server. ``arun`` uses
    ``asyncio.to_thread`` to dispatch the blocking work to the default
    thread pool, freeing the event loop to handle health checks,
    other concurrent requests, etc.

    Performance note: this trades parallelism-on-the-loop for one
    blocking operation per request. For sustained throughput you still
    want a separate worker process; ``arun`` is the minimum-viable
    fix to keep the HTTP tier responsive when Nanonets (or any
    synchronous backend) is the chosen LLM.
    """
    import asyncio
    return await asyncio.to_thread(self.run, doc)

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 
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
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 
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
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 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
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,
        )

    async def arun(self, doc: Document) -> PipelineResult:
        """Async variant of ``run()`` — runs the blocking pipeline in a
        worker thread so the FastAPI event loop stays responsive.

        Without this, ``Pipeline.run()`` (which calls sync LLM clients and
        on Nanonets runs torch inference) blocks the event loop for
        5–15 s per request, freezing the entire server. ``arun`` uses
        ``asyncio.to_thread`` to dispatch the blocking work to the default
        thread pool, freeing the event loop to handle health checks,
        other concurrent requests, etc.

        Performance note: this trades parallelism-on-the-loop for one
        blocking operation per request. For sustained throughput you still
        want a separate worker process; ``arun`` is the minimum-viable
        fix to keep the HTTP tier responsive when Nanonets (or any
        synchronous backend) is the chosen LLM.
        """
        import asyncio
        return await asyncio.to_thread(self.run, doc)

    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

arun async 

arun(doc: Document) -> PipelineResult

Async variant of run() — runs the blocking pipeline in a worker thread so the FastAPI event loop stays responsive.

Without this, Pipeline.run() (which calls sync LLM clients and on Nanonets runs torch inference) blocks the event loop for 5–15 s per request, freezing the entire server. arun uses asyncio.to_thread to dispatch the blocking work to the default thread pool, freeing the event loop to handle health checks, other concurrent requests, etc.

Performance note: this trades parallelism-on-the-loop for one blocking operation per request. For sustained throughput you still want a separate worker process; arun is the minimum-viable fix to keep the HTTP tier responsive when Nanonets (or any synchronous backend) is the chosen LLM.

Source code in src/idp/pipeline/pipeline.py 
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
async def arun(self, doc: Document) -> PipelineResult:
    """Async variant of ``run()`` — runs the blocking pipeline in a
    worker thread so the FastAPI event loop stays responsive.

    Without this, ``Pipeline.run()`` (which calls sync LLM clients and
    on Nanonets runs torch inference) blocks the event loop for
    5–15 s per request, freezing the entire server. ``arun`` uses
    ``asyncio.to_thread`` to dispatch the blocking work to the default
    thread pool, freeing the event loop to handle health checks,
    other concurrent requests, etc.

    Performance note: this trades parallelism-on-the-loop for one
    blocking operation per request. For sustained throughput you still
    want a separate worker process; ``arun`` is the minimum-viable
    fix to keep the HTTP tier responsive when Nanonets (or any
    synchronous backend) is the chosen LLM.
    """
    import asyncio
    return await asyncio.to_thread(self.run, doc)

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 
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
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 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
@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 
314
315
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.

Adding a new backend

Subclass Backend and decorate it with register_backend("name"). name becomes the canonical lookup key for get_backend("name"), and the class is then listed by list_backends(). Aliases for backward compatibility go in _REGISTRY_ALIASES.

Backend

Bases: ABC

Pluggable LLM backend.

Backends implement :meth:complete (sync). For async use, backends also expose :meth:acomplete, which by default delegates to :meth:complete via :func:asyncio.to_thread so every existing backend transparently supports await. Backends with a native async HTTP client (e.g. openai.AsyncOpenAI, anthropic.AsyncAnthropic) can override :meth:acomplete for better performance.

Source code in src/idp/llm/backend.py 
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
class Backend(abc.ABC):
    """Pluggable LLM backend.

    Backends implement :meth:`complete` (sync). For async use, backends
    also expose :meth:`acomplete`, which by default delegates to
    :meth:`complete` via :func:`asyncio.to_thread` so every existing
    backend transparently supports ``await``. Backends with a native
    async HTTP client (e.g. ``openai.AsyncOpenAI``,
    ``anthropic.AsyncAnthropic``) can override :meth:`acomplete` for
    better performance.
    """

    name: str = "base"

    def __repr__(self) -> str:
        # Default repr. Subclasses with interesting state (model name,
        # base_url, API-key presence) override this for debuggability.
        cls = type(self).__name__
        return f"<{cls} name={self.name!r}>"

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

    async def acomplete(self, req: CompletionRequest) -> str:
        """Async version of :meth:`complete`.

        Default implementation runs the sync :meth:`complete` in a
        worker thread via :func:`asyncio.to_thread` so every existing
        backend automatically supports async. Backends with a native
        async client (anthropic, openai) can override this for better
        performance.
        """
        import asyncio
        return await asyncio.to_thread(self.complete, req)

    @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 
194
195
196
197
198
199
200
201
202
203
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)

acomplete async 

acomplete(req: CompletionRequest) -> str

Async version of :meth:complete.

Default implementation runs the sync :meth:complete in a worker thread via :func:asyncio.to_thread so every existing backend automatically supports async. Backends with a native async client (anthropic, openai) can override this for better performance.

Source code in src/idp/llm/backend.py 
205
206
207
208
209
210
211
212
213
214
215
async def acomplete(self, req: CompletionRequest) -> str:
    """Async version of :meth:`complete`.

    Default implementation runs the sync :meth:`complete` in a
    worker thread via :func:`asyncio.to_thread` so every existing
    backend automatically supports async. Backends with a native
    async client (anthropic, openai) can override this for better
    performance.
    """
    import asyncio
    return await asyncio.to_thread(self.complete, req)

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

mock | mock-ideal | mock-random | mock-omits (no API key needed) openai | ollama | vllm | compat | anthropic (international) intl:groq | intl:mistral | intl:together | (international providers) intl:cohere | intl:fireworks china:deepseek | china:qwen | china:zhipu | (China providers) china:moonshot | china:yi | china:doubao | china:hunyuan | china:baichuan nanonets (gated, self-hosted OCR) unlimited-ocr (gated, self-hosted OCR, v0.6) 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 enabled when IDP_ENABLE_SLOWMOCK=1 is set in the environment. Production deployments never set this; it's intended for load-testing only.

To register a new backend without editing this file, use @register_backend("name") (see module docstring).

Source code in src/idp/llm/backend.py 
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
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)
        intl:groq | intl:mistral | intl:together |      (international providers)
        intl:cohere | intl:fireworks
        china:deepseek | china:qwen | china:zhipu |     (China providers)
        china:moonshot | china:yi | china:doubao |
        china:hunyuan | china:baichuan
        nanonets                                            (gated, self-hosted OCR)
        unlimited-ocr                                       (gated, self-hosted OCR, v0.6)
        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 enabled when ``IDP_ENABLE_SLOWMOCK=1``
    is set in the environment. Production deployments never set this; it's
    intended for load-testing only.

    To register a new backend without editing this file, use
    ``@register_backend("name")`` (see module docstring).
    """
    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."
            )
    # Gated backends: slowmock (load-test only). Looked up by canonical
    # name, NOT in the registry (gating is enforced here so production
    # never accidentally enables them).
    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)
    if name == "unlimited-ocr":
        # Gated like nanonets. Same gate pattern (env var), same deps
        # (the [hf-vlm] extra covers both). Reuses the [hf-vlm] install
        # so no new pip dependencies.
        if os.environ.get("IDP_ENABLE_UNLIMITED_OCR") != "1":
            raise ValueError(
                "unlimited-ocr backend is gated. To enable:\n"
                "  pip install py-idp[hf-vlm]   # installs torch + transformers\n"
                "  export IDP_ENABLE_UNLIMITED_OCR=1\n"
                "  IDP_BACKEND=unlimited-ocr  # or: backend = UnlimitedOcrBackend()"
            )
        from idp.llm.unlimited_ocr import UnlimitedOcrBackend as _UnlimitedOcrBackend
        return _UnlimitedOcrBackend(**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 presets (groq, mistral, together, cohere,
    # fireworks). Resolved before the alias table because the provider list is
    # open-ended and cannot be enumerated as static aliases.
    if name.startswith("intl:"):
        from idp.llm.international import get_international_backend

        provider = name.split(":", 1)[1]
        multimodal = kwargs.pop("multimodal", False)
        return get_international_backend(
            provider,
            model=kwargs.pop("model", None),
            multimodal=multimodal,
            api_key=kwargs.pop("api_key", None),
            timeout=kwargs.pop("timeout", 120.0),
        )
    # Ollama is its own thing as of v0.4: it speaks Ollama's native
    # /api/chat shape (not OpenAI-compatible chat-completions), and
    # ``OllamaBackend`` talks to it directly over plain HTTP. The other
    # aliases (``openai``, ``compat``, ``vllm``, ``lm-studio``) all
    # DO speak OpenAI chat-completions so we keep them on the
    # OpenAICompatBackend. This must stay *above* the alias table, which
    # would otherwise rewrite "ollama" onto "openai-compat".
    if name == "ollama":
        from idp.llm.ollama import OllamaBackend as _OllamaBackend

        return _OllamaBackend(**kwargs)
    # Aliases -> canonical name (preserves backward compat for "openai",
    # "vllm", "lm-studio", "mock-ideal", "mock-random", ...).
    original_name = name
    name = _REGISTRY_ALIASES.get(name, name)
    # Registry lookup
    cls = _REGISTRY.get(name)
    if cls is None:
        known = sorted(set(_REGISTRY) | set(_REGISTRY_ALIASES))
        raise ValueError(
            f"Unknown backend: {original_name!r}. "
            f"Registered: {known}"
        )
    # Honor the legacy mock-* alias by selecting mode from the original alias
    # unless the caller already specified one.
    if name == "mock" and "mode" not in kwargs and original_name.startswith("mock-"):
        kwargs["mode"] = original_name.split("-", 1)[1]
    # OpenAI-compatible aliases need a sane default base_url, and the
    # local-server presets (vLLM / LM Studio) need an explicit opt-in to
    # the SSRF guard. That guard rejects loopback / private IPs by
    # default; these presets normally point at a server on the same
    # machine or LAN, so they opt in here. The operator can still
    # override either setting through the environment.
    if cls is OpenAICompatBackend:
        kwargs.setdefault(
            "base_url",
            "https://api.openai.com/v1"
            if original_name in ("openai", "compat", "openai-compat")
            else "http://localhost:11434/v1",
        )
        kwargs.setdefault(
            "_allow_private_network",
            original_name in ("vllm", "lm-studio"),
        )
    return cls(**kwargs)

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, 'W2': W2, 'IDCard': IDCard, 'MedicalClaim': MedicalClaim}

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.

Source code in src/idp/core/schemas.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
class Invoice(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.
    """

    # --- required for a usable invoice ---
    invoice_number: str
    vendor_name: str
    total_amount: float
    # --- everything else stays optional (partial extraction is fine) ---
    invoice_date: str | None = None
    due_date: str | None = None
    vendor_address: str | None = None
    customer_name: str | None = None
    customer_address: str | None = None
    subtotal: float | None = None
    tax_amount: float | None = None
    currency: str | None = None
    line_items: list[LineItem] = Field(default_factory=list)

Contract

Bases: BaseModel

Basic contract extraction schema.

Title is required: a contract without a title is unidentifiable. Parties and effective dates are optional — partial extraction still surfaces what was found for downstream HITL review.

Source code in src/idp/core/schemas.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
class Contract(BaseModel):
    """Basic contract extraction schema.

    Title is required: a contract without a title is unidentifiable.
    Parties and effective dates are optional — partial extraction still
    surfaces what was found for downstream HITL review.
    """

    title: str
    effective_date: str | None = None
    expiration_date: str | None = None
    parties: list[ContractParty] = Field(default_factory=list)
    governing_law: str | None = None
    total_value: float | None = None
    currency: str | None = None
    key_obligations: list[str] = Field(default_factory=list)

BankStatement

Bases: BaseModel

Bank statement extraction schema.

Account holder is required — a statement without an identified account holder is unactionable. Transactions and balances are optional and may be partially extracted.

Source code in src/idp/core/schemas.py 
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
class BankStatement(BaseModel):
    """Bank statement extraction schema.

    Account holder is required — a statement without an identified
    account holder is unactionable. Transactions and balances are
    optional and may be partially extracted.
    """

    account_holder: str
    account_number_last4: str | None = None
    statement_period_start: str | None = None
    statement_period_end: str | None = None
    opening_balance: float | None = None
    closing_balance: float | None = None
    transactions: list[BankTransaction] = Field(default_factory=list)

BankTransaction

Bases: BaseModel

Source code in src/idp/core/schemas.py 
77
78
79
80
81
class BankTransaction(BaseModel):
    date: str
    description: str
    amount: float
    balance: float | None = None

Receipt

Bases: BaseModel

Receipt extraction schema (CORD: Consolidated Receipt Dataset shape).

merchant_name is required (a receipt without an identified merchant is unactionable). Other fields are optional and may be partially extracted — handwritten or faded receipts often miss subtotals, tips, or tax.

Source code in src/idp/core/schemas.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
class Receipt(BaseModel):
    """Receipt extraction schema (CORD: Consolidated Receipt Dataset shape).

    `merchant_name` is required (a receipt without an identified merchant
    is unactionable). Other fields are optional and may be partially
    extracted — handwritten or faded receipts often miss subtotals, tips,
    or tax.
    """

    merchant_name: str
    date: str | None = None
    time: str | None = None
    line_items: list[ReceiptLineItem] = Field(default_factory=list)
    subtotal: float | None = None
    tax_amount: float | None = None
    tip_amount: float | None = None
    total: float | None = None
    payment_method: str | None = None
    credit_card_last4: str | None = None

LineItem

Bases: BaseModel

Source code in src/idp/core/schemas.py 
22
23
24
25
26
class LineItem(BaseModel):
    description: str
    quantity: float = 1.0
    unit_price: float
    total: float

ReceiptLineItem

Bases: BaseModel

One line on a receipt (CORD-style).

Source code in src/idp/core/schemas.py 
101
102
103
104
105
106
107
class ReceiptLineItem(BaseModel):
    """One line on a receipt (CORD-style)."""

    description: str
    quantity: float = 1.0
    unit_price: float
    total: float

HITL

store

Storage interface.

Keeps the framework decoupled from any specific database / object store. Default in-memory implementation supports tests + single-node demos. Swap in Postgres + S3 (or whatever) for production.

JsonFileStorage

Bases: Storage

Line-delimited JSON store on disk. Trivially inspectable, zero-deps.

Memory profile: keeps an in-memory cache of the file's parsed StoredResults. Cache is invalidated on put() and mark_reviewed(). For a file with N entries, peak memory is roughly 2-3× the on-disk JSON size (raw dicts + StoredResult objects + the cache dict itself).

For workloads with >10k stored results, switch to SqlStorage instead — JSONL doesn't index either and the full-file cache becomes the dominant cost.

Source code in src/idp/storage/store.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
class JsonFileStorage(Storage):
    """Line-delimited JSON store on disk. Trivially inspectable, zero-deps.

    Memory profile: keeps an in-memory cache of the file's parsed
    StoredResults. Cache is invalidated on ``put()`` and
    ``mark_reviewed()``. For a file with N entries, peak memory is
    roughly 2-3× the on-disk JSON size (raw dicts + StoredResult
    objects + the cache dict itself).

    For workloads with >10k stored results, switch to ``SqlStorage``
    instead — JSONL doesn't index either and the full-file cache
    becomes the dominant cost.
    """

    # Cache invalidation: byte-offset of last file size we read.
    # If the file has grown (someone wrote outside our lock), drop the cache.
    _CACHE_FILE_STALE = -1

    def __init__(self, path: str | Path) -> None:
        self.path = Path(path)
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self.path.touch(exist_ok=True)
        self._lock = threading.Lock()
        # In-memory cache: id -> StoredResult
        self._cache: dict[str, StoredResult] | None = None
        self._cache_size_bytes: int = self._CACHE_FILE_STALE

    def _cache_valid(self) -> bool:
        """True iff the cache exists AND the file hasn't changed on disk."""
        if self._cache is None or self._cache_size_bytes == self._CACHE_FILE_STALE:
            return False
        try:
            current_size = self.path.stat().st_size
        except FileNotFoundError:
            return False
        return current_size == self._cache_size_bytes

    def _invalidate_cache(self) -> None:
        self._cache = None
        self._cache_size_bytes = self._CACHE_FILE_STALE

    def _read_all(self) -> dict[str, StoredResult]:
        """Return cached entries if valid; otherwise re-read the file.

        Crashes during a `put()` (process kill, full disk) can leave a
        partial trailing line. Reading the whole file should never take
        down the app — we skip and continue. Use a real DB for transactions.
        """
        # Hot path: cache is valid -> return it (no I/O)
        if self._cache_valid():
            return self._cache  # type: ignore[return-value]

        # Cold path: re-read the file
        by_id: dict[str, StoredResult] = {}
        with self.path.open() as f:
            for line_no, raw in enumerate(f, start=1):
                s = raw.strip()
                if not s:
                    continue
                try:
                    d = json.loads(s)
                    by_id[d["id"]] = StoredResult(**d)
                except Exception as e:  # noqa: BLE001
                    log.warning(
                        "skipping corrupt line %d in %s: %s",
                        line_no, self.path, e,
                    )

        # Cache + remember file size at this point
        self._cache = by_id
        self._cache_size_bytes = self.path.stat().st_size
        return by_id

    def put(self, result: StoredResult) -> str:
        if not result.id:
            result.id = uuid.uuid4().hex[:16]
        if not result.created_at:
            result.created_at = time.time()
        try:
            with self._lock, self.path.open("a") as f:
                f.write(json.dumps(asdict(result), default=str) + "\n")
        except OSError as e:
            log.error("JsonFileStorage.put failed for %s: %s", self.path, e)
            raise StorageError(
                f"could not append to {self.path}: {e}",
            ) from e
        # Cache is now stale; drop it. The next read will rebuild.
        self._invalidate_cache()
        return result.id

    def get(self, result_id: str) -> StoredResult | None:
        return self._read_all().get(result_id)

    def list(
        self,
        doc_id: str | None = None,
        limit: int = 50,
        *,
        reviewed_only: bool = False,
        reviewed_since: float | None = None,
        schema_name: str | None = None,
    ) -> list[StoredResult]:
        all_results = sorted(self._read_all().values(), key=lambda r: -r.created_at)
        if doc_id:
            all_results = [r for r in all_results if r.doc_id == doc_id]
        if reviewed_only:
            all_results = [r for r in all_results if r.reviewed]
        if schema_name:
            all_results = [r for r in all_results if r.schema_name == schema_name]
        if reviewed_since is not None:
            # created_at on JsonFileStorage is float; last_reviewed_at we
            # don't track there yet, so fall back to created_at.
            # (Pre-fix note: this also includes never-reviewed rows because
            # their last_reviewed_at is None and falls through to
            # created_at. Acceptable for the JsonFileStorage path
            # because we don't expect rich filtering on it — SqlStorage
            # is the recommended backend for that.)
            all_results = [
                r for r in all_results
                if (getattr(r, "last_reviewed_at", None) or r.created_at) >= reviewed_since
            ]
        return all_results[:limit]

    def mark_reviewed(
        self, result_id: str, edited: dict[str, Any], reviewer: str
    ) -> None:
        # For an audit-grade store you'd append a separate review event;
        # here we just append a fresh line with updated state.
        original = self.get(result_id)
        if original is None:
            return
        import time as _t
        original.reviewed = True
        original.reviewed_extraction = edited
        original.reviewer = reviewer
        original.last_reviewed_at = _t.time()
        try:
            with self._lock, self.path.open("a") as f:
                f.write(json.dumps(asdict(original), default=str) + "\n")
        except OSError as e:
            log.error("JsonFileStorage.mark_reviewed failed for %s: %s", self.path, e)
            raise StorageError(
                f"could not append review for {result_id}: {e}",
            ) from e
        # Cache is now stale (the new append could shadow an earlier entry)
        self._invalidate_cache()

StoredResult dataclass

A pipeline run, persisted.

Source code in src/idp/storage/store.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
@dataclass
class StoredResult:
    """A pipeline run, persisted."""

    id: str
    doc_id: str
    schema_name: str
    backend_name: str
    mode: str | None
    classification: str | None
    extraction: dict[str, Any]
    confidence: dict[str, float] | None
    validation: dict[str, Any] | None
    source_path: str
    created_at: float
    reviewed: bool = False
    reviewed_extraction: dict[str, Any] | None = None
    reviewer: str | None = None
    last_reviewed_at: float | None = None  # epoch seconds, set by mark_reviewed/submit_review