Extract structured data

Turning an invoice, a receipt or a scanned form into rows you can insert. The technique is a JSON schema plus a validation step you do not skip.

The extractor#

extract.py

1import json
2from openai import OpenAI
3from pydantic import BaseModel, ValidationError, Field
4 
5client = OpenAI(base_url="https://ai.upfyn.com/v1", api_key=KEY, max_retries=0)
6 
7 
8class LineItem(BaseModel):
9    description: str
10    amount: float
11 
12 
13class Invoice(BaseModel):
14    invoice_number: str
15    issued_on: str = Field(description="ISO date, YYYY-MM-DD")
16    currency: str
17    total: float
18    line_items: list[LineItem]
19 
20 
21SCHEMA = {
22    "type": "object",
23    "properties": {
24        "invoice_number": {"type": "string"},
25        "issued_on": {"type": "string", "description": "ISO date, YYYY-MM-DD"},
26        "currency": {"type": "string", "enum": ["INR", "USD", "EUR", "GBP"]},
27        "total": {"type": "number"},
28        "line_items": {
29            "type": "array",
30            "items": {
31                "type": "object",
32                "properties": {
33                    "description": {"type": "string"},
34                    "amount": {"type": "number"},
35                },
36                "required": ["description", "amount"],
37                "additionalProperties": False,
38            },
39        },
40    },
41    "required": ["invoice_number", "issued_on", "currency", "total", "line_items"],
42    "additionalProperties": False,
43}
44 
45 
46def extract_invoice(text: str) -> Invoice:
47    response = client.chat.completions.create(
48        model="upfyn-yuva",
49        messages=[
50            {"role": "system",
51             "content": "Extract invoice fields exactly as written. Do not calculate or infer."},
52            {"role": "user", "content": text},
53        ],
54        response_format={
55            "type": "json_schema",
56            "json_schema": {"name": "invoice", "schema": SCHEMA, "strict": True},
57        },
58        max_tokens=2000,          # room for a long line-item list
59        temperature=0,            # extraction is not a creative task
60    )
61 
62    choice = response.choices[0]
63    if choice.finish_reason == "length":
64        # Truncated JSON will not parse. This is a token problem, not a schema problem.
65        raise ValueError("ran out of output tokens — raise max_tokens")
66 
67    return Invoice.model_validate_json(choice.message.content)

The schema constrains the model, but nothing guarantees the string you get back is what you expected — a truncated response is still a string. Parsing into a Pydantic model turns a wrong shape into an exception at the boundary instead of a KeyError three functions later.

Temperature zero

Extraction has one right answer. Sampling variety is exactly what you do not want, and it makes the same document produce different rows on different runs.

Check finish_reason first

A parse failure is usually a token failure

If finish_reason is "length", the model ran out of room mid-object and the JSON is genuinely incomplete. The fix is more max_tokens, not a more forgiving parser. Checking this first saves you from debugging the wrong problem.

Enums for anything from a fixed set

"currency" as a free string invites "Rupees", "Rs." and "₹". An enum removes the whole category.

Tell it not to calculate

“Extract exactly as written. Do not calculate or infer.” Without that, a model asked for a total will happily add the line items up and hand you a number the document does not contain — which is worse than a missing field, because it looks right.

Straight from a PDF#

Upload it and reference it — no OCR step of your own.

Extracting from a file

1uploaded = client.files.create(file=open("invoice.pdf", "rb"), purpose="assistants")
2 
3response = client.chat.completions.create(
4    model="upfyn-yuva",
5    messages=[{
6        "role": "user",
7        "content": [
8            {"type": "text", "text": "Extract the invoice fields."},
9            {"type": "file", "file": {"file_id": uploaded.id}},
10        ],
11    }],
12    response_format={"type": "json_schema",
13                     "json_schema": {"name": "invoice", "schema": SCHEMA, "strict": True}},
14)

Yuva reads PDFs, images and documents. Bala cannot — it is text only, and attaching a file to it returns 400 media_not_supported. See Multimodal input.

Many documents#

Collecting failures rather than raising

1def extract_many(documents: list[str]) -> tuple[list[Invoice], list[tuple[int, str]]]:
2    """One bad document must not lose the whole batch."""
3    good, bad = [], []
4    for i, text in enumerate(documents):
5        try:
6            good.append(extract_invoice(text))
7        except (ValidationError, ValueError, json.JSONDecodeError) as err:
8            bad.append((i, str(err)))
9    return good, bad

Return the failures with their index so you can re-run only those.
Watch your rate limit — a loop over a thousand documents needs throttling.
There is no batch API here; concurrency with a semaphore is the way to go faster.