Instructor: The Python Library That Forces LLMs to Output Valid JSON 100% of the Time — 2026 Guide

# Install Instructor
pip install instructor

# Install your preferred LLM client (OpenAI shown)
pip install openai

import instructor
from openai import OpenAI
from pydantic import BaseModel

# Patch the OpenAI client with Instructor
client = instructor.from_openai(OpenAI())

# Define your output schema as a Pydantic model
class UserProfile(BaseModel):
    name: str
    age: int
    email: str
    interests: list[str]

# Extract structured data from natural language
def extract_profile(user_description: str) -> UserProfile:
    return client.chat.completions.create(
        model="gpt-4o",
        response_model=UserProfile,
        messages=[
            {
                "role": "user",
                "content": f"Extract a user profile from this description: {user_description}"
            }
        ]
    )

# Usage
profile = extract_profile(
    "Sarah is a 28-year-old software engineer from Seattle. "
    "She loves hiking, photography, and reading sci-fi novels. "
    "Her email is sarah.chen@example.com"
)

print(profile)
# UserProfile(name='Sarah', age=28, email='sarah.chen@example.com', 
#             interests=['hiking', 'photography', 'reading sci-fi novels'])

# Access typed fields directly
print(f"Name: {profile.name}, Age: {profile.age}")
print(f"Email valid: {'@' in profile.email}")

from pydantic import BaseModel, Field, field_validator

class ValidatedProduct(BaseModel):
    name: str = Field(description="Product name, max 50 characters")
    price: float = Field(description="Price in USD, must be positive")
    category: str = Field(description="One of: electronics, clothing, food, books")
    
    @field_validator('category')
    @classmethod
    def validate_category(cls, v):
        allowed = {'electronics', 'clothing', 'food', 'books'}
        if v.lower() not in allowed:
            raise ValueError(f"Category must be one of: {allowed}")
        return v.lower()
    
    @field_validator('price')
    @classmethod
    def validate_price(cls, v):
        if v <= 0:
            raise ValueError("Price must be positive")
        return round(v, 2)

# Instructor automatically retries on validation failure
def parse_product(description: str) -> ValidatedProduct:
    return client.chat.completions.create(
        model="gpt-4o",
        response_model=ValidatedProduct,
        max_retries=3,  # Retry up to 3 times with feedback
        messages=[
            {"role": "user", "content": f"Parse this product: {description}"}
        ]
    )

# This will work even if the first attempt has issues
product = parse_product(
    "Wireless Bluetooth headphones with noise cancellation, priced at $79.99. "
    "Electronics category."
)
print(product)
# ValidatedProduct(name='Wireless Bluetooth Headphones', 
#                  price=79.99, category='electronics')

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

class Address(BaseModel):
    street: str
    city: str
    state: str = Field(description="2-letter state code")
    zip_code: str
    country: str = "US"

class OrderItem(BaseModel):
    product_name: str
    quantity: int = Field(ge=1, description="Must be at least 1")
    unit_price: float = Field(gt=0)
    
    @property
    def total(self) -> float:
        return self.quantity * self.unit_price

class CustomerOrder(BaseModel):
    customer_name: str
    customer_email: str
    shipping_address: Address
    billing_address: Optional[Address] = None
    items: List[OrderItem]
    order_notes: Optional[str] = None
    
    @property
    def grand_total(self) -> float:
        return sum(item.total for item in self.items)

def extract_order(email_text: str) -> CustomerOrder:
    return client.chat.completions.create(
        model="gpt-4o",
        response_model=CustomerOrder,
        messages=[
            {"role": "user", "content": f"Extract order from email:\n\n{email_text}"}
        ]
    )

order = extract_order("""
Hi, I'd like to place an order.

Customer: John Smith (john.smith@email.com)
Ship to: 123 Oak Street, San Francisco, CA 94102

Items:
- MacBook Pro M3, qty 1, $1999
- USB-C Hub, qty 2, $49 each

Please gift wrap the laptop.
""")

print(f"Customer: {order.customer_name}")
print(f"Shipping to: {order.shipping_address.city}")
print(f"Order total: ${order.grand_total:.2f}")

# Optional fields with default values are handled gracefully
from pydantic import BaseModel
from typing import Optional
from datetime import datetime

class Event(BaseModel):
    name: str
    start_time: datetime
    end_time: Optional[datetime] = None
    location: Optional[str] = None
    description: Optional[str] = ""

event = client.chat.completions.create(
    model="gpt-4o-mini",
    response_model=Event,
    messages=[{
        "role": "user",
        "content": "Team standup meeting tomorrow at 10 AM in Conference Room B"
    }]
)
print(f"Event: {event.name}")
print(f"Starts: {event.start_time}")
print(f"Location: {event.location}")  # Conference Room B
print(f"Description: '{event.description}'")  # Uses default empty string

# --- Anthropic Claude ---
import anthropic
import instructor

anthropic_client = instructor.from_anthropic(anthropic.Anthropic())

result = anthropic_client.messages.create(
    model="claude-sonnet-4-20250514",
    response_model=UserProfile,
    max_tokens=1024,
    messages=[{"role": "user", "content": "Extract: Mike is 35, likes tennis"}]
)
print(result)

# --- Google Gemini ---
import google.generativeai as genai
import instructor

gemini_client = instructor.from_gemini(
    genai.GenerativeModel("gemini-2.5-pro")
)

result = gemini_client.generate_content(
    response_model=UserProfile,
    contents=["Extract: Lisa is 29, likes cooking and yoga"]
)
print(result)

# --- Cohere ---
import cohere
import instructor

cohere_client = instructor.from_cohere(cohere.Client())

result = cohere_client.chat(
    response_model=UserProfile,
    message="Extract: David is 42, likes golf and fishing"
)
print(result)

# Model configuration with system prompts and temperature
class CodeReview(BaseModel):
    quality_score: int  # 1-10
    issues_found: list[str]
    suggestions: list[str]
    is_safe_to_merge: bool

review = client.chat.completions.create(
    model="gpt-4o",
    response_model=CodeReview,
    temperature=0.2,  # Lower temperature for more deterministic output
    messages=[
        {
            "role": "system",
            "content": "You are a senior software engineer conducting code reviews."
        },
        {
            "role": "user",
            "content": "Review this function:\n\ndef divide(a, b):\n    return a / b"
        }
    ]
)
print(f"Quality: {review.quality_score}/10")
print(f"Safe to merge: {review.is_safe_to_merge}")

import asyncio
import instructor
from openai import AsyncOpenAI
from pydantic import BaseModel

# Use async client for batch processing
async_client = instructor.from_openai(AsyncOpenAI())

class SentimentResult(BaseModel):
    text: str
    sentiment: str  # "positive", "negative", "neutral"
    confidence: float
    key_phrases: list[str]

async def analyze_single(text: str) -> SentimentResult:
    return await async_client.chat.completions.create(
        model="gpt-4o-mini",
        response_model=SentimentResult,
        messages=[
            {"role": "user", "content": f"Analyze sentiment: {text}"}
        ]
    )

async def analyze_batch(texts: list[str]) -> list[SentimentResult]:
    """Process multiple texts concurrently."""
    tasks = [analyze_single(text) for text in texts]
    results = await asyncio.gather(*tasks)
    return results

# Process 100 reviews concurrently
texts = [
    "This product exceeded my expectations!",
    "Terrible quality, broke after one day.",
    "It's okay, nothing special but does the job.",
    # ... 97 more items
]

results = asyncio.run(analyze_batch(texts))
positive = sum(1 for r in results if r.sentiment == "positive")
print(f"Positive: {positive}/{len(results)}")

from typing import Iterable
from pydantic import BaseModel

class PartialArticle(BaseModel):
    title: str
    sections: list[str]
    key_points: list[str]

# Stream structured data as it's generated
def stream_article(topic: str) -> Iterable[PartialArticle]:
    return client.chat.completions.create_partial(
        model="gpt-4o",
        response_model=PartialArticle,
        stream=True,
        messages=[
            {"role": "user", "content": f"Write an article outline about: {topic}"}
        ]
    )

# Consume partial results as they arrive
for partial in stream_article("renewable energy trends 2026"):
    print(f"Title: {partial.title}")
    print(f"Sections so far: {len(partial.sections)}")
    print("---")

from pydantic import BaseModel, field_validator

class StrictDateRange(BaseModel):
    start_date: str = Field(description="YYYY-MM-DD format")
    end_date: str = Field(description="YYYY-MM-DD format, must be after start")
    
    @field_validator('start_date', 'end_date')
    @classmethod
    def validate_date_format(cls, v):
        from datetime import datetime
        datetime.strptime(v, "%Y-%m-%d")
        return v
    
    @field_validator('end_date')
    @classmethod
    def validate_order(cls, end, info):
        start = info.data.get('start_date')
        if start and end <= start:
            raise ValueError("end_date must be after start_date")
        return end

# Instructor will retry with specific validation error feedback
def extract_date_range(text: str) -> StrictDateRange:
    return client.chat.completions.create(
        model="gpt-4o",
        response_model=StrictDateRange,
        max_retries=3,
        messages=[
            {"role": "user", "content": f"Extract date range: {text}"}
        ]
    )

# Even if the model initially swaps dates or uses wrong format,
# Instructor will re-ask with the specific error message
try:
    result = extract_date_range(
        "The project ran from March 15, 2026 to January 10, 2026"
    )
    print(result)
except Exception as e:
    print(f"Failed after max retries: {e}")

from typing import Literal

class SupportTicket(BaseModel):
    customer_query: str
    category: Literal[
        "billing", 
        "technical_support", 
        "account_access",
        "feature_request",
        "refund",
        "general_inquiry"
    ]
    priority: Literal["low", "medium", "high", "urgent"]
    suggested_response: str

def classify_ticket(ticket_text: str) -> SupportTicket:
    return client.chat.completions.create(
        model="gpt-4o-mini",
        response_model=SupportTicket,
        messages=[
            {
                "role": "system",
                "content": "You are a customer support triage specialist."
            },
            {"role": "user", "content": f"Classify this ticket:\n\n{ticket_text}"}
        ]
    )

# Classification is guaranteed to be one of the allowed values
ticket = classify_ticket(
    "I was charged twice for my subscription this month. "
    "Please refund the duplicate charge immediately."
)
print(f"Category: {ticket.category}")  # Always "billing"
print(f"Priority: {ticket.priority}")  # Always one of the 4 values

# Extracting structured data from long documents
from pydantic import BaseModel

class ExtractedFact(BaseModel):
    subject: str
    predicate: str
    object_: str
    confidence: float

class DocumentExtraction(BaseModel):
    title: str
    facts: list[ExtractedFact]
    entities: list[str]
    summary: str

extraction = client.chat.completions.create(
    model="gpt-4o",
    response_model=DocumentExtraction,
    messages=[{
        "role": "user",
        "content": (
            "Extract structured information from this text:\n\n"
            "Apple Inc. was founded by Steve Jobs and Steve Wozniak "
            "on April 1, 1976. The company is headquartered in "
            "Cupertino, California. Tim Cook became CEO in 2011."
        )
    }]
)
print(f"Title: {extraction.title}")
print(f"Entities found: {extraction.entities}")
print(f"Total facts: {len(extraction.facts)}")

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import instructor
from openai import OpenAI

app = FastAPI(title="Structured LLM API")
client = instructor.from_openai(OpenAI())

# Request schema
class ExtractionRequest(BaseModel):
    text: str
    extract_fields: list[str]

# Response schema
class ExtractedData(BaseModel):
    entities: list[dict]
    relationships: list[dict]
    summary: str

@app.post("/extract", response_model=ExtractedData)
async def extract_entities(request: ExtractionRequest):
    """Extract structured entities from unstructured text."""
    try:
        result = client.chat.completions.create(
            model="gpt-4o",
            response_model=ExtractedData,
            messages=[
                {
                    "role": "user",
                    "content": (
                        f"Extract entities from this text. "
                        f"Focus on: {', '.join(request.extract_fields)}\n\n"
                        f"Text: {request.text}"
                    )
                }
            ]
        )
        return result
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

# Run with: uvicorn main:app --reload

from typing import Type

class SearchQuery(BaseModel):
    """Generated search query with parameters"""
    keywords: list[str]
    filters: dict[str, str]
    sort_by: Literal["relevance", "date", "price_asc", "price_desc"]
    
def generate_search(user_request: str) -> SearchQuery:
    return client.chat.completions.create(
        model="gpt-4o",
        response_model=SearchQuery,
        messages=[
            {
                "role": "user",
                "content": f"Generate an optimized search query for: {user_request}"
            }
        ]
    )

query = generate_search(
    "Find wireless earbuds under $100 with good battery life, newest first"
)
print(query.keywords)  # ['wireless earbuds', 'bluetooth']
print(query.filters)   # {'max_price': '100'}
print(query.sort_by)   # 'date'

import logging
import instructor

# Enable detailed logging
instructor.enable_logging()
logging.basicConfig(level=logging.DEBUG)

# Or configure specific loggers
logger = logging.getLogger("instructor")
logger.setLevel(logging.INFO)

# Add file handler for production
handler = logging.FileHandler("instructor.log")
handler.setFormatter(logging.Formatter(
    '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
))
logger.addHandler(handler)

# All retry attempts and validation errors are now logged
result = client.chat.completions.create(
    model="gpt-4o",
    response_model=UserProfile,
    max_retries=3,
    messages=[{"role": "user", "content": "Extract: Jane, age 25, likes art"}]
)