AI Document Processing with Python

Somewhere on your machine there is a folder of PDFs — supplier invoices, signed contracts, monthly statements, scanned delivery notes — and the only way anyone gets numbers out of it is by opening each file and typing what they see into a spreadsheet. This guide replaces that with a script. You will read text out of every file, keep track of which page each fact came from, ask a language model to pull out named fields, check the model's answer against a schema before you believe it, and write one clean row per document plus a separate log of everything that failed. The result is a folder you can point at a directory and walk away from.

The hard part of document work is not the AI call. It is everything around it: files that turn out to be photographs of paper, a total that arrives as 12,450.00 when your database wants a number, a contract that quietly runs to eighty pages and blows past the request size limit. A pipeline that survives real input has to expect all three. That is why the four steps below spend as much attention on detection and validation as on prompting.

This is one section of Building AI-Powered Business Applications. If you have never sent a request to a model from Python, read Understanding LLM APIs first so the API calls here look familiar rather than magical.

Prerequisites

You need Python 3.10 or newer, because the code uses the list[dict] and X | None type hints introduced in that version. Check what you have:

python --version

Work inside a virtual environment — an isolated folder of libraries that belongs to this project alone. If that phrase is new, Create a Python Virtual Environment for AI walks through it. Then install the four packages this guide uses:

python -m venv .venv
source .venv/bin/activate        # Windows: .venv\Scripts\activate
pip install "pypdf>=5.0" "pydantic>=2.7" "openai>=1.40" "python-dotenv>=1.0"
pip freeze > requirements.txt

pypdf reads PDFs, pydantic checks that data matches a shape you declare, openai talks to the model, and python-dotenv loads secrets from a file. Put your key in .env in the project folder:

OPENAI_API_KEY=sk-your-openai-key-here

Then add that file to .gitignore before you make a single commit:

echo ".env" >> .gitignore

A key that reaches a public repository is scraped and billed by strangers within minutes, so this line is not optional housekeeping.

Finally, create the two folders the scripts expect — documents/ for input and output/ for results — and drop a handful of real PDFs into the first one. Test with files that are genuinely messy: a scan, a very long report, something in a language you do not speak. A pipeline tuned only on tidy examples fails on its first real day.

Step 1 — Extract text reliably and spot the scanned files

A PDF is a container, and what is inside it varies enormously. When a PDF is produced by software — exported from an accounting package, printed to PDF from a word processor — it carries a text layer: the actual characters, positioned on the page. pypdf reads that layer directly, instantly, and for free. When a PDF is produced by a scanner or a phone camera, there is no text layer at all. The page is a photograph, and asking for its text returns an empty string.

Both kinds land in the same folder and look identical in a file browser, so your first job is to tell them apart automatically. The test is simple: pull the text of each page and count the characters. A real page of prose returns hundreds or thousands. A scan returns zero, or a handful of stray marks that the encoder happened to leave behind. Anything under roughly thirty characters is a scanned page.

from pathlib import Path

from pypdf import PdfReader


def read_pdf_pages(path: Path) -> list[dict]:
    """Return one dict per page: its number, its text, and whether it needs OCR."""
    reader = PdfReader(str(path))
    pages = []
    for number, page in enumerate(reader.pages, start=1):
        # extract_text() returns None on some pages, so coerce to a string first.
        text = (page.extract_text() or "").strip()
        pages.append({
            "page": number,
            "text": text,
            "needs_ocr": len(text) < 30,
        })
    return pages


if __name__ == "__main__":
    for page in read_pdf_pages(Path("documents/sample.pdf")):
        label = "OCR" if page["needs_ocr"] else "text"
        print(f"page {page['page']:>3}  {label:>4}  {len(page['text']):>6} chars")

Run that against your folder and you immediately learn something useful: how much of your archive is actually machine-readable. If every page prints text, the rest of this guide covers you completely. If some print OCR, those files need optical character recognition — software that reads letters out of an image — before any of the later steps can touch them. Common choices are the pytesseract package driving the open-source Tesseract engine, or a hosted vision model that accepts a page image directly. Both cost more per page than reading a text layer, in money or in time, which is exactly why you detect first and only send the pages that need it.

Two wrinkles show up in real archives. The first is the mixed file: a digital contract with a scanned signature page bolted on the end. Because read_pdf_pages decides page by page rather than file by file, you can process the twenty readable pages and send only the last one for image treatment. The second is the password-protected file. PdfReader raises an error when it meets one, so wrap the call in a try block and log the filename instead of letting a single locked document halt a batch of four hundred.

Resist the urge to concatenate every page into one long string at this stage. Keeping the per-page structure costs nothing, and it is what makes the traceability in the next step possible.

The decision below is the one your code makes for every single page.

Step 2 — Normalise the text and chunk it with page numbers intact

Raw PDF text is ugly. Justified layouts leave words split across lines with a hyphen, column layouts interleave fragments, and headers and footers repeat on every page. None of that stops a model outright, but it costs tokens and it makes matching harder. A short normalising pass fixes the worst of it.

The more important half of this step is chunking: splitting a long document into pieces small enough to fit inside a single request. Every model has a context limit, and a two-hundred-page report will not fit. The naive fix is to cut the text every few thousand characters, which works but throws away the one thing an auditor will ask for — where did this number come from? So insert a page marker into the text before you split, and carry the page range on every chunk. Now any fact the model reports can be traced to a page you can open.

import re


def normalise(text: str) -> str:
    """Repair the usual PDF text-layer damage before the model sees it."""
    text = text.replace("\u00ad", "")            # invisible soft hyphens
    text = re.sub(r"-\n(\w)", r"\1", text)       # re-join words split across lines
    text = re.sub(r"[ \t]+", " ", text)          # collapse runs of spaces
    text = re.sub(r"\n{3,}", "\n\n", text)       # collapse blank-line pile-ups
    return text.strip()


def chunk_pages(pages: list[dict], max_chars: int = 6000, overlap: int = 400) -> list[dict]:
    """Group page texts into chunks of at most max_chars, tagged with their page range."""
    chunks: list[dict] = []
    buffer = ""
    first_page = None
    for page in pages:
        body = normalise(page["text"])
        if not body:
            continue
        block = f"\n\n[page {page['page']}]\n{body}"
        if buffer and len(buffer) + len(block) > max_chars:
            chunks.append({"pages": (first_page, page["page"] - 1), "text": buffer.strip()})
            buffer = buffer[-overlap:]           # carry a tail so sentences are not cut in half
            first_page = page["page"]
        if first_page is None:
            first_page = page["page"]
        buffer += block
    if buffer.strip():
        chunks.append({"pages": (first_page, pages[-1]["page"]), "text": buffer.strip()})
    return chunks

Two numbers control the behaviour. max_chars sets how much text goes into one request — 6000 characters is roughly 1500 tokens, comfortably inside any current model's limit while leaving room for the reply. overlap copies the last few hundred characters into the next chunk so a sentence, or a table row, that straddles the boundary still appears whole somewhere. For a deeper treatment of the trade-offs, including token-aware splitting, see Split Long Text into Chunks for AI APIs.

There is one cleanup normalise deliberately leaves alone: repeated headers and footers. A page that ends with "Acme Holdings — Confidential — Page 7 of 84" contributes that line to every chunk, which is pure cost and a small distraction for the model. The tidy fix is to collect the first and last line of every page, count how often each string appears, and delete any line that shows up on more than about eighty per cent of pages. Do that after you have the page list and before you chunk. It is worth writing once for a long report and skipping entirely for one-page invoices, where the header often is the vendor name you are trying to extract.

One habit worth building now: log the character count of every chunk you send. When a bill surprises you, that log tells you instantly whether the cause was more documents or fatter chunks. Count Tokens in Python Before You Send shows how to convert those character counts into the tokens you are actually charged for.

Step 3 — Extract fields against a schema, then validate

Now the model earns its keep. You hand it a chunk and ask for named fields back as JSON. Three settings turn this from a demo into something dependable.

Set temperature=0. Temperature is the dial that controls randomness in word choice; at zero the model always takes its highest-probability option, so the same page yields the same answer on every run. That repeatability is what lets you re-run a batch and compare results. What Is Temperature in an LLM API? covers the dial in full.

Set response_format={"type": "json_object"}. This tells the API that the reply must be a syntactically valid JSON object, which eliminates the single most common failure in extraction scripts: a model that helpfully wraps its JSON in a code fence or prefaces it with "Here is the data you asked for."

Then — and this is the part people skip — do not trust the JSON. Valid JSON is not correct data. The model can omit a key, return a total as the string "12,450.00", or invent a date format nobody asked for. Declare the shape you expect as a pydantic model, a Python class where each field has a type and optional constraints, and let pydantic parse the reply. Anything that does not match raises an error you can catch and log.

import os
from datetime import date

from dotenv import load_dotenv
from openai import OpenAI
from pydantic import BaseModel, Field, ValidationError

load_dotenv()                                    # .env is in .gitignore
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))


class InvoiceFields(BaseModel):
    vendor_name: str
    invoice_number: str
    invoice_date: date                           # rejects "next Tuesday"
    total_amount: float                          # rejects "12,450.00 EUR"
    currency: str = Field(min_length=3, max_length=3)
    source_page: int                             # which [page N] marker it came from
    confidence: float = Field(ge=0.0, le=1.0)


SYSTEM = (
    "You extract fields from business documents. "
    "Reply with a single JSON object and nothing else. "
    "Every value must be copied from the document, never guessed. "
    "source_page is the number in the nearest [page N] marker above the value. "
    "confidence is your own 0-1 estimate that every field is correct."
)


def extract_invoice(chunk_text: str) -> InvoiceFields | None:
    """Return a validated InvoiceFields, or None if the reply failed the schema."""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        temperature=0,
        max_tokens=500,
        response_format={"type": "json_object"},
        messages=[
            {"role": "system", "content": SYSTEM},
            {"role": "user", "content": f"Keys: {list(InvoiceFields.model_fields)}\n\n{chunk_text}"},
        ],
    )
    raw = response.choices[0].message.content
    try:
        return InvoiceFields.model_validate_json(raw)
    except ValidationError as err:
        print(f"schema rejected the reply: {err.error_count()} problem(s)")
        return None

Notice what the type annotations buy you. invoice_date: date means pydantic parses 2026-03-14 into a real date object and refuses anything it cannot parse. total_amount: float means a thousands separator is caught here, at the door, instead of corrupting a sum three steps later. confidence is constrained to the range zero to one, so a model that answers "high" fails loudly.

The confidence field deserves a note. A model's self-reported confidence is an opinion, not a measurement, and it should never be the only check you run. What it is genuinely good at is sorting: the twenty documents that came back at 0.4 are far more likely to be wrong than the four hundred that came back at 0.98, so a human reviewer starts at the bottom of the sorted list and stops when the errors dry up. That is a real productivity win even though the number itself is soft.

One structural question comes up as soon as documents get longer than a page: which chunk do you send? For an invoice the answer is easy, because everything you want sits on the first page. For a twelve-page purchase agreement it is not. The pattern that works is to run the extractor on each chunk in turn and merge the results, taking the first non-null value for each field and keeping the source_page that came with it. That way a payment term buried on page nine still lands in the row, and you still know where it came from. It costs one call per chunk, so apply it only to the document types that need it — which is exactly why classification comes before extraction rather than after.

The path from chunk to stored row looks like this.

Step 4 — Route by document type and write the results

A real folder is not all invoices. Contracts, statements and reports arrive in the same drop, and each needs a different set of fields — asking an invoice extractor for a contract's end date returns nonsense with a straight face. So before extraction, classify.

Classification is the cheapest call in the whole pipeline. Send the first chunk, cap the reply at a handful of tokens, and demand a single word from a fixed list. Then use that word to look up a handler function. Adding support for a new document type becomes two small edits: write the pydantic model, register the handler.

from extract import client, extract_invoice     # the client and handler from Step 3

HANDLERS = {
    "invoice": extract_invoice,
    # "contract": extract_contract,   # add your own, one pydantic model each
}


def classify(chunk_text: str) -> str:
    """Return one of: invoice, contract, report, unknown."""
    reply = client.chat.completions.create(
        model="gpt-4o-mini",
        temperature=0,
        max_tokens=5,
        messages=[
            {"role": "system", "content":
                "Answer with exactly one word: invoice, contract, report or unknown."},
            {"role": "user", "content": chunk_text[:2000]},
        ],
    )
    word = reply.choices[0].message.content.strip().lower()
    return word if word in {"invoice", "contract", "report"} else "unknown"

Two details make this reliable. The reply is truncated to a handful of tokens, so the model cannot ramble even if it wants to. And the last line rejects anything outside the allowed set, which means a surprising answer becomes unknown and lands in the review queue rather than crashing a dictionary lookup. Never let a model's free text index directly into your code.

Where the results go depends on how many you have. CSV is right for the first few thousand documents: it opens in any spreadsheet, it diffs in version control, and a colleague can check it without installing anything. Move to a database when you need concurrent writers, or when you want to re-run one document without rewriting the whole file. Either way, keep the two output streams separate — successes in the table, failures in an exceptions log with the filename and the reason. Merging them is tempting and always ends with a half-empty row that somebody averages by accident.

One more thing is worth storing that most tutorials drop: the model's raw reply. Write it to a per-document JSON file next to the CSV, named after the source PDF. It costs a few kilobytes and it saves an afternoon the first time somebody disputes a figure, because you can see exactly what the model returned and whether the mistake was the extraction, the schema, or the document itself. It also lets you re-validate an entire archive against a tightened schema without paying for a single new API call.

The routing table below is the whole of Step 4 in one picture.

Parameter reference

These are the knobs you will actually turn while tuning a pipeline. Everything else can stay at its default.

Troubleshooting

1. AttributeError: 'NoneType' object has no attribute 'strip' — page.extract_text() returned None rather than an empty string, which happens on image-only pages. Cause: calling a string method on the result directly. Fix: keep the (page.extract_text() or "") pattern from Step 1 everywhere you read a page.
2. pydantic_core._pydantic_core.ValidationError: 1 validation error for InvoiceFields — the model returned a value your schema rejects, most often total_amount as "12,450.00". Cause: the document formats numbers with separators. Fix: log the document to the exceptions file, and add an explicit instruction to the system prompt that numbers must be returned without separators or currency symbols.
3. json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0) — the reply started with prose or a code fence instead of {. Cause: response_format was omitted. Fix: add response_format={"type": "json_object"} to the call; Fix JSONDecodeError with AI API Responses in Python covers the fallback parsing if you cannot.
4. openai.BadRequestError: ... maximum context length ... — a chunk plus its prompt exceeded what the model accepts. Cause: max_chars set too high, or a page of dense tables. Fix: lower max_chars to 4000 and re-run; Fix the Context-Length-Exceeded Error in Python explains the arithmetic.
5. openai.RateLimitError: Error code: 429 — you fired requests faster than your account tier allows. Cause: looping over hundreds of files with no pause. Fix: retry with an increasing wait between attempts, as described in Fix the 429 Rate-Limit Error in Python.
6. pypdf.errors.PdfReadError: EOF marker not found — the file is truncated or is not really a PDF. Cause: an interrupted download or a renamed file. Fix: wrap PdfReader in a try block, log the filename to the exceptions file, and move on rather than letting one bad file stop the batch.

Worked example

Here is the whole pipeline as one script. It assumes the four snippets above live beside it in reading.py, chunking.py, extract.py and routing.py, walks the documents/ folder, classifies each file, runs the matching handler, and writes one CSV row per success and one JSON line per failure.

"""process_folder.py — turn a folder of PDFs into one CSV plus an exceptions log."""
import csv
import json
from pathlib import Path

from chunking import chunk_pages
from extract import InvoiceFields, extract_invoice
from reading import read_pdf_pages
from routing import HANDLERS, classify

IN_DIR = Path("documents")
OUT_CSV = Path("output/extracted.csv")
EXCEPTIONS = Path("output/exceptions.jsonl")
COLUMNS = list(InvoiceFields.model_fields)


def process_folder() -> None:
    OUT_CSV.parent.mkdir(parents=True, exist_ok=True)
    with OUT_CSV.open("w", newline="", encoding="utf-8") as csv_file, \
            EXCEPTIONS.open("w", encoding="utf-8") as log:

        def reject(name: str, reason: str) -> None:
            log.write(json.dumps({"file": name, "error": reason}) + "\n")
            print(f"  skipped {name}: {reason}")

        writer = csv.DictWriter(csv_file, fieldnames=["source_file", "doc_type", *COLUMNS])
        writer.writeheader()

        for pdf in sorted(IN_DIR.glob("*.pdf")):
            print(f"reading {pdf.name}")
            try:
                pages = read_pdf_pages(pdf)
            except Exception as err:                      # corrupt or non-PDF file
                reject(pdf.name, f"unreadable: {err}")
                continue
            if all(page["needs_ocr"] for page in pages):  # a scan, no text layer
                reject(pdf.name, "no text layer; OCR required")
                continue

            chunks = chunk_pages(pages)
            doc_type = classify(chunks[0]["text"])        # one cheap word back
            handler = HANDLERS.get(doc_type)
            if handler is None:                           # unknown type -> human queue
                reject(pdf.name, f"no handler for type '{doc_type}'")
                continue

            record = handler(chunks[0]["text"])           # returns None if invalid
            if record is None:
                reject(pdf.name, "reply failed schema validation")
                continue

            writer.writerow({
                "source_file": pdf.name,
                "doc_type": doc_type,
                **record.model_dump(mode="json"),         # dates become ISO strings
            })
            print(f"  {record.vendor_name}: {record.total_amount} "
                  f"{record.currency} (confidence {record.confidence})")


if __name__ == "__main__":
    process_folder()

Run it with python process_folder.py. You get output/extracted.csv, ready to open in a spreadsheet and sort by the confidence column, and output/exceptions.jsonl, one line per file that did not make it with the reason attached. Work that second file top to bottom: every entry is either a scan that needs optical character recognition, a document type you have not written a handler for yet, or a prompt that needs sharpening. Each is a concrete, finishable task rather than a vague sense that the script "sometimes fails".

Two changes make this script safe to run repeatedly. As written it opens the CSV with "w", which truncates the file, so every run reprocesses everything and pays for everything again. Switch to "a" and write the header only when the file does not yet exist, then keep a small set of filenames you have already handled — read it back from the CSV's source_file column at startup — and skip anything in it. Now a crashed run resumes instead of restarting, which matters the first time a rate limit stops you two hundred files into a batch of a thousand.

The other change is to make the loop tell you what it costs. Each response carries a usage object with input and output token counts; add them to a running total and print it at the end. After one full pass you know the price per document to a useful precision, and you can answer the only question a manager will actually ask about this script before you have to guess.

Once the batch runs cleanly by hand, the natural next move is to stop running it by hand. Schedule Python AI Jobs with GitHub Actions shows how to point the same script at a folder every morning without leaving a laptop open.

Where to go next

Four guides take each part of this pipeline further. Start with Extract Invoice Data from PDFs with Python and AI if the folder in front of you is accounts payable — it goes deeper on line items, tax lines and the multi-currency edge cases that a single total_amount field glosses over. If your documents are long rather than numerous, Summarize Long PDFs with Python and Chunking shows the summarise-then-combine pattern that keeps an eighty-page report inside the context limit without losing the thread.

When people want to ask questions of the archive rather than pull fixed fields from it, Build a Document Question-Answering Tool turns the same page-tagged chunks into a searchable index that answers with citations, and Connect a Chatbot to Your Docs with RAG puts a conversational front end on top of it. If your bottleneck is sorting the incoming pile before anything else can happen, Classify Incoming Documents with Python and AI expands the one-word classifier above into something that handles overlapping categories and knows when to abstain.

And when the script has earned its keep on your machine, Deploying Python AI Apps covers turning it into a service, keeping your keys safe outside a .env file, and watching what it does in production.

Back to Building AI-Powered Business Applications.

Related guides

• Extract Invoice Data from PDFs with Python and AI — the full field set for accounts payable, including line items and tax.
• Split Long Text into Chunks for AI APIs — token-aware splitting when character counts stop being precise enough.
• Classify Incoming Documents with Python and AI — a sturdier router for mixed inboxes and overlapping document types.
• Fix the Context-Length-Exceeded Error in Python — what to do the first time a long report refuses to fit.
• Deploying Python AI Apps — take the finished pipeline off your laptop and run it on a schedule.