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
expecteddict 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)