Read time: ~11 minutes. What you’ll build: a real batch pipeline that reads a folder of receipt/invoice images (and multi-page PDFs), extracts structured JSON against a typed schema with Qwen3-VL, validates every result, retries transient failures, runs documents concurrently, and writes results.json + results.csv. Every code block below is taken from a pipeline that actually runs — the outputs shown are real, not mocked.

Most “extract data from documents with an LLM” tutorials stop at a single client.chat.completions.create() call on one image and call it a pipeline. Real document work isn’t one image — it’s a folder of 500, some of which the model will misread, return as broken JSON, or hallucinate a total for. A pipeline is the boring 80% nobody shows: the schema, the validation, the retries, the concurrency, and the CSV your finance team actually opens.

This guide builds that 80% with Qwen3-VL — Alibaba’s vision-language model — accessed through an OpenAI-compatible endpoint. The model swap is one line, so everything here works just as well against a self-hosted NuExtract 3 server or any other OpenAI-compatible VLM. The pipeline is the durable part; the model behind it is replaceable.

---

1. What you’re building

The data flow is a fan-out/fan-in:

samples/                    extract_one()            results.json
  receipt_01.png  ──┐      ┌─ to data URL           results.csv
  receipt_02.png  ──┼─►    ├─ call Qwen3-VL    ──►   (one row per
  invoice_03.pdf  ──┘      ├─ parse + repair JSON     document)
                           ├─ validate (schema +
                           │   business rules)
                           └─ retry on failure

Five responsibilities, each a section below:

1. A typed schema (Pydantic) so output is validated, not just “JSON-shaped.”
2. Extraction — image/PDF → JSON, with prompt design that forces clean output.
3. Validation — schema typing plus business rules the schema can’t express (does the line-item sum exceed the stated total?).
4. Resilience — retries with backoff for the calls that fail, because at scale some will.
5. Batch + export — concurrency over the folder, then results.csv.

---

2. Setup

You need Python 3.10+, an API key, and three packages:

python -m venv .venv && source .venv/bin/activate
pip install openai pydantic pymupdf

• openai — the client. Qwen3-VL is served on a DashScope endpoint that speaks the OpenAI chat-completions protocol, so the official OpenAI SDK talks to it directly.
• pydantic — typed validation of the model’s output.
• pymupdf — rasterizes PDF pages to images (VLMs read pixels, not PDF structure).

Get a DashScope API key from the Alibaba Cloud Model Studio console and export it:

export DASHSCOPE_API_KEY="sk-..."

The client points at the compatible-mode base URL:

from openai import OpenAI

client = OpenAI(
    api_key=os.environ["DASHSCOPE_API_KEY"],
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)

MODEL = "qwen3-vl-plus"  # vision-capable; swap for any OpenAI-compatible VLM

That base_url is the only line tying this to DashScope. Point it at http://localhost:8000/v1 and you’re driving a self-hosted vLLM model instead — see running NuExtract 3 locally for that server setup.

---

3. The typed schema (this is what makes it a pipeline)

The difference between a demo and a pipeline is that a pipeline refuses malformed data. Define the shape once as Pydantic models and every extraction is validated against it for free:

from pydantic import BaseModel, field_validator
from datetime import datetime

class LineItem(BaseModel):
    name: str
    price: float

class Receipt(BaseModel):
    store: str
    date: str
    currency: str
    total: float
    payment_method: str | None = None
    items: list[LineItem]

    @field_validator("date")
    @classmethod
    def date_parseable(cls, v):
        for fmt in ("%Y-%m-%d", "%Y-%m-%dT%H:%M:%S", "%Y-%m-%dT%H:%M"):
            try:
                datetime.strptime(v, fmt)
                return v
            except ValueError:
                continue
        raise ValueError(f"unparseable date: {v!r}")

If the model returns "total": "thirteen dollars" or a date it invented in the wrong format, Receipt(**data) raises instead of silently poisoning your dataset. That single line — constructing the Pydantic model — is your data quality gate.

Separately, you hand the model a human-readable version of the schema as part of the prompt. Keep these two in sync; the prompt schema tells the model what to produce, the Pydantic model enforces it:

SCHEMA_FOR_MODEL = {
    "store": "string (exact store name)",
    "date": "string (ISO 8601 date or datetime)",
    "currency": "string (ISO 4217 code, e.g. USD/JPY)",
    "total": "number",
    "payment_method": "string or null",
    "items": [{"name": "string", "price": "number"}],
}

---

4. Extraction: image → JSON

A VLM reads the image directly — no OCR step. The prompt does three jobs: set the role, hand over the schema, attach the image. The system prompt is where you kill the two most common failure modes — chatty preambles and markdown fences:

SYSTEM = (
    "You are a precise document-extraction engine. Given a receipt image and a "
    "JSON schema, return ONLY valid JSON matching the schema. Copy values "
    "verbatim where possible. Use null for missing fields. No commentary, no "
    "markdown fences."
)

The image is passed as a base64 data URL, and temperature=0 makes extraction near-deterministic:

import base64, json
from pathlib import Path

def image_data_url(path: Path) -> str:
    b64 = base64.b64encode(path.read_bytes()).decode()
    return f"data:image/png;base64,{b64}"

def call_model(data_urls: list[str]) -> str:
    content = [{"type": "text", "text": "Schema:\n" + json.dumps(SCHEMA_FOR_MODEL, indent=2)}]
    content += [{"type": "image_url", "image_url": {"url": u}} for u in data_urls]
    resp = client.chat.completions.create(
        model=MODEL,
        temperature=0,
        messages=[
            {"role": "system", "content": SYSTEM},
            {"role": "user", "content": content},
        ],
    )
    return resp.choices[0].message.content

Even with “no markdown fences” in the system prompt, models occasionally wrap output in ```json. Don’t trust the instruction — strip defensively before parsing:

def clean_json(text: str) -> str:
    t = text.strip()
    if t.startswith("```"):
        t = t.split("\n", 1)[1] if "\n" in t else t   # drop the ```json line
        t = t.removeprefix("json").strip()
        t = t.rsplit("```", 1)[0].strip()              # drop the closing fence
    return t

On a real receipt image, the raw output comes back as exactly this — clean, typed, verbatim store name, ISO datetime, line items with prices:

{
  "store": "BLUE BOTTLE COFFEE",
  "date": "2026-06-10T14:32:00",
  "currency": "USD",
  "total": 13.9,
  "payment_method": "VISA ****4821",
  "items": [
    {"name": "Latte (12oz)", "price": 5.5},
    {"name": "Croissant", "price": 4.25},
    {"name": "Sparkling Water", "price": 3.0}
  ]
}

---

5. Validation the schema can’t express

Pydantic guarantees types. It can’t tell you the extraction is wrong. The classic failure: the model reads the receipt correctly except it grabs a line-item price as the total, and every type is still valid. You need business rules.

The most useful one for receipts: the line-item sum should never exceed the stated total. A total higher than the sum is normal (tax, tip, fees); a total lower than the line items means the model misread something.

TOTAL_TOLERANCE = 0.02  # ignore sub-2-cent rounding noise

def validate_business_rules(r: Receipt) -> dict:
    flags = []
    items_sum = round(sum(i.price for i in r.items), 2)
    if items_sum - r.total > TOTAL_TOLERANCE:
        flags.append(f"items_sum {items_sum} > total {r.total}")
    if not r.items:
        flags.append("no line items")
    if r.total <= 0:
        flags.append("non-positive total")
    return {"passed": not flags, "flags": flags, "items_sum": items_sum}

This catches the silent errors typing never will. A flagged receipt isn’t discarded — it’s marked for human review, which is exactly the queue you want: clean rows auto-process, suspicious ones get eyes.

Why a one-directional check? On a real US receipt, total = subtotal + tax, so total > items_sum almost always. Flagging total != items_sum would mark every taxed receipt as broken. Flagging only items_sum > total catches genuine misreads without the false positives. Tune the rule to your documents — that’s the point of keeping it separate from the schema.

---

6. Resilience: retry the calls that fail

At one document, retries feel like over-engineering. At 500, some calls will hit a rate limit, a timeout, or return JSON that won’t parse. The pipeline must distinguish two failure classes and treat them differently:

• Parse/validation errors — retry immediately (a re-roll at temperature=0 can still differ enough to parse).
• API/transport errors — back off before retrying, so you don’t hammer a rate-limited endpoint.

from pydantic import ValidationError

MAX_RETRIES = 3

def extract_one(path: Path) -> dict:
    data_urls = file_to_data_urls(path)
    last_err = None
    for attempt in range(1, MAX_RETRIES + 1):
        try:
            raw = call_model(data_urls)
            data = json.loads(clean_json(raw))
            receipt = Receipt(**data)                 # type gate
            return {
                "file": path.name, "status": "ok", "attempts": attempt,
                "validation": validate_business_rules(receipt),
                "data": receipt.model_dump(),
            }
        except (json.JSONDecodeError, ValidationError, KeyError, TypeError) as e:
            last_err = f"parse/validation: {e}"        # retry now
        except Exception as e:
            last_err = f"api: {e}"
            time.sleep(1.5 * attempt)                  # back off, then retry
    return {"file": path.name, "status": "failed",
            "attempts": MAX_RETRIES, "error": last_err}

Every record carries its attempts count — useful telemetry. If attempts creep up across a batch, your prompt or your image quality needs work.

---

7. Handling PDFs

Real invoices are multi-page PDFs. VLMs don’t read PDF structure — rasterize each page to PNG with PyMuPDF and pass them as a list of images in one request. The model reads across all pages and returns a single merged JSON:

def file_to_data_urls(path: Path, dpi: int = 170) -> list[str]:
    if path.suffix.lower() == ".pdf":
        import fitz  # PyMuPDF
        urls = []
        with fitz.open(path) as doc:
            for page in doc:
                png = page.get_pixmap(dpi=dpi, alpha=False).tobytes("png")
                urls.append("data:image/png;base64," + base64.b64encode(png).decode())
        return urls
    b64 = base64.b64encode(path.read_bytes()).decode()
    mime = "image/png" if path.suffix.lower() == ".png" else "image/jpeg"
    return [f"data:{mime};base64,{b64}"]

dpi=170 is the sweet spot — high enough for small print, low enough to keep the vision-token count (and your bill) down. Because both images and PDFs come back as a list of data URLs, the rest of the pipeline doesn’t care which it got.

---

8. Batch + concurrency

Documents are independent, so process them concurrently. A thread pool is the right tool — the work is I/O-bound (waiting on the API), not CPU-bound:

from concurrent.futures import ThreadPoolExecutor, as_completed

def run_batch(samples: Path):
    files = sorted(p for p in samples.iterdir()
                   if p.suffix.lower() in {".png", ".jpg", ".jpeg", ".pdf"})
    results = []
    with ThreadPoolExecutor(max_workers=4) as pool:
        futs = {pool.submit(extract_one, f): f for f in files}
        for fut in as_completed(futs):
            res = fut.result()
            mark = "OK " if res["status"] == "ok" else "ERR"
            if res["status"] == "ok" and not res["validation"]["passed"]:
                mark = "WARN"
            print(f"  [{mark}] {res['file']} (try {res['attempts']})")
            results.append(res)
    return results

Keep max_workers modest (4–8) — it’s bounded by your API rate limit, not your CPU. Push it too high and you’ll trade parse retries for rate-limit backoffs and net out slower.

Running the whole thing over a folder of three receipts:

[pipeline] model=qwen3-vl-plus  files=3  workers=4
  [OK ] receipt_03.png (try 1)
  [OK ] receipt_01.png (try 1)
  [OK ] receipt_02.png (try 1)
[pipeline] done in 5.7s  ok=3 warn=0 failed=0
[pipeline] wrote results.json + results.csv

Three documents — including a Japanese-yen receipt — extracted correctly on the first attempt, concurrently, in 5.7 seconds. Note the out-of-order completion: that’s the concurrency working.

---

9. Export to CSV

JSON is for the next program; CSV is for the human. Collapse the nested line items into one JSON-string column so each document is a single row:

import csv

def write_csv(results, path="results.csv"):
    cols = ["file", "status", "store", "date", "currency", "total",
            "payment_method", "items_sum", "validation_passed", "items_json"]
    with open(path, "w", newline="") as f:
        w = csv.DictWriter(f, fieldnames=cols)
        w.writeheader()
        for r in results:
            if r["status"] != "ok":
                w.writerow({"file": r["file"], "status": r["status"]})
                continue
            d = r["data"]
            w.writerow({
                "file": r["file"], "status": "ok",
                "store": d["store"], "date": d["date"], "currency": d["currency"],
                "total": d["total"], "payment_method": d["payment_method"],
                "items_sum": r["validation"]["items_sum"],
                "validation_passed": r["validation"]["passed"],
                "items_json": json.dumps(d["items"], ensure_ascii=False),
            })

The real output, with the validation column that tells your reviewer which rows to trust:

file	store	currency	total	items_sum	validation_passed
receipt_01.png	BLUE BOTTLE COFFEE	USD	13.9	12.75	True
receipt_02.png	DAISO JAPAN	JPY	550	550	True
receipt_03.png	THE HOME DEPOT	USD	42.0	38.44	True

The USD receipts show total > items_sum (sales tax); the JPY receipt shows total == items_sum (tax-included pricing). The one-directional rule from §5 passes all three correctly — no false “broken” flags on the taxed receipts.

---

10. Qwen3-VL (API) vs NuExtract 3 (self-hosted): which to reach for

This pipeline is model-agnostic by design, so the real question is what to plug into it.

Reach for Qwen3-VL through an API when you want zero infrastructure, your volume is moderate, your documents can leave your network, or you need a general VLM that also handles odd layouts and reasoning. You pay per call, but there’s no GPU to run.

Reach for a self-hosted model like NuExtract 3 when volume is high enough that per-page API costs dominate, documents can’t leave your infrastructure (legal, medical, financial compliance), or you want a fixed-cost pipeline instead of a metered one. NuExtract is a 4B model purpose-built for extraction that fits on a single consumer GPU — and because it serves an OpenAI-compatible endpoint too, you change one base_url line in this exact pipeline to switch.

That’s the payoff of investing in the pipeline rather than the prompt: the schema, validation, retries, and CSV export are identical either way. The model becomes a deployment decision, not a rewrite. For the local-serving side of that decision — the vLLM command, the template language, the VRAM math — see the NuExtract 3 local setup guide, and for the broader “what fits on my GPU” question, the local model hardware walkthrough.

---

The takeaway

A document-extraction pipeline is five things a one-shot demo skips: a typed schema that rejects malformed output, business rules that catch the errors typing can’t, retries that split parse failures from API failures, concurrency over the folder, and a CSV your reviewers can open. Get those right once and the model behind them — Qwen3-VL over an API today, a self-hosted NuExtract tomorrow — is a single-line swap. Start with one receipt, a Pydantic model, and temperature=0; you’ll have validated rows coming out in minutes, and the same code will hold when the folder has 500.

Sources

• Alibaba Cloud Model Studio (DashScope) docs — OpenAI-compatible endpoint, Qwen-VL model names, and rate limits
• OpenAI Python SDK — the client used for the compatible endpoint
• Pydantic docs — schema validation and field validators
• PyMuPDF docs — PDF-to-image rasterization
• Pipeline built and run on 2026-06-13; the outputs shown (3 receipts, 5.7s, all validated) are from a real run against qwen3-vl-plus. Sample receipts are synthetic; extraction results are not edited.