You've built a pipeline that calls an LLM to extract key information from customer support tickets — the product name, issue category, sentiment, and suggested priority level. The model is smart enough to get the information right. But then it returns something like: "Based on my analysis, this ticket appears to be about a billing issue with the Pro Plan subscription. The customer seems frustrated, so I'd rate the priority as high." Now you're writing regex to parse a sentence. Your pipeline is brittle, your downstream database insert is failing, and you're wondering why you didn't just write a rules engine.
This is the gap between AI as a demo and AI as infrastructure. When language models are woven into real automation workflows — feeding databases, triggering downstream APIs, populating dashboards, or chaining into multi-step agents — they need to return data in a shape your code can consume without heroics. The solution is structured output: the practice of constraining model responses to a predictable, machine-readable format, most commonly JSON, that your application can parse, validate, and act on reliably.
By the end of this lesson, you'll know how to do that properly — not the "just tell it to return JSON" approach that breaks in production, but the schema-enforced, validated, production-ready approach that belongs in real systems.
What you'll learn:
You should be comfortable with:
requests or the openai SDKThe instinct is reasonable. If you want the model to return JSON, you tell it to return JSON. Something like:
Return your answer as a JSON object with keys: product, issue_category, sentiment, priority.
And it works — some of the time. Here's what happens the rest of the time:
Sure! Here's the JSON you requested:
{
"product": "Pro Plan",
"issue_category": "billing",
"sentiment": "frustrated",
"priority": "high"
}
Let me know if you need anything else!
That wrapping text breaks json.loads(). Or the model decides to nest things differently than you specified. Or it adds fields you didn't ask for. Or it returns valid JSON that violates your business logic — "priority": "urgent" when you only accept low, medium, or high. Or, on long outputs, the model just... stops mid-response with the JSON bracket unclosed.
These aren't edge cases you can prompt-engineer away entirely. They're probabilistic failures inherent to text generation. What you need is a mechanism that enforces structure at the generation level, not just at the instruction level.
There are two real solutions:
Let's build up to both, starting with the problem space they solve.
Before you can enforce a schema, you need to know how to write one. JSON Schema is a standard for describing the shape of a JSON document — what fields exist, what types they have, what values are allowed. You don't need to learn the full specification here, but you do need the core vocabulary.
A minimal schema for a support ticket extraction looks like this:
{
"type": "object",
"properties": {
"product_name": {
"type": "string",
"description": "The name of the product the customer is having an issue with"
},
"issue_category": {
"type": "string",
"enum": ["billing", "technical", "account_access", "feature_request", "other"]
},
"sentiment": {
"type": "string",
"enum": ["positive", "neutral", "frustrated", "angry"]
},
"priority": {
"type": "string",
"enum": ["low", "medium", "high", "critical"]
},
"summary": {
"type": "string",
"description": "A one-sentence summary of the issue"
}
},
"required": ["product_name", "issue_category", "sentiment", "priority", "summary"],
"additionalProperties": false
}
A few things worth noting here:
"enum" is your best friend for controlled vocabularies. If priority can only be low, medium, high, or critical, you say so explicitly. The model can't hallucinate "urgent" into existence if the schema forbids it."required" forces the model to include every field you need. Without it, the model might omit fields it considers irrelevant."additionalProperties": false prevents the model from adding fields you didn't ask for. This matters for downstream systems that fail on unexpected keys."description" fields in your schema act as embedded instructions. The model reads them. Use them.Tip: Think of your JSON schema as a second system prompt that operates at the type level. Your system prompt guides the model on what to extract and how to reason about it. Your schema enforces the shape of what comes out. Both are important and they work together.
OpenAI's JSON Mode is a model-level setting that guarantees the model's response will be valid, parseable JSON. That's all it guarantees — not the fields, not the types, not the values. Just syntactic validity.
To enable it with the OpenAI Python SDK:
from openai import OpenAI
import json
client = OpenAI()
ticket_text = """
I've been charged twice for my Pro Plan subscription this month.
I've tried contacting billing support but haven't heard back in 3 days.
This is completely unacceptable. Please fix this immediately or I'm canceling.
"""
response = client.chat.completions.create(
model="gpt-4o",
response_format={"type": "json_object"}, # JSON Mode
messages=[
{
"role": "system",
"content": """You are a support ticket classifier. Extract structured information from support tickets.
Return a JSON object with these exact fields:
- product_name (string): The product being discussed
- issue_category (string): One of: billing, technical, account_access, feature_request, other
- sentiment (string): One of: positive, neutral, frustrated, angry
- priority (string): One of: low, medium, high, critical
- summary (string): One sentence describing the issue"""
},
{
"role": "user",
"content": ticket_text
}
]
)
result = json.loads(response.choices[0].message.content)
print(result)
This will return something like:
{
"product_name": "Pro Plan",
"issue_category": "billing",
"sentiment": "angry",
"priority": "high",
"summary": "Customer was charged twice for their Pro Plan subscription and has not received a response from billing support after three days."
}
No prose. No wrapping text. Clean JSON, every time. json.loads() will not throw.
The catch: JSON Mode doesn't validate against your schema. If you asked for "sentiment" to be one of four values but the model returns "very frustrated", JSON Mode won't stop it. You're still responsible for validating field presence and value correctness after parsing.
That's where you write a simple validation layer:
import json
from typing import Literal
VALID_CATEGORIES = {"billing", "technical", "account_access", "feature_request", "other"}
VALID_SENTIMENTS = {"positive", "neutral", "frustrated", "angry"}
VALID_PRIORITIES = {"low", "medium", "high", "critical"}
REQUIRED_FIELDS = {"product_name", "issue_category", "sentiment", "priority", "summary"}
def validate_ticket_extraction(data: dict) -> tuple[bool, list[str]]:
errors = []
missing = REQUIRED_FIELDS - set(data.keys())
if missing:
errors.append(f"Missing required fields: {missing}")
if "issue_category" in data and data["issue_category"] not in VALID_CATEGORIES:
errors.append(f"Invalid issue_category: {data['issue_category']}")
if "sentiment" in data and data["sentiment"] not in VALID_SENTIMENTS:
errors.append(f"Invalid sentiment: {data['sentiment']}")
if "priority" in data and data["priority"] not in VALID_PRIORITIES:
errors.append(f"Invalid priority: {data['priority']}")
return len(errors) == 0, errors
result = json.loads(response.choices[0].message.content)
is_valid, errors = validate_ticket_extraction(result)
if not is_valid:
print(f"Validation failed: {errors}")
else:
print("Valid extraction:", result)
JSON Mode is the right tool when you want a quick, reliable JSON guarantee and you're prepared to validate the contents yourself. For production systems at scale, you want the next level up.
OpenAI's Structured Outputs feature (available on gpt-4o and later models) goes further. Instead of just guaranteeing valid JSON, it guarantees valid JSON that conforms to a schema you provide. The enforcement happens at the decoding level through constrained generation — the model's sampling is restricted so it literally cannot produce tokens that would violate your schema.
This is architecturally different from JSON Mode. With Structured Outputs, you're not hoping the model follows instructions about structure. You're constraining the generation process itself.
Here's the same ticket classifier using Structured Outputs:
from openai import OpenAI
import json
client = OpenAI()
ticket_schema = {
"type": "object",
"properties": {
"product_name": {
"type": "string",
"description": "The name of the product the customer is having an issue with"
},
"issue_category": {
"type": "string",
"enum": ["billing", "technical", "account_access", "feature_request", "other"],
"description": "The category that best describes the customer's issue"
},
"sentiment": {
"type": "string",
"enum": ["positive", "neutral", "frustrated", "angry"],
"description": "The emotional tone of the customer's message"
},
"priority": {
"type": "string",
"enum": ["low", "medium", "high", "critical"],
"description": "Suggested priority level based on urgency and customer impact"
},
"summary": {
"type": "string",
"description": "A one-sentence summary of the customer's issue"
}
},
"required": ["product_name", "issue_category", "sentiment", "priority", "summary"],
"additionalProperties": false
}
ticket_text = """
I've been charged twice for my Pro Plan subscription this month.
I've tried contacting billing support but haven't heard back in 3 days.
This is completely unacceptable. Please fix this immediately or I'm canceling.
"""
response = client.chat.completions.create(
model="gpt-4o-2024-08-06", # Structured Outputs requires this version or later
response_format={
"type": "json_schema",
"json_schema": {
"name": "ticket_extraction",
"schema": ticket_schema,
"strict": True # Enables constrained generation
}
},
messages=[
{
"role": "system",
"content": "You are a support ticket classifier. Extract structured information from the support ticket provided."
},
{
"role": "user",
"content": ticket_text
}
]
)
result = json.loads(response.choices[0].message.content)
print(result)
With "strict": True, the model cannot produce output that violates your schema. "priority" will always be one of your four allowed values. Every required field will be present. No additional fields will appear. The response will always parse cleanly.
Warning: Structured Outputs with
strict: Truehas limitations on which JSON Schema features are supported. Notably, you cannot useoneOf,anyOf,not,if/then/else, or certain recursive structures. The OpenAI documentation maintains the current list. For complex schemas, test carefully and fall back to JSON Mode plus manual validation if needed.
The ticket classifier is a clean example, but real data problems are messier. Let's look at a few patterns you'll actually encounter.
You're extracting structured data from job postings to feed a recruitment analytics database. A flat structure won't cut it — compensation, required skills, and company details each need their own structure.
job_posting_schema = {
"type": "object",
"properties": {
"job_title": {"type": "string"},
"company": {
"type": "object",
"properties": {
"name": {"type": "string"},
"industry": {"type": "string"},
"size": {
"type": "string",
"enum": ["startup", "small", "mid-size", "enterprise", "unknown"]
}
},
"required": ["name", "industry", "size"],
"additionalProperties": False
},
"compensation": {
"type": "object",
"properties": {
"min_salary_usd": {"type": ["integer", "null"]},
"max_salary_usd": {"type": ["integer", "null"]},
"equity_offered": {"type": "boolean"},
"remote_stipend": {"type": "boolean"}
},
"required": ["min_salary_usd", "max_salary_usd", "equity_offered", "remote_stipend"],
"additionalProperties": False
},
"required_skills": {
"type": "array",
"items": {"type": "string"},
"description": "Technical skills explicitly required (not preferred) in the posting"
},
"experience_years_min": {"type": ["integer", "null"]},
"remote_policy": {
"type": "string",
"enum": ["fully_remote", "hybrid", "on_site", "unspecified"]
}
},
"required": [
"job_title", "company", "compensation",
"required_skills", "experience_years_min", "remote_policy"
],
"additionalProperties": False
}
Notice "type": ["integer", "null"] — this handles the case where salary data simply isn't present in the posting. Using null as an allowed type instead of omitting the field entirely means your database insert always has a value to work with, even if that value is null. This is almost always the right pattern for optional data in strict schemas.
You're parsing invoices to extract line items. Each line item has its own structure, and there are multiple of them.
invoice_schema = {
"type": "object",
"properties": {
"invoice_number": {"type": "string"},
"vendor_name": {"type": "string"},
"invoice_date": {
"type": "string",
"description": "Date in ISO 8601 format: YYYY-MM-DD"
},
"line_items": {
"type": "array",
"items": {
"type": "object",
"properties": {
"description": {"type": "string"},
"quantity": {"type": "number"},
"unit_price_usd": {"type": "number"},
"line_total_usd": {"type": "number"}
},
"required": ["description", "quantity", "unit_price_usd", "line_total_usd"],
"additionalProperties": False
}
},
"subtotal_usd": {"type": "number"},
"tax_usd": {"type": "number"},
"total_usd": {"type": "number"}
},
"required": [
"invoice_number", "vendor_name", "invoice_date",
"line_items", "subtotal_usd", "tax_usd", "total_usd"
],
"additionalProperties": False
}
Tip: For date and time fields, always specify the format in the
description. The JSON Schemaformatkeyword ("format": "date") is not enforced by OpenAI's constrained generation — only thetypeis. But the model will follow formatting instructions indescriptionreliably when using Structured Outputs.
In production, you need to handle failures gracefully. Even with Structured Outputs, API calls can fail due to rate limits, network issues, or content filter triggers. Here's a production-grade wrapper that handles all of this:
import json
import time
import logging
from typing import Optional
from openai import OpenAI, RateLimitError, APIError
logger = logging.getLogger(__name__)
client = OpenAI()
def extract_structured(
content: str,
schema: dict,
schema_name: str,
system_prompt: str,
model: str = "gpt-4o-2024-08-06",
max_retries: int = 3,
base_delay: float = 1.0
) -> Optional[dict]:
"""
Extract structured data from text content using OpenAI Structured Outputs.
Returns parsed dict on success, None on failure after retries.
"""
for attempt in range(max_retries):
try:
response = client.chat.completions.create(
model=model,
response_format={
"type": "json_schema",
"json_schema": {
"name": schema_name,
"schema": schema,
"strict": True
}
},
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": content}
],
temperature=0 # Deterministic extraction — never use temperature > 0 for structured extraction
)
# Check for refusal (content filter triggered)
message = response.choices[0].message
if message.refusal:
logger.warning(f"Model refused to process content: {message.refusal}")
return None
result = json.loads(message.content)
return result
except RateLimitError as e:
delay = base_delay * (2 ** attempt)
logger.warning(f"Rate limit hit on attempt {attempt + 1}. Retrying in {delay}s")
time.sleep(delay)
except APIError as e:
logger.error(f"API error on attempt {attempt + 1}: {e}")
if attempt == max_retries - 1:
raise
time.sleep(base_delay)
except json.JSONDecodeError as e:
# Should not happen with Structured Outputs, but log if it does
logger.error(f"JSON parse error (unexpected): {e}")
if attempt == max_retries - 1:
return None
time.sleep(base_delay)
return None
# Usage
system_prompt = """You are a support ticket classifier.
Extract structured information from the support ticket.
Be precise — do not infer information that isn't present in the ticket."""
ticket = """
Hi, I can't log into my account. I've tried resetting my password twice
but the reset email never arrives. I need access for a client presentation tomorrow.
— Sarah K., Enterprise Plan
"""
result = extract_structured(
content=ticket,
schema=ticket_schema,
schema_name="ticket_extraction",
system_prompt=system_prompt
)
if result:
print(f"Extracted: {json.dumps(result, indent=2)}")
else:
print("Extraction failed after retries")
A few design decisions worth explaining:
temperature=0 is not optional for structured extraction. You want deterministic behavior, not creative variation. A non-zero temperature doesn't improve extraction quality — it introduces noise.
The refusal check handles a real production case: some content (especially if it contains offensive text from angry customers) may trigger the model's content filters. Rather than crashing, you catch it and log it.
Exponential backoff on rate limits is standard practice. Don't just sleep a fixed interval — the exponential pattern (1s, 2s, 4s) gives the API time to recover while not wasting excessive time on short-duration limits.
Here's where this becomes genuinely powerful. Let's build a realistic pipeline: ingest support tickets from an email queue, extract structured data, route tickets to the appropriate team queue, and log everything to a database.
import json
import sqlite3
from datetime import datetime
from openai import OpenAI
client = OpenAI()
# --- Database setup ---
def init_db(db_path: str = "tickets.db"):
conn = sqlite3.connect(db_path)
conn.execute("""
CREATE TABLE IF NOT EXISTS processed_tickets (
id INTEGER PRIMARY KEY AUTOINCREMENT,
raw_ticket TEXT NOT NULL,
product_name TEXT,
issue_category TEXT,
sentiment TEXT,
priority TEXT,
summary TEXT,
assigned_queue TEXT,
processed_at TEXT,
extraction_success INTEGER
)
""")
conn.commit()
return conn
# --- Routing logic ---
ROUTING_MAP = {
"billing": "billing_team",
"technical": "engineering_support",
"account_access": "security_team",
"feature_request": "product_team",
"other": "general_support"
}
ESCALATION_THRESHOLD = {"critical", "high"}
def route_ticket(extraction: dict) -> str:
base_queue = ROUTING_MAP.get(extraction["issue_category"], "general_support")
if extraction["priority"] in ESCALATION_THRESHOLD:
return f"{base_queue}_escalated"
return base_queue
# --- Main pipeline ---
def process_ticket(raw_ticket: str, conn: sqlite3.Connection) -> dict:
extraction = extract_structured(
content=raw_ticket,
schema=ticket_schema,
schema_name="ticket_extraction",
system_prompt="""You are a support ticket classifier for a SaaS company.
Extract the structured information from the ticket precisely.
If information isn't present, use your best judgment based on context.
For priority: critical = service down or data loss, high = blocking work,
medium = significant inconvenience, low = minor issue or question."""
)
if extraction is None:
# Log failed extraction
conn.execute("""
INSERT INTO processed_tickets
(raw_ticket, processed_at, extraction_success)
VALUES (?, ?, 0)
""", (raw_ticket, datetime.utcnow().isoformat()))
conn.commit()
return {"success": False, "error": "Extraction failed"}
assigned_queue = route_ticket(extraction)
conn.execute("""
INSERT INTO processed_tickets
(raw_ticket, product_name, issue_category, sentiment,
priority, summary, assigned_queue, processed_at, extraction_success)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, 1)
""", (
raw_ticket,
extraction["product_name"],
extraction["issue_category"],
extraction["sentiment"],
extraction["priority"],
extraction["summary"],
assigned_queue,
datetime.utcnow().isoformat()
))
conn.commit()
return {
"success": True,
"extraction": extraction,
"assigned_queue": assigned_queue
}
# --- Run it ---
conn = init_db()
sample_tickets = [
"""The API is completely down. We can't process any transactions.
This is a production outage affecting our entire business.
Using the Enterprise API plan. Need immediate help.""",
"""Quick question — does the Business Plan include unlimited API calls?
I'm evaluating plans for our team. Thanks!""",
"""I've been billed for a team seat for an employee who left 6 months ago.
I've submitted three support requests with no response.
Pro Team plan, account owner."""
]
for ticket in sample_tickets:
result = process_ticket(ticket, conn)
if result["success"]:
print(f"Queue: {result['assigned_queue']}")
print(f"Summary: {result['extraction']['summary']}")
print(f"Priority: {result['extraction']['priority']}\n")
else:
print(f"Failed: {result['error']}\n")
This pipeline is doing real work: structured extraction feeding typed database inserts feeding routing logic. Every component is deterministic and auditable. The database logs both successes and failures, so you can review what broke and why.
Build a document intelligence extractor for financial data. You'll extract structured data from earnings report summaries and load it into a format ready for a financial analytics database.
Your task:
Write a JSON schema for an earnings report extraction with the following fields:
company_ticker (string)fiscal_quarter (string, format "Q1 2024")revenue_usd_millions (number or null)net_income_usd_millions (number or null)yoy_revenue_growth_pct (number or null — positive means growth, negative means decline)eps_diluted (number or null)guidance_raised (boolean — true if the company raised forward guidance)key_risks (array of strings, each a brief risk factor mentioned in the report)analyst_sentiment (enum: "beat", "met", "missed", "mixed")Write a system prompt that instructs the model on how to handle ambiguous cases (e.g., when guidance language is hedged, when revenue is reported in a different currency).
Implement the extraction using Structured Outputs and test it against this sample text:
Acme Corp (ACME) reported Q3 2024 earnings that handily beat Wall Street expectations.
Revenue came in at $847 million, up 23% year-over-year, surpassing the consensus
estimate of $812 million. Net income was $94 million with diluted EPS of $1.42,
compared to analyst expectations of $1.28.
Management raised full-year revenue guidance from $3.1B-$3.3B to $3.4B-$3.5B,
citing strong enterprise demand. Key risks mentioned include slowing SMB segment
growth, potential currency headwinds in EMEA, and increased competition in the
core analytics product line.
Stretch goal: Process three different earnings summaries and load them into a SQLite table called earnings_extractions. Write a query that shows the average YoY revenue growth across all extracted reports.
JSON Mode guarantees syntactic validity, not semantic correctness. If your downstream system depends on specific fields or controlled vocabulary values, JSON Mode is not enough. Use Structured Outputs with a schema. The extra effort of writing the schema pays off immediately in reduced validation code and fewer production failures.
Without this, the model may add helpful-sounding fields you didn't ask for. These extra fields can cause downstream systems to fail if they're strict about schema conformance, or worse, silently get passed into your database. Always include it.
If you mark salary_usd as required for a job posting extractor, the model will hallucinate a salary when none is mentioned in the posting. The correct pattern is to mark the field required but allow null: "type": ["number", "null"]. This forces the model to include the field but express absence properly.
Temperature above 0 is appropriate for generative tasks. Extraction is deterministic by nature — you want the model to read what's there and report it faithfully, not generate creative variations. Always use temperature=0 for structured extraction.
If your schema includes derived fields like line_total_usd = quantity * unit_price_usd, the model will attempt to compute it — and sometimes get it wrong due to floating-point representation in text. For numeric derivations, extract the base values and compute the derived ones in your application code.
A field like "status": {"type": "string"} is nearly useless for automation because "Active", "active", "ACTIVE", and "is currently active" are all valid strings. Use enum whenever you have a controlled vocabulary. If you genuinely can't predict all values, at least specify the format in the description field.
When something goes wrong, check in this order:
gpt-4o-2024-08-06 or later. JSON Mode works on older models.response.choices[0].message.refusal — some content triggers filters and returns a refusal instead of structured output.You now have a complete, production-oriented mental model for structured output with LLMs. Let's recap the key ideas:
JSON Mode gives you syntactic validity — guaranteed parseable JSON, but no schema enforcement. It's fast to implement and works on a wider range of models. Pair it with explicit validation logic.
Structured Outputs give you semantic validity — guaranteed conformance to a JSON Schema you define. Constrained generation at the decoding level means the model literally cannot violate your schema when strict: True is set. This is the right tool for production pipelines.
JSON Schema design is where most of the real work happens. Use enum for controlled vocabularies, ["type", "null"] for optional fields, additionalProperties: false to prevent drift, and description fields as embedded instructions.
Retry logic and failure handling are non-negotiable in production. Rate limits happen. Refusals happen. Network errors happen. Build the wrapper before you need it.
Temperature zero for extraction. Always.
instructor (by Jason Liu) wrap OpenAI's API with Pydantic model validation, letting you define your output schema as a Python class. This is often cleaner for larger projects.The fundamental shift this lesson represents is treating language models as typed data components rather than chatbots. Once you can reliably force a model's output into a machine-readable shape, you can wire it into any system that consumes data — which is almost every system worth building.
Learning Path: Intro to AI & Prompt Engineering