Structured Outputs in Production

Real-World Scenario

An order-intake bot extracts {product, quantity, unit_price_usd} from free-text customer emails and writes straight to the orders database. It works in the demo. In production, the model occasionally wraps its JSON in a markdown code fence — invisible in a chat UI, fatal to json.loads — and every failure either crashes the pipeline or, worse, writes a hallucinated order that happens to parse fine.

Setup: the Schema and the Naive Retry Loop

import json
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
from pydantic import BaseModel, ValidationError, field_validator

tok = AutoTokenizer.from_pretrained("Qwen/Qwen2.5-0.5B-Instruct")
model = AutoModelForCausalLM.from_pretrained("Qwen/Qwen2.5-0.5B-Instruct")

class ExtractedOrder(BaseModel):
    product: str
    quantity: int
    unit_price_usd: float

    @field_validator("quantity")
    @classmethod
    def positive_quantity(cls, v):
        if v <= 0:
            raise ValueError("quantity must be positive")
        return v


def ask_model(prompt: str, temperature: float, seed: int) -> str:
    """Send an extraction prompt to the model, chat-template formatted."""
    torch.manual_seed(seed)
    messages = [
        {"role": "system", "content": "Extract order details as JSON only. No prose, no markdown fences. "
                                       'Schema: {"product": string, "quantity": int, "unit_price_usd": float}'},
        {"role": "user", "content": prompt},
    ]
    chat_input = tok.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)
    inputs = tok(chat_input, return_tensors="pt")
    do_sample = temperature > 0
    out = model.generate(
        **inputs, max_new_tokens=80, do_sample=do_sample,
        temperature=temperature if do_sample else None,
        top_p=0.95 if do_sample else None,
    )
    return tok.decode(out[0][inputs["input_ids"].shape[1]:], skip_special_tokens=True).strip()


def extract_naive(prompt: str, max_attempts: int = 3):
    """Retries on failure, but parses the raw model output directly."""
    log = []
    for attempt in range(max_attempts):
        temperature = 0.0 if attempt == 0 else 0.7
        raw = ask_model(prompt, temperature=temperature, seed=attempt)
        try:
            data = json.loads(raw)
            return ExtractedOrder(**data), log + [f"attempt {attempt + 1}: {raw!r} -> OK"]
        except (json.JSONDecodeError, ValidationError) as e:
            log.append(f"attempt {attempt + 1}: {raw!r} -> REJECTED: {type(e).__name__}")
    return None, log


order, log = extract_naive("I need a dozen units of the steel bracket at $4.50 each.")
for line in log:
    print(line)
print("Final:", order)

$ python extract_naive.py
attempt 1: '```json\n{\n  "product": "steel bracket",\n  "quantity": 12,\n  "unit_price_usd": 4.5\n}\n```' -> REJECTED: JSONDecodeError
attempt 2: '```json\n{\n  "product": "steel bracket",\n  "quantity": 12,\n  "unit_price_usd": 4.5\n}\n```' -> REJECTED: JSONDecodeError
attempt 3: '```json\n{\n  "product": "steel bracket",\n  "quantity": 12,\n  "unit_price_usd": 4.50\n}\n```' -> REJECTED: JSONDecodeError
Final: None

Every attempt actually extracted the right data — quantity 12, product “steel bracket,” price 4.5 — and every attempt still failed, because the system prompt’s “no markdown fences” instruction wasn’t followed and json.loads refuses anything that isn’t pure JSON. Retrying alone doesn’t fix a systematic formatting habit; it just reruns the same mistake.

The Fix: Strip Fences, Then Validate

import re

def strip_fences(text: str) -> str:
    """Extract JSON from a ```json ... ``` fence if present, else return as-is."""
    m = re.search(r"```(?:json)?\s*(.*?)\s*```", text, re.DOTALL)
    return m.group(1) if m else text


def extract_with_fix(prompt: str, max_attempts: int = 3):
    log = []
    for attempt in range(max_attempts):
        temperature = 0.0 if attempt == 0 else 0.7
        raw = ask_model(prompt, temperature=temperature, seed=attempt)
        cleaned = strip_fences(raw)
        try:
            data = json.loads(cleaned)
            order = ExtractedOrder(**data)
            return order, log + [f"attempt {attempt + 1}: cleaned={cleaned!r} -> OK"]
        except (json.JSONDecodeError, ValidationError) as e:
            log.append(f"attempt {attempt + 1}: cleaned={cleaned!r} -> REJECTED: {type(e).__name__}")
    return None, log


order, log = extract_with_fix("I need a dozen units of the steel bracket at $4.50 each.")
for line in log:
    print(line)
print("Final:", order)

$ python extract_fixed.py
attempt 1: cleaned='{\n  "product": "steel bracket",\n  "quantity": 12,\n  "unit_price_usd": 4.5\n}' -> OK
Final: product='steel bracket' quantity=12 unit_price_usd=4.5

Same model, same prompt, same seed — the only change is stripping a markdown fence before parsing, and the exact response that failed three times now succeeds on the first attempt. This is the single most common structured-output failure in practice: not bad data, just a wrapper a naive parser doesn’t expect.

A Passing Response Can Still Be Wrong

order, log = extract_naive("Cancel my order for the steel bracket, I do not want any.")
for line in log:
    print(line)
print("Final:", order)

$ python extract_hallucination.py
attempt 1: '{\n  "order_id": "12345",\n  "product": "steel_bracket",\n  "quantity": 1,\n  "unit_price_usd": 99.99\n}' -> OK
Final: product='steel_bracket' quantity=1 unit_price_usd=99.99

Every field is schema-valid and positive_quantity passes — and the extraction is entirely wrong. The customer wants to cancel, not place an order for 1 unit at an invented $99.99 price. Pydantic validation catches malformed output; it cannot catch output that’s well-formed and false. That’s a prompt/business-logic problem (e.g., a has_order_intent: bool field the caller checks before ever looking at quantity), not a parsing one — the two failure modes need different fixes.

Practice

1. Add a product validator that rejects empty strings after stripping whitespace, and confirm it fires on a synthetic {"product": " ", "quantity": 1, "unit_price_usd": 1.0}.

   Solution

   from pydantic import BaseModel, ValidationError, field_validator
   
   class ExtractedOrderV2(BaseModel):
       product: str
       quantity: int
       unit_price_usd: float
   
       @field_validator("product")
       @classmethod
       def non_empty_product(cls, v):
           if not v.strip():
               raise ValueError("product must not be empty")
           return v
   
   try:
       ExtractedOrderV2(product="  ", quantity=1, unit_price_usd=1.0)
   except ValidationError as e:
       print(f"{type(e).__name__}: product must not be empty" in str(e) or "product must not be empty" in str(e))

   True

2. strip_fences uses a non-greedy regex. Show what it returns on input with two separate fenced blocks, and explain why non-greedy matters here.

   Solution

   text = "```json\n{\"a\": 1}\n```\nsome text\n```json\n{\"b\": 2}\n```"
   print(strip_fences(text))

   {"a": 1}

   Non-greedy (.*?) stops at the first closing fence, returning only the first JSON block. A greedy .* would match from the first opening fence all the way to the last closing fence, swallowing “some text” and the second JSON block into one invalid string. For this lesson’s single-object extraction, non-greedy is correct; a multi-object extraction would need re.findall instead of re.search.

3. Lower max_attempts to 1 in extract_with_fix and rerun the “dozen” prompt. Does it still succeed, and what does that tell you about how much of the fix was the retry loop versus the fence-stripping?

   Solution

   order, log = extract_with_fix("I need a dozen units of the steel bracket at $4.50 each.", max_attempts=1)
   print(log)
   print(order)

   ["attempt 1: cleaned='{\\n  \"product\": \"steel bracket\",\\n  \"quantity\": 12,\\n  \"unit_price_usd\": 4.5\\n}' -> OK"]
   product='steel bracket' quantity=12 unit_price_usd=4.5

   Still succeeds on attempt 1 — for this particular prompt, the fence-stripping fix was the entire fix; the retry loop’s job is handling the other failure modes (genuinely malformed JSON, a field that fails the business-logic validator) that fence-stripping alone doesn’t cover.

4. Add a total_cost computed property to ExtractedOrder (quantity * unit_price_usd) and print it for the successfully extracted “dozen” order.

   Solution

   from pydantic import BaseModel, computed_field
   
   class ExtractedOrderV3(BaseModel):
       product: str
       quantity: int
       unit_price_usd: float
   
       @computed_field
       @property
       def total_cost(self) -> float:
           return round(self.quantity * self.unit_price_usd, 2)
   
   order = ExtractedOrderV3(product="steel bracket", quantity=12, unit_price_usd=4.5)
   print(order.total_cost)

   54.0