Ustrukturyzowane dane wyjściowe z Pydantic

Dlaczego Pydantic do danych wyjściowych LLM?

Definiowanie podstawowego schematu Pydantic

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

class Sentiment(str, Enum):
    positive = 'positive'
    negative = 'negative'
    neutral = 'neutral'

class ReviewAnalysis(BaseModel):
    sentiment: Sentiment
    confidence: float = Field(ge=0.0, le=1.0, description='Confidence score 0-1')
    key_themes: List[str] = Field(description='Main topics mentioned in the review')
    summary: str = Field(max_length=200, description='One-sentence summary')
    product_name: Optional[str] = Field(default=None, description='Product mentioned, if any')
    would_recommend: Optional[bool] = None

# Pydantic validates types and constraints at instantiation
example = ReviewAnalysis(
    sentiment=Sentiment.positive,
    confidence=0.95,
    key_themes=['fast delivery', 'good quality'],
    summary='Customer loves the product and quick shipping.',
    product_name='Wireless Headphones',
    would_recommend=True
)
print(example.model_dump_json(indent=2))

Pydantic z ustrukturyzowanymi danymi wyjściowymi OpenAI

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

client = openai.OpenAI()

class Sentiment(str, Enum):
    positive = 'positive'
    negative = 'negative'
    neutral = 'neutral'

class ReviewAnalysis(BaseModel):
    sentiment: Sentiment
    confidence: float
    key_themes: List[str]
    summary: str
    would_recommend: Optional[bool]

review_text = '''
I bought this laptop for my design work and I am blown away. It handles Photoshop
like a dream, the screen colors are beautiful, and it has not slowed down once in
three months. Battery life could be better but overall highly recommend!
'''

result = client.beta.chat.completions.parse(
    model='gpt-4o-mini',
    messages=[
        {'role': 'system', 'content': 'Analyze the customer review and extract structured information.'},
        {'role': 'user', 'content': review_text}
    ],
    response_format=ReviewAnalysis
)

analysis = result.choices[0].message.parsed
print(f'Sentiment: {analysis.sentiment.value}')
print(f'Confidence: {analysis.confidence}')
print(f'Themes: {analysis.key_themes}')
print(f'Recommend: {analysis.would_recommend}')

Zagnieżdżone modele Pydantic

from pydantic import BaseModel
from typing import List, Optional

class Address(BaseModel):
    street: Optional[str]
    city: str
    country: str
    postal_code: Optional[str]

class ContactInfo(BaseModel):
    email: Optional[str]
    phone: Optional[str]
    address: Optional[Address]

class Person(BaseModel):
    full_name: str
    age: Optional[int]
    job_title: Optional[str]
    contact: ContactInfo
    skills: List[str]

# When you pass Person to response_format, the API generates:
# {
#   "full_name": "...",
#   "contact": {
#     "email": "...",
#     "address": { "city": "...", "country": "..." }
#   },
#   "skills": ["...", "..."]
# }
print('Nested model defined - pass to response_format for extraction')

Walidacja pól za pomocą walidatorów Pydantic

from pydantic import BaseModel, Field, field_validator
from typing import Optional
import re

class ExtractedContact(BaseModel):
    name: str
    email: Optional[str] = None
    phone: Optional[str] = None
    confidence: float = Field(ge=0.0, le=1.0)

    @field_validator('email')
    @classmethod
    def validate_email(cls, v):
        if v is not None:
            # Basic email format check
            if not re.match(r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$', v):
                raise ValueError(f'Invalid email format: {v}')
        return v

    @field_validator('phone')
    @classmethod
    def normalize_phone(cls, v):
        if v is not None:
            # Remove non-digit characters for normalization
            digits = re.sub(r'[^0-9+]', '', v)
            return digits
        return v

try:
    contact = ExtractedContact(name='Alice', email='not-an-email', confidence=0.9)
except Exception as e:
    print(f'Validation error: {e}')

Ekstrakcja list obiektów

import openai
from pydantic import BaseModel
from typing import List

client = openai.OpenAI()

class ActionItem(BaseModel):
    task: str
    assignee: str
    due_date: str  # or use datetime with proper parsing
    priority: str  # high / medium / low

class MeetingNotes(BaseModel):
    meeting_title: str
    action_items: List[ActionItem]
    key_decisions: List[str]

meeting_transcript = '''
Q3 Planning Meeting - June 2025
Decision: Launch new feature in July.
Decision: Extend free trial to 30 days.
Action: Alice to finalize designs by June 30th - High priority.
Action: Bob to write API docs by July 5th - Medium priority.
Action: Carol to set up staging environment by June 28th - High priority.
'''

result = client.beta.chat.completions.parse(
    model='gpt-4o-mini',
    messages=[
        {'role': 'system', 'content': 'Extract structured data from meeting notes.'},
        {'role': 'user', 'content': meeting_transcript}
    ],
    response_format=MeetingNotes
)
notes = result.choices[0].message.parsed
for item in notes.action_items:
    print(f'[{item.priority.upper()}] {item.task} -> {item.assignee} by {item.due_date}')

Pola opcjonalne i wartości domyślne

Konwertowanie modeli Pydantic na schemat JSON

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

class ProductExtraction(BaseModel):
    name: str = Field(description='Product name as mentioned in the text')
    price_usd: Optional[float] = Field(default=None, description='Price in USD')
    features: List[str] = Field(default_factory=list)
    in_stock: bool = Field(description='Whether the product is currently available')

# See the JSON Schema that will be sent to the API
schema = ProductExtraction.model_json_schema()
print(json.dumps(schema, indent=2))
# This shows exactly what constraints the API will enforce

Obsługa niepowodzeń ekstrakcji

import openai
from pydantic import BaseModel, ValidationError
from typing import Optional

client = openai.OpenAI()

class ContactExtraction(BaseModel):
    name: Optional[str] = None
    email: Optional[str] = None
    phone: Optional[str] = None

try:
    result = client.beta.chat.completions.parse(
        model='gpt-4o-mini',
        messages=[
            {'role': 'system', 'content': 'Extract contact information.'},
            {'role': 'user', 'content': 'I would like to discuss partnership opportunities.'}
        ],
        response_format=ContactExtraction
    )
    msg = result.choices[0].message
    if msg.refusal:
        print('Refused:', msg.refusal)
    else:
        contact = msg.parsed
        if not any([contact.name, contact.email, contact.phone]):
            print('No contact information found in text')
        else:
            print(contact.model_dump())
except ValidationError as e:
    print('Validation failed:', e)

Używanie Pydantic z biblioteką instructor

# pip install instructor
import instructor
import openai
from pydantic import BaseModel, Field
from typing import Optional

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

class ProductInfo(BaseModel):
    name: str
    price_usd: float = Field(gt=0, description='Price must be positive')
    brand: Optional[str] = None

# instructor automatically retries if Pydantic validation fails
product = client.chat.completions.create(
    model='gpt-4o-mini',
    messages=[
        {'role': 'user', 'content': 'The Sony WH-1000XM5 headphones cost $279.99 at Best Buy.'}
    ],
    response_model=ProductInfo,  # instructor-specific parameter
    max_retries=3
)
print(f'{product.name}: ${product.price_usd} by {product.brand}')

Unie rozróżniane i dynamiczne schematy

from pydantic import BaseModel
from typing import Union, Literal, Optional

class FlightExpense(BaseModel):
    expense_type: Literal['flight']
    airline: str
    departure_city: str
    arrival_city: str
    amount_usd: float

class HotelExpense(BaseModel):
    expense_type: Literal['hotel']
    hotel_name: str
    check_in: str
    check_out: str
    amount_usd: float

class MealExpense(BaseModel):
    expense_type: Literal['meal']
    restaurant: Optional[str]
    amount_usd: float

class ExpenseReport(BaseModel):
    submitter: str
    expenses: list[Union[FlightExpense, HotelExpense, MealExpense]]
    total_usd: float

print('Discriminated union schema - model selects correct subtype per item')

Szybkie sprawdzenie

Podsumowanie lekcji