FastAPI and Pydantic: validating LLM structured outputs

When you build agents or extractors on top of language models, you quickly learn that the model will occasionally emit malformed JSON or drop a field. This tutorial shows how to build a FastAPI service that uses Pydantic to validate structured outputs from an LLM, giving you a hard contract at the boundary. We focus on fastapi pydantic structured outputs llm patterns that catch errors before they reach your callers.

Prerequisites

Python 3.11 or newer
fastapi, uvicorn, pydantic v2, openai (>=1.0), python-dotenv
An API key for an OpenAI-compatible endpoint (OpenAI, or a gateway like n4n.ai)
curl or httpie for local testing

Install with:

pip install fastapi uvicorn pydantic openai python-dotenv

Set your key in .env:

OPENAI_API_KEY=sk-...
BASE_URL=https://api.openai.com/v1

If you use a gateway, point BASE_URL at its OpenAI-compatible endpoint.

Define the output contract

Start with the shape you expect. Suppose we extract invoice data from unstructured text. Pydantic v2 models give you strict types and validators.

from pydantic import BaseModel, Field, ValidationInfo, field_validator
from typing import List

class LineItem(BaseModel):
    quantity: int = Field(gt=0)
    unit_price_cents: int = Field(ge=0)

class Invoice(BaseModel):
    vendor: str
    invoice_number: str
    total_cents: int = Field(ge=0)
    items: List[LineItem] = Field(min_length=1)

    @field_validator("total_cents")
    def check_total(cls, v, info: ValidationInfo):
        if "items" in info.data:
            computed = sum(i.quantity * i.unit_price_cents for i in info.data["items"])
            if computed != v:
                raise ValueError(f"total {v} does not match sum {computed}")
        return v

This model rejects missing fields, negative quantities, and totals that don’t match line items. That’s the core of fastapi pydantic structured outputs llm validation: the schema is the test.

Request structured output from the LLM

Modern LLM APIs accept a JSON schema for response_format. The OpenAI Python client supports response_format={"type": "json_schema", "json_schema": {...}}. Build the schema from the Pydantic model with model_json_schema().

from openai import OpenAI
import os
from dotenv import load_dotenv

load_dotenv()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"), base_url=os.getenv("BASE_URL"))

def extract_invoice(text: str) -> str:
    schema = Invoice.model_json_schema()
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "Extract invoice data as JSON."},
            {"role": "user", "content": text}
        ],
        response_format={
            "type": "json_schema",
            "json_schema": {"name": "invoice", "schema": schema}
        },
        temperature=0
    )
    return resp.choices[0].message.content

Call it with a sample:

sample = """
Acme Corp
Invoice #INV-2024-005
2x Widget @ $3.00
1x Sprocket @ $5.50
"""
raw = extract_invoice(sample)
print(raw)

Expected raw output (truncated):

{"vendor":"Acme Corp","invoice_number":"INV-2024-005","total_cents":1150,"items":[{"description":"Widget","quantity":2,"unit_price_cents":300},{"description":"Sprocket","quantity":1,"unit_price_cents":550}]}

Validate and parse

Never trust the string. Parse it with Pydantic.

from pydantic import ValidationError

def parse_invoice(raw: str) -> Invoice:
    try:
        return Invoice.model_validate_json(raw)
    except ValidationError as e:
        raise ValueError(f"LLM returned invalid structure: {e.errors()}") from e

invoice = parse_invoice(raw)
print(invoice.total_cents)

If the model drops total_cents or mismatches the sum, ValidationError fires before the data propagates.

Build the FastAPI endpoint

Wire the two functions into a POST endpoint. Use a Pydantic input model for the request body, and return the validated Invoice. FastAPI will serialize it automatically.

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

class ExtractRequest(BaseModel):
    text: str

class ExtractResponse(Invoice):
    pass  # reuse the validated shape

@app.post("/extract", response_model=ExtractResponse)
def extract_endpoint(req: ExtractRequest):
    raw = extract_invoice(req.text)
    try:
        invoice = parse_invoice(raw)
    except ValueError as e:
        raise HTTPException(status_code=422, detail=str(e))
    return invoice

Run it:

uvicorn main:app --reload --port 8000

Test with curl:

curl -s -X POST http://localhost:8000/extract \
  -H "Content-Type: application/json" \
  -d '{"text":"Acme Corp\nInvoice #INV-2024-005\n2x Widget @ $3.00\n1x Sprocket @ $5.50"}'

Expected response:

{
  "vendor": "Acme Corp",
  "invoice_number": "INV-2024-005",
  "total_cents": 1150,
  "items": [
    {"description": "Widget", "quantity": 2, "unit_price_cents": 300},
    {"description": "Sprocket", "quantity": 1, "unit_price_cents": 550}
  ]
}

If the LLM returns total_cents: 1000, the endpoint returns 422 with the validation detail. That’s fastapi pydantic structured outputs llm enforcement in action.

Make the pipeline resilient

A single provider call can fail from rate limits or schema drift. Wrap the LLM call with retry and fallback. If you route through n4n.ai, its OpenAI-compatible endpoint fronts 240+ models and automatically falls back when a provider is rate-limited or degraded, so your fastapi pydantic structured outputs llm service keeps serving valid contracts without custom retry code.

For self-hosted resilience, add a simple retry:

from tenacity import retry, stop_after_attempt, wait_fixed

@retry(stop=stop_after_attempt(3), wait=wait_fixed(1))
def extract_invoice_retry(text: str) -> str:
    return extract_invoice(text)

Swap extract_invoice for extract_invoice_retry in the endpoint.

Validate before you trust

Pydantic v2 also lets you run async validators and coerce types. If your LLM returns "3.00" as a string for a float, use field_validator to cast. Keep the API boundary strict: the LLM is an untrusted data source.

from pydantic import field_validator

class LineItem(BaseModel):
    quantity: int = Field(gt=0)
    unit_price_cents: int = Field(ge=0)

    @field_validator("unit_price_cents", mode="before")
    def parse_money(cls, v):
        if isinstance(v, str):
            return int(float(v.replace("$", "")) * 100)
        return v

This absorbs minor formatting noise from the model.

Testing the contract

Write a pytest that feeds known-good and known-bad LLM outputs:

def test_parse_good():
    raw = '{"vendor":"X","invoice_number":"1","total_cents":100,"items":[{"description":"a","quantity":1,"unit_price_cents":100}]}'
    inv = parse_invoice(raw)
    assert inv.total_cents == 100

def test_parse_bad_total():
    raw = '{"vendor":"X","invoice_number":"1","total_cents":50,"items":[{"description":"a","quantity":1,"unit_price_cents":100}]}'
    try:
        parse_invoice(raw)
        assert False
    except ValueError:
        pass

Run pytest. Green means your fastapi pydantic structured outputs llm boundary is locked.

Key takeaways

Define the output shape with Pydantic before calling the model.
Pass model_json_schema() as response_format to constrain the LLM.
Parse with model_validate_json and catch ValidationError at the edge.
Return the validated model from FastAPI; let it serialize.
Add retries or a gateway with fallback for production traffic.

That’s the full loop: from prompt to validated JSON contract in a FastAPI app.