Structured Outputs

Free-form Output                 Structured Output
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
"The sentiment is positive       {"sentiment": "positive",
and confidence is around          "confidence": 0.92,
92%. The key topics are           "topics": ["service", "quality"]}
service and quality."

from openai import OpenAI

client = OpenAI()

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {
            "role": "system",
            "content": "Extract information and return valid JSON."
        },
        {
            "role": "user", 
            "content": "John Smith, 35 years old, software engineer at Google"
        }
    ],
    response_format={"type": "json_object"}
)

import json
data = json.loads(response.choices[0].message.content)
print(data)
# {"name": "John Smith", "age": 35, "job": "software engineer", "company": "Google"}

def extract_with_schema(text: str, schema: dict) -> dict:
    """Extract structured data following a schema"""
    
    schema_str = json.dumps(schema, indent=2)
    
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": f"""Extract information from the text and return JSON matching this schema:

{schema_str}

Rules:
- Return ONLY valid JSON
- Include all required fields
- Use null for missing optional fields
- Match the exact field names and types"""
            },
            {"role": "user", "content": text}
        ],
        response_format={"type": "json_object"}
    )
    
    return json.loads(response.choices[0].message.content)

# Usage
schema = {
    "name": "string",
    "email": "string", 
    "age": "integer",
    "interests": ["string"]
}

result = extract_with_schema(
    "Jane Doe (jane@example.com) is 28 and loves hiking and photography",
    schema
)

from pydantic import BaseModel
from typing import Optional, List

class Person(BaseModel):
    name: str
    age: int
    email: Optional[str] = None
    skills: List[str]

response = client.beta.chat.completions.parse(
    model="gpt-4o-2024-08-06",
    messages=[
        {"role": "system", "content": "Extract person information."},
        {"role": "user", "content": "Bob is 30, knows Python and JavaScript, email: bob@test.com"}
    ],
    response_format=Person
)

person = response.choices[0].message.parsed
print(f"Name: {person.name}, Age: {person.age}")
print(f"Skills: {', '.join(person.skills)}")

from pydantic import BaseModel, Field
from typing import List, Optional
from enum import Enum

class Priority(str, Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"

class Task(BaseModel):
    title: str = Field(description="Short task title")
    description: str = Field(description="Detailed task description")
    priority: Priority
    estimated_hours: float = Field(ge=0, le=100)

class ProjectPlan(BaseModel):
    project_name: str
    objective: str
    tasks: List[Task]
    total_hours: float
    risks: List[str]
    dependencies: Optional[List[str]] = None

response = client.beta.chat.completions.parse(
    model="gpt-4o-2024-08-06",
    messages=[
        {
            "role": "system",
            "content": "You are a project planning assistant. Create detailed project plans."
        },
        {
            "role": "user",
            "content": "Plan a website redesign project for a small business"
        }
    ],
    response_format=ProjectPlan
)

plan = response.choices[0].message.parsed
print(f"Project: {plan.project_name}")
for task in plan.tasks:
    print(f"  - [{task.priority.value}] {task.title}: {task.estimated_hours}h")

pip install instructor

import instructor
from openai import OpenAI
from pydantic import BaseModel, Field
from typing import List

# Patch OpenAI client
client = instructor.from_openai(OpenAI())

class ExtractedEntity(BaseModel):
    name: str
    entity_type: str = Field(description="person, organization, location, etc.")
    context: str = Field(description="How this entity is mentioned")

class DocumentAnalysis(BaseModel):
    summary: str = Field(description="One paragraph summary")
    entities: List[ExtractedEntity]
    sentiment: str = Field(description="positive, negative, or neutral")
    key_topics: List[str]
    language: str

# Use like normal, but with response_model
analysis = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "user", "content": "Analyze this article: Apple announced..."}
    ],
    response_model=DocumentAnalysis
)

print(analysis.summary)
for entity in analysis.entities:
    print(f"  {entity.entity_type}: {entity.name}")

from pydantic import BaseModel, Field, field_validator
from typing import List
import instructor

class ValidatedExtraction(BaseModel):
    email: str
    phone: str
    website: str
    
    @field_validator("email")
    @classmethod
    def validate_email(cls, v):
        if "@" not in v:
            raise ValueError("Invalid email format")
        return v.lower()
    
    @field_validator("phone")
    @classmethod
    def validate_phone(cls, v):
        digits = "".join(c for c in v if c.isdigit())
        if len(digits) < 10:
            raise ValueError("Phone must have at least 10 digits")
        return digits

client = instructor.from_openai(OpenAI())

# Instructor will retry if validation fails -- the key production feature.
# When the model returns "555-123" (too few digits), the validator rejects it,
# Instructor sends the error message back to the model ("Phone must have at 
# least 10 digits"), and the model self-corrects on the next attempt.
result = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "user", "content": "Contact: john@example.com, 555-123-4567, www.example.com"}
    ],
    response_model=ValidatedExtraction,
    max_retries=3  # Usually succeeds on retry 1; 3 is a safe ceiling
)

from pydantic import BaseModel
from typing import List
import instructor

class StoryChapter(BaseModel):
    title: str
    content: str
    characters: List[str]

class Story(BaseModel):
    title: str
    chapters: List[StoryChapter]

client = instructor.from_openai(OpenAI())

# Stream partial objects
for partial in client.chat.completions.create_partial(
    model="gpt-4o",
    messages=[
        {"role": "user", "content": "Write a 3 chapter short story about a robot"}
    ],
    response_model=Story
):
    # Partial object updates as it streams
    if partial.title:
        print(f"Title: {partial.title}")
    if partial.chapters:
        print(f"Chapters so far: {len(partial.chapters)}")

from openai import OpenAI
import json

client = OpenAI()

tools = [
    {
        "type": "function",
        "function": {
            "name": "create_calendar_event",
            "description": "Create a calendar event",
            "strict": True,  # Enable strict mode
            "parameters": {
                "type": "object",
                "properties": {
                    "title": {
                        "type": "string",
                        "description": "Event title"
                    },
                    "start_time": {
                        "type": "string",
                        "description": "Start time in ISO format"
                    },
                    "end_time": {
                        "type": "string", 
                        "description": "End time in ISO format"
                    },
                    "attendees": {
                        "type": "array",
                        "items": {"type": "string"},
                        "description": "List of attendee emails"
                    },
                    "location": {
                        "type": "string",
                        "description": "Event location"
                    }
                },
                "required": ["title", "start_time", "end_time"],
                "additionalProperties": False
            }
        }
    }
]

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "user", "content": "Schedule a team meeting tomorrow at 2pm for 1 hour with alice@co.com and bob@co.com"}
    ],
    tools=tools,
    tool_choice={"type": "function", "function": {"name": "create_calendar_event"}}
)

# Parse the structured function call
tool_call = response.choices[0].message.tool_calls[0]
event_data = json.loads(tool_call.function.arguments)
print(event_data)

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

class OutputFormat(str, Enum):
    JSON = "json"
    PYDANTIC = "pydantic"
    FUNCTION_CALL = "function_call"

class StructuredOutputPipeline:
    """Unified pipeline for structured outputs"""
    
    def __init__(self, model: str = "gpt-4o"):
        self.model = model
        self.raw_client = OpenAI()
        self.instructor_client = instructor.from_openai(OpenAI())
    
    def extract(
        self,
        text: str,
        schema: type[BaseModel],
        system_prompt: str = "Extract information accurately.",
        max_retries: int = 2
    ) -> BaseModel:
        """Extract structured data using Instructor"""
        return self.instructor_client.chat.completions.create(
            model=self.model,
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": text}
            ],
            response_model=schema,
            max_retries=max_retries
        )
    
    def extract_batch(
        self,
        texts: List[str],
        schema: type[BaseModel],
        system_prompt: str = "Extract information accurately."
    ) -> List[BaseModel]:
        """Extract from multiple texts"""
        import asyncio
        from openai import AsyncOpenAI
        
        async_client = instructor.from_openai(AsyncOpenAI())
        
        async def extract_one(text: str) -> BaseModel:
            return await async_client.chat.completions.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": system_prompt},
                    {"role": "user", "content": text}
                ],
                response_model=schema
            )
        
        async def extract_all():
            return await asyncio.gather(*[extract_one(t) for t in texts])
        
        return asyncio.run(extract_all())
    
    def extract_with_confidence(
        self,
        text: str,
        schema: type[BaseModel]
    ) -> tuple[BaseModel, float]:
        """Extract with confidence score"""
        
        class WithConfidence(BaseModel):
            data: schema
            confidence: float = Field(ge=0, le=1, description="Confidence in extraction accuracy")
            reasoning: str = Field(description="Why this confidence level")
        
        result = self.instructor_client.chat.completions.create(
            model=self.model,
            messages=[
                {
                    "role": "system",
                    "content": "Extract information and rate your confidence (0-1) in the extraction accuracy."
                },
                {"role": "user", "content": text}
            ],
            response_model=WithConfidence
        )
        
        return result.data, result.confidence

# Usage
pipeline = StructuredOutputPipeline()

class ProductReview(BaseModel):
    product_name: str
    rating: float = Field(ge=1, le=5)
    pros: List[str]
    cons: List[str]
    recommendation: bool

review, confidence = pipeline.extract_with_confidence(
    "The new iPhone 15 is amazing! Great camera, fast processor. "
    "Battery could be better though. 4.5 stars, definitely recommend!",
    ProductReview
)

print(f"Product: {review.product_name}")
print(f"Rating: {review.rating}/5")
print(f"Confidence: {confidence:.0%}")

from pydantic import BaseModel, Field
from typing import Optional, List

class FlexibleExtraction(BaseModel):
    """Schema that handles missing data gracefully -- prefer Optional fields
    over required fields when the source text may not contain the data."""
    
    name: Optional[str] = Field(None, description="Person's name if mentioned")
    email: Optional[str] = Field(None, description="Email if found")
    phone: Optional[str] = Field(None, description="Phone if found")
    
    extracted_fields: List[str] = Field(
        default_factory=list,
        description="List of fields that were successfully extracted"
    )
    missing_fields: List[str] = Field(
        default_factory=list,
        description="List of fields that could not be found"
    )
    extraction_notes: str = Field(
        "",
        description="Any notes about ambiguity or uncertainty"
    )

from pydantic import BaseModel
from typing import Union, Literal

class SuccessResponse(BaseModel):
    status: Literal["success"]
    data: dict
    message: str

class ErrorResponse(BaseModel):
    status: Literal["error"]
    error_code: str
    error_message: str

class APIResponse(BaseModel):
    response: Union[SuccessResponse, ErrorResponse]