Python automation for OCR + LLM: the full workflow behind 80,000 folios

Python automation for OCR + LLM is usually explained as two calls: read the text from an image and ask a model what it says. In production, those two calls are the easy part.

The hard part is everything around them: preparing the image, forcing the model to answer in a predictable format, deciding what to do when it is unsure, and knowing, without logging into the server, whether the system is still healthy.

This article walks through that workflow end to end, starting from a real case. The code is illustrative: it shows the pattern in Python, not the client's code.

The case: verifying receipts with nobody reviewing them

The starting point is the Panini Fan Zone FWC 2026 system. It managed a fixed pool of 80,000 folios (unique entry codes) and verified the purchase receipts that participants submitted.

Verification had two steps. Tesseract extracted the text from the image, and a LLaMA 3.3 70B model, served by Groq, validated it against the promotion rules.

It ran unattended, on a cron job every 2 minutes, on an AlmaLinux VPS. Nobody was reviewing receipts one by one.

That last sentence defines the design. If nobody reviews, the system has to tell "approved", "rejected" and "I'm not sure" apart. And it has to leave a trail for every decision.

I already wrote about the performance of this kind of script in Python optimization for 80,000 folios: data structures, memory and concurrency. This one is about something else: how to build the workflow so the automation can be trusted.

Task automation with Python: think in states

The first mindset shift is to stop seeing the pipeline as a linear script. It is a state machine. Each receipt moves through stages, and each stage can end well, badly, or in doubt.

A minimal state model looks like this:

• pending: the receipt arrived and is waiting its turn.
• ocr_done: there is extracted text and a measure of its quality.
• approved or rejected: the model and the rules agree with enough confidence.
• review: something does not add up and the system prefers not to decide.
• error: a technical failure, with the attempt count and the last message.

Keeping the state in the database rather than in memory has three advantages. If the process dies, the next run picks up where it left off. You can reprocess an entire state when you change a rule. And every number on your dashboard comes from a GROUP BY.

The review state deserves an honest note. In the real case the pipeline ran without manual review. Even so, when I design a new one, I build that exit in from the start: it is cheaper to have the door and not use it than to add it with the system already in production.

OCR integration: quality is decided before Tesseract

Tesseract works much better with clean images: dark text on a light background, at sufficient resolution. A receipt photographed with a phone almost never arrives like that. It comes with shadows, tilt, low light and compression.

That is why preprocessing is the highest-return step in the whole pipeline. A language model cannot fix numbers the OCR never read.

from PIL import Image, ImageOps
import pytesseract
def extract_text(path: str) -> tuple[str, float]:
    with Image.open(path) as img:
        img = ImageOps.exif_transpose(img)        # phone photos arrive rotated
        gray = ImageOps.grayscale(img)            # color only adds noise for OCR
        if gray.width < 1500:                     # tiny text = invented characters
            gray = gray.resize((gray.width * 2, gray.height * 2), Image.LANCZOS)
        bw = gray.point(lambda p: 255 if p > 160 else 0)  # threshold: text vs background
        data = pytesseract.image_to_data(bw, lang="spa", config="--psm 6", output_type=pytesseract.Output.DICT)
    confs = [float(c) for c in data["conf"] if float(c) >= 0]
    return " ".join(w for w in data["text"] if w.strip()), (sum(confs) / len(confs) if confs else 0.0)

There are three decisions in that block worth understanding.

Rotation comes first. Phone photos store their orientation in EXIF metadata. Ignore it and Tesseract reads a sideways image and confidently returns garbage.

A fixed threshold is a starting point. With highly variable lighting, an adaptive threshold (for example, with OpenCV) gives better results. The way to choose is to measure against real samples, not to guess.

The function returns confidence, not just text. image_to_data provides a confidence score per word. That average is the first signal for deciding whether calling the model is worth it at all.

If confidence is very low or the text comes out nearly empty, the receipt does not go to the LLM. It is rejected or sent to review. Paying for a model call to read garbage only adds cost and noise.

LLM integration: structured output or nothing

The most common mistake when integrating an LLM is asking for a natural-language answer and then searching it for keywords. It works in the demo. In production, the model changes a comma or adds an explanation, and the parser breaks.

The alternative is to treat the model like an API with a contract. You ask for JSON with explicit fields, and you validate that JSON against a schema before believing a single word of it.

from pydantic import BaseModel, Field, ValidationError
class Reading(BaseModel):
    is_receipt: bool
    store: str | None = None
    date: str | None = Field(default=None, pattern=r"^\d{4}-\d{2}-\d{2}$")
    total: float | None = Field(default=None, ge=0)
    confidence: float = Field(ge=0, le=1)         # the model says how sure it is
def interpret(raw_response: str) -> Reading | None:
    try:
        return Reading.model_validate_json(raw_response)
    except ValidationError:
        return None    # a response that breaks the contract does not exist

The schema does three jobs at once. It documents what you expect from the model, rejects malformed responses, and converts every field to the right type.

A few practices that make the contract sturdier:

• Put the schema in the prompt. Describe every field, its format, and what to do if it is not in the text: return null, never invent.
• Use the provider's structured output if it exists. JSON mode drastically reduces malformed responses. Validation is still mandatory.
• Low temperature. For extracting data you want the most likely answer, not creativity.
• One retry, not a loop. If the response does not validate, retry once. If it fails again, the receipt goes to error or review.

Deterministic rules after the model

The model reads. It does not decide. That separation is what makes the system auditable.

The promotion rules —which stores take part, which dates count, what the minimum amount is— live in code. They are applied to the already-validated reading, and every rejection carries a concrete reason that gets stored.

STORES = {"store-a", "store-b"}             # real catalog, in config, not in the prompt
VALIDITY = ("2026-05-01", "2026-07-31")
def decide(r: Reading, ocr_conf: float) -> tuple[str, str]:
    if not r.is_receipt:
        return "rejected", "does not look like a receipt"
    if r.confidence < 0.7 or ocr_conf < 60:
        return "review", "uncertain reading"  # doubting is a valid outcome
    if (r.store or "").lower() not in STORES:
        return "rejected", "store not in the promotion"
    if not r.date or not (VALIDITY[0] <= r.date <= VALIDITY[1]):
        return "rejected", "date outside the promotion period"
    return "approved", "meets every rule"

The sample values are made up; yours come from the promotion's terms.

What matters is the order. First whatever invalidates everything, then doubt, and finally the business rules. And every outcome returns a readable reason: when a participant asks why they were rejected, the answer is already in the database.

There is one more advantage. If a rule changes tomorrow, you do not need to call the model again. Just rerun decide over the stored readings.

Scheduling: cron or a systemd timer

In the real case, the pipeline ran on cron every 2 minutes. Cron is simple and available everywhere. Its weakness is that it knows nothing about the previous run and keeps no logs on its own.

On a server with systemd, a timer solves that with no extra code. If the previous run is still alive, it does not start another. Output goes to journald. And you can cap memory and CPU so the pipeline never competes with the website.

# /etc/systemd/system/verifier.service
[Service]
Type=oneshot
User=verifier
ExecStart=/opt/verifier/venv/bin/python /opt/verifier/run.py
MemoryMax=1G        # if it overflows, the pipeline dies, not the website
CPUQuota=50%
# /etc/systemd/system/verifier.timer
[Timer]
OnUnitInactiveSec=2min   # counts from when it finished: runs never overlap
[Install]
WantedBy=timers.target

Turning it on in production calls for care:

# ⚠️ Production: back up the receipts table before the first start.
mysqldump --single-transaction promo receipts > "receipts-$(date +%F).sql"

sudo systemctl daemon-reload
sudo systemctl enable --now verifier.timer

# Check when it runs and what the last execution said.
systemctl list-timers verifier.timer
journalctl -u verifier.service --since "15 min ago" --no-pager

OnUnitInactiveSec counts the interval from when the previous run finished, not from when it started. If one execution takes longer than usual, the next one simply waits.

Best practices for bulk processing

With the workflow in place, what remains is being able to trust it weeks later, when nobody is watching.

• Store everything in between. OCR text, confidence, the model's raw response, the validated reading and the decision. Without that, an error cannot be reproduced.
• Version the prompt and the rules. Every decision records which version produced it. If you change the prompt, you know which receipts were evaluated with which.
• Build a labeled sample set. A few dozen real receipts with the correct answer. Every change to preprocessing, prompt or model is measured against them before deploying.
• Measure per state, not per log line. How many are pending, how many were approved, rejected or left in doubt per hour, and how long each stage takes.
• Alert on trends. A pending queue that grows run after run, or an error rate that climbs, warns you before anyone complains.
• Hash every image. A resubmitted receipt should not cost another round of OCR and model, or be counted twice.

Observability does not require an expensive platform. A query grouped by state and by hour, checked by a script that alerts when something falls out of range, covers most cases.

What NOT to do

• Put business rules in the prompt. "Reject if the date is outside the promotion period" feels convenient, but the model can be wrong with total confidence and leaves no auditable reason. The model extracts; the code decides.
• Send raw images to OCR and blame the model. If the extracted text is bad, no prompt will fix it. Invest in preprocessing first and measure its effect.
• Parse free text with regular expressions. Searching for "approved" in a natural-language answer breaks at the first variation. Demand JSON and validate it against a schema.

The next step

Automating an OCR + LLM pipeline with Python is not chaining two calls. It is designing states, preparing the input, holding the model to a contract, deciding in code, and measuring everything that happens.

If your team needs to automate document verification, remove manual work from an image-based process, or make an LLM integration reliable when today it fails silently, at Conqr Digital Agency we design architectures that scale. Let's review your case.

The system that inspired this article is documented in the Panini Fan Zone case study. And if bringing AI into real processes is your goal, take a look at the AI automation service.

FAQ

What preprocessing does an image need before Tesseract?

At minimum: convert it to grayscale, upscale it if it is small, and apply a threshold to separate text from background. For phone photos, fixing the rotation also helps. Every step is measured against a set of real samples.

How do I force an LLM to always answer in the same format?

Ask for JSON with explicit fields, use your provider's structured output mode if it has one, and validate the response with a schema, for example a pydantic model. If validation fails, the response is treated as an error and either retried or sent to review.

Is cron or a systemd timer better for scheduling the pipeline?

Both work. Cron is universal and simple. A systemd timer adds journald logging, resource limits, and prevents overlapping runs with no extra code. On a server with systemd, the timer gives more control for little effort.