Your Extraction Pipeline Doesn't Need a Smarter Model. It Needs a Schema Contract Test.

Last quarter I inherited a small pipeline that does something increasingly common: pull unstructured text (vendor emails, PDF invoices, changelog entries), ask a language model to extract structured fields, and insert the result into Postgres. The original author had picked a model, eyeballed twenty outputs, declared it "pretty reliable," and shipped. Three weeks later, a silent 4 a.m. failure: the model had returned {"total": "1,240.50 USD"} - a string with a thousands separator and a currency symbol - into a numeric column. The insert failed, the retry loop failed identically, and the dead-letter queue filled up. Nobody noticed for two days because the text of the extraction looked perfect in the logs. The fix wasn't a better model. It was treating the model's output the way we'd treat any external API response: validate it against a contract before it touches anything downstream, and continuously measure how often the contract holds. This article is the harness I built for that, why it changed which models I'll trust in pipelines, and where free model access fits in. Why code-generation intuition fails here Most LLM evaluation advice (including things I've written myself) targets code generation: does the function pass its tests? Extraction pipelines are a different beast in three ways: - The failure mode is formatting, not logic. The model usually extracts the right value; it just wraps it in the wrong shape. Wrong types, missing keys, extra commentary around the JSON, dates in five formats. - Failures are silent and delayed. Bad code fails loudly when you run it. A malformed record fails at the database boundary, or worse, passes validation and corrupts aggregations downstream. - You can't manually review at volume. A pipeline processing 500 documents a day has no human in the loop by design. So the evaluation question isn't "is the output good?" It's "what percentage of outputs parse, validate, and coerce cleanly - and what does the failure distribution look like?" The contract test harness The design: take a fixed corpus of real inputs with known-correct expected values, run each through the model, and score three separate gates. The separation matters - it tells you which layer is breaking. # contract_probe.py import json, os, time, urllib.request from pydantic import BaseModel, ValidationError, field_validator BASE_URL = os.environ["PROBE_BASE_URL"] # any OpenAI-compatible endpoint API_KEY = os.environ.get("PROBE_API_KEY", "not-needed") MODEL = os.environ["PROBE_MODEL"] class Invoice(BaseModel): vendor: str total: float currency: str due_date: str # ISO 8601 @field_validator("currency") @classmethod def iso_currency(cls, v: str) -> str: if len(v) != 3 or not v.isalpha() or not v.isupper(): raise ValueError("must be ISO 4217, e.g. USD") return v @field_validator("due_date") @classmethod def iso_date(cls, v: str) -> str: from datetime import date date.fromisoformat(v) # raises if not YYYY-MM-DD return v def chat(prompt: str) -> str: body = json.dumps({ "model": MODEL, "messages": [{"role": "user", "content": prompt}], "temperature": 0.0, # pipelines should be boring }).encode() req = urllib.request.Request( f"{BASE_URL}/chat/completions", data=body, headers={"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}) with urllib.request.urlopen(req, timeout=120) as r: return json.loads(r.read())["choices"][0]["message"]["content"] def gate1_parse(raw: str): """Gate 1: is there parseable JSON in there at all?""" text = raw.strip() # models love wrapping JSON in prose or fences; strip fences only if text.startswith(""): text = text.split("")[1] if text.startswith("json"): text = text[4:] try: return json.loads(text), None except json.JSONDecodeError as e: return None, f"parse: {e}" def gate2_validate(obj) -> tuple: """Gate 2: does it satisfy the schema?""" try: return Invoice(**obj), None except ValidationError as e: return None, f"schema: {e.errors()[0]['loc']} {e.errors()[0]['type']}" def gate3_correct(model: Invoice, expected: dict) -> list[str]: """Gate 3: are the extracted VALUES right?""" misses = [] if abs(model.total - expected["total"]) > 0.005: misses.append(f"total {model.total} != {expected['total']}") if model.currency != expected["currency"]: misses.append(f"currency {model.currency}") if model.due_date != expected["due_date"]: misses.append(f"due_date {model.due_date}") if model.vendor.strip().lower() != expected["vendor"].strip().lower(): misses.append(f"vendor '{model.vendor}'") return misses PROMPT = """Extract invoice fields from the text below. Return ONLY a JSON object with keys: vendor (string), total (number), currency (ISO 4217 code), due_date (YYYY-MM-DD). TEXT: {text}""" def probe(corpus_path: str = "corpus.json") -> None: corpus = json.load(open(corpus_path)) # [{"text": ..., "expected": {...}}] tally = {"parse_fail": 0, "schema_fail": 0, "value_fail": 0, "clean": 0} failure_log = [] for i, item in enumerate(corpus): raw = chat(PROMPT.format(text=item["text"])) obj, err = gate1_parse(raw) if err: tally["parse_fail"] += 1 failure_log.append({"i": i, "gate": "parse", "detail": err, "raw": raw[:200]}) else: inv, err = gate2_validate(obj) if err: tally["schema_fail"] += 1 failure_log.append({"i": i, "gate": "schema", "detail": err}) else: misses = gate3_correct(inv, item["expected"]) if misses: tally["value_fail"] += 1 failure_log.append({"i": i, "gate": "value", "detail": misses}) else: tally["clean"] += 1 time.sleep(1) n = len(corpus) print(json.dumps({ "model": MODEL, "n": n, "clean_rate": round(tally["clean"] / n, 3), "breakdown": tally, }, indent=2)) json.dump(failure_log, open("failures.json", "w"), indent=2) if name == "main": probe() The corpus is the part you can't skip: 30-50 real inputs from your actual domain, with hand-verified expected values. Mine includes the nasty cases - the email with two totals (which one is due?), the invoice where the date is "net 30 from 03/14", the vendor name with an ampersand. Synthetic happy-path inputs will tell you every model is great. What the gate breakdown reveals Running this across a few models on my 40-item corpus produced a pattern I didn't expect. I'll describe the shape without naming winners, because your corpus will differ and mine is small: | Gate | Failure share (my corpus) | Typical cause | |---|---|---| | Parse | 2-15% | Prose preamble ("Sure! Here's the JSON:"), truncated output | | Schema | 5-25% | Numbers as strings, due_date as "March 14", currency as "$" | | Value | 3-10% | Wrong total when two appear, hallucinated vendor suffix | Two findings changed my pipeline design: 1. Schema failures dominated over value failures - and they're the cheap kind to fix. A model that gets the right values in the wrong shape is one validation-error-feedback retry away from success. So I added exactly one retry that appends the Pydantic error verbatim to the prompt: "Your previous output failed validation: total: string type expected . Return corrected JSON only." In my runs, that single repair loop recovered roughly 70-80% of schema failures. Two retries recovered almost nothing more - diminishing returns arrive immediately. 2. Temperature 0 did not eliminate schema drift. Even at temperature 0, a small fraction of outputs varied across repeated runs of the same input, and occasionally the variation crossed the schema boundary. Determinism is a spectrum, not a switch - so the validator stays in the serving path permanently, not just at evaluation time. The contract test isn't a one-off selection tool; the same Invoice model ships in production as the boundary guard. Where free models fit - honestly A 40-document corpus, 3 candidate models, plus a repair-retry pass, is a few hundred completions. Trivial for one run, but I re-run the probe whenever a provider rotates a model version, and that cadence adds up. This is where I've been using MonkeyCode: Disclosure: This article was prepared as part of MonkeyCode's product outreach. Its free model access covers the probe runs, and the free server option means the harness can sit somewhere always-on and re-run nightly against a cron instead of living on my laptop. That's the honest scope of the fit - a contract probe is quota-hungry and latency-insensitive, which is precisely the workload profile free tiers absorb well. I can't speak to limits or longevity, so the harness treats the endpoint as swappable configuration, not a dependency. Limitations, and who should skip this - The corpus is a bottleneck and a bias. 40 hand-labeled examples can rank models for your domain; it says nothing about anyone else's. Below ~25 items, differences between models are mostly noise. - Gate 3 only checks what you thought to check. My expected dict can't catch a confidently wrong value in a field I didn't verify. Spot-audit raw outputs periodically even when the clean rate looks good. - A passing contract is not correctness. Schema validation guarantees shape, never truth. For high-stakes extraction (financial totals, legal dates), you still need human review sampling or a second-model cross-check. - Skip this approach entirely if your volume is low enough to review every output by hand, or if your extraction target has no stable schema - contracts require something to contract against. The mindset shift is the actual deliverable: the model's output is an untrusted API response from a third party that occasionally rewrites its own documentation. Validate at the boundary, log the failures by gate, repair once, and let the dead-letter queue tell you the truth. If you build a corpus for your own domain, I'd genuinely like to hear whether schema failures dominate for you too, or whether that's an artifact of my invoice-shaped corner of the world. Top comments (0)