End-to-End JSON Extraction Pipeline: A Working Code Example

The previous three subtopics each covered one piece of reliable JSON prompting in isolation — why it matters, the techniques that improve reliability, and how to recover from failures. This subtopic puts all of it together into one complete, reusable pipeline you could drop into a real application.

The pipeline has five stages, each addressing a specific failure mode covered earlier: (1) a schema-in-prompt template with placeholders for the actual input and a couple of few-shot examples baked in for edge cases specific to the domain; (2) an API call with native JSON mode enabled as a structural backstop; (3) lenient extraction of the JSON object from the raw response text, in case trailing commentary sneaks through despite JSON mode; (4) schema validation via a typed model (Pydantic here), catching type mismatches or missing fields that json.loads() alone wouldn't catch; and (5) a retry loop that, on any failure in stages 3 or 4, feeds the specific error back to the model and asks it to correct course, rather than either giving up immediately or blindly retrying with no additional information.

Wrapping all five stages into a single reusable function means every call site in your application gets the full reliability stack automatically, rather than each call site having to remember to re-implement error handling from scratch — which is exactly the kind of consistency and reusability this course's 'reusable templates' pillar of JSON prompting is pointing at: once you've built this pipeline once, it becomes a dependable building block your whole team or codebase can plug into APIs, databases, and downstream systems without manual reformatting or ad hoc error handling at every call site.

import json
import re
from typing import Type, TypeVar
from pydantic import BaseModel, ValidationError
from openai import OpenAI

client = OpenAI()
T = TypeVar("T", bound=BaseModel)

def extract_json_object(text: str) -> str:
    match = re.search(r"\{.*\}", text, re.DOTALL)
    if not match:
        raise ValueError("No JSON object found in response")
    return match.group(0)

def extract_structured(
    prompt: str, schema: Type[T], examples: str = "", max_retries: int = 2,
) -> T:
    """Full pipeline: schema-in-prompt + few-shot + JSON mode + lenient
    parse + schema validation + retry-with-error."""
    full_prompt = f"{examples}\n\n{prompt}" if examples else prompt
    messages = [{"role": "user", "content": full_prompt}]

    for attempt in range(max_retries + 1):
        resp = client.chat.completions.create(
            model="gpt-4.1",
            messages=messages,
            response_format={"type": "json_object"},  # structural backstop
        )
        raw = resp.choices[0].message.content
        try:
            return schema(**json.loads(extract_json_object(raw)))
        except (ValueError, json.JSONDecodeError, ValidationError) as e:
            if attempt == max_retries:
                raise
            messages += [
                {"role": "assistant", "content": raw},
                {"role": "user", "content": f"That failed to parse: {e}. Resend valid JSON."},
            ]

class Invoice(BaseModel):
    vendor: str
    amount: float
    due_date: str

result = extract_structured(
    "Extract as JSON {vendor, amount, due_date}: 'Acme Corp invoice, $4,200, due July 15.'",
    schema=Invoice,
)
print(result)  # Invoice(vendor='Acme Corp', amount=4200.0, due_date='July 15')