Extraction you can measure
By the end: You can measure extraction accuracy field by field, and run it over thousands of documents without surprises.
- A labelled set of documents
- Field-level accuracy, not document-level
- Normalising before comparing
- Lists: precision and recall
- Retries, review queues and batch pricing
New to structured output? Often you do not want prose back, you want data your program can use: named fields with values, of the kind JSON holds. Structured output is how you ask for that and, when the API supports a schema, how you enforce it — you hand over the shape you expect and the model is prevented from producing anything that would not fit, so what arrives always parses. Why it matters
Extraction works on the ten documents you tried it on. This section is about finding out whether it works on ten thousand. Score fields separately rather than marking a whole document wrong, because a pipeline that misreads only the total looks identical to one that is slightly wrong everywhere. Normalise before comparing, or you will count correct answers as failures and spend a week chasing a problem that does not exist. Then the operational half: what happens when validation fails, and what running a backlog actually costs.
A pipeline you can run#
Everything below is measured against the same forty synthetic invoices and a stub extractor that stands in for the model call. The stub is deliberately wrong in one realistic way: some layouts have two lines beginning with the word Total, one before VAT and one after, and it takes the first. That single flaw is enough to produce every symptom this section is about, and because it is deterministic you can run the numbers yourself and get the same ones.
import re
MONTHS = ["January", "February", "March", "April", "May", "June"]
def make_invoice(i):
"""One synthetic invoice and the truth to score it against."""
day, month, year = 1 + i % 28, i % 6, 2024
net, tax = 100.0 + 7 * i, round((100.0 + 7 * i) * 0.19, 2)
gross = round(net + tax, 2)
printed_date = (f"{day} {MONTHS[month]} {year}" if i % 3 == 0
else f"{day:02d}/{month + 1:02d}/{year}")
items = [("Licence", net * .5), ("Support", net * .3), ("Hosting", net * .2)][: 2 + i % 2]
lines = [f"Invoice INV-{1000 + i}", f"Issued {printed_date}"]
lines += [f"- {name} EUR {amount:.2f}" for name, amount in items]
if i % 5 == 0: # a layout with two lines starting Total
lines += [f"Total excl. VAT EUR {net:.2f}", f"VAT EUR {tax:.2f}",
f"Total incl. VAT EUR {gross:.2f}"]
else:
lines += [f"Subtotal EUR {net:.2f}", f"VAT EUR {tax:.2f}",
f"Total EUR {gross:.2f}"]
truth = {"invoice_number": f"INV-{1000 + i}",
"date": f"{year:04d}-{month + 1:02d}-{day:02d}",
"total": f"{gross:.2f}", "currency": "EUR",
"items": [name for name, _ in items]}
return "\n".join(lines), truth
DOCS, LABELS = {}, {}
for i in range(40):
text, truth = make_invoice(i)
DOCS[f"doc-{i:03d}"], LABELS[f"doc-{i:03d}"] = text, truth
TOTAL_LINE = re.compile(r"(?mi)^Total[^0-9]*(EUR|USD|GBP) ([0-9]+\.[0-9]{2})")
def extract(doc_id, take_last=False):
"""Stands in for one model call. Deterministic, and wrong the way a model
is wrong: on a document with two lines beginning Total it picks one."""
text = DOCS[doc_id]
money = TOTAL_LINE.findall(text)
chosen = money[-1] if take_last and money else (money[0] if money else (None, None))
items = re.findall(r"(?m)^- (\w+) EUR", text)[:2]
if doc_id.endswith("0"):
items = items + ["Delivery"] # an item the document does not contain
return {"invoice_number": re.search(r"(INV-[0-9]+)", text).group(1),
"date": re.search(r"Issued (.+)", text).group(1),
"total": chosen[1], "currency": chosen[0],
"source_text": None if chosen[1] is None else f"EUR {chosen[1]}",
"items": items}
row = extract("doc-005")
print(DOCS["doc-005"])
print({k: row[k] for k in ("total", "currency", "items")})
# Invoice INV-1005
# Issued 06/06/2024
# - Licence EUR 67.50
# - Support EUR 40.50
# - Hosting EUR 27.00
# Total excl. VAT EUR 135.00
# VAT EUR 25.65
# Total incl. VAT EUR 160.65
# {'total': '135.00', 'currency': 'EUR', 'items': ['Licence', 'Support']}Normalise first#
'2024-03-05', '5 March 2024' and '05/03/2024' may all be right. Without normalising, your score counts correct answers as wrong and you chase problems that do not exist. Normalise dates, numbers, currency symbols, case and whitespace before comparing, and decide explicitly what counts as the same.
from datetime import datetime
FORMATS = ["%Y-%m-%d", "%d/%m/%Y", "%d %B %Y", "%B %d, %Y"]
def normalise(value):
if value is None:
return None
s = str(value).strip()
for fmt in FORMATS:
try:
return datetime.strptime(s, fmt).strftime("%Y-%m-%d")
except ValueError:
pass
cleaned = s.replace(",", "")
for code in ("EUR", "USD", "GBP"):
cleaned = cleaned.replace(code, "")
try:
return f"{float(cleaned.strip()):.2f}" # money, to a fixed number of places
except ValueError:
return s.lower() # everything else, case-folded
for raw in ["5 March 2024", "05/03/2024", "2024-03-05", " EUR 1,234.50 ", "EUR"]:
print(f"{raw!r:>18} -> {normalise(raw)!r}")
# '5 March 2024' -> '2024-03-05'
# '05/03/2024' -> '2024-03-05'
# '2024-03-05' -> '2024-03-05'
# ' EUR 1,234.50 ' -> '1234.50'
# 'EUR' -> 'eur'Score each field#
Scoring a whole document as right or wrong hides everything useful. A pipeline that gets the total wrong on 20% of invoices and every other field right looks the same as one that is slightly wrong everywhere. Score each field separately.
from collections import defaultdict
def score(labelled, predicted):
right, total = defaultdict(int), defaultdict(int)
for doc_id, truth in labelled.items():
pred = predicted[doc_id]
for field, expected in truth.items():
if isinstance(expected, list):
continue # lists need precision and recall
total[field] += 1
right[field] += normalise(pred.get(field)) == normalise(expected)
rates = {f: right[f] / total[f] for f in total}
worst = min(rates, key=rates.get)
for field, rate in rates.items():
print(f"{field:16} {rate:5.0%}{' <- start here' if field == worst else ''}")
return rates
PRED = {d: extract(d) for d in DOCS}
rates = score(LABELS, PRED)
raw = sum(PRED[d]["date"] == LABELS[d]["date"] for d in DOCS) / len(DOCS)
print(f"date, compared as raw strings: {raw:.0%}")
# invoice_number 100%
# date 100%
# total 80% <- start here
# currency 100%
# date, compared as raw strings: 0%Lists need precision and recall#
For a list such as invoice line items, the model can miss items or invent them. Report both: of the items it returned, how many were real (precision), and of the real items, how many it found (recall). One number hides which mistake it makes.
def list_scores(labelled, predicted, field="items"):
tp = fp = fn = 0
for doc_id, truth in labelled.items():
got, want = set(predicted[doc_id][field]), set(truth[field])
tp += len(got & want) # found and real
fp += len(got - want) # returned but not in the document
fn += len(want - got) # in the document and missed
return tp / (tp + fp), tp / (tp + fn)
p, r = list_scores(LABELS, PRED)
print(f"line items precision {p:.2f} recall {r:.2f}")
# line items precision 0.95 recall 0.80Validate, and decide where the failures go#
A schema check is cheap and catches the rows that are not merely wrong but unusable: a null where a number is required, a date the parser cannot read, a currency outside the enum. Run it on every row before anything reaches a database, and count the failures by field rather than in total, because the field is what tells you which part of the prompt to change.
from pydantic import BaseModel, ValidationError, field_validator
class Invoice(BaseModel):
invoice_number: str
date: str
total: float
currency: str
@field_validator("date")
@classmethod
def iso(cls, v):
return datetime.strptime(normalise(v), "%Y-%m-%d").strftime("%Y-%m-%d")
def validate_all(predicted):
ok, failed = {}, defaultdict(list)
for doc_id, row in predicted.items():
keep = {k: row[k] for k in ("invoice_number", "date", "total", "currency")}
try:
ok[doc_id] = Invoice(**keep)
except ValidationError as e:
failed[e.errors()[0]["loc"][0]].append(doc_id)
return ok, failed
broken = dict(PRED)
broken["doc-003"] = dict(PRED["doc-003"], total=None) # nothing found
broken["doc-004"] = dict(PRED["doc-004"], date="not a date") # unparseable
ok, failed = validate_all(broken)
print(f"{len(ok)} rows stored, {sum(len(v) for v in failed.values())} to the review queue")
for field, ids in failed.items():
print(f" {field}: {ids}")
# 38 rows stored, 2 to the review queue
# total: ['doc-003']
# date: ['doc-004']- Retry once, sending back the validation error. This fixes most formatting slips and costs little.
- If it fails again, send the document to a review queue rather than storing a guess.
- Track the retry rate and the review rate. Either rising is an early warning.
- Never let a retry loop run unbounded. One retry, then the queue, or a bad document costs you the same call fifty times.
Concurrency, and how many workers to run#
Forty documents run one at a time is a minute. Two hundred and fifty thousand is a week. The work is entirely waiting on the network, so a small pool of concurrent calls with a semaphore holding the ceiling is all the machinery you need — but the ceiling is not a number you guess. It comes from the rate limit and the call latency, by Little's law: the number of requests in flight equals the rate multiplied by how long each one takes.
import asyncio
async def worker_pool(doc_ids, concurrency=8):
gate = asyncio.Semaphore(concurrency)
inflight = peak = 0
async def one(doc_id):
nonlocal inflight, peak
async with gate:
inflight += 1
peak = max(peak, inflight)
await asyncio.sleep(0.005) # stands in for the model call
result = extract(doc_id)
inflight -= 1
return doc_id, result
done = await asyncio.gather(*(one(d) for d in doc_ids))
return dict(done), peak
results, peak = asyncio.run(worker_pool(list(DOCS)))
print(f"{len(results)} documents, peak concurrency {peak}")
for rpm, latency in [(4000, 2.5), (4000, 0.4), (500, 2.5)]:
print(f"limit {rpm:>5}/min, {latency:.1f}s per call -> {rpm / 60 * latency:5.1f} workers")
# 40 documents, peak concurrency 8
# limit 4000/min, 2.5s per call -> 166.7 workers
# limit 4000/min, 0.4s per call -> 26.7 workers
# limit 500/min, 2.5s per call -> 20.8 workersRate limits are usually two numbers, requests per minute and tokens per minute, and the token one binds first on extraction because each document is long. When you are throttled the provider says so with a 429 and often a retry-after header; honour it, back off exponentially with jitter, and treat a sustained 429 rate as a signal to shrink the pool rather than to retry harder. A pool that retries harder under throttling converts a slow job into a stalled one.
Resuming a run that died#
A job of that size will be interrupted: a deploy, a provider incident, a laptop lid. The rule is that nothing is in memory that you would mind losing. Append each finished row to a file as it completes, flush, and start every run by reading back what is already there. The job becomes restartable without any coordination, and the second run pays only for what the first one did not finish.
import json, pathlib
CHECKPOINT = pathlib.Path("extracted.jsonl")
def run(doc_ids, stop_after=None):
done = set()
if CHECKPOINT.exists():
done = {json.loads(line)["id"] for line in CHECKPOINT.read_text().splitlines()}
todo = [d for d in doc_ids if d not in done]
made = 0
with CHECKPOINT.open("a") as fh:
for doc_id in todo:
if stop_after is not None and made == stop_after:
raise RuntimeError("provider returned 500")
fh.write(json.dumps({"id": doc_id, "row": extract(doc_id)}) + "\n")
fh.flush() # the flush is the whole point
made += 1
return len(done), made
CHECKPOINT.unlink(missing_ok=True)
try:
run(list(DOCS), stop_after=17)
except RuntimeError as e:
print("crashed:", e)
skipped, made = run(list(DOCS))
print(f"second run: skipped {skipped} already done, made {made} calls")
# crashed: provider returned 500
# second run: skipped 17 already done, made 23 callsMeasuring quality when you have no labels#
Forty labelled documents is an afternoon. Two hundred and fifty thousand is not, so most of a backlog is scored by proxies that need no labels at all. Two are worth having. The first is agreement: run the extraction twice under settings that ought not to change the answer and count the fields that differ, because a field the pipeline cannot agree with itself about is a field it does not know. The second is grounding: check that the text the model quoted really appears in the document.
second = {d: extract(d, take_last=True) for d in DOCS} # the other plausible reading
def disagreement(a, b, fields):
return {f: sum(normalise(a[d][f]) != normalise(b[d][f]) for d in a) / len(a)
for f in fields}
def grounded(pred):
return sum(bool(p["source_text"]) and p["source_text"] in DOCS[d]
for d, p in pred.items()) / len(pred)
fields = ["invoice_number", "date", "total", "currency"]
dis = disagreement(PRED, second, fields)
print(f"{'field':16}{'disagree':>10}{'accuracy':>10}")
for f in fields:
print(f"{f:16}{dis[f]:>10.0%}{rates[f]:>10.0%}")
print(f"grounded (quoted text really in the document): {grounded(PRED):.0%}")
# field disagree accuracy
# invoice_number 0% 100%
# date 0% 100%
# total 20% 80%
# currency 0% 100%
# grounded (quoted text really in the document): 100%Agreement is a floor, not an estimate
Disagreement finds fields the pipeline is unsure about; it cannot find fields it is confidently wrong about. Here grounding reports a clean 100% because the wrong total, 135.00, genuinely is printed in the document — it is the wrong number correctly quoted. Both proxies are worth logging daily and neither is a substitute for a labelled set. Use them to decide where to spend labelling effort, and keep a few hundred labels as the only number you would quote to anyone.
Running it over a backlog#
For work that does not need an answer in seconds, most major providers offer a batch interface that processes requests within hours at a large discount, often around half price. Extraction over an archive is the perfect case. Run the eval on a sample first, since a mistake in the prompt is repeated ten thousand times.
IN_PER_M, OUT_PER_M = 3.00, 15.00 # usd per million tokens, check yours
in_tok = 4800 / 4 # a one-page invoice, about 4 chars per token
out_tok = 60 # the filled schema, measured on the eval set
per_doc = (in_tok * IN_PER_M + out_tok * OUT_PER_M) / 1e6
print(f"{in_tok:.0f} input tokens per document, {out_tok} out")
for n, label, factor in [(40, "the eval set", 1.0),
(250_000, "the backlog", 1.0),
(250_000, "the backlog, batch", 0.5),
(250_000, "the backlog, batch + 1 retry in 20", 0.525)]:
print(f"{label:38} {n:>7} docs {n * per_doc * factor:9.2f} usd")
# 1200 input tokens per document, 60 out
# the eval set 40 docs 0.18 usd
# the backlog 250000 docs 1125.00 usd
# the backlog, batch 250000 docs 562.50 usd
# the backlog, batch + 1 retry in 20 250000 docs 590.62 usdTwo levers move that figure more than the model choice does. Cutting input tokens is the first: send the page that has the totals on it rather than the whole document, and the bill falls in proportion. The second is not running the call at all — a document identical to one already processed, which happens more often in an archive than anyone expects, needs a lookup rather than an extraction.
Where people go wrong
Comparing raw strings. Half your 'errors' turn out to be formatting differences, and the real errors get lost among them.
If this comes up in an interview
How would you measure the accuracy of an invoice extraction pipeline?
Hand-label fifty or more real invoices, normalise values before comparing, and report accuracy per field. For lists such as line items report precision and recall separately. Then read the failures for the worst field first.
Then they push further
Your total field is 79% accurate. What do you look at?
The failures, grouped. Common causes are picking the subtotal instead of the total, missing tax, reading the wrong currency, or documents that parsed badly. Each has a different fix, so count them before changing anything.
How do you keep it accurate after launch?
Sample processed documents every week for review, route validation failures to people, and add every confirmed mistake to the labelled set. Watch the review rate, since new document layouts show up there first.
What gets marked up. Strong answers score per field, normalise before comparing, and look at failures before fixing. Document-level accuracy only, or no labelled set, is marked down.
When would you use a batch API?
For work that can wait hours, such as processing a backlog or nightly jobs. It is much cheaper per request. It is wrong for anything a user is waiting on.
How would you run an extraction over 250,000 documents?
Size the worker pool from the rate limit and the call latency rather than by guessing: requests in flight equals rate times latency, so a 4,000 per minute limit with 2.5 second calls justifies about 167 workers. Append each finished row to a file and start every run by reading back what is already there, so a crash costs only the unfinished part. Price it first on the eval set, and use the batch interface if hours are acceptable.
Then they push further
The job is being throttled. What do you change?
Honour the retry-after header, back off exponentially with jitter, and shrink the pool. Retrying harder under throttling turns a slow job into a stalled one. Check which limit is binding, since the token limit usually binds before the request limit on long documents.
How do you watch quality across all 250,000 when you can only label a few hundred?
Run two proxies that need no labels. Agreement: extract twice under settings that should not change the answer and count the fields that differ, since a field the pipeline cannot agree with itself about is one it does not know. Grounding: check the quoted source text really appears in the document. Both are floors rather than estimates, so keep the labelled set as the number you would quote.
What gets marked up. Strong answers derive the concurrency from the limit rather than picking a number, make the job restartable, and separate the unlabelled proxies from real accuracy instead of presenting them as the same thing.
Build this: Label thirty documents for an extraction task, write a field-level scorer with normalisation, and produce the per-field table. Fix the worst field and show the table again.
Watch it explained
This is one section of Structured output, 10 hours across 3 sections. Read it in full to get the whole topic in order.
Stop 11 of 50 on the AI Engineer route
0 of 50 topics done · 0h behind you, 603h to go
What you have marked done is stored in this browser only. It does not reach a server, and clearing site data will clear it.