Document extraction to JSON

Turn invoices, forms and letters into JSON that matches your schema. The schema rides on the request, routing serves only models with verified structured-output support, retries are idempotent, validation decides what goes to a reviewer, and every record carries the model and the cost that produced it. Built as a worker pipeline on Postgres.

What you build

A pipeline that takes documents from a bucket or a queue, gets their text (the PDF text layer, OCR, or a vision model for scans), asks Inferbase for JSON that matches a schema you define in code, validates the result, writes a record with the model and the cost that produced it, and sends what fails validation to a review queue. Every extraction carries an idempotency key derived from the document and the schema version, so a worker can retry anything without paying twice or writing twice. The model is chosen per document under a policy that only admits models with verified structured-output support.

When you are done

• A worker that turns a document into a validated record, with the schema as code and its version on every record
• Retries that are safe by construction: the same document and schema replay the original response
• A review queue that holds exactly the records validation rejected, with the model's output beside the document
• Cost, model and request id on every record, so the expensive document types are visible
• An evaluation that scores field accuracy, schema validity and repair rate on a labelled set, for auto against a pinned model

The components

Intake

A queue message or a bucket event per document, carrying where the bytes are and the document type that selects the schema.

Text acquisition

The PDF text layer when there is one; OCR for scans; or the page sent as an image to a vision model through the same API when layout matters.

Schema registry

One Pydantic model per document type, exported as a strict JSON Schema, with a version string that is part of the idempotency key and the record.

Extraction call

One chat completion with the model set to auto, response_format json_schema, thinking off, and the idempotency key in the header. The policy decides what happens to an answer that does not parse.

Validation

Pydantic parses the JSON; business rules check what the schema cannot: totals that add up, dates that parse, currencies that exist.

Records and review

Postgres: a records table and a review table. A record that passes is written; one that fails goes to review with the raw output and the reason.

Inferbase

The key with a spend cap, the routing policy with a rule pool on structured output and extract-then-retry for invalid JSON, and the chat completions endpoint.

Evaluation

A labelled set of documents with gold records. Field-level accuracy, validity and repair rates, and cost per document, for auto against a pinned model.

The problem

Extraction is a pipeline step, and a pipeline step cannot take a model that returns prose.

A document pipeline turns text into fields: the vendor, the total, the line items, the dates. The output goes into a database or a downstream system, so it must be valid JSON in a known shape every time, not most of the time. A model that wraps its answer in a code fence, adds a sentence of commentary, or invents a field breaks the step that reads it.

Documents also vary in difficulty. A clean invoice with three lines is a simple read; a scanned letter with a table spread over two pages and a correction in the margin is not. One model for every document is the usual compromise, and it is paid for twice: in the price of the hard-document model on the easy documents, and in the retries when the easy-document model meets a hard one.

The third problem is retries. A worker that times out and runs the document again pays again and may write the record twice. In a pipeline, a retry has to be free and idempotent, or operators stop retrying.

This guide puts the schema on the request as a JSON Schema and sends the model as auto. Routing keeps only models with verified structured-output support, serves the one that wins for the document in front of it, and the routing policy says what happens when an answer still does not parse: unwrap it, ask once more, or return it as is and let validation decide. Every request carries an idempotency key, so a retry replays the response instead of running the extraction again.

Who has it

• A back office that keys invoices, receipts or forms by hand today
• A product that ingests customer documents and needs fields, not summaries
• A pipeline that already calls a model for extraction and handles its failures with regular expressions

Architecture

System overview

Every component, grouped by who runs it. The product sits at the generation call; everything in your lanes is yours and unchanged by this design.

One request, end to end

The schema is yours and travels with each request. Inferbase chooses a model that honours it and tells you when it had to unwrap the answer.

Deployment

Where each part runs, and what talks to what over the network.

What stays in your stack

• The schema and its versioning
• OCR and page splitting
• Validation and what happens to a rejected record
• The queue and its retry policy

Technology stack

One choice per layer, the alternative that was considered, and the reason. Swap any row; the Inferbase call does not change.

What carries it

The Inferbase features this design rests on. Each exists in the product today and is linked to its documentation.

json_schema on the request

response_format with a JSON Schema is forwarded to the serving model as written. With the model set to auto, routing serves only models whose structured-output support is verified. Documentation

Invalid JSON handling in the policy

When an answer does not parse, the policy decides: extract the JSON from fences or commentary, extract and then ask the same model once more, or serve the answer as it came. Nothing is rewritten to make it parse. Documentation

Repair disclosed

When the JSON was taken out of a wrapped answer, usage.structured_output_repair says so, so a pipeline can count how often each model needed it. Documentation

An idempotency key per document

A retry with the same Idempotency-Key replays the original response instead of running the extraction again; the same key with a different body is refused with a 422. A failed request releases the key at once. Documentation

Image input for scanned pages

A page sent as an image content part is routed to models that accept images. A request no pool model can accept gets a clear 503 rather than a text-only guess. Documentation

Cost and model on every record

The response names the model that served and carries the cost in micro-dollars, so each extracted record can store what produced it and what it cost. Documentation

Zero retention

Documents carry names and account numbers. Zero retention on the key keeps request content out of the log; the decision record is kept without it. Documentation

Build it step by step

Eight steps from a bucket of documents to validated records. The code is Python; the extraction call alone is also given in TypeScript after the policy, and both were run against the API. Each step ends with the thing it turns on.

Before you start

• An Inferbase account, a project, and a key with a spend cap for this pipeline.
• A bucket of documents and a queue, or a table to poll. Postgres for the records.
• Python 3.12 with openai, pydantic and a PDF library (pypdf); Tesseract for scans if you will OCR.

from pydantic import BaseModel, ConfigDict, Field

SCHEMA_VERSION = "invoice-v3"


class LineItem(BaseModel):
    model_config = ConfigDict(extra="forbid")
    description: str
    quantity: float
    amount: float


class Invoice(BaseModel):
    model_config = ConfigDict(extra="forbid")
    vendor: str
    invoice_number: str
    date: str = Field(description="ISO 8601 date")
    currency: str = Field(description="ISO 4217 code")
    total: float
    line_items: list[LineItem]
    purchase_order: str | None = Field(description="null when the document shows none")


def response_format() -> dict:
    """The schema as the request carries it: strict, every field required, no extras."""
    return {"type": "json_schema", "json_schema": {"name": "invoice", "strict": True, "schema": Invoice.model_json_schema()}}

import base64
from pypdf import PdfReader


def document_content(path: str, kind: str) -> list[dict]:
    """The user message content: text when there is text, an image part when layout matters."""
    if kind == "scan-layout":
        with open(path, "rb") as f:
            data = base64.b64encode(f.read()).decode()
        return [
            {"type": "text", "text": "Extract the fields from this page."},
            {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{data}"}},
        ]
    if path.endswith(".pdf"):
        text = "\n".join(page.extract_text() or "" for page in PdfReader(path).pages)
        if text.strip():
            return [{"type": "text", "text": text}]
    return [{"type": "text", "text": ocr(path)}]  # tesseract, or your OCR service

import hashlib
import json
from openai import OpenAI

client = OpenAI(api_key="inf_your_api_key", base_url="https://api.inferbase.ai/api/v1/inference")


def idempotency_key(document_bytes: bytes, schema_version: str) -> str:
    return f"extract-{schema_version}-{hashlib.sha256(document_bytes).hexdigest()[:32]}"


def extract(content: list[dict], document_bytes: bytes, schema_version: str, fmt: dict) -> dict:
    completion = client.chat.completions.create(
        model="auto",
        max_tokens=1_500,
        messages=[
            {"role": "system", "content": "Extract the fields from the document. Every field is required; use null where the document shows nothing."},
            {"role": "user", "content": content},
        ],
        response_format=fmt,
        extra_headers={"Idempotency-Key": idempotency_key(document_bytes, schema_version)},
        extra_body={"extended_thinking": False},
    )
    usage = completion.usage.model_extra
    return {
        "raw": completion.choices[0].message.content or "",
        "model": completion.model,
        "request_id": completion.id,
        "cost_usd": usage["cost_micros"] / 1e6,
        "repaired": bool(usage.get("structured_output_repair")),
    }

from datetime import date
from pydantic import ValidationError

KNOWN_CURRENCIES = {"EUR", "USD", "GBP", "INR", "JPY"}


def validate(raw: str) -> tuple[Invoice | None, str | None]:
    try:
        invoice = Invoice.model_validate_json(raw)
    except ValidationError as exc:
        return None, f"schema: {exc.errors()[0]['loc']}: {exc.errors()[0]['msg']}"
    if abs(sum(i.amount for i in invoice.line_items) - invoice.total) > 0.01:
        return None, "line items do not sum to the total"
    try:
        date.fromisoformat(invoice.date)
    except ValueError:
        return None, f"date not ISO 8601: {invoice.date}"
    if invoice.currency not in KNOWN_CURRENCIES:
        return None, f"unknown currency: {invoice.currency}"
    return invoice, None

CREATE TABLE records (
  idempotency_key text PRIMARY KEY,
  document_ref    text NOT NULL,
  document_type   text NOT NULL,
  schema_version  text NOT NULL,
  record          jsonb NOT NULL,
  model           text NOT NULL,
  cost_usd        numeric(12, 6) NOT NULL,
  request_id      text NOT NULL,
  repaired        boolean NOT NULL,
  text_path       text NOT NULL,          -- text_layer | ocr | image
  created_at      timestamptz NOT NULL DEFAULT now()
);

CREATE TABLE review (
  id              bigserial PRIMARY KEY,
  idempotency_key text NOT NULL,
  document_ref    text NOT NULL,
  raw_output      text NOT NULL,
  reason          text NOT NULL,
  model           text NOT NULL,
  request_id      text NOT NULL,
  created_at      timestamptz NOT NULL DEFAULT now()
);

def handle(message: dict) -> None:
    document_bytes = fetch(message["ref"])
    kind = message["type"]  # invoice, receipt, ...
    schema = REGISTRY[kind]  # the Pydantic model, its version and its response format
    content = document_content_from_bytes(document_bytes, message.get("path", "auto"))

    result = extract(content, document_bytes, schema.version, schema.response_format())  # idempotent: a retry replays
    record, reason = validate(result["raw"])
    key = idempotency_key(document_bytes, schema.version)

    if record is None:
        db.execute(
            "INSERT INTO review (idempotency_key, document_ref, raw_output, reason, model, request_id) VALUES (%s, %s, %s, %s, %s, %s)",
            (key, message["ref"], result["raw"], reason, result["model"], result["request_id"]),
        )
        return
    db.execute(
        "INSERT INTO records (idempotency_key, document_ref, document_type, schema_version, record, model, cost_usd, request_id, repaired, text_path) "
        "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) ON CONFLICT (idempotency_key) DO NOTHING",
        (key, message["ref"], kind, schema.version, record.model_dump_json(), result["model"], result["cost_usd"], result["request_id"], result["repaired"], message.get("path", "auto")),
    )

SELECT model,
       count(*)                                   AS records,
       round(100.0 * avg(repaired::int), 1)       AS repair_pct,
       round(avg(cost_usd), 6)                    AS mean_cost_usd
FROM records
WHERE created_at > now() - interval '7 days'
GROUP BY model
ORDER BY records DESC;

The routing policy

Balanced as the objective, a rule pool that requires structured output, High eligibility, extract-then-retry for answers that do not parse, and zero retention for documents that carry personal data.

Create it under Policies in the portal and attach it to the key the application uses; the steps are in Create a routing policy.

The call

One extraction: the schema on the request, the document as the user turn, an idempotency key derived from the document so a retry replays rather than re-runs, and the repair flag read from the response. The full pipeline above wraps this call in validation, records and a review queue.

import hashlib
import json
from openai import OpenAI

client = OpenAI(api_key="inf_your_api_key", base_url="https://api.inferbase.ai/api/v1/inference")

SCHEMA = {
    "name": "invoice",
    "strict": True,
    "schema": {
        "type": "object",
        "properties": {
            "vendor": {"type": "string"},
            "invoice_number": {"type": "string"},
            "date": {"type": "string", "description": "ISO 8601 date"},
            "currency": {"type": "string"},
            "total": {"type": "number"},
            "line_items": {
                "type": "array",
                "items": {
                    "type": "object",
                    "properties": {
                        "description": {"type": "string"},
                        "quantity": {"type": "number"},
                        "amount": {"type": "number"},
                    },
                    "required": ["description", "quantity", "amount"],
                    "additionalProperties": False,
                },
            },
        },
        "required": ["vendor", "invoice_number", "date", "currency", "total", "line_items"],
        "additionalProperties": False,
    },
}

document = """INVOICE
Northwind Paper Co.
Invoice no. NP-2041 - Date: 3 October 2026

2 x A4 copier paper, 5 reams    EUR 48.00
1 x Toner cartridge, black       EUR 89.50
Shipping                         EUR 12.00

TOTAL DUE                        EUR 149.50
Payment within 30 days."""  # a PDF text layer or an OCR result

completion = client.chat.completions.create(
    model="auto",
    max_tokens=400,
    messages=[
        {"role": "system", "content": "Extract the invoice fields from the document. Use null for nothing else; every field is required."},
        {"role": "user", "content": document},
    ],
    response_format={"type": "json_schema", "json_schema": SCHEMA},
    # The same document, the same key: a queue retry replays the response instead of paying again.
    extra_headers={"Idempotency-Key": "invoice-" + hashlib.sha256(document.encode()).hexdigest()[:24]},
    extra_body={"extended_thinking": False},
)

invoice = json.loads(completion.choices[0].message.content)
usage = completion.usage.model_extra
print(json.dumps(invoice, indent=2))
print(completion.model, f'{usage["cost_micros"] / 1e6:.6f} USD', "repaired" if usage.get("structured_output_repair") else "parsed as returned")

Both snippets were run against api.inferbase.ai on 2026-10-13: every field came back as the schema states, with the date normalised to ISO 8601 and three line items. The TypeScript run was made on a second copy of the document, because the key the Python run had used for the same document was refused for a different request body, which is the idempotency rule doing its job.

Open in PlaygroundAPI reference

Security and privacy

Documents are the most personal data most pipelines touch. The controls are about what is kept, who may see it, and what a retry can do.

Zero retention on the key

Request content is kept for up to 30 days by default. With zero retention, the document text and the model's answer are not kept; the decision record is, without the content. For invoices, forms and letters, this is the setting.

The pool is the allow-list

Only models in the policy pool ever see a document. A rule pool is a rule you can show an auditor; a chosen pool is a list. Your own provider keys or cloud accounts keep the call inside a contract you already have.

The idempotency key carries no content

It is a hash of the bytes and the schema version: it proves identity without revealing the document, and it is safe in a log.

Redact before you send

What the record never needs, such as card numbers on a receipt, can be masked in the text before the call. What is masked is never sent and never kept anywhere.

The review queue is sensitive

It holds raw model output beside document references. Put it behind the same access control as the documents, and purge items once corrected.

The key

One key per pipeline with a spend cap and a rate limit, held by the workers. A runaway queue meets the cap before it meets your budget.

Evaluation

A labelled set: documents with gold records, by type. Run them through the pipeline and score four things per document: whether the output was valid JSON against the schema, whether it needed repair, field-level accuracy against the gold record, and cost. Then run the set pinned to one model and compare.

Validity rate

The fraction of documents whose output parsed against the schema without review. With a strict schema and the structured-output pool it should be close to all of them; what is left is the model breaking the schema or the policy serving an unparseable answer as it came.

Repair rate

The fraction whose JSON had to be unwrapped, from the repair flag on each response. Read it by model; a model that needs it often is a candidate to leave the pool.

Field accuracy

Per field, exact match for strings and codes, tolerance for amounts, parsed equality for dates. Report by document type: the types with low accuracy are the ones that need the strong models or a better prompt.

Cost per document

From the receipts, by type and by model. The saving from routing is here: simple types on cheap models, hard types on strong ones.

Auto against pinned

The same set with the model pinned to what you use today. Compare field accuracy and cost per document; the decision records say which model handled which document.

def field_accuracy(predicted: dict, gold: dict) -> float:
    hits = 0
    for key, want in gold.items():
        got = predicted.get(key)
        if isinstance(want, float):
            hits += abs((got or 0) - want) < 0.01
        elif key == "line_items":
            hits += len(got or []) == len(want)
        else:
            hits += got == want
    return hits / len(gold)


scores = [field_accuracy(run(doc), gold) for doc, gold in labelled]
print(f"mean field accuracy {sum(scores) / len(scores):.1%}")

Operations

The records table is the dashboard. Everything an operator needs is a query on it.

Review queue size

The number of items in review by type and by reason. A reason that grows is a schema or prompt problem; a type that grows is a document source that changed.

Repair and validity by model

The weekly query in step 8. A model whose repair rate climbs is a model whose structured-output support changed; the pool rule will drop it when Inferbase's verification does, but your own number is earlier.

Cost by type

Mean cost per document by type and model. The expensive types are where a cheaper model or a shorter prompt pays.

Replays

A retried message that replayed costs nothing and writes nothing new. Count replays by comparing messages handled with records written; a high replay rate is a queue with a visibility timeout too short for the call.

Read one decision

The request id on a record reads back why that model was chosen: the task, the complexity, the pool after filtering. Start there when a record looks wrong.

Schema versions

Bump the version with the schema. Old records keep their version; the new key space means nothing replays across versions.

Scaling and limits

The pipeline scales by workers; the limits are the key, document size and the model's window.

Real time, measured

A batch pipeline, and latency only matters at the queue. Measured on 2026-10-14: a short invoice extracted in 1.8 to 3.2 s per document, non-streamed, under Balanced. A worker handles 20 to 30 documents a minute; the limit at volume is the key's rate limit, not the model. Interactive use, a person waiting on one document, is fine at 2 to 3 s; sub-second is not what this is.

Workers and the rate limit

One document is one request. A key at 600 requests a minute carries 600 documents a minute across any number of workers; raise the limit before the queue backs up.

Long documents

A fifty-page contract does not fit one call. Split by page or section, extract per part with the same schema, and merge by rule; or set min_context_window on the request so only long-context models qualify.

Images

A page as an image costs more than its text and routes to fewer models. Reserve the image path for layouts that need it; send text for the rest.

Prompt caching

The system prompt and the schema repeat on every call. Turn caching on in the policy; the decision says whether the chosen provider caches.

Batch windows

A nightly batch can run at full rate for an hour and stop; a daily cap on the key sized to the batch is the safety net.

Many document types

One schema per type in the registry, one prompt line per type, the same worker. A type with different compliance needs gets its own key and policy.

Troubleshooting

What goes wrong when this is built, what it means, and what to do. Each one was met while building or running it.

Extending

The pipeline is a worker and a schema registry. Each extension is a schema, a path, or a step.

Classify first

When the document type is unknown, a first call with a small schema (type, confidence) picks the schema; the task can be stated as general and the objective as cost. One cheap call saves a wrong extraction.

Tables

A schema with an array of rows and named columns; send the page as an image when the table's structure is visual.

A review interface

The review table is the backlog; a page that shows the document beside the raw output and lets a reviewer correct fields writes the record and an evaluation case.

Confidence

Ask for a per-field confidence in the schema, and route low-confidence records to review even when validation passes.

Your own keys

Build the pool from a provider key or cloud account. The worker does not change; the policy decides whose credential serves.

Downstream events

Publish an event per written record with the request id; a consumer can read the decision and the cost without touching the pipeline.

What was measured

What one extraction costs in time and money on the public API. A long document classifies and extracts slower than a short one.

Measured from our own infrastructure in India against api.inferbase.ai on 2026-10-14, three runs each, client-side wall time including the network hop (about 0.25 s each way), thinking off. Every response returns its own figures.

Further reading

• The Inferbase quickstarts repository; the document-extraction folder for this guide is planned
• Structured output on chat completions
• Invalid JSON handling in the policy
• Authentication and idempotency
• Client compatibility: images and structured output
• Security and privacy policies: zero retention

Frequently asked questions