Structured outputs z AI: Pydantic, Instructor i JSON Schema w produkcji
Jak przestać parsować stringi z GPT i zacząć dostawać dane gotowe do użycia w kodzie — JSON Schema, Pydantic i biblioteka Instructor.
Wtorek rano, deployment na produkcję. Model zwraca: `{"name": "Jan Kowalski", "age": "trzydzieści dwa", "tags": "python, django"}`. Twój kod oczekiwał `age` jako int, `tags` jako list — i rzuca wyjątek. Model "starał się", ale nie mógł wiedzieć, że lista to `["python", "django"]`, nie string. To nie edge case — to codzienność gdy LLM i kod komunikują się przez string.
Trzy podejścia — i dlaczego pierwsze dwa zawodzą
Większość zespołów przechodzi przez te fazy. Faza 1 — "powiedz GPT żeby zwrócił JSON" — działa tydzień, potem model dodaje markdown fence lub komentarz i `json.loads` wybucha. Faza 2 — JSON mode (`response_format={"type": "json_object"}`) — stabilny JSON, ale bez schematu model sam decyduje o kształcie pól. Faza 3 — Structured Outputs z JSON Schema lub Instructor — dostajesz dokładnie to co opisałeś, z walidacją po stronie API lub kodu.
/// EWOLUCJA STRUCTURED OUTPUTS
3 podejścia — od chaosu do gwarantowanej struktury
JSON Schema i strict mode — walidacja po stronie API
OpenAI Structured Outputs (od GPT-4o) wymuszają schemat na poziomie tokenizacji — model generuje tylko tokeny pasujące do struktury. `strict: true` + `response_format` z `json_schema` gwarantuje, że odpowiedź zawsze parsuje się bez błędu. Wymagania: każdy obiekt musi mieć `additionalProperties: false` i wszystkie pola w `required` — opcjonalność realizujesz przez `anyOf` z `{"type": "null"}`.
from openai import OpenAIimport jsonclient = OpenAI()SCHEMA = { "name": "order_extraction", "strict": True, "schema": { "type": "object", "properties": { "customer_name": {"type": "string"}, "order_id": {"type": "string"}, "items": { "type": "array", "items": { "type": "object", "properties": { "product": {"type": "string"}, "quantity": {"type": "integer"}, "price_pln": {"type": "number"} }, "required": ["product", "quantity", "price_pln"], "additionalProperties": False } }, "total_pln": {"type": "number"} }, "required": ["customer_name", "order_id", "items", "total_pln"], "additionalProperties": False }}resp = client.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": "Wyodrębnij: Jan Kowalski, ORD-001234, 3x kawa 12.99 PLN, 1x herbata 8.50 PLN"}], response_format={"type": "json_schema", "json_schema": SCHEMA})order = json.loads(resp.choices[0].message.content)print(order["total_pln"])
Wynik jest zawsze poprawnym JSON zgodnym ze schematem — zero wyjątków z `json.loads`. Minusem jest verbosity: dla złożonych obiektów JSON Schema szybko staje się nieczytelny i trudny w utrzymaniu.
Pydantic jako warstwa opisu schematu
Zamiast pisać JSON Schema ręcznie, opisz strukturę jako klasę Pydantic. `Model.model_json_schema()` generuje schemat automatycznie z type hints i walidatorów. Klucz: `Field(description=...)` — model LLM czyta opisy pól i znacznie lepiej wypełnia dane gdy wie czego oczekujesz. `field_validator` pozwala dodać reguły biznesowe których JSON Schema nie obsługuje — walidację sumy, format ID, reguły warunkowe.
from pydantic import BaseModel, Field, field_validatorfrom typing import Optionalimport reclass OrderItem(BaseModel): product: str = Field(description="Nazwa produktu dokładnie jak w tekście") quantity: int = Field(ge=1, description="Liczba sztuk, min. 1") price_pln: float = Field(gt=0, description="Cena jednostkowa w PLN")class Order(BaseModel): customer_name: str = Field(description="Imię i nazwisko klienta") order_id: str = Field(description="ID zamówienia w formacie ORD-XXXXXX") items: list[OrderItem] = Field(description="Lista wszystkich pozycji zamówienia") total_pln: float = Field(description="Suma wszystkich pozycji w PLN") notes: Optional[str] = Field(default=None, description="Uwagi jeśli podane, inaczej null") @field_validator("order_id") @classmethod def validate_order_id(cls, v: str) -> str: if not re.match(r"ORD-d{6}$", v): raise ValueError(f"order_id musi być ORD-XXXXXX, got: {v}") return v @field_validator("total_pln") @classmethod def validate_total(cls, v: float, info) -> float: if "items" in info.data: expected = sum(i.price_pln * i.quantity for i in info.data["items"]) if abs(v - expected) > 0.01: raise ValueError(f"total_pln {v} != suma items {expected:.2f}") return v
`field_validator` pozwala zdefiniować reguły biznesowe — walidacja sumy, format ID, zakres dat — których JSON Schema nie obsługuje. Błąd walidacji to konkretny komunikat który możesz przekazać modelowi w następnej próbie.
Instructor — 3 linie kodu zamiast własnego parsera
Instructor opakowuje klienta OpenAI (i 10+ innych providerów) i zamienia odpowiedź bezpośrednio w zwalidowany obiekt Pydantic. Nie potrzebujesz `json.loads`, `model.model_validate` ani ręcznego retry — biblioteka robi to za Ciebie z domyślnie 3 próbami, przesyłając komunikat błędu walidacji z powrotem do modelu jako kontekst.
import instructorfrom openai import OpenAIfrom pydantic import BaseModel, Fieldfrom typing import Literalclient = instructor.from_openai(OpenAI())class ProductReview(BaseModel): sentiment: Literal["positive", "negative", "neutral"] score: int = Field(ge=1, le=5, description="Ocena 1–5") key_issues: list[str] = Field(description="Lista głównych problemów lub zalet, max 5 punktów") would_recommend: bool summary: str = Field(max_length=200, description="Jedno zdanie podsumowania")review = client.chat.completions.create( model="gpt-4o", response_model=ProductReview, messages=[ {"role": "user", "content": "Przeanalizuj: 'Produkt dotarł uszkodzony, obsługa nie odbierała przez 3 dni, w końcu zwrot ale strata czasu. Nigdy więcej.'"} ])print(review.sentiment)print(review.score)print(review.key_issues)
`response_model=ProductReview` to wszystko — Instructor generuje JSON Schema z klasy, wywołuje API, parsuje odpowiedź, waliduje Pydanticiem i przy błędzie automatycznie robi retry z błędem dołączonym do kontekstu rozmowy.
/// INSTRUCTOR — PIPELINE WALIDACJI
Od klasy Pydantic do zwalidowanego obiektu
Wzorce: ekstrakcja, klasyfikacja, normalizacja
Trzy główne zastosowania różnią się podejściem do schematu. Ekstrakcja (wyciąganie danych z tekstu) — używaj `Optional` dla pól które mogą nie wystąpić; nigdy nie wymuszaj pól których model nie może wypełnić. Klasyfikacja — użyj `Literal` lub `Enum` zamiast `str`, model wybierze tylko z dozwolonych wartości. Normalizacja — opisz w `description` dokładny format wyjściowy z przykładem i użyj `field_validator` do weryfikacji.
| Wzorzec | Typ pola | Kluczowy trick | Pułapka |
|---|---|---|---|
| Ekstrakcja | Optional[str] | null gdy pola nie ma w tekście | Wymuszanie pól których nie ma |
| Klasyfikacja | Literal["a","b","c"] | Enum zamiast str | Zbyt wiele klas (>10) — jakość spada |
| Normalizacja daty | str + validator | Format z przykładem w description | Strefy czasowe — zawsze UTC |
| Lista elementów | list[Model] | "Wyodrębnij WSZYSTKIE" w description | Duplikaty — deduplikuj w validatorze |
| Zagnieżdżone obiekty | BaseModel w BaseModel | Flat schema szybszy i dokładniejszy | Głębokość >3 — halucynacje |
from enum import Enumfrom typing import Optional, Literalfrom pydantic import BaseModel, Field, field_validatorfrom datetime import datetimeimport instructorfrom openai import OpenAIclass Priority(str, Enum): LOW = "low" MEDIUM = "medium" HIGH = "high" CRITICAL = "critical"class TicketExtraction(BaseModel): title: str = Field(max_length=100, description="Krótki tytuł zgłoszenia") priority: Priority = Field(description="Priorytet na podstawie pilności i wpływu biznesowego") category: Literal["bug", "feature", "question", "billing"] affected_users: Optional[int] = Field(default=None, ge=1, description="Liczba dotkniętych użytkowników jeśli podana, inaczej null") reported_at: Optional[str] = Field(default=None, description="Data w formacie ISO 8601 np. 2026-06-05T10:30:00Z, null jeśli nieznana") is_regression: bool = Field(description="True jeśli wcześniej działało") @field_validator("reported_at") @classmethod def validate_date(cls, v: Optional[str]) -> Optional[str]: if v is None: return v try: datetime.fromisoformat(v.replace("Z", "+00:00")) except ValueError: raise ValueError(f"reported_at musi być ISO 8601, got: {v}") return vclient = instructor.from_openai(OpenAI())ticket = client.chat.completions.create( model="gpt-4o", response_model=TicketExtraction, messages=[{"role": "user", "content": "PILNE: logowanie przestało działać o 10:30, ok 500 użytkowników, wcześniej działało"}])print(ticket.priority)print(ticket.affected_users)
Instructor działa z wieloma providerami — `instructor.from_anthropic()`, `instructor.from_gemini()`, `instructor.from_mistral()` — ten sam kod Pydantic, inny klient.
Kiedy structured output zawodzi — 4 scenariusze
Nawet z Instructorem trafiasz na ściany. Oto cztery główne i jak z nich wychodzić.
1. Model nie może wypełnić wymaganego pola. Symptom: retry loop, model hallucynuje wartość żeby "cokolwiek" wpisać. Fix: zmień pole na `Optional` i dodaj `description="null jeśli nieznane"` — pozwól modelowi przyznać brak informacji.
2. Schemat jest zbyt złożony. Symptom: model wypełnia pole losową wartością zamiast null. Fix: uprość do flat structure. Jeśli potrzebujesz złożoności, podziel na dwa wywołania — pierwsze wyciąga płaskie dane, drugie klasyfikuje lub normalizuje.
3. Walidacja biznesowa nie przechodzi po 3 próbach. Symptom: `InstructorRetryException`. Fix: złap wyjątek i zaloguj ostatnią próbę modelu — często okaże się, że reguła jest zbyt restrykcyjna lub prompt nie zawiera informacji których validator oczekuje. Poluzuj walidator lub wzbogać prompt o przykład poprawnej odpowiedzi.
4. Lista zawiera za mało elementów. Symptom: `items` ma 2 zamiast 5 pozycji. Fix: dodaj `"Wyodrębnij WSZYSTKIE elementy — nie pomijaj żadnego"` w `description`. Instructor obsługuje też `Iterable[Model]` jako `response_model` — model streamuje obiekty inkrementalnie.
---
Buduję systemy ekstrakcji i klasyfikacji danych dla firm — od prostych pipeline'ów po złożone architektury wieloetapowe z walidacją biznesową i monitoringiem. Napisz do mnie — zaczynam od analizy Twoich danych wejściowych i projektu schematu.
/// RELATED_RECORDS
Jak AI czyta faktury z maila i wprowadza je do ERP
AI odczytuje fakturę z załącznika e-mail — PDF, skan lub zdjęcie z telefonu — i wprowadza dane bezpośrednio do ERP bez ręcznego przepisywania. Pełna automatyzacja obiegu faktur kosztowych: od skrzynki mailowej do zaksięgowania dokumentu.
Od czego zacząć wdrażanie AI w firmie?
Wdrażanie AI w firmie zaczyna się nie od wyboru narzędzia, lecz od jednego powtarzalnego procesu, który dziś zabiera najwięcej czasu. Dowiedz się jak krok po kroku wybrać, opisać i zautomatyzować ten proces.
Jak zbudować wewnętrzną bazę wiedzy firmy z AI (RAG w praktyce)
Wewnętrzna baza wiedzy oparta na RAG pozwala stworzyć własnego chatbota firmowego, który odpowiada wyłącznie na podstawie dokumentów Twojej firmy — nie domysłów modelu. Bezpieczne, aktualne, precyzyjne AI z pełną kontrolą nad danymi.
Signal received?
Przerwij
Ciszę
Zainicjuj protokół. Nawiąż połączenie. Zbudujmy coś głośnego.