	In Browser
	StumbleUpon
	del.icio.us
	Google
	Google Buzz
	reddit
	LinkedIn

	Facebook
	Twitter
	Linkedin
	E-Mail

Generative AI > OpenAI API > OpenAI Structured Outputs with Pydantic

OpenAI Structured Outputs with Pydantic

Author: Venkata Sudhakar

OpenAI Structured Outputs is a feature that guarantees the model will return a response that exactly matches a JSON schema you define. Unlike prompting the model to "respond in JSON" (which sometimes fails or produces malformed JSON), Structured Outputs constrains the model at the token generation level - it can only emit tokens that are valid according to your schema. This makes it reliable enough to use in production without any error handling for JSON parsing failures. Structured Outputs was introduced in gpt-4o-2024-08-06 and later models.

The OpenAI Python SDK integrates directly with Pydantic - you pass your Pydantic model class to client.beta.chat.completions.parse() and the SDK automatically converts it to a JSON schema, sends it to the API, and deserialises the response back into a validated Pydantic model instance. This gives you full type safety end-to-end: the LLM output is automatically a Python object with the correct types, validated by Pydantic, with no manual JSON parsing needed. This is ideal for data extraction, classification, and any task where you need machine-readable structured output from an LLM.

The below example shows three practical use cases: extracting migration metadata from unstructured text, classifying support tickets with confidence scores, and extracting multiple entities from a paragraph.

# pip install openai pydantic
from openai import OpenAI
from pydantic import BaseModel, Field
from typing import Optional, List
from enum import Enum

client = OpenAI(api_key="your-api-key-here")

# Example 1: Extract structured migration metadata from a free-text description
class DatabaseType(str, Enum):
    MYSQL = "MySQL"
    POSTGRESQL = "PostgreSQL"
    ORACLE = "Oracle"
    SQLSERVER = "SQL Server"
    MONGODB = "MongoDB"

class MigrationSpec(BaseModel):
    source_database: DatabaseType
    target_database: DatabaseType
    estimated_rows: Optional[int] = Field(None, description="Estimated row count if mentioned")
    timeline_weeks: Optional[int] = Field(None, description="Timeline in weeks if mentioned")
    key_challenges: List[str] = Field(default_factory=list,
                                      description="List of challenges or risks mentioned")
    recommended_approach: str = Field(description="Recommended migration approach")

text = """
We need to migrate our legacy Oracle 11g database to Aurora PostgreSQL on AWS.
The main tables have about 50 million rows and we have 200+ stored procedures
that will need conversion. The business cannot afford more than 30 minutes of downtime.
We want to complete this in about 3 months.
"""

completion = client.beta.chat.completions.parse(
    model="gpt-4o-2024-08-06",
    messages=[
        {"role": "system", "content": "Extract structured migration information from the text."},
        {"role": "user", "content": text}
    ],
    response_format=MigrationSpec
)

spec = completion.choices[0].message.parsed
print("Source:", spec.source_database)
print("Target:", spec.target_database)
print("Rows:", spec.estimated_rows)
print("Timeline:", spec.timeline_weeks, "weeks")
print("Challenges:", spec.key_challenges)
print("Approach:", spec.recommended_approach)

It gives the following output,

Source: Oracle
Target: PostgreSQL
Rows: 50000000
Timeline: 12 weeks
Challenges: ['200+ stored procedures requiring conversion',
             '30-minute maximum downtime constraint']
Approach: 'Use AWS DMS with full-load-and-cdc mode for near-zero-downtime migration,
           combined with AWS SCT for stored procedure conversion.'

from openai import OpenAI
from pydantic import BaseModel, Field
from typing import List
from enum import Enum

client = OpenAI(api_key="your-api-key-here")

# Example 2: Classify and extract multiple entities from support tickets
class TicketCategory(str, Enum):
    BILLING = "BILLING"
    TECHNICAL = "TECHNICAL"
    ACCOUNT = "ACCOUNT"
    MIGRATION = "MIGRATION"
    OTHER = "OTHER"

class Priority(str, Enum):
    LOW = "LOW"
    MEDIUM = "MEDIUM"
    HIGH = "HIGH"
    CRITICAL = "CRITICAL"

class TicketAnalysis(BaseModel):
    category: TicketCategory
    priority: Priority
    summary: str = Field(description="One-sentence summary of the issue")
    affected_tables: List[str] = Field(default_factory=list,
                                       description="Database tables mentioned")
    sentiment: str = Field(description="Customer sentiment: positive, neutral, or negative")
    suggested_action: str = Field(description="Recommended next action for support team")

tickets = [
    "Our CDC pipeline stopped syncing the orders and customers tables 2 hours ago. "
    "We are losing production data and need this fixed immediately!",
    "Hi, I was charged twice for my Pro plan this month."
]

for ticket in tickets:
    result = client.beta.chat.completions.parse(
        model="gpt-4o-2024-08-06",
        messages=[
            {"role": "system", "content": "Analyse this support ticket and extract structured information."},
            {"role": "user", "content": ticket}
        ],
        response_format=TicketAnalysis
    )
    analysis = result.choices[0].message.parsed
    print(f"Category: {analysis.category} | Priority: {analysis.priority}")
    print(f"Summary: {analysis.summary}")
    print(f"Affected tables: {analysis.affected_tables}")
    print(f"Sentiment: {analysis.sentiment}")
    print(f"Action: {analysis.suggested_action}\n")

It gives the following output,

Category: MIGRATION | Priority: CRITICAL
Summary: CDC pipeline has stopped syncing orders and customers tables, causing data loss.
Affected tables: ['orders', 'customers']
Sentiment: negative
Action: Escalate to on-call engineer immediately, check Debezium connector status
        and Kafka consumer lag for the affected topics.

Category: BILLING | Priority: MEDIUM
Summary: Customer was charged twice for their Pro plan subscription.
Affected tables: []
Sentiment: neutral
Action: Review billing records, issue refund for duplicate charge, send confirmation email.

Structured Outputs vs JSON mode vs function calling:

Structured Outputs (response_format=YourPydanticModel) - Guarantees 100% schema compliance. The model cannot deviate from the schema. Use for production data extraction where parsing failures are not acceptable.

JSON mode (response_format={"type":"json_object"}) - Guarantees valid JSON but not a specific schema. The model may include or omit fields. Use when you need valid JSON but do not have a strict schema requirement.

Function calling / tool calling - The model chooses when to call a function and populates its arguments. Use for agentic workflows where the model needs to decide which action to take, not just extract data from text.

Send your comments, suggestions or queries regarding this site to [email protected].