Build the new ELT: 200 plain-text invoices go in, validated warehouse rows come out, and nothing reaches the invoices table without passing a deterministic validation gate — schema conformance, allowed currency codes, date sanity, line-arithmetic reconciliation, and a model-confidence floor.
Rejects land in a human-review queue directory with machine-readable reasons, and you finish by measuring field-level extraction accuracy on a 30-invoice labeled sample. Twelve invoices in the corpus are seeded with realistic defects (wrong printed totals, junk currency strings, missing and future dates); your gate must catch all twelve without you hard-coding their IDs. The discipline this lab trains is the one that separates a demo from a pipeline: LLM output is untrusted input, and the warehouse only ever sees rows that survived verification.
~150 minutes hands-on · Advanced tier (maps to module M8).
Laptop-only, $0 in dry-run mode. Extraction has two implementations behind one interface: extract_with_claude (structured outputs via the API — used when you have a key; the 200-document run costs roughly a dollar at Opus-tier prices) and extract_dry_run (a deterministic local parser that stands in for the model so the pipeline — gate, queue, warehouse, audit — is fully buildable and testable offline). The gate does not know or care which extractor produced the record; that indifference is the design point.
mkdir -p ~/labs/de-lab-12 && cd ~/labs/de-lab-12
python3 -m venv .venv && source .venv/bin/activate
pip install duckdb faker pydantic anthropic
Create make_invoices.py — a deterministic generator (seeded Faker) for 200 invoices with ground-truth labels and 12 seeded defects at fixed positions:
"""Generate 200 plain-text invoices with seeded defects + ground-truth labels."""
import json, random, pathlib
from faker import Faker
fake = Faker("en_IN")
Faker.seed(7)
random.seed(7)
OUT = pathlib.Path("invoices"); OUT.mkdir(exist_ok=True)
CURRENCIES = ["INR", "USD", "EUR"]
ITEMS = ["Annual support plan", "Sensor calibration", "Spare bearing kit", "On-site inspection",
"Firmware upgrade", "Coolant refill service", "Vibration audit", "Filter replacement set"]
truth = []
defect_rows = []
for n in range(200):
inv_id = f"INV-2026-{n:04d}"
vendor = fake.company()
date = fake.date_between(start_date="-120d", end_date="today").isoformat()
currency = random.choice(CURRENCIES)
n_lines = random.randint(1, 4)
lines = []
for _ in range(n_lines):
qty = random.randint(1, 5)
unit = round(random.uniform(500, 20000), 2)
lines.append({"desc": random.choice(ITEMS), "qty": qty, "unit_price": unit,
"amount": round(qty * unit, 2)})
total = round(sum(l["amount"] for l in lines), 2)
defect = None
body_total, body_currency, body_date = total, currency, date
if n in (11, 47, 83, 141): # arithmetic mismatch: printed total is wrong
body_total = round(total * random.uniform(1.05, 1.30), 2); defect = "total_mismatch"
elif n in (23, 96, 168): # invalid currency code
body_currency = random.choice(["Rs.", "US$", "EURO"]); defect = "bad_currency"
elif n in (59, 120): # missing invoice date
body_date = ""; defect = "missing_date"
elif n in (74, 155, 189): # future-dated invoice
body_date = fake.date_between(start_date="+30d", end_date="+120d").isoformat()
defect = "future_date"
text_lines = [f"TAX INVOICE {inv_id}", f"Vendor: {vendor}"]
if body_date:
text_lines.append(f"Invoice date: {body_date}")
text_lines.append(f"Currency: {body_currency}")
text_lines.append("-" * 40)
for l in lines:
text_lines.append(f"{l['desc']:28s} x{l['qty']} @ {l['unit_price']:.2f} = {l['amount']:.2f}")
text_lines.append("-" * 40)
text_lines.append(f"GRAND TOTAL: {body_total:.2f} {body_currency}")
(OUT / f"{inv_id}.txt").write_text("\n".join(text_lines))
truth.append({"invoice_id": inv_id, "vendor": vendor, "date": date, "currency": currency,
"line_count": n_lines, "grand_total": total, "defect": defect})
if defect:
defect_rows.append((inv_id, defect))
pathlib.Path("ground_truth.jsonl").write_text("\n".join(json.dumps(t) for t in truth))
print(f"wrote 200 invoices; {len(defect_rows)} seeded defects")
Run it: python make_invoices.py → wrote 200 invoices; 12 seeded defects. Do not read ground_truth.jsonl yet — the gate must catch the defects on its own merits.
Define the contract as a Pydantic schema. Create extract_pipeline.py and start with the record types. The schema is the single source of truth for both extraction modes and for validation. The confidence field is part of the contract: the extractor must self-report, and the gate enforces a floor.
import argparse, json, pathlib, re
from datetime import date
from pydantic import BaseModel, Field
import duckdb
class InvoiceLine(BaseModel):
description: str
quantity: int = Field(ge=1)
unit_price: float = Field(ge=0)
amount: float = Field(ge=0)
class InvoiceRecord(BaseModel):
invoice_id: str
vendor: str
invoice_date: str = Field(description="ISO date YYYY-MM-DD, empty string if absent")
currency: str
lines: list[InvoiceLine]
grand_total: float
confidence: float = Field(ge=0, le=1,
description="Model's confidence the extraction is complete and correct")
Write both extractors. The API version uses structured outputs — client.messages.parse validates the response against the schema and hands back a typed object. Note the prompt line "copy values exactly as printed; do not correct arithmetic": you want the wrong printed total extracted faithfully so the gate can catch it — an extractor that silently fixes data destroys your ability to audit the source.
def extract_with_claude(text: str) -> InvoiceRecord:
import anthropic
client = anthropic.Anthropic()
response = client.messages.parse(
model="claude-opus-4-8",
max_tokens=2048,
thinking={"type": "adaptive"},
messages=[{
"role": "user",
"content": (
"Extract the invoice below into the structured schema. "
"Copy values exactly as printed; do not correct arithmetic. "
"Use an empty string for a missing date. Set confidence lower if any "
"field is ambiguous or missing.\n\n<invoice>\n" + text + "\n</invoice>"
),
}],
output_format=InvoiceRecord,
)
return response.parsed_output
def extract_dry_run(text: str) -> InvoiceRecord:
"""Deterministic stand-in for the LLM: parses the invoice layout directly."""
inv = re.search(r"TAX INVOICE\s+(\S+)", text).group(1)
vendor = re.search(r"Vendor: (.+)", text).group(1)
m = re.search(r"Invoice date: (.+)", text)
inv_date = m.group(1).strip() if m else ""
currency = re.search(r"Currency: (.+)", text).group(1).strip()
lines = []
for lm in re.finditer(r"(.+?)\s+x(\d+)\s+@ ([\d.]+)\s+= ([\d.]+)", text):
lines.append(InvoiceLine(description=lm.group(1).strip(), quantity=int(lm.group(2)),
unit_price=float(lm.group(3)), amount=float(lm.group(4))))
total = float(re.search(r"GRAND TOTAL: ([\d.]+)", text).group(1))
conf = 0.95 if inv_date else 0.60
return InvoiceRecord(invoice_id=inv, vendor=vendor, invoice_date=inv_date,
currency=currency, lines=lines, grand_total=total, confidence=conf)
Write the validation gate. Every check is deterministic and independent of the extractor. Returning a list of reasons (not a boolean) is what makes the review queue useful to a human.
ALLOWED_CURRENCIES = {"INR", "USD", "EUR"}
CONFIDENCE_FLOOR = 0.80
TOLERANCE = 0.01
def validate(rec: InvoiceRecord) -> list[str]:
reasons = []
if rec.currency not in ALLOWED_CURRENCIES:
reasons.append(f"currency '{rec.currency}' not in {sorted(ALLOWED_CURRENCIES)}")
if not rec.invoice_date:
reasons.append("missing invoice_date")
else:
try:
d = date.fromisoformat(rec.invoice_date)
if d > date.today():
reasons.append(f"future-dated invoice: {rec.invoice_date}")
except ValueError:
reasons.append(f"unparseable date: {rec.invoice_date}")
line_sum = round(sum(l.amount for l in rec.lines), 2)
if abs(line_sum - rec.grand_total) > TOLERANCE:
reasons.append(f"line sum {line_sum} != grand_total {rec.grand_total}")
if rec.confidence < CONFIDENCE_FLOOR:
reasons.append(f"confidence {rec.confidence} below floor {CONFIDENCE_FLOOR}")
return reasons
Wire the pipeline: extract → gate → warehouse or review queue. Batch-style processing here is a plain loop over documents calling the extractor once each — with retries handled by the SDK. Append to extract_pipeline.py:
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--dry-run", action="store_true")
args = ap.parse_args()
extract = extract_dry_run if args.dry_run else extract_with_claude
review = pathlib.Path("review_queue"); review.mkdir(exist_ok=True)
con = duckdb.connect("finance.duckdb")
con.execute("""CREATE TABLE IF NOT EXISTS invoices (
invoice_id VARCHAR PRIMARY KEY, vendor VARCHAR, invoice_date DATE,
currency VARCHAR, line_count INTEGER, grand_total DOUBLE,
confidence DOUBLE, extracted_by VARCHAR)""")
accepted = rejected = 0
for path in sorted(pathlib.Path("invoices").glob("*.txt")):
rec = extract(path.read_text())
reasons = validate(rec)
if reasons:
rejected += 1
(review / f"{rec.invoice_id}.json").write_text(json.dumps(
{"record": rec.model_dump(), "reasons": reasons, "source": str(path)}, indent=2))
else:
accepted += 1
con.execute("INSERT OR REPLACE INTO invoices VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
[rec.invoice_id, rec.vendor, rec.invoice_date, rec.currency,
len(rec.lines), rec.grand_total, rec.confidence,
"dry-run" if args.dry_run else "claude-opus-4-8"])
print(f"accepted={accepted} rejected={rejected} (review_queue/)")
print(con.execute("""SELECT currency, count(*), round(sum(grand_total),2)
FROM invoices GROUP BY 1 ORDER BY 1""").fetchall())
if __name__ == "__main__":
main()
Run python extract_pipeline.py --dry-run. Expected: accepted=188 rejected=12, and a currency breakdown over three valid codes only. Open two files in review_queue/ and check that the reasons array would let an accounts-payable clerk act without seeing your code.
Reconcile the gate against ground truth. Now read ground_truth.jsonl: python -c "import json,pathlib; [print(t['invoice_id'], t['defect']) for t in map(json.loads, pathlib.Path('ground_truth.jsonl').read_text().splitlines()) if t['defect']]". Compare against ls review_queue/. All 12 seeded IDs must be present and — equally important — nothing else: a gate that also rejects clean invoices has a precision problem that silently starves the warehouse. If any defect slipped through, fix the gate, not the generator.
Measure field-level accuracy on a labeled sample. Create audit_accuracy.py — it samples 30 clean invoices and compares warehouse rows to ground truth per field:
import json, random, pathlib, duckdb
random.seed(30)
truth = {t["invoice_id"]: t for t in
(json.loads(l) for l in pathlib.Path("ground_truth.jsonl").read_text().splitlines())}
sample = random.sample([i for i, t in truth.items() if t["defect"] is None], 30)
con = duckdb.connect("finance.duckdb", read_only=True)
fields = ["vendor", "invoice_date", "currency", "line_count", "grand_total"]
correct = {f: 0 for f in fields}
for inv in sample:
row = con.execute("""SELECT vendor, invoice_date, currency, line_count, grand_total
FROM invoices WHERE invoice_id = ?""", [inv]).fetchone()
t = truth[inv]
got = {"vendor": row[0], "invoice_date": str(row[1]), "currency": row[2],
"line_count": row[3], "grand_total": round(row[4], 2)}
exp = {"vendor": t["vendor"], "invoice_date": t["date"], "currency": t["currency"],
"line_count": t["line_count"], "grand_total": round(t["grand_total"], 2)}
for f in fields:
if got[f] == exp[f]: correct[f] += 1
for f in fields:
print(f"{f:14s} {correct[f]}/30 = {100*correct[f]/30:.0f}%")
In dry-run mode expect 100% on all five fields — the deterministic parser is exact on a clean layout. With the real API expect high-but-imperfect numbers; the per-field table tells you which field needs prompt work (vendor names with punctuation are the classic offender). Either way, the audit method — random labeled sample, field-level exact match — is the deliverable habit.
Write the cost and scale notes. In quality-report.md, estimate the API-mode cost: ~350 input + ~250 output tokens per invoice at $5/$25 per million tokens ≈ $0.008 per invoice, ~$1.60 for the corpus. Then note the two production levers: provider batch tiers process non-urgent extraction at roughly half price (this pipeline is the definition of non-urgent), and prompt caching pays off when a large shared instruction block precedes many documents. Add the operational shape: this loop becomes an orchestrated task processing the day's new documents (Lab 11's change detection tells you which are new), the review queue becomes a table your ops tool reads, and the confidence floor becomes a tunable dial — raise it and more goes to humans, lower it and more risk lands in the warehouse.
Never let the extraction prompt "fix" the document. The gate caught the four wrong-total invoices only because the extractor copied the printed total faithfully. An extractor instructed to be helpful will repair the arithmetic, the gate will pass the row, and you will have laundered a source-document error into a trusted warehouse table.
make_invoices.py reports 200 invoices, 12 seeded defects; invoices/ contains 200 .txt files.extract_pipeline.py --dry-run reports accepted=188 rejected=12.review_queue/ holds exactly the 12 seeded invoice IDs, each JSON naming at least one correct reason.SELECT count(*) FROM invoices returns 188; no row has a currency outside INR/USD/EUR; no future invoice_date.audit_accuracy.py prints five field-accuracy lines over a 30-invoice sample.A de-lab-12/ folder containing make_invoices.py, extract_pipeline.py, audit_accuracy.py, the populated review_queue/, and quality-report.md — accepted/rejected counts, the reconciliation of rejects vs seeded defects, the field-level accuracy table on the 30-doc sample, and the cost estimate with batch/caching notes. Self-grading rubric (pass = 8/10):
accepted ≠ 188 in dry-run: diff ls review_queue/ against the seeded IDs from step 5 — an over-strict tolerance (float rounding) or a date-parse bug are the usual causes.pydantic.ValidationError in dry-run: an invoice line regex missed — check you kept the exact line format from the generator.response.parsed_output is None — the model refused or hit max_tokens; check response.stop_reason before retrying.Constraint Error on rerun: expected — INSERT OR REPLACE handles it; if you see it, you changed the insert statement.confidence: your dry-run confidence for dated invoices must be ≥ the 0.80 floor (the listing uses 0.95).The 12 seeded defects: INV-2026-0011, -0047, -0083, -0141 (printed grand total inflated 5–30% → line-sum reconciliation), INV-2026-0023, -0096, -0168 (currency printed as Rs. / US$ / EURO → allowed-values check), INV-2026-0059, -0120 (date line absent → missing-date check, and dry-run confidence drops to 0.60, so these trip two rules), INV-2026-0074, -0155, -0189 (invoice dated 30–120 days in the future → date-sanity check). A correct gate catches all twelve with zero knowledge of these IDs; if yours needed the list, it is pattern-matching the test, not validating the data.
In dry-run mode nothing leaves the laptop, which makes the full pipeline buildable inside any client boundary. When the real extractor is enabled, the invoice text is sent to the model — and under Indian DPDP or client MSAs, invoices are personal and commercial data. The compliant pattern to propose: run PII scrubbing before the LLM stage where feasible, prefer an LLM endpoint inside the client's own cloud tenancy or warehouse-native extraction functions (processing stays inside the governed platform boundary), keep the validation gate and review queue exactly as built (they are vendor-neutral), and log every document sent, to whom, and under which approval. The gate is your compliance asset too: it is documented evidence that model output is verified before entering financial systems.