In this article, you will learn how to get reliable structured outputs from large language models by validating them against a contract and the source text, retrying only when it helps, and constraining generation.
Topics we will cover include:
- How to define a strict Pydantic contract and check every validated object against the source it came from.
- How to build a bounded retry loop that regenerates only on shape errors and keeps the SDK’s hidden transport retries under control.
- How to use constrained decoding with OpenAI’s Responses API without dropping validation, and how to test each failure path with a mocked client.
Structured output is a standard feature now. OpenAI’s Responses API takes a JSON schema or a Pydantic model and promises a response that fits it, vLLM does the same on a local server with xgrammar behind it, and the Python SDKs hand you a parsed object instead of a string. Most tutorials stop at the moment the JSON parses. That’s the wrong place to stop, and here’s the whole problem in four lines of invoice:
INVOICE 2026-0417
Bill to: Northwind Traders
Total due: EUR 1,250.00
Due 30 days from issue.
1234
INVOICE 2026-0417Bill to: Northwind TradersTotal due: EUR 1,250.00Due 30 days from issue.
Ask a model for the invoice id, the currency, the total in cents and the line it took the total from, and you might get this back:
{"invoice_id": "2026-0417", "currency": "EUR", "total_cents": 12500, "total_quote": "Total due: EUR 1,250.00"}
1
{"invoice_id": "2026-0417", "currency": "EUR", "total_cents": 12500, "total_quote": "Total due: EUR 1,250.00"}
It’s valid JSON, every key is present, the types are right, EUR is in the allowed set, and the quoted line really is in the invoice. The total is 12,500 cents, which is 125 euros, for an invoice that says 1,250. No parser catches that and no schema catches it, because 12500 is a perfectly good integer. The only thing that catches it is a comparison with the source text, and your application has to make that comparison itself.
The three terms at the end of the title are the three layers that do the catching. Validation means your code checks the returned object against a contract you own, then against the invoice it came from. Retries means a second attempt is allowed only when another attempt can fix what went wrong, with a hard stop at three. Constrained decoding means the generator can only emit tokens that keep the output inside your schema, so a whole class of shape failures never reaches validation. The examples below were run on Python 3.10.12 and Pydantic 2.14.1; the code extracts fields and nothing more, and never writes to a real system.
Define the Contract Before Calling the Model #
The contract has four fields, and your application owns it, not the prompt:
from typing import Literal
from pydantic import BaseModel, ConfigDict
class Invoice(BaseModel):
model_config = ConfigDict(strict=True, extra="forbid")
invoice_id: str
currency: Literal["USD", "EUR"] | None
total_cents: int | None
total_quote: str | None
1234567891011
from typing import Literalfrom pydantic import BaseModel, ConfigDict class Invoice(BaseModel): model_config = ConfigDict(strict=True, extra="forbid") invoice_id: str currency: Literal["USD", "EUR"] | None total_cents: int | None total_quote: str | None
Every key is required, so a response that leaves one out fails. Three fields are nullable without a default, which is different from optional: the model may say “the invoice doesn’t state this” by sending null, but it may not stay silent and it may not invent a number to avoid a null. Two currencies is a tutorial scope, not a claim about invoices. Cents are stored as an integer so that 1,250.00 never becomes 1249.9999.
The two configuration options do most of the work. Strict mode switches off Pydantic’s default coercion, which would otherwise accept the string "125000" for an integer field and convert it silently; in strict mode that string fails with the error type int_type, and so does the boolean True, which the default mode would turn into 1. extra="forbid" turns an unexpected key into an extra_forbidden error instead of ignoring it. Feed Invoice four bad inputs and you get four distinct error types back:
from pydantic import ValidationError
bad_inputs = [
{"invoice_id": "2026-0417", "currency": "GBP", "total_cents": 125000, "total_quote": "x"},
{"invoice_id": "2026-0417", "currency": "EUR", "total_cents": "125000", "total_quote": "x"},
{"invoice_id": "2026-0417", "currency": "EUR", "total_cents": True, "total_quote": "x"},
{"invoice_id": "2026-0417", "currency": "EUR", "total_cents": 125000, "total_quote": "x", "vendor": "v"},
]
for item in bad_inputs:
try:
Invoice.model_validate(item)
except ValidationError as e:
print([(err["loc"], err["type"]) for err in e.errors()])
12345678910111213
from pydantic import ValidationError bad_inputs = [ {"invoice_id": "2026-0417", "currency": "GBP", "total_cents": 125000, "total_quote": "x"}, {"invoice_id": "2026-0417", "currency": "EUR", "total_cents": "125000", "total_quote": "x"}, {"invoice_id": "2026-0417", "currency": "EUR", "total_cents": True, "total_quote": "x"}, {"invoice_id": "2026-0417", "currency": "EUR", "total_cents": 125000, "total_quote": "x", "vendor": "v"},]for item in bad_inputs: try: Invoice.model_validate(item) except ValidationError as e: print([(err["loc"], err["type"]) for err in e.errors()])
[(('currency',), 'literal_error')]
[(('total_cents',), 'int_type')]
[(('total_cents',), 'int_type')]
[(('vendor',), 'extra_forbidden')]
1234
[(('currency',), 'literal_error')][(('total_cents',), 'int_type')][(('total_cents',), 'int_type')][(('vendor',), 'extra_forbidden')]
You can use dataclasses for application-owned data, but model output needs validation before it becomes a trusted record, and Invoice is the gate it passes through. One thing the gate can’t hold: Invoice.model_json_schema() exports four properties, all required, additionalProperties false, and not a word about the invoice. The rule that the cents have to match the quoted line can’t be written in JSON Schema, so a decoder that enforces this schema perfectly will still produce the 12,500-cent total from the opening. A valid object is the entry ticket to the next check, not the result.
Validate the Object, Then Check It Against the Source #
accept() runs four steps and ends in one of four states: accepted, needs_review, rejected or shape_error.
import json
from dataclasses import dataclass
from pydantic import ValidationError
@dataclass
class Outcome:
status: str # "accepted" | "needs_review" | "rejected" | "shape_error"
invoice: Invoice | None
reasons: list[str]
def accept(response_text: str, source_text: str, expected_invoice_id: str) -> Outcome:
try:
candidate = Invoice.model_validate(json.loads(response_text))
except json.JSONDecodeError as e:
return Outcome("shape_error", None, [f"not JSON: {e.msg} at position {e.pos}"])
except ValidationError as e:
reasons = [f"{'.'.join(str(p) for p in err['loc'])}: {err['msg']}" for err in e.errors()]
return Outcome("shape_error", None, reasons)
reasons: list[str] = []
if candidate.total_quote is not None and candidate.total_quote not in source_text:
reasons.append("total_quote does not appear in the source")
if candidate.invoice_id != expected_invoice_id:
reasons.append(f"invoice_id {candidate.invoice_id!r} is not {expected_invoice_id!r}")
if candidate.currency is not None and candidate.currency not in source_text:
reasons.append(f"currency {candidate.currency} is not stated in the source")
if candidate.total_cents is not None and candidate.total_quote is not None:
digits = "".join(ch for ch in candidate.total_quote if ch.isdigit())
if digits and int(digits) != candidate.total_cents:
reasons.append(f"total_cents {candidate.total_cents} does not match the quoted line")
if reasons:
return Outcome("rejected", candidate, reasons)
if candidate.total_cents is None or candidate.currency is None:
return Outcome("needs_review", candidate, ["source did not state the amount or currency"])
return Outcome("accepted", candidate, [])
123456789101112131415161718192021222324252627282930313233343536373839404142434445
import jsonfrom dataclasses import dataclassfrom pydantic import ValidationError @dataclassclass Outcome: status: str # "accepted" | "needs_review" | "rejected" | "shape_error" invoice: Invoice | None reasons: list[str] def accept(response_text: str, source_text: str, expected_invoice_id: str) -> Outcome: # Step 1: parse and validate the shape. Any failure here is a shape error. try: candidate = Invoice.model_validate(json.loads(response_text)) except json.JSONDecodeError as e: return Outcome("shape_error", None, [f"not JSON: {e.msg} at position {e.pos}"]) except ValidationError as e: reasons = [f"{'.'.join(str(p) for p in err['loc'])}: {err['msg']}" for err in e.errors()] return Outcome("shape_error", None, reasons) reasons: list[str] = [] # Step 2: the quoted line has to exist in the source, verbatim. if candidate.total_quote is not None and candidate.total_quote not in source_text: reasons.append("total_quote does not appear in the source") # Step 3: identity and currency have to match what the source says. if candidate.invoice_id != expected_invoice_id: reasons.append(f"invoice_id {candidate.invoice_id!r} is not {expected_invoice_id!r}") if candidate.currency is not None and candidate.currency not in source_text: reasons.append(f"currency {candidate.currency} is not stated in the source") # Step 4: the amount has to be consistent with the quoted line. if candidate.total_cents is not None and candidate.total_quote is not None: digits = "".join(ch for ch in candidate.total_quote if ch.isdigit()) if digits and int(digits) != candidate.total_cents: reasons.append(f"total_cents {candidate.total_cents} does not match the quoted line") if reasons: return Outcome("rejected", candidate, reasons) if candidate.total_cents is None or candidate.currency is None: return Outcome("needs_review", candidate, ["source did not state the amount or currency"]) return Outcome("accepted", candidate, [])
Step one is the parser and the contract, and it’s the only step that can return shape_error, which matters because a shape error is the only outcome the next section is allowed to retry. Step two checks that total_quote is in the invoice character for character. Step three checks the id you asked about and that the currency the model picked is actually printed in the text. Step four catches the opening’s mistake: the digits in “Total due: EUR 1,250.00” with the separators stripped are 125000; the model sent 12500, so the object is rejected with the reason “total_cents 12500 does not match the quoted line”. For a real invoice format you’d write a check that understands your decimals; the principle holds either way: the number has to be derivable from the evidence or it isn’t accepted.
needs_review is where a null goes when the invoice genuinely didn’t state an amount or a currency; under this policy a missing fact goes to a person, not to a guessed zero. rejected is terminal, because another call to the model can’t change what the invoice says. Expected values in the test fixtures are written by hand from the source text, never copied from a model’s output.
One boundary before the loop. Everything here starts from text that already exists. When you’re extracting text from scanned invoices, inspect the transcription before you ask the model to populate the schema, because a wrong digit upstream passes every check downstream as long as the object is internally consistent. What happens to a shape_error from accept() is the next question, and the answer is a policy rather than a loop that runs until something parses.
Retry Only When Another Attempt Can Help #
extract_with_retries() has three exits, and only one of them is a retry. A shape_error gets another attempt with the errors fed back. accepted, needs_review and rejected are terminal, because the source won’t change. A refusal or a configuration error stops at once, because repeating the request won’t help.
from typing import Callable
MAX_ATTEMPTS = 3 # the first attempt plus at most two regenerations
class GenerationRefused(Exception):
"""The provider refused or the request is not repairable by another attempt."""
def extract_with_retries(generate: Callable[[str, dict, list[str]], str],
source_text: str, expected_invoice_id: str) -> tuple[Outcome, int]:
schema = Invoice.model_json_schema()
feedback: list[str] = []
attempts = 0
while attempts < MAX_ATTEMPTS:
attempts += 1
try:
response_text = generate(source_text, schema, feedback)
except GenerationRefused as e:
return Outcome("rejected", None, [f"stopped: {e}"]), attempts
outcome = accept(response_text, source_text, expected_invoice_id)
if outcome.status != "shape_error":
return outcome, attempts # terminal: the source will not change
feedback = outcome.reasons[:5] # bounded: the relevant errors, nothing else
return Outcome("rejected", None, ["attempt budget exhausted"]), attempts
12345678910111213141516171819202122232425
from typing import Callable MAX_ATTEMPTS = 3 # the first attempt plus at most two regenerations class GenerationRefused(Exception): """The provider refused or the request is not repairable by another attempt.""" def extract_with_retries(generate: Callable[[str, dict, list[str]], str], source_text: str, expected_invoice_id: str) -> tuple[Outcome, int]: schema = Invoice.model_json_schema() feedback: list[str] = [] attempts = 0 while attempts < MAX_ATTEMPTS: attempts += 1 try: response_text = generate(source_text, schema, feedback) except GenerationRefused as e: return Outcome("rejected", None, [f"stopped: {e}"]), attempts outcome = accept(response_text, source_text, expected_invoice_id) if outcome.status != "shape_error": return outcome, attempts # terminal: the source will not change feedback = outcome.reasons[:5] # bounded: the relevant errors, nothing else return Outcome("rejected", None, ["attempt budget exhausted"]), attempts
generate is whatever calls your model; it gets the invoice, the schema and the feedback list, and returns text. The schema and the source never change between attempts. The feedback is capped at five reasons, enough to tell the model that vendor isn’t a field or that total_cents has to be an integer, and not enough to turn the prompt into a transcript of everything that went wrong. Use the field-level validation errors to identify the failed check and send only that; each error carries a loc, a type and a msg, and the message is usually all the model needs. Every candidate is validated from scratch, because a repair call can change the amount while it fixes the key.
The budget of three is only real if you also own the transport retries. The OpenAI Python SDK retries connection errors, timeouts, 409, 429 and 5xx responses twice by default with exponential backoff, so three application attempts can be nine HTTP requests without your code noticing. Set max_retries=0 on the client for a budget you can demonstrate, add a per-call timeout (the default is ten minutes, far too long for an extraction) and keep an overall deadline in the loop; the SDK retry configuration is documented per client and per request. Three attempts is this tutorial’s policy, not a universal optimum; what matters is counting the hidden work and stopping on the terminal cases, which the table below sorts.
Table 1. Match the failure to the next step.
| Failure | What to check | Next step in this tutorial |
|---|---|---|
| Malformed complete response, or shape or type mismatch | The parser or validator error and the remaining application budget | Bounded content regeneration with the source, the contract and the relevant errors; validate again |
| Valid shape, wrong or unsupported source value | The source text and the acceptance rule | rejected , or a specifically justified bounded re-extraction; never accept because it parses |
| Missing or ambiguous source fact sent as null | Whether the business information exists in the source | needs_review ; don’t invent a default |
| Refusal or policy-filtered generation | The provider’s documented outcome | Stop or escalate; don’t rephrase to defeat the refusal |
| Truncation or unfinished stream | The completion status and its reason | Don’t commit; adjust a justified output budget or simplify scope within the deadline |
| Unsupported schema or model, invalid credentials, other configuration error | The exact API error | Fix the configuration; no blind content-repair loop |
| Eligible transient transport failure | The SDK’s current retry behavior, backoff and budget | A controlled transport retry, kept distinct from content repair |
| Application budget exhausted | Total attempts and elapsed deadline | Explicit failure or review; no extra call and no downstream write |
A loop like this fixes shape failures. It can’t fix a model that was never constrained, and that’s the third layer.
Constrain Generation Without Dropping Validation #
Asking for JSON in the prompt and restricting what the generator can emit are different things. A prompt is a request, and the model can still answer with a code fence, a stray sentence or an extra key. A schema constraint runs inside decoding: at each step the generator may only choose from the tokens that keep the output valid. Adrian Tam’s walkthrough of decoding strategies puts it in one line: “The idea is to mask out tokens that would make the output invalid”; this example uses a supported schema constraint rather than implementing the masking. On that path the malformed, extra-key and wrong-type cases from the fixtures stop occurring, because the decoder can’t produce them.
def extract_constrained(client, model: str, source_text: str, expected_invoice_id: str) -> Outcome:
response = client.responses.parse(
model=model,
input=[
{"role": "system", "content": "Extract the invoice fields. Use null when the text does not state a value."},
{"role": "user", "content": source_text},
],
text_format=Invoice,
)
if response.status != "completed": # anything else is an outcome, not data
return Outcome("rejected", None, [f"response status {response.status}"])
if response.output_parsed is None: # no parsed output: treat as a refusal
return Outcome("rejected", None, ["no parsed output; treat as a refusal"])
return accept(response.output_parsed.model_dump_json(), source_text, expected_invoice_id)
123456789101112131415
The call is the one OpenAI’s structured-output guide documents for the Python SDK as of October 2026: client.responses.parse(...) with text_format=Invoice, which accepts the Pydantic model directly and returns output_parsed as an instance of it. The guide’s examples name gpt-6-astra; pin whichever model you run and record it. I wrote this adapter from the documentation and exercised it with a mocked client in the fixtures below, not against a paid API, and I’d rather say that than imply a live result.
Two lines do the work the decoder can’t. The status check runs before anything reads the object: a response that isn’t completed is routed, not used. A completed response with no parsed output is a refusal, not an empty invoice. Then the object goes through accept() unchanged, because the decoder enforced the shape and only the shape; the constraint guarantees that total_cents is an integer, not that it’s 125000 rather than 12500. Keep the contract plain too, since not every Pydantic feature survives the trip into a provider’s schema subset.
Run Both Paths Through the Same Failure Cases #
The fixture suite turns the sentences above into fourteen test cases, from a missing required key and a code fence around the JSON to a fabricated quote, a refusal and exhaustion with no fourth call. Four of them, with the mocked client, look like this:
import unittest
from types import SimpleNamespace
SOURCE = "INVOICE 2026-0417\nBill to: Northwind Traders\nTotal due: EUR 1,250.00\nDue 30 days from issue."
ID = "2026-0417"
GOOD = {"invoice_id": ID, "currency": "EUR", "total_cents": 125000, "total_quote": "Total due: EUR 1,250.00"}
class Fixtures(unittest.TestCase):
def test_schema_valid_wrong_amount_is_rejected(self):
o = accept(json.dumps({**GOOD, "total_cents": 12500}), SOURCE, ID)
self.assertEqual(o.status, "rejected")
def test_exhaustion_makes_no_fourth_call(self):
calls = []
def gen(src, schema, feedback):
calls.append(1); return "not json at all"
outcome, attempts = extract_with_retries(gen, SOURCE, ID)
self.assertEqual((outcome.status, attempts, len(calls)), ("rejected", 3, 3))
def test_wrong_amount_is_not_retried(self):
calls = []
def gen(src, schema, feedback):
calls.append(1); return json.dumps({**GOOD, "total_cents": 12500})
outcome, attempts = extract_with_retries(gen, SOURCE, ID)
self.assertEqual((outcome.status, len(calls)), ("rejected", 1))
def test_incomplete_response_is_not_data(self):
resp = SimpleNamespace(status="incomplete", output_parsed=Invoice(**GOOD))
client = SimpleNamespace(responses=SimpleNamespace(parse=lambda **kw: resp))
self.assertEqual(extract_constrained(client, "m", SOURCE, ID).status, "rejected")
12345678910111213141516171819202122232425262728293031
import unittestfrom types import SimpleNamespace SOURCE = "INVOICE 2026-0417\nBill to: Northwind Traders\nTotal due: EUR 1,250.00\nDue 30 days from issue."ID = "2026-0417"GOOD = {"invoice_id": ID, "currency": "EUR", "total_cents": 125000, "total_quote": "Total due: EUR 1,250.00"} class Fixtures(unittest.TestCase): def test_schema_valid_wrong_amount_is_rejected(self): o = accept(json.dumps({**GOOD, "total_cents": 12500}), SOURCE, ID) self.assertEqual(o.status, "rejected") def test_exhaustion_makes_no_fourth_call(self): calls = [] def gen(src, schema, feedback): calls.append(1); return "not json at all" outcome, attempts = extract_with_retries(gen, SOURCE, ID) self.assertEqual((outcome.status, attempts, len(calls)), ("rejected", 3, 3)) def test_wrong_amount_is_not_retried(self): calls = [] def gen(src, schema, feedback): calls.append(1); return json.dumps({**GOOD, "total_cents": 12500}) outcome, attempts = extract_with_retries(gen, SOURCE, ID) self.assertEqual((outcome.status, len(calls)), ("rejected", 1)) def test_incomplete_response_is_not_data(self): resp = SimpleNamespace(status="incomplete", output_parsed=Invoice(**GOOD)) client = SimpleNamespace(responses=SimpleNamespace(parse=lambda **kw: resp)) self.assertEqual(extract_constrained(client, "m", SOURCE, ID).status, "rejected")
Ran 17 tests in 0.002s
OK
123
Ran 17 tests in 0.002s OK
Read what the mock proves and what it doesn’t. It proves the routing: a wrong amount is rejected on the first attempt and never retried, exhaustion returns after exactly three calls, an incomplete response never reaches the source checks, and a refusal stops after one. It proves nothing about how often a real model produces the 12,500-cent total, and nothing about cost or latency; there are no percentages in this article because a success rate from canned responses would describe the fixtures, not a model. When you run the suite against a real adapter, record the model name, the SDK version, the outcome class and the attempt count per case, mask the source text in the logs, and report parsed, structurally valid and correct-against-the-source as three separate columns, with needs_review in the third as not yet correct.
Let Only Accepted Results Reach the Next Step #
The pipeline consumes three outcomes and treats them differently. accepted goes downstream. needs_review goes to a person with the candidate and the reasons attached. rejected, whether from a contradiction, a refusal or an exhausted budget, goes to an error path that records what happened and does nothing else; a retry never repeats a write, and a repair attempt never acquires permissions the first attempt didn’t have.
Prefer constrained generation when the provider supports your contract, because it removes the shape failures and the retries they’d cause; fall back to the prompt-only loop with the same accept() when it doesn’t. In both cases I’d keep the typed contract and the source check even though the generator already enforces structure, because structure is the one thing it can enforce and correctness is the thing you need. The deployment test is a single question you can ask of a pipeline you already run: does every failure path stop before the downstream action? If the evidence is missing, the response is incomplete or the three attempts are spent, the result has to be an explicit review or error outcome, not a plausible object that happened to parse.