Extracting Structured Data From Messy Documents

The model call is the easy part of an extraction pipeline and the part every tutorial shows. The accuracy lives in what you do to the document before the call and what you assert about the record after it.

The shape of the pipeline

raw file
   -> normalise to text        (PDF/HTML/OCR -> plain text, reading order fixed)
   -> chunk                    (overlapping, on structural boundaries)
   -> extract per chunk        (strict schema, temperature 0)
   -> ground                   (every quoted span must occur in the chunk)
   -> validate                 (types, ranges, cross-field invariants)
   -> repair once, or drop     (bounded; never a loop)
   -> merge + dedupe           (across chunks, by a stable key)
   -> store with provenance    (chunk id, char offsets, model, schema version)

Two stages in there are the ones usually missing. Grounding is what separates “the model read it” from “the model wrote something plausible”. Provenance is what lets you answer “where did this number come from” six months later, and without it a wrong record is unfalsifiable.

Preparation is most of the accuracy

• Reading order. Naive PDF text extraction returns content in the order it appears in the file, not the order a human reads it. Two-column layouts interleave; a table becomes a column of numbers with no headers attached. If your accuracy is bad and your prompt looks fine, print the text you actually sent. This is the single most common cause and it is invisible from the model side.
• Chunk on structure, not on length. Split at headings, page breaks or table boundaries, then merge small pieces up to your budget. A field split across a chunk boundary is unfindable; overlap of a few hundred characters makes it findable twice, which dedupe handles and a miss does not.
• Keep the offsets. Carry each chunk’s start offset in the source document. Grounding gives you an offset within the chunk; adding the two gives you a citation into the original, which is what a reviewer needs.
• Do not strip the noise you think is noise. Headers, footers and page numbers often carry the invoice number or the date. Strip after extraction if at all.

Span grounding: the check that costs nothing

Require a verbatim quote alongside every extracted value, emitted before it, then assert the quote is a substring of the chunk you sent. It is a string comparison. It needs no labels, no judge model and no ground truth, and it catches the failure mode people fear most — a value that is not in the document at all.

It is not a proof of correctness. A quote can be real and the value derived from it can still be wrong, and a model can quote the wrong real span. What it does give you is a hard floor: a record whose quote is not present is not a record. In practice you also get a second, unexpected benefit — the quotes make human review fast, because the reviewer reads two lines instead of ten pages.

Normalise whitespace on both sides before comparing, or you will fail on line wrapping. Do not normalise case; the model should be copying.

Watch the ungrounded rate as a first-class metric rather than as a log line, because it separates two problems that otherwise look identical. A rate that is low and stable is the check doing its job. A rate that spikes on one document type usually means the quote is real and your whitespace normalisation is wrong — ligatures, non-breaking spaces and soft hyphens from PDF extraction are the usual culprits, and they make a perfectly copied quote fail a substring test. A rate that rises gradually across all types is the interesting one, and it is worth investigating before any accuracy metric moves.

The pipeline

import json, re
from dataclasses import dataclass
from openai import OpenAI

client = OpenAI()
MODEL = "your-model-id"

SCHEMA = {
  "type": "object",
  "additionalProperties": False,
  "required": ["findings"],
  "properties": {
    "findings": {
      "type": "array",
      "items": {
        "type": "object",
        "additionalProperties": False,
        "required": ["quote", "field", "value"],
        "properties": {
          "quote": {"type": "string",
                    "description": "Text copied EXACTLY from the document, including punctuation."},
          "field": {"type": "string",
                    "enum": ["invoice_number", "total", "issue_date", "customer_name"]},
          "value": {"type": "string",
                    "description": "The normalised value. Dates as YYYY-MM-DD, totals as digits and one dot."}
        }
      }
    }
  }
}

def ws(s: str) -> str:
    return re.sub(r"\s+", " ", s).strip()

@dataclass
class Finding:
    field: str
    value: str
    quote: str
    offset: int          # into the original document

def extract_chunk(text: str, chunk_start: int) -> tuple[list[Finding], list[str]]:
    resp = client.chat.completions.create(
        model=MODEL,
        temperature=0,
        messages=[
            {"role": "system",
             "content": "Extract only what is literally printed. Quote before you answer. "
                        "If a field is not present, do not emit a finding for it."},
            {"role": "user", "content": text},
        ],
        response_format={"type": "json_schema",
                         "json_schema": {"name": "extraction", "strict": True, "schema": SCHEMA}},
    )
    choice = resp.choices[0]
    if choice.finish_reason == "length":
        raise Truncated(chunk_start)          # your bug: raise max_tokens or shrink the chunk

    findings, rejected = [], []
    haystack = ws(text)
    for f in json.loads(choice.message.content)["findings"]:
        pos = haystack.find(ws(f["quote"]))
        if pos < 0:
            rejected.append(f"ungrounded {f['field']}={f['value']!r} quote={f['quote']!r}")
            continue
        findings.append(Finding(f["field"], f["value"], f["quote"], chunk_start + pos))
    return findings, rejected

class Truncated(Exception): pass

Note what is not there. There is no retry loop, no “ask the model to fix it”, and no fallback that accepts an ungrounded finding because the run would otherwise be empty. Rejected findings go into a list you count and look at; whether to repair or retry is a separate decision with its own arithmetic, and it belongs outside this function.

Merging chunks

Overlapping chunks produce duplicates by design. Dedupe on (field, value) rather than on the quote, since the same fact may be quoted from two places, and keep the lowest offset so citations point at the first occurrence.

Conflicts are the interesting case: two chunks give different values for total. Do not silently take the first. Depending on the domain, the right rule is usually one of “prefer the occurrence nearest a heading that names the field”, “prefer the last page”, or “flag for review”. All three are defensible and all three are better than an arbitrary winner, because a conflict rate is a metric and an arbitrary winner is not.

Store the provenance alongside the value — chunk id, offset, model id, schema version. It costs a few bytes per record and it is the difference between a data quality investigation that takes ten minutes and one that takes a week. The schema version in particular is what makes a later migration possible at all.

Two operational details that decide whether this pipeline survives contact with a real corpus. Make the unit of work the chunk, not the document, and give it an idempotency key of (document_hash, chunk_index, schema_version, prompt_version). A run that dies two thirds of the way through then resumes instead of restarting, and a re-run after a prompt change re-extracts only what the change affects. Second, bound concurrency at the provider rather than at your worker count, and treat a 429 as backpressure rather than as an error to retry immediately — an extraction backfill is the classic way to discover your rate limit, and an unbounded retry storm turns a slow job into a failed one.