方寸 Portal

AI Engineer Portal

Your personal operating system for career transition.

Private mode
Back to Learning Paths
Lesson

Pydantic for AI applications

Structured output validation from LLMs, settings management with BaseSettings, schema generation for tool definitions, and response models for provider APIs.

40 min
python-for-ai-engineersphase-1portfolio

Why this matters

Pydantic is the connective tissue of modern Python AI systems. Every provider SDK, every FastAPI endpoint, every settings file you touch is already using it. But most engineers use Pydantic at 20% capacity. This lesson covers the patterns that unlock the rest: parsing LLM JSON reliably, managing configuration across environments, generating tool schemas automatically, and building response models that survive provider API changes.

Core concepts

Parsing LLM JSON output into Pydantic models

LLMs asked to return JSON produce almost-JSON: markdown fences, trailing commas, mixed quotes, leading prose. A robust parser handles the common failure modes before reaching model_validate:

import json
import re
from pydantic import BaseModel, ValidationError


class ExtractionResult(BaseModel):
    entities: list[str]
    sentiment: str
    confidence: float

    model_config = {"str_strip_whitespace": True}


def extract_json_block(text: str) -> str | None:
    text = text.strip()
    text = re.sub(r"^```(?:json)?\s*", "", text)
    text = re.sub(r"\s*```$", "", text)
    match = re.search(r"(\{[\s\S]*\}|\[[\s\S]*\])", text)
    return match.group(1) if match else None


def parse_llm_output(
    raw_text: str,
    model_class: type[BaseModel],
    fallback: BaseModel | None = None,
) -> BaseModel | None:
    json_str = extract_json_block(raw_text)
    if json_str is None:
        return fallback
    try:
        data = json.loads(json_str)
        return model_class.model_validate(data)
    except (json.JSONDecodeError, ValidationError) as exc:
        import logging
        logging.getLogger(__name__).warning(
            "llm_parse_failed",
            extra={"error": str(exc), "raw_length": len(raw_text)},
        )
        return fallback


# Usage:
fallback = ExtractionResult(entities=[], sentiment="unknown", confidence=0.0)
result = parse_llm_output(llm_response_text, ExtractionResult, fallback=fallback)

The fallback parameter lets callers decide whether a parse failure is fatal or recoverable at the call site.

Settings management with BaseSettings

pydantic-settings reads from environment variables, .env files, and secrets, with validation and type coercion built in:

from pydantic import Field, SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict


class AppSettings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        case_sensitive=False,
    )

    anthropic_api_key: SecretStr = Field(..., description="Anthropic API key")
    openai_api_key: SecretStr | None = Field(None, description="OpenAI API key (optional)")
    default_model: str = Field("claude-3-5-sonnet-20241022")
    max_tokens: int = Field(1024, ge=1, le=8192)
    temperature: float = Field(0.7, ge=0.0, le=2.0)
    redis_url: str = Field("redis://localhost:6379/0")
    log_level: str = Field("INFO", pattern="^(DEBUG|INFO|WARNING|ERROR|CRITICAL)$")


settings = AppSettings()
api_key = settings.anthropic_api_key.get_secret_value()

SecretStr fields display as ********** in logs and repr. Validation runs at startup — a missing required key crashes immediately rather than failing silently at first use.

Schema generation for tool definitions

Pydantic models generate JSON Schema automatically. This eliminates manually maintaining tool definitions that drift from your actual parameter models:

from pydantic import BaseModel, Field


class SearchOrdersParams(BaseModel):
    query: str = Field(..., description="Order ID or customer email address")
    status: str | None = Field(
        None,
        description="Filter by order status",
        pattern="^(pending|shipped|delivered|cancelled)$",
    )
    limit: int = Field(10, ge=1, le=100, description="Maximum results to return")


def to_anthropic_tool(name: str, description: str, params_model: type[BaseModel]) -> dict:
    schema = params_model.model_json_schema()
    schema.pop("title", None)
    return {"name": name, "description": description, "input_schema": schema}


def to_openai_tool(name: str, description: str, params_model: type[BaseModel]) -> dict:
    schema = params_model.model_json_schema()
    schema.pop("title", None)
    return {"type": "function", "function": {"name": name, "description": description, "parameters": schema}}

Add a field, add a Field(description=...), and the tool schema updates automatically.

Response models for provider APIs

Provider API responses change. A response model that normalizes across providers shields the rest of your code:

from pydantic import BaseModel, Field, model_validator


class TokenUsage(BaseModel):
    input_tokens: int = Field(ge=0)
    output_tokens: int = Field(ge=0)

    @property
    def total(self) -> int:
        return self.input_tokens + self.output_tokens


class NormalizedResponse(BaseModel):
    id: str
    content: str
    model: str
    stop_reason: str
    usage: TokenUsage
    latency_ms: int = Field(ge=0)

    @model_validator(mode="after")
    def content_must_not_be_empty(self) -> "NormalizedResponse":
        if not self.content.strip():
            raise ValueError("Response content is empty")
        return self


def from_anthropic(raw: dict, latency_ms: int) -> NormalizedResponse:
    return NormalizedResponse(
        id=raw["id"],
        content=raw["content"][0]["text"],
        model=raw["model"],
        stop_reason=raw["stop_reason"],
        usage=TokenUsage(
            input_tokens=raw["usage"]["input_tokens"],
            output_tokens=raw["usage"]["output_tokens"],
        ),
        latency_ms=latency_ms,
    )


def from_openai(raw: dict, latency_ms: int) -> NormalizedResponse:
    return NormalizedResponse(
        id=raw["id"],
        content=raw["choices"][0]["message"]["content"],
        model=raw["model"],
        stop_reason=raw["choices"][0]["finish_reason"],
        usage=TokenUsage(
            input_tokens=raw["usage"]["prompt_tokens"],
            output_tokens=raw["usage"]["completion_tokens"],
        ),
        latency_ms=latency_ms,
    )

Provider-specific parsing is isolated to two adapter functions. Everything downstream only ever sees NormalizedResponse.

Common mistakes

Catching ValidationError and swallowing it. Validation errors contain precise field-level messages. Log them with the original text; do not silently return None without recording why.

Storing secrets in regular str fields. Use SecretStr for API keys and credentials. It prevents the key from appearing in logs, tracebacks, and model_dump() output by default.

Manually maintaining tool schemas. Hand-written JSON schema drifts from your actual code. Generate schemas from Pydantic models and treat the schema as a derived artifact.

Using model_dump() to pass data between layers. If you are converting a Pydantic model to a dict to pass it to another function, consider whether that function should accept the Pydantic model directly.

Try it yourself

  1. Write a parse_llm_output function for a model of your choice. Test it against: valid JSON, JSON wrapped in backtick fences, JSON embedded in prose, and completely invalid text. Verify the fallback behavior in each case.
  2. Create a BaseSettings subclass for an AI project. Include at least one SecretStr field and one field with a regex pattern constraint. Verify that it raises a clear error for an invalid value.
  3. Take a tool definition you have written by hand and replace it with auto-generated schema from a Pydantic model. Compare the output to verify they match.
Lesson Deep Dive

Ask a follow-up question about this lesson and get an AI-powered explanation.

Loading previous questions...