Type-safe LLM calls in Python with OpenAI SDK and Pydantic

Most LLM integrations start as a quick chat.completions.create call and degrade into stringly-typed chaos. Achieving openai python sdk pydantic type safety means treating model outputs as structured data from the first line, not after a regex scrape.

Why type safety matters for LLM outputs

LLMs return text. Your downstream code wants objects with known fields, types, and constraints. Without a contract, you ship dict["maybe_key"] accesses that throw at 2 a.m. when the model hallucinates a different shape or omits a field. Pydantic gives you a single validation boundary; the OpenAI Python SDK gives you the transport. Together they turn probabilistic text into checked data.

Type safety here is not about pleasing mypy alone. It is about failing fast at the edge where the model output enters your system. A ValidationError is cheaper than a TypeError three layers deep in a billing job.

Install and configure the client

Use the official openai package. It speaks the OpenAI REST contract, which most inference gateways mirror.

pip install openai pydantic

Configure the client with your base URL and key. If you route through a gateway like n4n.ai, which exposes one OpenAI-compatible endpoint for 240+ models, the same client code works across providers without branching on vendor.

from openai import OpenAI

client = OpenAI(
    base_url="https://api.n4n.ai/v1",  # or https://api.openai.com/v1
    api_key="sk-...",  # pull from env in real code
)

For async services, use AsyncOpenAI and await the calls. Keep the key in environment variables or a secret manager. Never hardcode credentials in source.

Define strict Pydantic models

Model the smallest successful response you can. Add constraints with Field. Strict mode catches coercions you didn’t intend, such as a string "5" slipping into an int.

from pydantic import BaseModel, ConfigDict, Field, model_validator

class LineItem(BaseModel):
    model_config = ConfigDict(strict=True)
    sku: str = Field(min_length=3)
    qty: int = Field(ge=1)
    price_cents: int = Field(ge=0)

class Invoice(BaseModel):
    vendor: str
    items: list[LineItem]
    total_cents: int

    @model_validator(mode="after")
    def check_total(self):
        calc = sum(i.qty * i.price_cents for i in self.items)
        if calc != self.total_cents:
            raise ValueError("total does not match line items")
        return self

Cross-field validators belong in the model, not in scattered business logic. If you later need to accept "$5.00" as price_cents, write a field_validator that coerces explicitly rather than loosening strictness globally.

Force structure with JSON mode

Modern OpenAI-compatible models accept response_format={"type": "json_object"}. This reduces but does not eliminate malformed output. You still validate.

resp = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[
        {"role": "system", "content": "Return JSON matching the invoice schema only."},
        {"role": "user", "content": "Invoice from Acme: 2x SKU-123 @ 500, 1x SKU-999 @ 1200"}
    ],
    response_format={"type": "json_object"},
    temperature=0.0,
)
raw = resp.choices[0].message.content

JSON mode fails if the model can’t comply, and on some versions it forbids tool calls in the same request. The system prompt must explicitly demand the shape; the model will not infer your Pydantic schema from nowhere. Expect occasional wrapped objects like {"invoice": {...}}—unwrap in code or adjust the schema.

Use tool calls for stronger contracts

Function calling often beats JSON mode because the model emits arguments conforming to your JSON schema and the SDK parses them into a string you control. Define the tool with the Pydantic shape via model_json_schema().

tools = [{
    "type": "function",
    "function": {
        "name": "submit_invoice",
        "description": "Record a parsed invoice",
        "parameters": Invoice.model_json_schema(),
    }
}]

resp = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Invoice: Acme 2x SKU-123 @500, 1x SKU-999 @1200"}],
    tools=tools,
    tool_choice={"type": "function", "function": {"name": "submit_invoice"}},
)

The response carries tool_calls with function.arguments as a JSON string. Parse and validate identically to the JSON-mode path. Tool calls also let you enforce a specific function name, which simplifies routing in code.

Parse and validate at the boundary

Wrap parsing in a function that returns your typed object or raises a clear error. Do not swallow exceptions.

import json
from pydantic import ValidationError

def parse_invoice(raw: str) -> Invoice:
    try:
        data = json.loads(raw)
    except json.JSONDecodeError as e:
        raise ValueError(f"model returned non-JSON: {e}") from e
    try:
        return Invoice.model_validate(data)
    except ValidationError as e:
        raise ValueError(f"schema mismatch: {e}") from e

Distinguish JSONDecodeError (model ignored format) from ValidationError (shape wrong). That taxonomy drives retry vs. fallback logic.

Build a typed wrapper

Business logic should never call the SDK directly. Write one function that returns Invoice. This becomes your seam for mocking and testing.

def extract_invoice(text: str) -> Invoice:
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": text}],
        response_format={"type": "json_object"},
        temperature=0.0,
    )
    return parse_invoice(resp.choices[0].message.content)

An async variant is trivial:

async def extract_invoice_async(text: str) -> Invoice:
    resp = await async_client.chat.completions.create(...)
    return parse_invoice(resp.choices[0].message.content)

Handle fallback and degradation

Providers rate-limit and degrade. If you use a gateway with automatic fallback when a provider is degraded, you keep the same call shape and just swap model names. Otherwise implement retry with backoff.

import time
from openai import RateLimitError, APIConnectionError

def extract_invoice_with_retry(text: str, attempts=3) -> Invoice:
    for i in range(attempts):
        try:
            return extract_invoice(text)
        except (RateLimitError, APIConnectionError) as e:
            if i == attempts - 1:
                raise
            time.sleep(2 ** i)
    raise RuntimeError("unreachable")

When you control routing, set extra_headers to pass cache directives. n4n.ai forwards provider cache-control hints, so you can reuse prompt prefixes across calls and cut latency on repeated extractions.

Common pitfalls and tradeoffs

Schema drift

If you change the Pydantic model, cached responses break. Version your prompts and schemas together.

Token overhead

Sending the full JSON schema in every tool definition costs tokens. For high-volume jobs, cache the system prompt or use a smaller model for extraction before a larger one reasons over the result.

Strictness vs yield

Tight Pydantic constraints reject borderline outputs. A field_validator that coerces "$5.00" to cents saves a round-trip. Decide per field; strict mode is a default, not a religion.

JSON mode limitations

Some models silently return wrapped JSON. Your parser must unwrap or your schema must match. Test with real captured outputs.

Type safety is not correctness

Pydantic confirms shape, not truth. A valid total_cents can still be wrong. The cross-field model_validator above catches arithmetic mismatch, but semantic errors (wrong vendor) need other checks.

A reusable generic extractor

Avoid duplicating wrappers for every model. Use a generic function:

from typing import Type, TypeVar
from pydantic import BaseModel

T = TypeVar("T", bound=BaseModel)

def extract_model(cls: Type[T], text: str, model="gpt-4o-mini") -> T:
    resp = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": text}],
        response_format={"type": "json_object"},
        temperature=0.0,
    )
    raw = resp.choices[0].message.content
    try:
        return cls.model_validate(json.loads(raw))
    except (json.JSONDecodeError, ValidationError) as e:
        raise ValueError(f"extract {cls.__name__} failed: {e}") from e

Call it as extract_model(Invoice, user_text). This pattern scales to dozens of domains without new network code.

Testing the boundary

Write tests that feed raw model outputs—good and bad—to your parser. Use pytest and mock the network.

def test_parse_valid():
    raw = '{"vendor":"Acme","items":[{"sku":"SKU-123","qty":2,"price_cents":500}],"total_cents":1000}'
    inv = parse_invoice(raw)
    assert inv.vendor == "Acme"

def test_parse_invalid():
    import pytest
    with pytest.raises(ValueError):
        parse_invoice('{"vendor":"Acme"}')

Never call real APIs in unit tests. Record fixtures from staging if you need realistic malformed samples.

Recommendations

Adopt openai python sdk pydantic type safety as a default for any LLM output you persist. Start with JSON mode, move to tools when you need guaranteed arguments, and keep validation strict at the edge. The few lines of Pydantic pay back the first time a model shifts its format. The combination of openai python sdk pydantic type safety and a single client boundary keeps your system stable while the models behind it keep changing.